//

// This file is part of the aMule Project.

//

// Copyright (c) 2003-2026 aMule Team ( https://amule-org.github.io )

// Copyright (c) 2002-2011 Merkur ( [email protected] / http://www.emule-project.net )

//

// Any parts of this program derived from the xMule, lMule or eMule project,

// or contributed by third-party developers are copyrighted by their

// respective authors.

//

// 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., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301, USA

//

#include "updownclient.h" // Needed for CUpDownClient

#include <protocol/Protocols.h>

#include <protocol/ed2k/Client2Client/TCP.h>

#include <protocol/ed2k/Client2Client/UDP.h>

#include <common/EventIDs.h>

#include <common/Macros.h>

#include <common/Constants.h>

#include <zlib.h>

#include <cmath> // Needed for std:exp

#include "ClientCredits.h" // Needed for CClientCredits

#include "ClientUDPSocket.h" // Needed for CClientUDPSocket

#include "DownloadQueue.h" // Needed for CDownloadQueue

#include "Preferences.h" // Needed for thePrefs

#include "Packet.h" // Needed for CPacket

#include "MemFile.h" // Needed for CMemFile

#include "ClientTCPSocket.h" // Needed for CClientTCPSocket

#include "ListenSocket.h" // Needed for CListenSocket

#include "amule.h" // Needed for theApp

#include "PartFile.h" // Needed for CPartFile

#include "SharedFileList.h"

#include "Statistics.h" // Needed for theStats

#include "Logger.h"

#include "GuiEvents.h" // Needed for Notify_*

#include "UploadQueue.h" // Needed for CUploadQueue

#ifdef __MULE_UNUSED_CODE__

// This function is left as a reminder.

// Changes here _must_ be reflected in CClientList::FindMatchingClient.

bool CUpDownClient::Compare(const CUpDownClient *tocomp, bool bIgnoreUserhash) const

{

if (!tocomp) {

// should we wxASSERT here?

return false;

}

// Compare only the user hash..

if (!bIgnoreUserhash && HasValidHash() && tocomp->HasValidHash()) {

return GetUserHash() == tocomp->GetUserHash();

}

if (HasLowID()) {

// User is firewalled.. Must do two checks..

if (GetIP() != 0 && GetIP() == tocomp->GetIP()) {

// The IP of both match

if (GetUserPort() != 0 && GetUserPort() == tocomp->GetUserPort()) {

// IP-UserPort matches

return true;

}

if (GetKadPort() != 0 && GetKadPort() == tocomp->GetKadPort()) {

// IP-KadPort Matches

return true;

}

}

if (GetUserIDHybrid() != 0 && GetUserIDHybrid() == tocomp->GetUserIDHybrid() &&

GetServerIP() != 0 && GetServerIP() == tocomp->GetServerIP() &&

GetServerPort() != 0 && GetServerPort() == tocomp->GetServerPort()) {

// Both have the same lowID, Same serverIP and Port..

return true;

}

// Both IP, and Server do not match..

return false;

}

// User is not firewalled.

if (GetUserPort() != 0) {

// User has a Port, lets check the rest.

if (GetIP() != 0 && tocomp->GetIP() != 0) {

// Both clients have a verified IP..

if (GetIP() == tocomp->GetIP() && GetUserPort() == tocomp->GetUserPort()) {

// IP and UserPort match..

return true;

}

} else {

// One of the two clients do not have a verified IP

if (GetUserIDHybrid() == tocomp->GetUserIDHybrid() &&

GetUserPort() == tocomp->GetUserPort()) {

// ID and Port Match..

return true;

}

}

}

if (GetKadPort() != 0) {

// User has a Kad Port.

if (GetIP() != 0 && tocomp->GetIP() != 0) {

// Both clients have a verified IP.

if (GetIP() == tocomp->GetIP() && GetKadPort() == tocomp->GetKadPort()) {

// IP and KadPort Match..

return true;

}

} else {

// One of the users do not have a verified IP.

if (GetUserIDHybrid() == tocomp->GetUserIDHybrid() &&

GetKadPort() == tocomp->GetKadPort()) {

// ID and KadProt Match..

return true;

}

}

}

// No Matches..

return false;

}

#endif

bool CUpDownClient::AskForDownload()

{

// 0.42e

if (theApp->listensocket->TooManySockets()) {

if (!m_socket) {

if (GetDownloadState() != DS_TOOMANYCONNS) {

SetDownloadState(DS_TOOMANYCONNS);

}

return true;

} else if (!m_socket->IsConnected()) {

if (GetDownloadState() != DS_TOOMANYCONNS) {

SetDownloadState(DS_TOOMANYCONNS);

}

return true;

}

}

m_bUDPPending = false;

m_dwLastAskedTime = ::GetTickCount64();

SetDownloadState(DS_CONNECTING);

SetSentCancelTransfer(0);

return TryToConnect();

}

void CUpDownClient::SendStartupLoadReq()

