一个基于socket的资源共享平台的实现（二）

继续上次说的，其实任务调度宏观上普遍分为两种，实现上总的来说就是一个串行、一个并行，上次我们介绍的TCP传送服务我们是使用并行的实现的（任务池），这次我们再来一个串行的，这就简单多了，就是一个队列，FIFO，我们用它来实现下载任务（假设我们下载任务只能单独进行）。

 

我们用一个NS_Download_Pool类来封装对其的管理。

 

#ifndef NETSHARE_DOWNLOAD_POOL_H_
#define NETSHARE_DOWNLOAD_POOL_H_

#include "StdioFileEx/StdioFileEx.h"
#include "NetShareIndexManager.h"
#include "DownloadListDlg.h"
#include <string>
#include <vector>
#include <algorithm>

#include "cpplibbase.h"

using namespace std;

CString const DOWNLOAD_FILELIST_FILENAME("c:\\download_filelistfile");

//文件状态
enum FileStatus
{
	Ready = 0, //未下载
	Downloading, //下载中
	Failed, //下载失败
	Success, //完成
	Paused //暂停
};

//下载文件信息类
struct DownloadFileInfoType
{
	DownloadFileInfoType()
		: filename(_T(""))
		, size(_T(""))
		, ipaddr(_T(""))
		, status( Ready )
		, percent( 0 )
		, finish_size( 0 )
	{}

	bool operator==( DownloadFileInfoType& opt )
	{
		return ( filename == opt.filename &&
			size == opt.size &&
			ipaddr == opt.ipaddr &&
			targetdir == opt.targetdir );
	}

	CString filename;//文件名
	CString size;//大小
	CString ipaddr;//源IP
	CString targetdir;//保存地址
	FileStatus status;//文件状态
	double percent;//下载完成百分比
	DWORD finish_size;//完成的字节数
};

namespace
{
	bool IsComplete( DownloadFileInfoType const& unit )
	{
		return ( unit.status == Success );
	}
}

//下载池类
class NS_Download_Pool
{
public:
	NS_Download_Pool()
		: m_pDlg( NULL )
		, m_pList( NULL )
	{ Load(); }

	~NS_Download_Pool(){ Save(); }

	//增加
	void Add( DownloadFileInfoType info )
	{
		cpplib::resource::MutexLock MyMutex(FileListMutex);
		if( Exist( info ) )
			return;
		else
			files_.push_back( info );

	}

	void Save()
	{
		CStdioFileEx fp;
		fp.Open(DOWNLOAD_FILELIST_FILENAME,CFile::modeWrite|CFile::modeCreate);   //|CFile::modeCreate
		for( size_t i=0;i<files_.size();++i)
		{
			DownloadFileInfoType& opt = files_[i];
			fp.WriteString( opt.filename );
			fp.WriteString(_T("\n"));
			fp.WriteString( opt.size );
			fp.WriteString(_T("\n"));
			fp.WriteString( opt.ipaddr );
			fp.WriteString(_T("\n"));
			fp.WriteString( opt.targetdir );
			fp.WriteString(_T("\n"));
			CString write("");
			write.Format(_T("%d"),opt.status);
			fp.WriteString( write );
			fp.WriteString(_T("\n"));
			write.Format(_T("%.2lf"),opt.percent);
			fp.WriteString( write );
			fp.WriteString(_T("\n"));
			write.Format(_T("%d"),opt.finish_size);
			fp.WriteString( write );
			fp.WriteString(_T("\n"));
		}
		fp.Close(); 
	}

	void Load()
	{
		CStdioFileEx fp;
		if(fp.Open(DOWNLOAD_FILELIST_FILENAME,CFile::modeRead))   
		{
			CString test;
			while( NULL!= fp.ReadString( test ) )
			{
				if( test == "" )
					break;
				DownloadFileInfoType tmp;
				tmp.filename = test;
				fp.ReadString( tmp.size );
				fp.ReadString( tmp.ipaddr );
				fp.ReadString( tmp.targetdir );

				fp.ReadString( test );
				char* p = NetShareIndexManager::UnicodeToAnsi(test.GetBuffer(0));
				tmp.status = (FileStatus)(atoi( p ));
				delete p;

				fp.ReadString( test );
				p = NetShareIndexManager::UnicodeToAnsi(test.GetBuffer(0));
				tmp.percent = atof( p );
				delete p;

				fp.ReadString( test );
				p = NetShareIndexManager::UnicodeToAnsi(test.GetBuffer(0));
				tmp.finish_size = atoi( p );
				delete p;

				Add( tmp );
			}
			fp.Close();
		}
		else
		{
			Clear();
		}
	}

