uploadqueue.cpp

非常出名开源客户端下载的程序emule

//this file is part of eMule
//Copyright (C)2002 Merkur ( [e-mail] / http://www.emule-project.net )
//
//This program is free software; you can redistribute it and/or
//modify it under the terms of the GNU General Public License
//as published by the Free Software Foundation; either
//version 2 of the License, or (at your option) any later version.
//
//This program is distributed in the hope that it will be useful,
//but WITHOUT ANY WARRANTY; without even the implied warranty of
//MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
//GNU General Public License for more details.
//
//You should have received a copy of the GNU General Public License
//along with this program; if not, write to the Free Software
//Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.

#include "StdAfx.h"
#include "uploadqueue.h"
#include "packets.h"
#include "emule.h"
#include "knownfile.h"
#include "listensocket.h"
#include "ini2.h"
#include "math.h"

#ifdef _DEBUG
#undef THIS_FILE
static char THIS_FILE[]=__FILE__;
#define new DEBUG_NEW
#endif


static uint32 counter, sec,statsave;
static uint32 igraph, istats;
VOID CALLBACK TimerProc(HWND hwnd, UINT uMsg,UINT_PTR idEvent,DWORD dwTime);

//TODO rewrite the whole networkcode, use overlapped sockets

CUploadQueue::CUploadQueue(CPreferences* in_prefs){
	app_prefs = in_prefs;
	h_timer = SetTimer(0,141,100,TimerProc);
	if (!h_timer)
		theApp.emuledlg->AddDebugLogLine(false,GetResString(IDS_ERR_TIMERCREATEFAILED));
	estadatarate = 2000;
	datarate = 0;
	dataratems = 0;
	datarateave = 0;
	bannedcount = 0;
	counter=0;
	successfullupcount = 0;
	failedupcount = 0;
	totaluploadtime = 0;
	m_nUpDataRateMSOverhead = 0;
	m_nUpDatarateOverhead = 0;
	m_nUpDataOverheadSourceExchange = 0;
	m_nUpDataOverheadFileRequest = 0;
	m_nUpDataOverheadOther = 0;
	m_nUpDataOverheadServer = 0;
	m_nUpDataOverheadSourceExchangePackets = 0;
	m_nUpDataOverheadFileRequestPackets = 0;
	m_nUpDataOverheadOtherPackets = 0;
	m_nUpDataOverheadServerPackets = 0;
	m_nLastStartUpload = 0;
	m_nSumForAvgDataRate = 0;
	statsave=0;
	for (int i = 0;i != 400;i++)
		avarage_dr_list.AddHead((uint32)0);

}