{

// 0.42e

if (m_socket == NULL || m_reqfile == NULL) {

return;

}

SetDownloadState(DS_ONQUEUE);

CMemFile dataStartupLoadReq(16);

dataStartupLoadReq.WriteHash(m_reqfile->GetFileHash());

CPacket *packet = new CPacket(dataStartupLoadReq, OP_EDONKEYPROT, OP_STARTUPLOADREQ);

theStats::AddUpOverheadFileRequest(packet->GetPacketSize());

AddDebugLogLineN(logLocalClient, "Local Client: OP_STARTUPLOADREQ to " + GetFullIP());

SendPacket(packet, true, true);

}

bool CUpDownClient::IsSourceRequestAllowed()

{

// #warning REWRITE - Source swapping from eMule.

// 0.42e

uint64 dwTickCount = ::GetTickCount64() + CONNECTION_LATENCY;

uint64 nTimePassedClient = dwTickCount - GetLastSrcAnswerTime();

uint64 nTimePassedFile = dwTickCount - m_reqfile->GetLastAnsweredTime();

bool bNeverAskedBefore = (GetLastAskedForSources() == 0);

uint32 uSources = m_reqfile->GetSourceCount();

return (

// if client has the correct extended protocol

ExtProtocolAvailable() && (SupportsSourceExchange2() || GetSourceExchange1Version() > 1) &&

// AND if we need more sources

thePrefs::GetMaxSourcePerFileSoft() > uSources &&

// AND if...

(

// source is not complete and file is very rare

(!m_bCompleteSource &&

(bNeverAskedBefore || nTimePassedClient > SOURCECLIENTREASKS) &&

(uSources <= RARE_FILE / 5)) ||

// source is not complete and file is rare

(!m_bCompleteSource &&

(bNeverAskedBefore || nTimePassedClient > SOURCECLIENTREASKS) &&

(uSources <= RARE_FILE ||

uSources - m_reqfile->GetValidSourcesCount() <= RARE_FILE / 2) &&

(nTimePassedFile > SOURCECLIENTREASKF)) ||

// OR if file is not rare

((bNeverAskedBefore ||

nTimePassedClient > (unsigned)(SOURCECLIENTREASKS * MINCOMMONPENALTY)) &&

(nTimePassedFile > (unsigned)(SOURCECLIENTREASKF * MINCOMMONPENALTY)))));

}

void CUpDownClient::SendFileRequest()