	//删除
	void Remove( DownloadFileInfoType& info )
	{
		cpplib::resource::MutexLock MyMutex(FileListMutex);
		vector<DownloadFileInfoType>::iterator iter;
		for( iter = files_.begin(); iter != files_.end(); ++iter )
		{
			if( *iter == info )
			{
				files_.erase( iter );
				return;
			}
		}
		Save();

	}

	//删除容器src内指定序号的元素
	template<class T>
	void Remove( vector<T>& src, vector<size_t> const& indexs )
	{
		vector<T> result;
		int count = 0;
		for( vector<T>::iterator iter = src.begin();
			 iter != src.end();
			 ++iter )
		{
			if( find( indexs.begin(), indexs.end(), count ) == indexs.end() )
			{
				result.push_back( *iter );
			}
			count++;
		}
		src = result;
	}

	//根据索引删除
	void Remove( vector<size_t> delete_indexs )
	{
		cpplib::resource::MutexLock MyMutex(FileListMutex);
		Remove( files_, delete_indexs );
		Save();
	}



	//移除所有已完成的下载
	void RemoveAllCompleted()
	{
		cpplib::resource::MutexLock MyMutex(FileListMutex);
		files_.erase( remove_if( files_.begin(), files_.end(), IsComplete ), files_.end() );
		Save();
		Show();
	}


	//清空
	void Clear()
	{
		cpplib::resource::MutexLock MyMutex(FileListMutex);
		files_.clear();
	}

	//弹出第一个元素
	DownloadFileInfoType Pop()
	{
		for( vector<DownloadFileInfoType>::iterator iter = files_.begin(); iter != files_.end(); ++iter )
		{
			//找未下的文件
			if( iter->status == Ready )
			{
				DownloadFileInfoType tmp = *iter;
				iter->status = Downloading;
				curr_iter_ = iter;
				return tmp;
			}
		}
		assert( TRUE );//不可能发生
		DownloadFileInfoType tmp;
		return tmp;
	}

	//下载成功
	void DownloadSuccess()
	{
		cpplib::resource::MutexLock MyMutex(FileListMutex);
		curr_iter_->status = Success;
		Save();
	}

	//下载失败
	void DownloadFailed( DWORD CompleteSize )
	{
		cpplib::resource::MutexLock MyMutex(FileListMutex);
		curr_iter_->status = Failed;
		curr_iter_->finish_size = CompleteSize;
		Save();
	}

	//设置当前下载已完成字节数
	void CurrFinish( DWORD size )
	{
		cpplib::resource::MutexLock MyMutex(FileListMutex);
		curr_iter_->finish_size = size;
	}

	//检测是否为空
	bool Empty()
	{
		for( vector<DownloadFileInfoType>::iterator iter = files_.begin(); iter != files_.end(); ++iter )
		{
			//找未下的文件
			if( iter->status == Ready )
			{
				return FALSE;
			}
		}
		return TRUE;
	}

	//检测是否存在
	bool Exist( DownloadFileInfoType info )
	{
		vector<DownloadFileInfoType>::iterator iter;
		for( iter = files_.begin(); iter != files_.end(); ++iter )
		{
			if( *iter == info )
			{
				return true;
			}
		}
		return false;
	}

	void SetDlgPtr( DownloadListDlg* p_dlg, CListCtrl* pList )
	{
		m_pDlg = p_dlg;
		m_pList = pList;
	}

	//重置下载
	void SetReady( vector<size_t> select_indexs )
	{
		cpplib::resource::MutexLock MyMutex(FileListMutex);
		//若用户没有点选具体文件
		if( select_indexs.size() == 0 )
		{
			for( size_t i = 0; i < files_.size(); ++i )
				if( files_[ i ].status != Success && files_[ i ].status != Downloading )
					files_[ i ].status = Ready;
			return;
		}

		//对具体文件操作
		for( size_t i = 0; i < select_indexs.size(); ++i )
		{
			if( files_[ select_indexs[i]].status != Success && files_[ select_indexs[i]].status != Downloading )
				files_[ select_indexs[i]].status = Ready;
		}
	}

