peercacheclient.cpp

eMule0.44b的原代码

	uint32 uCacheIP = 0;
	uint16 uCachePort = 0;
	uint32 uPushId = 0;
	uchar aucFileHash[16];
	uint32 uRemoteIP = 0;
	md4clr(aucFileHash);

	CString strInfo;
	UINT uTags = dataRecv.ReadUInt8();
	while (uTags--)
	{
		CTag tag(&dataRecv, GetUnicodeSupport());
		if (tag.GetNameID() == PCTAG_CACHEIP && tag.IsInt())
		{
			uCacheIP = tag.GetInt();
			if (bDebug)
				strInfo.AppendFormat(_T("  CacheIP=%s"), ipstr(uCacheIP));
		}
		else if (tag.GetNameID() == PCTAG_CACHEPORT && tag.IsInt())
		{
			uCachePort = tag.GetInt();
			if (bDebug)
				strInfo.AppendFormat(_T("  CachePort=%u"), uCachePort);
		}
		else if (tag.GetNameID() == PCTAG_PUSHID && tag.IsInt())
		{
			uPushId = tag.GetInt();
			if (bDebug)
				strInfo.AppendFormat(_T("  PushId=%u"), uPushId);
		}
		else if (tag.GetNameID() == PCTAG_FILEID && tag.IsHash() && tag.GetHash() != NULL)
		{
			md4cpy(aucFileHash, tag.GetHash());
			if (bDebug)
				strInfo.AppendFormat(_T("  FileId=%s"), md4str(aucFileHash));
		}
		else if (tag.GetNameID() == PCTAG_PUBLICIP && tag.IsInt())
		{
			uRemoteIP = tag.GetInt();
			if (bDebug)
				strInfo.AppendFormat(_T("  PublicIP=%s"), ipstr(uRemoteIP));
		}
		else
		{
			if (bDebug)
				strInfo.AppendFormat(_T("  ***UnkTag: %s"), tag.GetFullInfo());
			ASSERT(0);
		}
	}

	if (bDebug)
	{
		if (dataRecv.GetPosition() < dataRecv.GetLength())
			strInfo.AppendFormat(_T("  ***AddData: %u bytes"), (UINT)(dataRecv.GetLength() - dataRecv.GetPosition()));
		Debug(_T("%s\n"), strInfo);
	}

	if (uCacheIP == 0 || uCachePort == 0 || uPushId == 0 || isnulmd4(aucFileHash)){
		if (thePrefs.GetVerbose())
			theApp.AddDebugLogLine(false, _T("Invalid PeerCacheQuery; %s"), DbgGetClientInfo());
		return false;
	}

	CKnownFile* pUploadFile = theApp.sharedfiles->GetFileByID(aucFileHash);
	if (pUploadFile == NULL){
		if (thePrefs.GetVerbose())
			theApp.AddDebugLogLine(false, _T("PeerCacheQuery reqfile does not match ed2k reqfile; %s"), DbgGetClientInfo());
		return false;
	}

	if (m_pPCUpSocket != NULL)
	{
        SetPeerCacheUpState(PCUS_NONE);
		m_pPCUpSocket->Safe_Delete();
		ASSERT( m_pPCUpSocket == NULL );
	}
	m_pPCUpSocket = new CPeerCacheUpSocket(this);
	m_pPCUpSocket->SetTimeOut(GetPeerCacheSocketUploadTimeout());
	m_pPCUpSocket->Create();

	SOCKADDR_IN sockAddr = {0};
	sockAddr.sin_family = AF_INET;
	sockAddr.sin_port = htons(uCachePort);
	sockAddr.sin_addr.S_un.S_addr = uCacheIP;
	//Try to always tell the socket to WaitForOnConnect before you call Connect.
	m_pPCUpSocket->WaitForOnConnect();
	m_pPCUpSocket->Connect((SOCKADDR*)&sockAddr, sizeof sockAddr);

	CStringA strPCRequest;
	strPCRequest.AppendFormat("GIVE %u\r\n", uPushId);

	if (thePrefs.GetDebugClientTCPLevel() > 0){
		DebugSend("PeerCache-GIVE", this, (char*)pUploadFile->GetFileHash());
		Debug(_T("  %hs\n"), strPCRequest);
	}
	
	CRawPacket* pHttpPacket = new CRawPacket(strPCRequest);
	theStats.AddUpDataOverheadFileRequest(pHttpPacket->size);
	m_pPCUpSocket->SendPacket(pHttpPacket);
	m_pPCUpSocket->SetHttpState(HttpStateRecvExpected);
	m_bPeerCacheUpHit = false;
	SetPeerCacheUpState(PCUS_WAIT_CACHE_REPLY);
	//theApp.uploadBandwidthThrottler->AddToStandardList(0, m_pPCUpSocket);

	CSafeMemFile dataSend(128);
	dataSend.WriteUInt8(PCPCK_VERSION);
	dataSend.WriteUInt8(PCOP_RES);
	dataSend.WriteUInt8(3);
	CTag tagPushId(PCTAG_PUSHID, uPushId);
	tagPushId.WriteNewEd2kTag(&dataSend);
	CTag tagPublicIP(PCTAG_PUBLICIP, theApp.GetPublicIP());
	tagPublicIP.WriteNewEd2kTag(&dataSend);
	CTag tagFileId(PCTAG_FILEID, (BYTE*)aucFileHash);
	tagFileId.WriteNewEd2kTag(&dataSend);
	
	if (thePrefs.GetDebugClientTCPLevel() > 0){
		DebugSend("OP__PeerCacheAnswer", this, (char*)aucFileHash);
		Debug(_T("  PushId=%u  PublicIP=%s  FileId=%s\n"), tagPushId.GetInt(), ipstr(tagPublicIP.GetInt()), md4str(tagFileId.GetHash()));
	}
	
	Packet* pEd2kPacket = new Packet(&dataSend, OP_EMULEPROT, OP_PEERCACHE_ANSWER);
	theStats.AddUpDataOverheadFileRequest(pEd2kPacket->size);
	socket->SendPacket(pEd2kPacket);
	return true;
}