{

wxCHECK_RET(m_reqfile, "Cannot request file when no reqfile is set");

CMemFile dataFileReq(16 + 16);

dataFileReq.WriteHash(m_reqfile->GetFileHash());

if (SupportMultiPacket()) {

DEBUG_ONLY(wxString sent_opcodes;)

if (SupportExtMultiPacket()) {

dataFileReq.WriteUInt64(m_reqfile->GetFileSize());

}

AddDebugLogLineN(logClient, "Sending file request to client");

dataFileReq.WriteUInt8(OP_REQUESTFILENAME);

DEBUG_ONLY(sent_opcodes += "|RFNM|";)

// Extended information

if (GetExtendedRequestsVersion() > 0) {

m_reqfile->WritePartStatus(&dataFileReq);

}

if (GetExtendedRequestsVersion() > 1) {

m_reqfile->WriteCompleteSourcesCount(&dataFileReq);

}

if (m_reqfile->GetPartCount() > 1) {

DEBUG_ONLY(sent_opcodes += "|RFID|";)

dataFileReq.WriteUInt8(OP_SETREQFILEID);

}

if (IsEmuleClient()) {

SetRemoteQueueFull(true);

SetRemoteQueueRank(0);

}

if (IsSourceRequestAllowed()) {

if (SupportsSourceExchange2()) {

DEBUG_ONLY(sent_opcodes += "|RSRC2|";)

dataFileReq.WriteUInt8(OP_REQUESTSOURCES2);

dataFileReq.WriteUInt8(SOURCEEXCHANGE2_VERSION);

const uint16 nOptions = 0; // 16 ... Reserved

dataFileReq.WriteUInt16(nOptions);

} else {

DEBUG_ONLY(sent_opcodes += "|RSRC|";)

dataFileReq.WriteUInt8(OP_REQUESTSOURCES);

}

m_reqfile->SetLastAnsweredTimeTimeout();

SetLastAskedForSources();

}

if (IsSupportingAICH()) {

DEBUG_ONLY(sent_opcodes += "|AFHR|";)

dataFileReq.WriteUInt8(OP_AICHFILEHASHREQ);

}

CPacket *packet = new CPacket(dataFileReq,

OP_EMULEPROT,

(SupportExtMultiPacket() ? OP_MULTIPACKET_EXT : OP_MULTIPACKET));

theStats::AddUpOverheadFileRequest(packet->GetPacketSize());

AddDebugLogLineN(logLocalClient,

CFormat("Local Client: %s (%s) to %s") %

(SupportExtMultiPacket() ? "OP_MULTIPACKET_EXT" : "OP_MULTIPACKET") %

sent_opcodes % GetFullIP());

SendPacket(packet, true);

} else {

// This is extended information

if (GetExtendedRequestsVersion() > 0) {

m_reqfile->WritePartStatus(&dataFileReq);

}

if (GetExtendedRequestsVersion() > 1) {

m_reqfile->WriteCompleteSourcesCount(&dataFileReq);

}

CPacket *packet = new CPacket(dataFileReq, OP_EDONKEYPROT, OP_REQUESTFILENAME);

theStats::AddUpOverheadFileRequest(packet->GetPacketSize());

AddDebugLogLineN(logLocalClient, "Local Client: OP_REQUESTFILENAME to " + GetFullIP());

SendPacket(packet, true);

// 26-Jul-2003: removed requesting the file status for files <= PARTSIZE for better

// compatibility with ed2k protocol (eDonkeyHybrid). if the remote client answers the

// OP_REQUESTFILENAME with OP_REQFILENAMEANSWER the file is shared by the remote client. if we

// know that the file is shared, we know also that the file is complete and don't need to

// request the file status.

// Sending the packet could have deleted the client, check m_reqfile

if (m_reqfile && (m_reqfile->GetPartCount() > 1)) {

CMemFile dataSetReqFileID(16);

dataSetReqFileID.WriteHash(m_reqfile->GetFileHash());

packet = new CPacket(dataSetReqFileID, OP_EDONKEYPROT, OP_SETREQFILEID);

theStats::AddUpOverheadFileRequest(packet->GetPacketSize());

AddDebugLogLineN(logLocalClient, "Local Client: OP_SETREQFILEID to " + GetFullIP());

SendPacket(packet, true);

}

if (IsEmuleClient()) {

SetRemoteQueueFull(true);

SetRemoteQueueRank(0);

}

// Sending the packet could have deleted the client, check m_reqfile

if (m_reqfile && IsSourceRequestAllowed()) {

m_reqfile->SetLastAnsweredTimeTimeout();

CMemFile packetdata;

if (SupportsSourceExchange2()) {

packetdata.WriteUInt8(SOURCEEXCHANGE2_VERSION);

packetdata.WriteUInt16(0 /* Reserved */);

}

packetdata.WriteHash(m_reqfile->GetFileHash());

packet = new CPacket(packetdata,

OP_EMULEPROT,

SupportsSourceExchange2() ? OP_REQUESTSOURCES2 : OP_REQUESTSOURCES);

theStats::AddUpOverheadSourceExchange(packet->GetPacketSize());

AddDebugLogLineN(logLocalClient, "Local Client: OP_REQUESTSOURCES to " + GetFullIP());

SendPacket(packet, true, true);

SetLastAskedForSources();

}

// Sending the packet could have deleted the client, check m_reqfile

if (m_reqfile && IsSupportingAICH()) {

packet = new CPacket(OP_AICHFILEHASHREQ, 16, OP_EMULEPROT);

packet->Copy16ToDataBuffer((const char *)m_reqfile->GetFileHash().GetHash());

theStats::AddUpOverheadOther(packet->GetPacketSize());

AddDebugLogLineN(

logLocalClient, "Local Client: OP_AICHFILEHASHREQ to " + GetFullIP());

SendPacket(packet, true, true);

}

}

}

void CUpDownClient::ProcessFileInfo(const CMemFile *data, const CPartFile *file)

{

// 0.42e

if (file == NULL) {

throw wxString("ERROR: Wrong file ID (ProcessFileInfo; file==NULL)");

}

if (m_reqfile == NULL) {

throw wxString("ERROR: Wrong file ID (ProcessFileInfo; m_reqfile==NULL)");

}

if (file != m_reqfile) {

throw wxString("ERROR: Wrong file ID (ProcessFileInfo; m_reqfile!=file)");

}

m_clientFilename = data->ReadString((GetUnicodeSupport() != utf8strNone));

// 26-Jul-2003: removed requesting the file status for files <= PARTSIZE for better compatibility with

// ed2k protocol (eDonkeyHybrid). if the remote client answers the OP_REQUESTFILENAME with

// OP_REQFILENAMEANSWER the file is shared by the remote client. if we know that the file is shared,

// we know also that the file is complete and don't need to request the file status.

if (m_reqfile->GetPartCount() == 1) {

m_nPartCount = m_reqfile->GetPartCount();

m_reqfile->UpdatePartsFrequency(this, false); // Decrement

m_downPartStatus.setsize(m_nPartCount, 1);

m_reqfile->UpdatePartsFrequency(this, true); // Increment

m_bCompleteSource = true;

UpdateDisplayedInfo();

// even if the file is <= PARTSIZE, we _may_ need the hashset for that file (if the file size

// == PARTSIZE)

if (m_reqfile->IsHashSetNeeded()) {

if (m_socket) {

CPacket *packet = new CPacket(OP_HASHSETREQUEST, 16, OP_EDONKEYPROT);

packet->Copy16ToDataBuffer((const char *)m_reqfile->GetFileHash().GetHash());

theStats::AddUpOverheadFileRequest(packet->GetPacketSize());

AddDebugLogLineN(

logLocalClient, "Local Client: OP_HASHSETREQUEST to " + GetFullIP());

SendPacket(packet, true, true);

SetDownloadState(DS_REQHASHSET);

m_fHashsetRequesting = 1;

m_reqfile->SetHashSetNeeded(false);

} else {

wxFAIL;

}

} else {

SendStartupLoadReq();

}

m_reqfile->UpdatePartsInfo();

}

}