void CUploadQueue::AddUpNextClient(CUpDownClient* directadd){
	POSITION toadd = 0;
	POSITION toaddlow = 0; // VQB - LowID Slot Patch
	uint32	bestscore = 0;
	uint32  bestlowscore = 0;   // VQB - LowID Slot Patch
	CUpDownClient* newclient;
	// select next client or use given client
	if (!directadd){
		POSITION pos1, pos2;
		for (pos1 = waitinglist.GetHeadPosition();( pos2 = pos1 ) != NULL;){
			waitinglist.GetNext(pos1);
			CUpDownClient* cur_client =	waitinglist.GetAt(pos2);
			// clear dead clients
			ASSERT ( cur_client->GetLastUpRequest() );
			if ((::GetTickCount() - cur_client->GetLastUpRequest() > MAX_PURGEQUEUETIME) || !theApp.sharedfiles->GetFileByID(cur_client->GetUploadFileID()) ){
				RemoveFromWaitingQueue(pos2,true);	
				if (!cur_client->socket)
					cur_client->Disconnected();
			}
			// finished clearing
			else if ( (cur_client->GetScore(true) > bestscore) && !cur_client->IsBanned()){ // VQB removed redundant cond now covered in GetScore
				bestscore = cur_client->GetScore(true);
				toadd = pos2;
			}
			else if ( (cur_client->GetScore(false) > bestlowscore) && !cur_client->IsBanned() && !cur_client->AddNextConnect){
				bestlowscore = cur_client->GetScore(false);
				toaddlow = pos2;
			}
		}
		if (!toadd)
			return;
		// VQB LowID Slot Patch -- mark LowID to add next connect
		if (bestlowscore > bestscore){
			newclient = waitinglist.GetAt(toaddlow);
			newclient->AddNextConnect = true;
		}
		// VQB end
		newclient = waitinglist.GetAt(toadd);
		RemoveFromWaitingQueue(toadd, true);
		theApp.emuledlg->transferwnd.ShowQueueCount(waitinglist.GetCount());
	}
	else
		newclient = directadd;

	if (IsDownloading(newclient)){
		return;
	}
	// tell the client that we are now ready to upload
	if (!newclient->socket || !newclient->socket->IsConnected()){
		newclient->SetUploadState(US_CONNECTING);
		newclient->TryToConnect(true);
	}
	else{
		Packet* packet = new Packet(OP_ACCEPTUPLOADREQ,0);
		theApp.uploadqueue->AddUpDataOverheadFileRequest(packet->size);
		newclient->socket->SendPacket(packet,true);
		newclient->SetUploadState(US_UPLOADING);
	}
	newclient->SetUpStartTime();
	newclient->ResetSessionUp();
	uploadinglist.AddTail(newclient);
	
	// statistic
	CKnownFile* reqfile = theApp.sharedfiles->GetFileByID((uchar*)newclient->GetUploadFileID());
	if (reqfile){
		reqfile->statistic.AddAccepted();
	}
	
	theApp.emuledlg->transferwnd.uploadlistctrl.AddClient(newclient);

}

void CUploadQueue::Process(){
	m_nSumForAvgDataRate -= avarage_dr_list.RemoveHead();
	avarage_dr_list.AddTail(dataratems);
	m_nSumForAvgDataRate += dataratems;
	datarate = 10 * m_nSumForAvgDataRate / avarage_dr_list.GetCount();

	dataratems = 0;
	if (AcceptNewClient() && waitinglist.GetCount()){
		m_nLastStartUpload = ::GetTickCount();
		AddUpNextClient();
	}
	if (!uploadinglist.GetCount())
		return;

	int16 clientsrdy = 0;
	for (POSITION pos = uploadinglist.GetHeadPosition();pos != 0;uploadinglist.GetNext(pos)){
		CUpDownClient* cur_client = uploadinglist.GetAt(pos);
		if ( (cur_client->socket) && (!cur_client->socket->IsBusy()) && cur_client->HasBlocks())
			clientsrdy++;
	}
	if (!clientsrdy){
		estadatarate -= 200;
		if (estadatarate < 100)
			estadatarate = 100;
		clientsrdy++;
	}
	else{
		estadatarate += 200;
		if (estadatarate > app_prefs->GetMaxUpload()*102)
			estadatarate = app_prefs->GetMaxUpload()*102;
	}
	uint32 sendperclient = estadatarate/clientsrdy;
	POSITION pos1,pos2;
	for (pos1 = uploadinglist.GetHeadPosition();( pos2 = pos1 ) != NULL; ){
		uploadinglist.GetNext(pos1);
		CUpDownClient* cur_client = uploadinglist.GetAt(pos2);
		dataratems += cur_client->SendBlockData(sendperclient);
	}
};