	//停止
	void StopAll()
	{
		cpplib::resource::MutexLock MyMutex(FileListMutex);
		for( vector<DownloadFileInfoType>::iterator iter = files_.begin(); iter != files_.end(); ++iter )
		{
			//找未下的文件
			if( iter->status != Success )
			{
				iter->status = Paused;
			}
		}
	}

	//显示
	void Show()
	{
		cpplib::resource::MutexLock MyMutex(FileListMutex);
		try
		{
			if( !m_pDlg || !m_pList )
				return;	
			/*if( ::IsWindowVisible( *m_pDlg ) == FALSE )
				return;*/

			int theItemCount = m_pList->GetItemCount();
			vector<size_t> select_indexs;

			for( int i=0; i < theItemCount; i++ )
			{
				if( m_pList->GetItemState(i, LVIS_SELECTED) == LVIS_SELECTED )
				{
					select_indexs.push_back(i);
				}
			}


			UINT ListIndex = 0;
			wchar_t* pTmp;

			m_pList->DeleteAllItems();

			vector<DownloadFileInfoType>::iterator iter;
			for( iter = files_.begin(); iter != files_.end(); ++iter )
			{
				DownloadFileInfoType const& MyUnit = *iter;
				m_pList->InsertItem(ListIndex,MyUnit.filename);//文件名
				m_pList->SetItemText(ListIndex,1,MyUnit.size);//文件大小
				m_pList->SetItemText(ListIndex,2,MyUnit.ipaddr);//IP地址
				m_pList->SetItemText(ListIndex,3,MyUnit.targetdir);//保存路径
				CString status("");
				switch( MyUnit.status )
				{
				case Ready:
					status = "待下载";
					break;
				case Downloading:
					status = "下载中";
					break;
				case Failed:
					status = "停止";
					break;
				case Success:
					status = "已完成";
					break;
				case Paused:
					status = "暂停";
					break;
				default:
					status = "格式错误";
					break;
				}
				m_pList->SetItemText(ListIndex,4,status);//状态
 
				if( Success == MyUnit.status )
				{
					m_pList->SetItemText(ListIndex,5,CString("100.00"));//百分比
				}
				else
				{
					CString process( "" );

					CString tmp = MyUnit.size;
					char* p = NetShareIndexManager::UnicodeToAnsi( tmp );
					double size = atof( p );
					delete p;
					double percent = 0;
					if( size < 0.01 )
						percent = 0;
					else
						percent = (double)MyUnit.finish_size * 100 / (double)(size * 1024 * 1024 );
					process.Format(_T("%.2lf%%"),percent );

					m_pList->SetItemText(ListIndex,5,process);//百分比
				}
				ListIndex++;
			}

			for( size_t i=0; i < select_indexs.size(); i++ )
			{
				m_pList->SetItemState(select_indexs[i], LVIS_SELECTED,LVIS_SELECTED );
			}

			m_pDlg->UpdateData( FALSE );
		}//try
		catch(...)
		{}
	}

private:
	vector<DownloadFileInfoType> files_;
	vector<DownloadFileInfoType>::iterator curr_iter_; //当前下载的文件结构迭代器

	cpplib::resource::Mutex FileListMutex;
	DownloadListDlg* m_pDlg;
	CListCtrl* m_pList;
};
#endif

 

 

接下来我们针对资源传送过程中限速进行分析和实现。

如果需要将发送速度限制在一个值，我们可以这么理解，单位时间内最多允许发送数据为N，若超过之，就需要降低速度，若不足，则需要提高速度。

如何控制速度？这里我们采用最朴素的方法，sleep，只要将sleep的具体值控制好，我们是可以控制速度的。

 

1. 若超过限制的速度，则增加sleep的时间；

2. 若不足限制的速度，则减少sleep的时间；

 

那么sleep的变化值怎么考虑？我们希望速度变化越快趋近于预期值越好，这里我们采用 计算机网络里提到的网络带宽探测的模型。

假设刚开始速度为0，采用一个指数型计算公式提升速度，当发现有一点超速了，退回到上一个状态，采用直线逼近。

横坐标t为时间轴，纵坐标s为速度，1为起始位置，速度为0，2点之后探测到速度超过限制，退回到2点，然后采用直线逼近，直到逼近速度极限。

 

细心的用户能发现，使用迅雷等下载软件时，也是呈现类似于这种的现象，先速度猛增，然后在到一个值之后，回退一点，开始线性缓慢增长，直到问题。

 

未完待续……