void CUpDownClient::ProcessFileStatus(bool bUdpPacket, const CMemFile *data, const CPartFile *file)

{

// 0.42e

wxString strReqFileNull("ERROR: Wrong file ID (ProcessFileStatus; m_reqfile==NULL)");

if (!m_reqfile || file != m_reqfile) {

if (!m_reqfile) {

throw strReqFileNull;

}

throw wxString("ERROR: Wrong file ID (ProcessFileStatus; m_reqfile!=file)");

}

uint16 nED2KPartCount = data->ReadUInt16();

m_reqfile->UpdatePartsFrequency(this, false); // Decrement

m_downPartStatus.clear();

bool bPartsNeeded = false;

if (!nED2KPartCount) {

m_nPartCount = m_reqfile->GetPartCount();

m_downPartStatus.setsize(m_nPartCount, 1);

bPartsNeeded = true;

m_bCompleteSource = true;

} else {

// Somehow this happened.

if (!m_reqfile) {

throw strReqFileNull;

}

if (m_reqfile->GetED2KPartCount() != nED2KPartCount) {

wxString strError;

strError << "ProcessFileStatus - wrong part number recv=" << nED2KPartCount

<< " expected=" << m_reqfile->GetED2KPartCount() << " "

<< m_reqfile->GetFileHash().Encode();

m_nPartCount = 0;

throw strError;

}

m_nPartCount = m_reqfile->GetPartCount();

m_bCompleteSource = false;

m_downPartStatus.setsize(m_nPartCount, 0);

uint16 done = 0;

try {

while (done != m_nPartCount) {

uint8 toread = data->ReadUInt8();

for (uint8 i = 0; i < 8; i++) {

bool status = ((toread >> i) & 1) ? 1 : 0;

m_downPartStatus.set(done, status);

if (status) {

if (!m_reqfile->IsComplete(done)) {

bPartsNeeded = true;

}

}

done++;

if (done == m_nPartCount) {

break;

}

}

}

} catch (...) {

// We want the counts to be updated, even if we fail to read everything

m_reqfile->UpdatePartsFrequency(this, true); // Increment

throw;

}

}

m_reqfile->UpdatePartsFrequency(this, true); // Increment

UpdateDisplayedInfo();

// NOTE: This function is invoked from TCP and UDP socket!

if (!bUdpPacket) {

if (!bPartsNeeded) {

SetDownloadState(DS_NONEEDEDPARTS);

} else if (m_reqfile->IsHashSetNeeded()) {

// If we are using the eMule filerequest packets, this is taken care of in the

// Multipacket!

if (m_socket) {

CPacket *packet = new CPacket(OP_HASHSETREQUEST, 16, OP_EDONKEYPROT);

packet->Copy16ToDataBuffer((const char *)m_reqfile->GetFileHash().GetHash());

theStats::AddUpOverheadFileRequest(packet->GetPacketSize());

AddDebugLogLineN(

logLocalClient, "Local Client: OP_HASHSETREQUEST to " + GetFullIP());

SendPacket(packet, true, true);

SetDownloadState(DS_REQHASHSET);

m_fHashsetRequesting = 1;

m_reqfile->SetHashSetNeeded(false);

} else {

wxFAIL;

}

} else {

SendStartupLoadReq();

}

} else {

if (!bPartsNeeded) {

SetDownloadState(DS_NONEEDEDPARTS);

} else {

SetDownloadState(DS_ONQUEUE);

}

}

m_reqfile->UpdatePartsInfo();

}

bool CUpDownClient::AddRequestForAnotherFile(CPartFile *file)

{

if (m_A4AF_list.find(file) == m_A4AF_list.end()) {

// When we access a non-existing entry entry, it will be zeroed by default,

// so we have to set NeededParts. All in one go.

m_A4AF_list[file].NeededParts = true;

file->AddA4AFSource(this);

return true;

} else {

return false;

}

}

bool CUpDownClient::DeleteFileRequest(CPartFile *file)

{

return (m_A4AF_list.erase(file) > 0);

}

void CUpDownClient::DeleteAllFileRequests()

{

m_A4AF_list.clear();

}

/* eMule 0.30c implementation, i give it a try (Creteil) BEGIN ... */

void CUpDownClient::SetDownloadState(uint8 byNewState)