bool CUpDownClient::ProcessPeerCacheAnswer(const char* packet, UINT size)
{
	const bool bDebug = (thePrefs.GetDebugClientTCPLevel() > 0);
	ASSERT( GetDownloadState() == DS_DOWNLOADING );
	ASSERT( m_ePeerCacheDownState == PCDS_WAIT_CLIENT_REPLY );

	if (bDebug)
		DebugRecv("OP_PeerCacheAnswer", this);

	if (socket == NULL || reqfile == NULL){
		ASSERT(0);
		return false;
	}

	CSafeMemFile dataRecv((const BYTE*)packet, size);
	uint8 uPCVersion = dataRecv.ReadUInt8();
	if (uPCVersion != PCPCK_VERSION){
		if (bDebug)
			Debug(_T("  ***Invalid packet version: 0x%02x\n"), uPCVersion);
		ASSERT(0);
		return false;
	}
	uint8 uPCOpcode = dataRecv.ReadUInt8();
	if (uPCOpcode == PCOP_NONE){
		if (thePrefs.GetVerbose())
			theApp.AddDebugLogLine(false, _T("Client does not support PeerCache; %s"), DbgGetClientInfo());
		return false;
	}
	if (uPCOpcode != PCOP_RES){
		if (bDebug)
			Debug(_T("  ***Invalid packet opcode: 0x%02x\n"), uPCOpcode);
		ASSERT(0);
		return false;
	}

	uint32 uPushId = 0;
	uint32 uRemoteIP = 0;
	uchar aucFileHash[16];
	md4clr(aucFileHash);

	CString strInfo;
	UINT uTags = dataRecv.ReadUInt8();
	while (uTags--)
	{
		CTag tag(&dataRecv, GetUnicodeSupport());
		if (tag.GetNameID() == PCTAG_PUSHID && tag.IsInt())
		{
			uPushId = tag.GetInt();
			if (bDebug)
				strInfo.AppendFormat(_T("  PushId=%u"), uPushId);
		}
		else if (tag.GetNameID() == PCTAG_PUBLICIP && tag.IsInt())
		{
			uRemoteIP = tag.GetInt();
			if (bDebug)
				strInfo.AppendFormat(_T("  RemoteIP=%s"), ipstr(uRemoteIP));
		}
		else if (tag.GetNameID() == PCTAG_FILEID && tag.IsHash() && tag.GetHash() != NULL)
		{
			md4cpy(aucFileHash, tag.GetHash());
			if (bDebug)
				strInfo.AppendFormat(_T("  FileId=%s"), md4str(aucFileHash));
		}
		else
		{
			if (bDebug)
				strInfo.AppendFormat(_T("  ***UnkTag: %s"), tag.GetFullInfo());
			ASSERT(0);
		}
	}

	if (bDebug)
	{
		if (dataRecv.GetPosition() < dataRecv.GetLength())
			strInfo.AppendFormat(_T("  ***AddData: %u bytes"), (UINT)(dataRecv.GetLength() - dataRecv.GetPosition()));
		Debug(_T("%s\n"), strInfo);
	}

	if (uPushId == 0 || uRemoteIP == 0 || isnulmd4(aucFileHash)){
		if (thePrefs.GetVerbose())
			theApp.AddDebugLogLine(false, _T("Invalid PeerCacheAnswer; %s"), DbgGetClientInfo());
		return false;
	}

	if (md4cmp(aucFileHash, reqfile->GetFileHash()) != 0){
		if (thePrefs.GetVerbose())
			theApp.AddDebugLogLine(false, _T("PeerCacheAnswer reqfile does not match ed2k reqfile; %s"), DbgGetClientInfo());
		return false;
	}

	m_uPeerCacheDownloadPushId = uPushId;
	m_uPeerCacheRemoteIP = uRemoteIP;

	if (!SendHttpBlockRequests())
		return false;

	theApp.m_pPeerCache->DownloadAttemptStarted();
	ASSERT( m_ePeerCacheDownState == PCDS_WAIT_CACHE_REPLY );
	return true;
}