bool CUploadQueue::AcceptNewClient(){
	// check if we can allow a new client to start downloading form us
	if (::GetTickCount() - m_nLastStartUpload < 1000 && datarate < 102400 )
		return false;
	if (uploadinglist.GetCount() < MIN_UP_CLIENTS_ALLOWED)
		return true;
	else if (uploadinglist.GetCount() >= MAX_UP_CLIENTS_ALLOWED)
		return false;
	uint32	upPerClient = UPLOAD_CLIENT_DATARATE + datarate/50;
	if( upPerClient > 10000 )
		upPerClient = 11000;
	//now the final check
	if (theApp.glob_prefs->GetMaxUpload() == UNLIMITED){
		if ((uint32)uploadinglist.GetCount() < ((datarate/upPerClient)+2))
			return true;
	}
	else{
		uint16 nMaxSlots = 0;
		if (theApp.glob_prefs->GetMaxUpload() > 10){
			nMaxSlots += 2;
			nMaxSlots += (uint16)ceil((float)((theApp.glob_prefs->GetMaxUpload() - 10)*1024) / upPerClient);
		}
		else
			nMaxSlots = MIN_UP_CLIENTS_ALLOWED;
		if ((uint32)uploadinglist.GetCount() < (datarate/UPLOAD_CHECK_CLIENT_DR) && uploadinglist.GetCount() <= nMaxSlots )
			return true;
		
	}
	//nope
	return false;
}

CUploadQueue::~CUploadQueue(){
	KillTimer(0,141);
}

CUpDownClient* CUploadQueue::GetWaitingClientByIP_UDP(uint32 dwIP, uint16 dwPort){
	for (POSITION pos = waitinglist.GetHeadPosition();pos != 0;waitinglist.GetNext(pos)){
		if (dwIP == waitinglist.GetAt(pos)->GetIP() && dwPort == waitinglist.GetAt(pos)->GetUDPPort())
			return waitinglist.GetAt(pos);
	}
	return 0;
}

void CUploadQueue::UpdateBanCount(){
	int count=0;
	for (POSITION pos = waitinglist.GetHeadPosition();pos != 0;waitinglist.GetNext(pos)){
		CUpDownClient* cur_client= waitinglist.GetAt(pos);
		if(cur_client->IsBanned())
			count++;
	}
	SetBanCount(count);
}
void CUploadQueue::AddClientToQueue(CUpDownClient* client, bool bIgnoreTimelimit){
	if (theApp.serverconnect->IsConnected() && theApp.serverconnect->IsLowID()
		&& !theApp.serverconnect->IsLocalServer(client->GetServerIP(),client->GetServerPort())
		&& client->GetDownloadState() == DS_NONE && !client->IsFriend()
		&& GetWaitingUserCount() > 50)
		return;
	client->AddAskedCount();
	client->SetLastUpRequest();
	if (!bIgnoreTimelimit){
		if (client->IsBanned()){
			if (::GetTickCount() - client->GetBanTime() > 18000000){
				client->UnBan();
			}
			else
				return;
		}
		client->AddRequestCount(client->GetUploadFileID());
	}
	// check for double
	for (POSITION pos = waitinglist.GetHeadPosition();pos != 0;waitinglist.GetNext(pos)){
		CUpDownClient* cur_client= waitinglist.GetAt(pos);
		if (cur_client == client){	//already on queue
// VQB LowID Slot Patch -- note:  should add limit so only if #slots < UL -or- UL+1 for Low UL (?)
			if (client->AddNextConnect && (uploadinglist.GetCount() < theApp.glob_prefs->GetMaxUpload())){
				client->AddNextConnect = false;
				RemoveFromWaitingQueue(client, true);
				AddUpNextClient(client);
//				theApp.emuledlg->AddLogLine(false,"Added Low ID User On Reconnect"); // VQB:  perhaps only add to debug log?
				return;
			}
// VQB end
			client->SendRankingInfo();
			theApp.emuledlg->transferwnd.queuelistctrl.RefreshClient(client);
			return;			
		}
		else if ( client->Compare(cur_client) ) {
			// another client with same ip or hash
			theApp.emuledlg->AddDebugLogLine(false,CString(GetResString(IDS_SAMEUSERHASH)),client->GetUserName(),cur_client->GetUserName(),cur_client->GetUserName() );
			RemoveFromWaitingQueue(pos,true);	
			if (!cur_client->socket)
				cur_client->Disconnected();	
			return;
		}