{

if (m_nDownloadState != byNewState) {

if (m_reqfile) {

// Notify the client that this source has changed its state

m_reqfile->ClientStateChanged(m_nDownloadState, byNewState);

if (byNewState == DS_DOWNLOADING) {

m_reqfile->AddDownloadingSource(this);

} else if (m_nDownloadState == DS_DOWNLOADING) {

m_reqfile->RemoveDownloadingSource(this);

}

}

if (byNewState == DS_DOWNLOADING) {

msReceivedPrev = GetTickCount64();

theStats::AddDownloadingSource();

} else if (m_nDownloadState == DS_DOWNLOADING) {

theStats::RemoveDownloadingSource();

}

if (m_nDownloadState == DS_DOWNLOADING) {

m_nDownloadState = byNewState;

ClearDownloadBlockRequests();

kBpsDown = 0.0;

bytesReceivedCycle = 0;

msReceivedPrev = 0;

if (byNewState == DS_NONE) {

if (m_reqfile) {

m_reqfile->UpdatePartsFrequency(this, false); // Decrement

}

m_downPartStatus.clear();

m_nPartCount = 0;

}

// (Old code disabled the per-socket download cap here on

// transition out of DS_DOWNLOADING. The cap is now a global

// budget in CDownloadBandwidthThrottler shared across all

// sockets, so there's no per-socket state to clear.)

}

m_nDownloadState = byNewState;

if (GetDownloadState() == DS_DOWNLOADING) {

if (IsEmuleClient()) {

SetRemoteQueueFull(false);

}

SetRemoteQueueRank(0); // eMule 0.30c set like this ...

}

UpdateDisplayedInfo(true);

}

}

/* eMule 0.30c implementation, i give it a try (Creteil) END ... */

void CUpDownClient::ProcessHashSet(const uint8_t *packet, uint32 size)

{

if ((!m_reqfile) || md4cmp(packet, m_reqfile->GetFileHash().GetHash())) {

throw wxString("Wrong fileid sent (ProcessHashSet)");

}

if (!m_fHashsetRequesting) {

throw wxString("Received unsolicited hashset, ignoring it.");

}

CMemFile data(packet, size);

if (m_reqfile->LoadHashsetFromFile(&data, true)) {

m_fHashsetRequesting = 0;

} else {

m_reqfile->SetHashSetNeeded(true);

throw wxString("Corrupted or invalid hashset received");

}

SendStartupLoadReq();

}

void CUpDownClient::SendBlockRequests()