bool CUpDownClient::ProcessPeerCacheAcknowledge(const char* packet, UINT size)
{
	const bool bDebug = (thePrefs.GetDebugClientTCPLevel() > 0);
	if (bDebug)
		DebugRecv("OP_PeerCacheAck", this);

	if (socket == NULL){
		ASSERT(0);
		return false;
	}

	m_bPeerCacheUpHit = false;
	CSafeMemFile data((BYTE*)packet, size);
	UINT uAck = data.ReadUInt8();
	if (uAck == 1)
	{
		// Cache hit
		if (bDebug)
			Debug(_T("  Cache hit\n"));

		// PC-TODO: If this socket is closed, PeerCache also closes the socket which it had opened to the
		// remote client! So, to give the remote client a chance to receive all the data from the PeerCache,
		// we have to keep this socket open, although it's not needed nor could it be reused!
		ASSERT( m_pPCUpSocket != NULL );
//		m_pPCUpSocket->Safe_Delete();
//		m_pPCUpSocket = NULL;
		m_pPCUpSocket->SetTimeOut(MIN2MS(60)); // set socket timeout to 1 hour ??
		m_bPeerCacheUpHit = true;
	}
	else if (uAck == 0)
	{
		// Cache miss, keep uploading
		if (bDebug)
			Debug(_T("  Cache miss\n"));
	}
	else{
		ASSERT(0);
		return false;
	}

	// PC-TODO: Since we can not close the PC socket, what exactly do we need this ACK-packet for?
	;

	UpdateDisplayedInfo();

	return true;
}

bool CUpDownClient::IsUploadingToPeerCache() const
{
	// this function should not check any socket ptrs, as the according sockets may already be closed/deleted
    return m_ePeerCacheUpState == PCUS_UPLOADING;
}

bool CUpDownClient::IsDownloadingFromPeerCache() const
{
	// this function should not check any socket ptrs, as the according sockets may already be closed/deleted
	return m_ePeerCacheDownState == PCDS_DOWNLOADING;
}

void CUpDownClient::OnPeerCacheDownSocketClosed(int nErrorCode)
{
	if (nErrorCode)
		return;
	
	// restart PC download if cache just closed the connection without obvious reason
	if (GetDownloadState() == DS_DOWNLOADING 
		&& m_ePeerCacheDownState == PCDS_DOWNLOADING
		&& !m_PendingBlocks_list.IsEmpty())
	{
		if (thePrefs.GetVerbose())
			AddDebugLogLine(DLP_HIGH, false, _T("PeerCache: Socket closed unexpedtedly, trying to reestablish connection"));
		theApp.m_pPeerCache->DownloadAttemptFailed();
		TRACE("+++ Restarting PeerCache download - socket closed\n");
		ASSERT( m_pPCDownSocket == NULL );
		SetPeerCacheDownState(PCDS_NONE);
		ClearDownloadBlockRequests();
		SendPeerCacheFileRequest();
	}
}

bool CUpDownClient::OnPeerCacheDownSocketTimeout()
{
	// restart PC download if cache just stalls
	if (GetDownloadState() == DS_DOWNLOADING 
		&& m_ePeerCacheDownState == PCDS_DOWNLOADING
		&& !m_PendingBlocks_list.IsEmpty())
	{
		if (thePrefs.GetVerbose())
			AddDebugLogLine(DLP_HIGH, false, _T("PeerCache Error: Socket TimeOut, trying to reestablish connection"));
		theApp.m_pPeerCache->DownloadAttemptFailed();
		TRACE("+++ Restarting PeerCache download - socket timeout\n");
		if (m_pPCDownSocket)
		{
			m_pPCDownSocket->Safe_Delete();
			ASSERT( m_pPCDownSocket == NULL );
		}
		SetPeerCacheDownState(PCDS_NONE);
		ClearDownloadBlockRequests();
		return SendPeerCacheFileRequest();
	}
	return false;
}

void CUpDownClient::SetPeerCacheDownState(EPeerCacheDownState eState)
{
	if (m_ePeerCacheDownState != eState)
	{
		m_ePeerCacheDownState = eState;
		if (m_ePeerCacheDownState == PCDS_NONE)
			m_bPeerCacheDownHit = false;
		UpdateDisplayedInfo();
	}
}

void CUpDownClient::SetPeerCacheUpState(EPeerCacheUpState eState)
{
	if (m_ePeerCacheUpState != eState)
	{
		//if (thePrefs.GetVerbose())
			//AddDebugLogLine(false, _T(" %s changed PeercacheState to %i"), DbgGetClientInfo(), eState);

		m_ePeerCacheUpState = eState;
		if (m_ePeerCacheUpState == PCUS_NONE)
			m_bPeerCacheUpHit = false;

        theApp.uploadqueue->ReSortUploadSlots(true);

		UpdateDisplayedInfo();
	}
}