2 // This file is part of the aMule Project.
4 // Copyright (c) 2003-2008 aMule Team ( admin@amule.org / http://www.amule.org )
5 // Copyright (c) 2002-2008 Merkur ( devs@emule-project.net / http://www.emule-project.net )
7 // Any parts of this program derived from the xMule, lMule or eMule project,
8 // or contributed by third-party developers are copyrighted by their
11 // This program is free software; you can redistribute it and/or modify
12 // it under the terms of the GNU General Public License as published by
13 // the Free Software Foundation; either version 2 of the License, or
14 // (at your option) any later version.
16 // This program is distributed in the hope that it will be useful,
17 // but WITHOUT ANY WARRANTY; without even the implied warranty of
18 // MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
19 // GNU General Public License for more details.
21 // You should have received a copy of the GNU General Public License
22 // along with this program; if not, write to the Free Software
23 // Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301, USA
26 #include "UploadQueue.h" // Interface declarations
28 #include <protocol/Protocols.h>
29 #include <protocol/ed2k/Client2Client/TCP.h>
30 #include <common/Macros.h>
31 #include <common/Constants.h>
35 #include "Types.h" // Do_not_auto_remove (win32)
38 #include <winsock.h> // Do_not_auto_remove
40 #include <sys/types.h> // Do_not_auto_remove
41 #include <netinet/in.h> // Do_not_auto_remove
42 #include <arpa/inet.h> // Do_not_auto_remove
45 #include "ServerConnect.h" // Needed for CServerConnect
46 #include "KnownFile.h" // Needed for CKnownFile
47 #include "Packet.h" // Needed for CPacket
48 #include "ClientTCPSocket.h" // Needed for CClientTCPSocket
49 #include "SharedFileList.h" // Needed for CSharedFileList
50 #include "updownclient.h" // Needed for CUpDownClient
51 #include "amule.h" // Needed for theApp
52 #include "Preferences.h"
53 #include "ClientList.h"
54 #include "Statistics.h" // Needed for theStats
56 #include <common/Format.h>
57 #include "UploadBandwidthThrottler.h"
58 #include "GuiEvents.h" // Needed for Notify_*
61 //TODO rewrite the whole networkcode, use overlapped sockets
63 CUploadQueue::CUploadQueue()
65 m_nLastStartUpload
= 0;
67 lastupslotHighID
= true;
68 m_allowKicking
= true;
72 CUpDownClient
* CUploadQueue::SortGetBestClient(bool sortonly
)
74 CUpDownClient
* newclient
= NULL
;
75 // Track if we purged any clients from the queue, as to only send one notify in total
77 uint32 tick
= GetTickCount();
79 CClientPtrList::iterator it
= m_waitinglist
.begin();
80 for (; it
!= m_waitinglist
.end(); ) {
81 CClientPtrList::iterator it2
= it
++;
82 CUpDownClient
* cur_client
= *it2
;
85 if (tick
- cur_client
->GetLastUpRequest() > MAX_PURGEQUEUETIME
86 || !theApp
->sharedfiles
->GetFileByID(cur_client
->GetUploadFileID())) {
88 cur_client
->ClearWaitStartTime();
89 RemoveFromWaitingQueue(it2
);
90 if (!cur_client
->GetSocket()) {
91 if (cur_client
->Disconnected(wxT("AddUpNextClient - purged"))) {
92 cur_client
->Safe_Delete();
99 if (cur_client
->IsBanned() || IsSuspended(cur_client
->GetUploadFileID())) { // Banned client or suspended upload ?
100 cur_client
->ClearScore();
105 // Calculate score of current client
106 uint32 cur_score
= cur_client
->CalculateScore();
107 // Check if it's better than that of a previous one, and move it up then.
108 CClientPtrList::iterator it1
= it2
;
109 while (it1
!= m_waitinglist
.begin()) {
111 if (cur_score
> (*it1
)->GetScore()) {
117 // no need to check further since list is already sorted
124 // - calculate queue rank
125 // - find best high id client
126 // - mark all better low id clients as enabled for upload
128 for (it
= m_waitinglist
.begin(); it
!= m_waitinglist
.end(); ) {
129 CClientPtrList::iterator it2
= it
++;
130 CUpDownClient
* cur_client
= *it2
;
131 cur_client
->SetUploadQueueWaitingPosition(rank
++);
133 // There's a better high id client
134 cur_client
->m_bAddNextConnect
= false;
136 if (cur_client
->HasLowID() && !cur_client
->IsConnected()) {
137 // No better high id client, so start upload to this one once it connects
138 cur_client
->m_bAddNextConnect
= true;
140 // We found a high id client (or a currently connected low id client)
141 newclient
= cur_client
;
142 cur_client
->m_bAddNextConnect
= false;
144 RemoveFromWaitingQueue(it2
);
146 lastupslotHighID
= true; // VQB LowID alternate
152 // Update the count on GUI if any clients were purged
153 if (purged
|| (newclient
&& !sortonly
)) {
154 Notify_ShowQueueCount(m_waitinglist
.size());
158 AddDebugLogLineN(logLocalClient
, CFormat(wxT("Current UL queue (%d):")) % (rank
- 1));
159 for (it
= m_waitinglist
.begin(); it
!= m_waitinglist
.end(); it
++) {
160 CUpDownClient
* c
= *it
;
161 AddDebugLogLineN(logLocalClient
, CFormat(wxT("%4d %7d %s %5d %s"))
162 % c
->GetUploadQueueWaitingPosition()
164 % (c
->HasLowID() ? (c
->IsConnected() ? wxT("LoCon") : wxT("LowId")) : wxT("High "))
175 void CUploadQueue::AddUpNextClient(CUpDownClient
* directadd
)
177 CUpDownClient
* newclient
= NULL
;
178 // select next client or use given client
180 newclient
= SortGetBestClient(false);
185 // Check if requested file is suspended or not shared (maybe deleted recently)
187 if (IsSuspended(directadd
->GetUploadFileID())
188 || !theApp
->sharedfiles
->GetFileByID(directadd
->GetUploadFileID())) {
191 newclient
= directadd
;
195 if (IsDownloading(newclient
)) {
198 // tell the client that we are now ready to upload
199 if (!newclient
->IsConnected()) {
200 newclient
->SetUploadState(US_CONNECTING
);
201 if (!newclient
->TryToConnect(true)) {
205 CPacket
* packet
= new CPacket(OP_ACCEPTUPLOADREQ
, 0, OP_EDONKEYPROT
);
206 theStats::AddUpOverheadFileRequest(packet
->GetPacketSize());
207 AddDebugLogLineN( logLocalClient
, wxT("Local Client: OP_ACCEPTUPLOADREQ to ") + newclient
->GetFullIP() );
208 newclient
->SendPacket(packet
,true);
209 newclient
->SetUploadState(US_UPLOADING
);
211 newclient
->SetUpStartTime();
212 newclient
->ResetSessionUp();
214 theApp
->uploadBandwidthThrottler
->AddToStandardList(m_uploadinglist
.size(), newclient
->GetSocket());
215 m_uploadinglist
.push_back(newclient
);
216 theStats::AddUploadingClient();
219 CKnownFile
* reqfile
= (CKnownFile
*) newclient
->GetUploadFile();
221 reqfile
->statistic
.AddAccepted();
225 void CUploadQueue::Process()
227 // Check if someone's waiting, if there is a slot for him,
228 // or if we should try to free a slot for him
229 uint32 tick
= GetTickCount();
230 // Nobody waiting or upload started recently
231 // (Actually instead of "empty" it should check for "no HighID clients queued",
232 // but the cost for that outweights the benefit. As it is, a slot will be freed
233 // even if it can't be taken because all of the queue is LowID. But just one,
234 // and the kicked client will instantly get it back if he has HighID.)
235 if (m_waitinglist
.empty() || tick
- m_nLastStartUpload
< 1000) {
236 m_allowKicking
= false;
237 // Already a slot free, try to fill it
238 } else if (m_uploadinglist
.size() < GetMaxSlots()) {
239 m_allowKicking
= false;
240 m_nLastStartUpload
= tick
;
242 // All slots taken, try to free one
244 m_allowKicking
= true;
247 // The loop that feeds the upload slots with data.
248 CClientPtrList::iterator it
= m_uploadinglist
.begin();
249 while (it
!= m_uploadinglist
.end()) {
250 // Get the client. Note! Also updates pos as a side effect.
251 CUpDownClient
* cur_client
= *it
++;
253 // It seems chatting or friend slots can get stuck at times in upload.. This needs looked into..
254 if (!cur_client
->GetSocket()) {
255 RemoveFromUploadQueue(cur_client
);
256 if(cur_client
->Disconnected(_T("CUploadQueue::Process"))){
257 cur_client
->Safe_Delete();
260 cur_client
->SendBlockData();
264 // Save used bandwidth for speed calculations
265 uint64 sentBytes
= theApp
->uploadBandwidthThrottler
->GetNumberOfSentBytesSinceLastCallAndReset();
266 (void)theApp
->uploadBandwidthThrottler
->GetNumberOfSentBytesOverheadSinceLastCallAndReset();
270 theStats::AddSentBytes(sentBytes
);
273 // Periodically resort queue if it doesn't happen anyway
274 if ((sint32
) (tick
- m_lastSort
) > MIN2MS(2)) {
275 SortGetBestClient(true);
280 uint16
CUploadQueue::GetMaxSlots() const
282 uint16 nMaxSlots
= 0;
283 float kBpsUpPerClient
= (float)thePrefs::GetSlotAllocation();
284 if (thePrefs::GetMaxUpload() == UNLIMITED
) {
285 float kBpsUp
= theStats::GetUploadRate() / 1024.0f
;
286 nMaxSlots
= (uint16
)(kBpsUp
/ kBpsUpPerClient
) + 2;
288 if (thePrefs::GetMaxUpload() >= 10) {
289 nMaxSlots
= (uint16
)floor((float)thePrefs::GetMaxUpload() / kBpsUpPerClient
+ 0.5);
290 // floor(x + 0.5) is a way of doing round(x) that works with gcc < 3 ...
291 if (nMaxSlots
< MIN_UP_CLIENTS_ALLOWED
) {
292 nMaxSlots
=MIN_UP_CLIENTS_ALLOWED
;
295 nMaxSlots
= MIN_UP_CLIENTS_ALLOWED
;
298 if (nMaxSlots
> MAX_UP_CLIENTS_ALLOWED
) {
299 nMaxSlots
= MAX_UP_CLIENTS_ALLOWED
;
305 CUploadQueue::~CUploadQueue()
307 wxASSERT(m_waitinglist
.empty());
308 wxASSERT(m_uploadinglist
.empty());
312 bool CUploadQueue::IsOnUploadQueue(const CUpDownClient
* client
) const
314 return std::find(m_waitinglist
.begin(), m_waitinglist
.end(), client
)
315 != m_waitinglist
.end();
319 bool CUploadQueue::IsDownloading(CUpDownClient
* client
) const
321 return std::find(m_uploadinglist
.begin(), m_uploadinglist
.end(), client
)
322 != m_uploadinglist
.end();
326 CUpDownClient
* CUploadQueue::GetWaitingClientByIP_UDP(uint32 dwIP
, uint16 nUDPPort
, bool bIgnorePortOnUniqueIP
, bool* pbMultipleIPs
)
328 CUpDownClient
* pMatchingIPClient
= NULL
;
332 CClientPtrList::iterator it
= m_waitinglist
.begin();
333 for (; it
!= m_waitinglist
.end(); ++it
) {
334 CUpDownClient
* cur_client
= *it
;
336 if ((dwIP
== cur_client
->GetIP()) && (nUDPPort
== cur_client
->GetUDPPort())) {
338 } else if ((dwIP
== cur_client
->GetIP()) && bIgnorePortOnUniqueIP
) {
339 pMatchingIPClient
= cur_client
;
345 *pbMultipleIPs
= cMatches
> 1;
348 if (pMatchingIPClient
&& cMatches
== 1) {
349 return pMatchingIPClient
;
356 void CUploadQueue::AddClientToQueue(CUpDownClient
* client
)
358 if (theApp
->serverconnect
->IsConnected() && theApp
->serverconnect
->IsLowID() && !theApp
->serverconnect
->IsLocalServer(client
->GetServerIP(),client
->GetServerPort()) && client
->GetDownloadState() == DS_NONE
&& !client
->IsFriend() && theStats::GetWaitingUserCount() > 50) {
359 // Well, all that issues finish in the same: don't allow to add to the queue
363 if ( client
->IsBanned() ) {
367 client
->AddAskedCount();
368 client
->SetLastUpRequest();
370 // Find all clients with the same user-hash
371 CClientList::SourceList found
= theApp
->clientlist
->GetClientsByHash( client
->GetUserHash() );
373 CClientList::SourceList::iterator it
= found
.begin();
374 while (it
!= found
.end()) {
375 CUpDownClient
* cur_client
= *it
++;
377 if ( IsOnUploadQueue( cur_client
) ) {
378 if ( cur_client
== client
) {
379 // This is where LowID clients get their upload slot assigned.
380 // They can't be contacted if they reach top of the queue, so they are just marked for uploading.
381 // When they reconnect next time AddClientToQueue() is called, and they get their slot
382 // through the connection they initiated.
383 // Since at that time no slot is free they get assigned an extra slot,
384 // so then the number of slots exceeds the configured number by one.
385 // To prevent a further increase no more LowID clients get a slot, until
386 // - a HighID client has got one (which happens only after two clients
387 // have been kicked so a slot is free again)
388 // - or there is a free slot, which means there is no HighID client on queue
389 if (client
->m_bAddNextConnect
) {
390 uint16 maxSlots
= GetMaxSlots();
391 if (lastupslotHighID
) {
394 if (m_uploadinglist
.size() < maxSlots
) {
395 client
->m_bAddNextConnect
= false;
396 RemoveFromWaitingQueue(client
);
397 AddUpNextClient(client
);
398 lastupslotHighID
= false; // LowID alternate
403 client
->SendRankingInfo();
404 Notify_SharedCtrlRefreshClient(client
, AVAILABLE_SOURCE
);
407 // Hash-clash, remove unidentified clients (possibly both)
409 if ( !cur_client
->IsIdentified() ) {
410 // Cur_client isn't identifed, remove it
411 theApp
->clientlist
->AddTrackClient( cur_client
);
413 RemoveFromWaitingQueue( cur_client
);
414 if ( !cur_client
->GetSocket() ) {
415 if (cur_client
->Disconnected( wxT("AddClientToQueue - same userhash") ) ) {
416 cur_client
->Safe_Delete();
421 if ( !client
->IsIdentified() ) {
422 // New client isn't identified, remove it
423 theApp
->clientlist
->AddTrackClient( client
);
425 if ( !client
->GetSocket() ) {
426 if ( client
->Disconnected( wxT("AddClientToQueue - same userhash") ) ) {
427 client
->Safe_Delete();
437 // Count the number of clients with the same IP-address
438 found
= theApp
->clientlist
->GetClientsByIP( client
->GetIP() );
441 for ( it
= found
.begin(); it
!= found
.end(); it
++ ) {
442 if ( ( *it
== client
) || IsOnUploadQueue( *it
) ) {
447 // We do not accept more than 3 clients from the same IP
450 } else if ( theApp
->clientlist
->GetClientsFromIP(client
->GetIP()) >= 3 ) {
455 CKnownFile
* reqfile
= (CKnownFile
*) client
->GetUploadFile();
457 reqfile
->statistic
.AddRequest();
460 if (client
->IsDownloading()) {
461 // he's already downloading and wants probably only another file
462 CPacket
* packet
= new CPacket(OP_ACCEPTUPLOADREQ
, 0, OP_EDONKEYPROT
);
463 theStats::AddUpOverheadFileRequest(packet
->GetPacketSize());
464 AddDebugLogLineN( logLocalClient
, wxT("Local Client: OP_ACCEPTUPLOADREQ to ") + client
->GetFullIP() );
465 client
->SendPacket(packet
,true);
469 // TODO find better ways to cap the list
470 if (m_waitinglist
.size() >= (thePrefs::GetQueueSize())) {
474 uint32 tick
= GetTickCount();
475 client
->ClearWaitStartTime();
476 // if possible start upload right away
477 if (m_waitinglist
.empty() && tick
- m_nLastStartUpload
>= 1000 && m_uploadinglist
.size() < GetMaxSlots()) {
478 AddUpNextClient(client
);
479 m_nLastStartUpload
= tick
;
481 // add to waiting queue
482 m_waitinglist
.push_back(client
);
483 // and sort it to update queue ranks
484 SortGetBestClient(true);
485 theStats::AddWaitingClient();
486 client
->ClearAskedCount();
487 client
->SetUploadState(US_ONUPLOADQUEUE
);
488 client
->SendRankingInfo();
489 //Notify_QlistAddClient(client);
494 bool CUploadQueue::RemoveFromUploadQueue(CUpDownClient
* client
)
496 // Keep track of this client
497 theApp
->clientlist
->AddTrackClient(client
);
499 CClientPtrList::iterator it
= std::find(m_uploadinglist
.begin(),
500 m_uploadinglist
.end(), client
);
502 if (it
!= m_uploadinglist
.end()) {
503 m_uploadinglist
.erase(it
);
504 theStats::RemoveUploadingClient();
505 if( client
->GetTransferredUp() ) {
506 theStats::AddSuccessfulUpload();
507 theStats::AddUploadTime(client
->GetUpStartTimeDelay() / 1000);
509 theStats::AddFailedUpload();
511 client
->SetUploadState(US_NONE
);
512 client
->ClearUploadBlockRequests();
520 bool CUploadQueue::CheckForTimeOver(CUpDownClient
* client
)
522 // Don't kick anybody if there's no need to
523 if (!m_allowKicking
) {
526 // First, check if it is a VIP slot (friend or Release-Prio).
527 if (client
->GetFriendSlot()) {
528 return false; // never drop the friend
530 // Release-Prio and nobody on queue for it?
531 if (client
->GetUploadFile()->GetUpPriority() == PR_POWERSHARE
) {
532 // Keep it unless half of the UL slots are occupied with friends or Release uploads.
534 for (CClientPtrList::iterator it
= m_uploadinglist
.begin(); it
!= m_uploadinglist
.end(); ++it
) {
535 CUpDownClient
* cur_client
= *it
;
536 if (cur_client
->GetFriendSlot() || cur_client
->GetUploadFile()->GetUpPriority() == PR_POWERSHARE
) {
540 // allow if VIP uploads occupy at most half of the possible upload slots
541 if (vips
<= GetMaxSlots() / 2) {
544 // Otherwise normal rules apply.
548 // "Transfer full chunks": drop client after 10 MB upload, or after an hour.
549 // (so average UL speed should at least be 2.84 kB/s)
550 // We don't track what he is downloading, but if it's all from one chunk he gets it.
551 if (client
->GetUpStartTimeDelay() > 3600000 // time: 1h
552 || client
->GetSessionUp() > 10485760) { // data: 10MB
553 m_allowKicking
= false; // kick max one client per cycle
562 * This function removes a file indicated by filehash from suspended_uploads_list.
564 void CUploadQueue::ResumeUpload( const CMD4Hash
& filehash
)
566 suspendedUploadsSet
.erase(filehash
);
567 AddLogLineN(CFormat( _("Resuming uploads of file: %s" ) )
568 % filehash
.Encode() );
572 * This function stops upload of a file indicated by filehash.
574 * a) teminate == false:
575 * File is suspended while a download completes. Then it is resumed after completion,
576 * so it makes sense to keep the client. Such files are kept in suspendedUploadsSet.
577 * b) teminate == true:
578 * File is deleted. Then the client is not added to the waiting list.
579 * Waiting clients are swept out with next run of AddUpNextClient,
580 * because their file is not shared anymore.
582 uint16
CUploadQueue::SuspendUpload(const CMD4Hash
& filehash
, bool terminate
)
584 AddLogLineN(CFormat( _("Suspending upload of file: %s" ) )
585 % filehash
.Encode() );
588 //Append the filehash to the list.
590 suspendedUploadsSet
.insert(filehash
);
593 CClientPtrList::iterator it
= m_uploadinglist
.begin();
594 while (it
!= m_uploadinglist
.end()) {
595 CUpDownClient
*potential
= *it
++;
596 //check if the client is uploading the file we need to suspend
597 if (potential
->GetUploadFileID() == filehash
) {
598 // remove the unlucky client from the upload queue
599 RemoveFromUploadQueue(potential
);
600 // if suspend isn't permanent add it to the waiting queue
602 potential
->SetUploadState(US_NONE
);
604 m_waitinglist
.push_back(potential
);
605 theStats::AddWaitingClient();
606 potential
->SetUploadState(US_ONUPLOADQUEUE
);
607 potential
->SendRankingInfo();
608 Notify_SharedCtrlRefreshClient(potential
, AVAILABLE_SOURCE
);
614 Notify_ShowQueueCount(m_waitinglist
.size());
619 bool CUploadQueue::RemoveFromWaitingQueue(CUpDownClient
* client
)
621 CClientPtrList::iterator it
= m_waitinglist
.begin();
624 while (it
!= m_waitinglist
.end()) {
625 CClientPtrList::iterator it1
= it
++;
626 if (*it1
== client
) {
627 RemoveFromWaitingQueue(it1
);
628 Notify_ShowQueueCount(m_waitinglist
.size());
629 // update ranks of remaining queue
630 while (it
!= m_waitinglist
.end()) {
631 (*it
)->SetUploadQueueWaitingPosition(rank
++);
642 void CUploadQueue::RemoveFromWaitingQueue(CClientPtrList::iterator pos
)
644 CUpDownClient
* todelete
= *pos
;
645 m_waitinglist
.erase(pos
);
646 theStats::RemoveWaitingClient();
647 if( todelete
->IsBanned() ) {
650 //Notify_QlistRemoveClient(todelete);
651 todelete
->SetUploadState(US_NONE
);
652 todelete
->ClearScore();
653 todelete
->SetUploadQueueWaitingPosition(0);
656 // File_checked_for_headers