{

m_dwLastBlockReceived = ::GetTickCount64();

if (!m_reqfile) {

return;

}

// RTT/BDP-adaptive request-pipeline depth. The number of outstanding block

// requests needed to keep a link busy is the bandwidth-delay product:

// bytes_in_flight = rate * RTT. On a low-latency link (LAN) the BDP is a

// fraction of a 180 KB block, so we stay shallow and avoid the burst/starve

// oscillation a deep pipeline provokes against a fast local peer; on a

// high-latency link the BDP is large, so we go deep and hide the round-trip.

// Unlike a speed-only ladder this distinguishes a fast LAN peer from a fast

// WAN peer -- the thing that actually determines the depth needed. m_minRTT is

// the min-filtered request->first-byte round-trip (see ProcessBlockPacket).

//

// The flat cap-24 clamp is deliberate, not a placeholder. Benchmarking showed the

// WAN ceiling is the OS TCP socket-buffer autotune limit (~4 MB by default), which

// pins throughput near 40 MB/s at 100 ms RTT no matter how deep the request

// pipeline runs -- and cap 24 (24 * 180 KB = 4.3 MB of authorised in-flight data)

// already covers that. A deeper pipe buys no throughput, and 24 still sits inside

// eMule's own pending range (its gate is 2*blockCount = 18, with a top-up batch

// reaching ~27); lifting the OS buffer is host tuning, not a client concern. (The

// 2x overshoot factor below is still an empirical constant.)

const float rttMs = (m_minRTT > 0) ? (float)m_minRTT : 1.0f;

const float bdpBlocks = ((float)GetKBpsDown() * 1024.0f) * (rttMs / 1000.0f) / (float)EMBLOCKSIZE;

// 2x overshoot + margin: sizing the pipe to exactly the current BDP is

// self-limiting (the measured rate is itself capped by the current pipe, so it

// can never grow past a low equilibrium). Over-provisioning gives headroom for

// the rate to climb until it hits the link's real ceiling, at which point the

// BDP -- and thus the depth -- settles at the bandwidth-delay product.

size_t pendingCap = 2 * (size_t)bdpBlocks + STANDARD_BLOCKS_REQUEST;

if (pendingCap < STANDARD_BLOCKS_REQUEST) {

// The floor doubles as the slow-source guard: a trickle source -- or one with

// no RTT sample yet (rttMs defaults to 1) -- has a near-zero BDP and is held at

// the 3-block minimum, never handed a deep pipe.

pendingCap = STANDARD_BLOCKS_REQUEST;

} else if (pendingCap > 24) {

pendingCap = 24; // OS-buffer ceiling (see above); still within eMule's pending range

}

// Smooth continuous refill: top the in-flight + staged block count up to

// pendingCap by the exact shortfall each call, never in bursts. Bursty refill

// makes the leecher emit requests in clusters that overfill then starve a fast

// peer's send queue, leaving it idle between bursts (and tanking throughput on

// a low-latency link). Requests still ride the wire 3 to an OP_REQUESTPARTS

// packet, emitted across successive calls (see below).

{

const size_t inSystem = m_PendingBlocks_list.size() + m_DownloadBlocks_list.size();

if (inSystem < pendingCap) {

uint16 count = (uint16)(pendingCap - inSystem);

std::vector<Requested_Block_Struct *> toadd;

if (m_reqfile->GetNextRequestedBlock(this, toadd, count)) {

for (uint16 i = 0; i < count; i++) {

m_DownloadBlocks_list.push_back(toadd[i]);

}

}

}

}

while (m_PendingBlocks_list.size() < pendingCap && !m_DownloadBlocks_list.empty()) {

Pending_Block_Struct *pblock = new Pending_Block_Struct;

pblock->block = m_DownloadBlocks_list.front();

pblock->zStream = NULL;

pblock->totalUnzipped = 0;

pblock->fZStreamError = 0;

pblock->fRecovered = 0;

pblock->fQueued = 0; // Block sits in our local queue, not yet asked over the wire.

pblock->sentTime = 0;

m_PendingBlocks_list.push_back(pblock);

m_DownloadBlocks_list.pop_front();

}

if (m_PendingBlocks_list.empty()) {

CUpDownClient *slower_client = NULL;

bool nearCompletion =

m_reqfile->GetPartCount() > 4 &&

(m_reqfile->GetFileSize() > m_reqfile->GetCompletedSize()) &&

((m_reqfile->GetFileSize() - m_reqfile->GetCompletedSize()) <= (4 * PARTSIZE));

if (thePrefs::GetDropSlowSources() || (nearCompletion && thePrefs::GetEndgame())) {

slower_client = m_reqfile->GetSlowerDownloadingClient(m_lastaverage, this);

}

if (slower_client == NULL) {

slower_client = this;

}

if (!slower_client->GetSentCancelTransfer()) {

CPacket *packet = new CPacket(OP_CANCELTRANSFER, 0, OP_EDONKEYPROT);

theStats::AddUpOverheadFileRequest(packet->GetPacketSize());

// if (slower_client != this) {

// printf("Dropped client %p to allow client %p to

// download\n",slower_client, this);

// }

slower_client->ClearDownloadBlockRequests();

slower_client->SendPacket(packet, true, true);

slower_client->SetSentCancelTransfer(1);

}

if (slower_client == this && nearCompletion) {

// requeue instead of self-banishing at endgame

slower_client->SetDownloadState(DS_ONQUEUE);

} else {

slower_client->SetDownloadState(DS_NONEEDEDPARTS);

}

if (slower_client != this) {

// Re-request freed blocks.

AddDebugLogLineN(logLocalClient,

"Local Client: OP_CANCELTRANSFER (faster source eager to transfer) to " +

slower_client->GetFullIP());

wxASSERT(m_DownloadBlocks_list.empty());

wxASSERT(m_PendingBlocks_list.empty());

uint16 count = (uint16)pendingCap;

std::vector<Requested_Block_Struct *> toadd;

if (m_reqfile->GetNextRequestedBlock(this, toadd, count)) {

for (int i = 0; i != count; i++) {

Pending_Block_Struct *pblock = new Pending_Block_Struct;

pblock->block = toadd[i];

pblock->zStream = NULL;

pblock->totalUnzipped = 0;

pblock->fZStreamError = 0;

pblock->fRecovered = 0;

pblock->fQueued = 0;

pblock->sentTime = 0;

m_PendingBlocks_list.push_back(pblock);

}

} else {

// It's possible the freed blocks were not available on our source.

// Just drop ourselves gracefully instead of crashing.

if (!GetSentCancelTransfer()) {

CPacket *packet = new CPacket(OP_CANCELTRANSFER, 0, OP_EDONKEYPROT);

theStats::AddUpOverheadFileRequest(packet->GetPacketSize());

ClearDownloadBlockRequests();

SendPacket(packet, true, true);

SetSentCancelTransfer(1);

}

AddDebugLogLineN(logLocalClient,

"Local Client: OP_CANCELTRANSFER (freed blocks not available "

"here) to " +

GetFullIP());

if (nearCompletion) {

// requeue instead of self-banishing at endgame

SetDownloadState(DS_ONQUEUE);

} else {

SetDownloadState(DS_NONEEDEDPARTS);

}

return;

}

} else {

// Drop this one.

AddDebugLogLineN(logLocalClient,

"Local Client: OP_CANCELTRANSFER (no free blocks) to " + GetFullIP());

// #warning Kry - Would be nice to swap A4AF here.

return;

}

}

// Collect up to 3 pending blocks that have not yet been written to a

// REQUESTPARTS packet (fQueued == 0). OP_REQUESTPARTS / OP_REQUESTPARTS_I64

// carry exactly 3 <start,end> pairs on the wire (see ProcessRequestPartsPacket

// in UploadClient.cpp, which unconditionally reads 3 pairs); asking for more

// in one packet corrupts the wire format and the sender ignores us. Pad with

// zero-pairs if we have fewer than 3 unqueued blocks.

//

// To take more than 3 blocks in flight we emit one 3-block packet per call to

// SendBlockRequests() and let the caller's existing re-invocation loop (after

// each OP_SENDINGPART, OP_ACCEPTUPLOADREQ, etc.) issue further packets until

// m_PendingBlocks_list is fully queued -- the same pattern eMule uses

// (eMule0.70b srchybrid/DownloadClient.cpp ::CreateBlockRequests +

// ::SendBlockRequests).

std::vector<Pending_Block_Struct *> toRequest;

bool bHasLongBlocks = false;

const size_t perPacketLimit = STANDARD_BLOCKS_REQUEST;

toRequest.reserve(perPacketLimit);

for (std::list<Pending_Block_Struct *>::iterator it = m_PendingBlocks_list.begin();

it != m_PendingBlocks_list.end() && toRequest.size() < perPacketLimit;

++it) {

Pending_Block_Struct *pending = *it;

if (pending->fQueued) {

continue; // Already asked for over the wire; sender owes us bytes.

}

wxASSERT(pending->block->StartOffset <= pending->block->EndOffset);

if (pending->block->StartOffset > 0xFFFFFFFF || pending->block->EndOffset > 0xFFFFFFFF) {

bHasLongBlocks = true;

if (!SupportsLargeFiles()) {

// Requesting a large block from a client that doesn't support large files?

if (!GetSentCancelTransfer()) {

CPacket *cancel_packet =

new CPacket(OP_CANCELTRANSFER, 0, OP_EDONKEYPROT);

theStats::AddUpOverheadFileRequest(cancel_packet->GetPacketSize());

AddDebugLogLineN(logLocalClient,

"Local Client: OP_CANCELTRANSFER to " + GetFullIP());

SendPacket(cancel_packet, true, true);

SetSentCancelTransfer(1);

}

SetDownloadState(DS_ERROR);

return;

}

}

toRequest.push_back(pending);

}

if (toRequest.empty()) {

// Every pending block is already in flight. Nothing to send;

// SendBlockRequests will be re-invoked as those blocks complete.

return;

}

CPacket *packet = NULL;

// OP_REQUESTPARTS / OP_REQUESTPARTS_I64: always 3 blocks on the wire. Pad

// missing entries with zero <start,end> pairs; the upload-side parser

// (UploadClient.cpp ProcessRequestPartsPacket) silently skips entries with

// end <= start, so zero pairs are inert.

const size_t kWireBlocks = STANDARD_BLOCKS_REQUEST;

const size_t offsetSize = bHasLongBlocks ? 8 : 4;

CMemFile data(16 + kWireBlocks * offsetSize * 2);

data.WriteHash(m_reqfile->GetFileHash());

for (size_t i = 0; i < kWireBlocks; ++i) {

uint64 start = 0;

if (i < toRequest.size()) {

Pending_Block_Struct *pending = toRequest[i];

pending->fZStreamError = 0;

pending->fRecovered = 0;

pending->fQueued = 1;

pending->sentTime = ::GetTickCount64();

start = pending->block->StartOffset;

}

if (bHasLongBlocks) {

data.WriteUInt64(start);

} else {

data.WriteUInt32((uint32)start);

}

}

for (size_t i = 0; i < kWireBlocks; ++i) {

uint64 end = 0;

if (i < toRequest.size()) {

end = toRequest[i]->block->EndOffset + 1;

}

if (bHasLongBlocks) {

data.WriteUInt64(end);

} else {

data.WriteUInt32((uint32)end);

}

}

packet = new CPacket(data,

(bHasLongBlocks ? OP_EMULEPROT : OP_EDONKEYPROT),

(bHasLongBlocks ? (uint8)OP_REQUESTPARTS_I64 : (uint8)OP_REQUESTPARTS));

AddDebugLogLineN(logLocalClient,

CFormat("Local Client: %s(%u of %u pending) to %s") %

(bHasLongBlocks ? "OP_REQUESTPARTS_I64" : "OP_REQUESTPARTS") %

(unsigned)toRequest.size() % (unsigned)m_PendingBlocks_list.size() % GetFullIP());

if (packet) {

theStats::AddUpOverheadFileRequest(packet->GetPacketSize());

SendPacket(packet, true, true);

}

}

/*

Barry - Originally this only wrote to disk when a full 180k block

had been received from a client, and only asked for data in

180k blocks.

This meant that on average 90k was lost for every connection

to a client data source. That is a lot of wasted data.

To reduce the lost data, packets are now written to a buffer

and flushed to disk regularly regardless of size downloaded.

This includes compressed packets.

Data is also requested only where gaps are, not in 180k blocks.

The requests will still not exceed 180k, but may be smaller to

fill a gap.

*/

void CUpDownClient::ProcessBlockPacket(const uint8_t *packet, uint32 size, bool packed, bool largeblocks)

{

// Ignore if no data required

if (!(GetDownloadState() == DS_DOWNLOADING || GetDownloadState() == DS_NONEEDEDPARTS)) {

return;

}

// This vars are defined here to be able to use them on the catch

int header_size = 16;

uint64 nStartPos = 0;

uint64 nEndPos = 0;

uint32 nBlockSize = 0;

uint32 lenUnzipped = 0;

// Update stats

m_dwLastBlockReceived = ::GetTickCount64();

try {

// Read data from packet

const CMemFile data(packet, size);

// Check that this data is for the correct file

if ((!m_reqfile) || data.ReadHash() != m_reqfile->GetFileHash()) {

throw wxString("Wrong fileid sent (ProcessBlockPacket)");

}

// Find the start & end positions, and size of this chunk of data

if (largeblocks) {

nStartPos = data.ReadUInt64();

header_size += 8;

} else {

nStartPos = data.ReadUInt32();

header_size += 4;

}

if (packed) {

nBlockSize = data.ReadUInt32();

header_size += 4;

nEndPos = nStartPos + (size - header_size);

} else {

if (largeblocks) {

nEndPos = data.ReadUInt64();

header_size += 8;

} else {

nEndPos = data.ReadUInt32();

header_size += 4;

}

}

// Check that packet size matches the declared data size + header size

if (nEndPos == nStartPos || size != ((nEndPos - nStartPos) + header_size)) {

throw wxString("Corrupted or invalid DataBlock received (ProcessBlockPacket)");

}

theStats::AddDownloadFromSoft(GetClientSoft(), size - header_size);

bytesReceivedCycle += size - header_size;

credits->AddDownloaded(size - header_size, GetIP(), theApp->CryptoAvailable());

// Move end back one, should be inclusive

nEndPos--;

// Loop through to find the reserved block that this is within

std::list<Pending_Block_Struct *>::iterator it = m_PendingBlocks_list.begin();

for (; it != m_PendingBlocks_list.end(); ++it) {

Pending_Block_Struct *cur_block = *it;

if ((cur_block->block->StartOffset <= nStartPos) &&

(cur_block->block->EndOffset >= nStartPos)) {

// Found reserved block

if (cur_block->block->StartOffset == nStartPos) {

// This block just started transferring. Set the start time.

m_last_block_start = ::GetTickCount64();

// RTT sample: the matched request->first-byte round-trip for

// this block. Min-filtered so pipeline queuing delay never

// inflates the estimate (BBR-style min-RTT floor); used to size

// the request pipeline to the bandwidth-delay product.

if (cur_block->sentTime != 0) {

uint64 sample = m_last_block_start - cur_block->sentTime;

if (m_minRTT == 0 || sample < m_minRTT) {

m_minRTT = sample;

}

cur_block->sentTime = 0;

}

}

if (cur_block->fZStreamError) {

AddDebugLogLineN(logZLib,

CFormat("Ignoring %u bytes of block %u-%u because of "

"erroneous zstream state for file: %s") %

(size - header_size) % nStartPos % nEndPos %

m_reqfile->GetFileName());

m_reqfile->RemoveBlockFromList(

cur_block->block->StartOffset, cur_block->block->EndOffset);

return;

}

// Remember this start pos, used to draw part downloading in list

m_lastDownloadingPart = nStartPos / PARTSIZE;

// Occasionally packets are duplicated, no point writing it twice

// This will be 0 in these cases, or the length written otherwise

uint32 lenWritten = 0;

// Handle differently depending on whether packed or not

if (!packed) {

// security sanitize check

if (nEndPos > cur_block->block->EndOffset) {

AddDebugLogLineN(logRemoteClient,

CFormat("Received Blockpacket exceeds requested "

"boundaries (requested end: %u, Part: %u, "

"received end: %u, Part: %u), file: %s "

"remote IP: %s") %

cur_block->block->EndOffset %

(uint32)(cur_block->block->EndOffset /

PARTSIZE) %

nEndPos % (uint32)(nEndPos / PARTSIZE) %

m_reqfile->GetFileName() %

Uint32toStringIP(GetIP()));

m_reqfile->RemoveBlockFromList(cur_block->block->StartOffset,

cur_block->block->EndOffset);

return;

}

// Write to disk (will be buffered in part file class)

lenWritten = m_reqfile->WriteToBuffer(size - header_size,

(uint8_t *)(packet + header_size),