src/UploadQueue.cpp

   1 //
   2 // This file is part of the aMule Project.
   3 //
   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 )
   6 //
   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
   9 // respective authors.
  10 //
  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.
  15 //
  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.
  20 //
  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
  24 //
  25
  26 #include "UploadQueue.h"        // Interface declarations
  27
  28 #include <protocol/Protocols.h>
  29 #include <protocol/ed2k/Client2Client/TCP.h>
  30 #include <common/Macros.h>
  31 #include <common/Constants.h>
  32
  33 #include <cmath>
  34
  35 #include "Types.h"              // Do_not_auto_remove (win32)
  36
  37 #ifdef __WXMSW__
  38         #include <winsock.h>    // Do_not_auto_remove
  39 #else
  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
  43 #endif
  44
  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
  55 #include "Logger.h"
  56 #include <common/Format.h>
  57 #include "UploadBandwidthThrottler.h"
  58 #include "GuiEvents.h"          // Needed for Notify_*
  59
  60
  61 //TODO rewrite the whole networkcode, use overlapped sockets
  62
  63 CUploadQueue::CUploadQueue()
  64 {
  65         m_nLastStartUpload = 0;
  66
  67         lastupslotHighID = true;
  68 }
  69
  70
  71 void CUploadQueue::AddUpNextClient(CUpDownClient* directadd)
  72 {
  73         CClientPtrList::iterator toadd = m_waitinglist.end();
  74         CClientPtrList::iterator toaddlow = m_waitinglist.end();
  75
  76         sint64  bestscore = -1;
  77         sint64  bestlowscore = -1;
  78
  79         CUpDownClient* newclient;
  80         // select next client or use given client
  81         if (!directadd) {
  82                 // Track if we purged any clients from the queue, as to only send one notify in total
  83                 bool purged = false;
  84
  85                 CClientPtrList::iterator it = m_waitinglist.begin();
  86                 for (; it != m_waitinglist.end(); ) {
  87                         CClientPtrList::iterator tmp_it = it++;
  88                         CUpDownClient* cur_client = *tmp_it;
  89
  90                         // clear dead clients
  91                         if ( (::GetTickCount() - cur_client->GetLastUpRequest() > MAX_PURGEQUEUETIME) || !theApp->sharedfiles->GetFileByID(cur_client->GetUploadFileID()) ) {
  92                                 purged = true;
  93                                 cur_client->ClearWaitStartTime();
  94                                 RemoveFromWaitingQueue(tmp_it);
  95                                 if (!cur_client->GetSocket()) {
  96                                         if(cur_client->Disconnected(wxT("AddUpNextClient - purged"))) {
  97                                                 cur_client->Safe_Delete();
  98                                                 cur_client = NULL;
  99                                         }
 100                                 }
 101                                 continue;
 102                         }
 103
 104                         suspendlist::iterator it2 = std::find( suspended_uploads_list.begin(),
 105                                                               suspended_uploads_list.end(),
 106                                                               cur_client->GetUploadFileID() );
 107                         if (cur_client->IsBanned() || it2 != suspended_uploads_list.end() ) { // Banned client or suspended upload ?
 108                                 continue;
 109                         }
 110                         // finished clearing
 111
 112                         sint64 cur_score = cur_client->GetScore(true);
 113                         if (cur_score > bestscore) {
 114                                 bestscore = cur_score;
 115                                 toadd = tmp_it;
 116                         } else {
 117                                 cur_score = cur_client->GetScore(false);
 118                                 if ((cur_score > bestlowscore) && !cur_client->m_bAddNextConnect){
 119                                         bestlowscore = cur_score;
 120                                         toaddlow = tmp_it;
 121                                 }
 122                         }
 123                 }
 124
 125                 // Update the count on GUI if any clients were purged
 126                 if (purged) {
 127                         Notify_ShowQueueCount(m_waitinglist.size());
 128                 }
 129
 130                 if (bestlowscore > bestscore){
 131                         newclient = *toaddlow;
 132                         newclient->m_bAddNextConnect = true;
 133                 }
 134
 135                 if (toadd == m_waitinglist.end()) {
 136                         return;
 137                 }
 138
 139                 newclient = *toadd;
 140                 lastupslotHighID = true; // VQB LowID alternate
 141                 RemoveFromWaitingQueue(toadd);
 142                 Notify_ShowQueueCount(m_waitinglist.size());
 143         } else {
 144                 //prevent another potential access of a suspended upload
 145
 146                 suspendlist::iterator it = std::find( suspended_uploads_list.begin(),
 147                                                       suspended_uploads_list.end(),
 148                                                       directadd->GetUploadFileID() );
 149                 if ( it != suspended_uploads_list.end() ) {
 150                         return;
 151                 } else {
 152                         newclient = directadd;
 153                 }
 154         }
 155
 156         if (IsDownloading(newclient)) {
 157                 return;
 158         }
 159         // tell the client that we are now ready to upload
 160         if (!newclient->IsConnected()) {
 161                 newclient->SetUploadState(US_CONNECTING);
 162                 if (!newclient->TryToConnect(true)) {
 163                         return;
 164                 }
 165         } else {
 166                 CPacket* packet = new CPacket(OP_ACCEPTUPLOADREQ, 0, OP_EDONKEYPROT);
 167                 theStats::AddUpOverheadFileRequest(packet->GetPacketSize());
 168                 AddDebugLogLineM( false, logLocalClient, wxT("Local Client: OP_ACCEPTUPLOADREQ to ") + newclient->GetFullIP() );
 169                 newclient->SendPacket(packet,true);
 170                 newclient->SetUploadState(US_UPLOADING);
 171         }
 172         newclient->SetUpStartTime();
 173         newclient->ResetSessionUp();
 174
 175         theApp->uploadBandwidthThrottler->AddToStandardList(m_uploadinglist.size(), newclient->GetSocket());
 176         m_uploadinglist.push_back(newclient);
 177         theStats::AddUploadingClient();
 178
 179         // Statistic
 180         CKnownFile* reqfile = (CKnownFile*) newclient->GetUploadFile();
 181         if (reqfile) {
 182                 reqfile->statistic.AddAccepted();
 183         }
 184         Notify_UploadCtrlAddClient(newclient);
 185
 186 }
 187
 188 void CUploadQueue::Process()
 189 {
 190         if (AcceptNewClient() && !m_waitinglist.empty()) {
 191                 m_nLastStartUpload = ::GetTickCount();
 192                 AddUpNextClient();
 193         }
 194
 195         // The loop that feeds the upload slots with data.
 196         CClientPtrList::iterator it = m_uploadinglist.begin();
 197         while (it != m_uploadinglist.end()) {
 198                 // Get the client. Note! Also updates pos as a side effect.
 199                 CUpDownClient* cur_client = *it++;
 200
 201                 // It seems chatting or friend slots can get stuck at times in upload.. This needs looked into..
 202                 if (!cur_client->GetSocket()) {
 203                         RemoveFromUploadQueue(cur_client, true);
 204                         if(cur_client->Disconnected(_T("CUploadQueue::Process"))){
 205                                 cur_client->Safe_Delete();
 206                         }
 207                 } else {
 208                         cur_client->SendBlockData();
 209                 }
 210         }
 211
 212         // Save used bandwidth for speed calculations
 213         uint64 sentBytes = theApp->uploadBandwidthThrottler->GetNumberOfSentBytesSinceLastCallAndReset();
 214         (void)theApp->uploadBandwidthThrottler->GetNumberOfSentBytesOverheadSinceLastCallAndReset();
 215
 216         // Update statistics
 217         if (sentBytes) {
 218                 theStats::AddSentBytes(sentBytes);
 219         }
 220 }
 221
 222
 223 bool CUploadQueue::AcceptNewClient()
 224 {
 225         // check if we can allow a new client to start downloading from us
 226         if (::GetTickCount() - m_nLastStartUpload < 1000 || m_uploadinglist.size() >= MAX_UP_CLIENTS_ALLOWED) {
 227                 return false;
 228         }
 229
 230         float kBpsUpPerClient = (float)thePrefs::GetSlotAllocation();
 231         float kBpsUp = theStats::GetUploadRate() / 1024.0f;
 232         if (thePrefs::GetMaxUpload() == UNLIMITED) {
 233                 if (m_uploadinglist.size() < ((uint32)((kBpsUp)/kBpsUpPerClient)+2)) {
 234                         return true;
 235                 }
 236         } else {
 237                 uint16 nMaxSlots = 0;
 238                 if (thePrefs::GetMaxUpload() >= 10) {
 239                         nMaxSlots = (uint16)floor((float)thePrefs::GetMaxUpload() / kBpsUpPerClient + 0.5);
 240                                 // floor(x + 0.5) is a way of doing round(x) that works with gcc < 3 ...
 241                         if (nMaxSlots < MIN_UP_CLIENTS_ALLOWED) {
 242                                 nMaxSlots=MIN_UP_CLIENTS_ALLOWED;
 243                         }
 244                 } else {
 245                         nMaxSlots = MIN_UP_CLIENTS_ALLOWED;
 246                 }
 247
 248                 if (m_uploadinglist.size() < nMaxSlots) {
 249                         return true;
 250                 }
 251         }
 252         return false;
 253 }
 254
 255
 256 CUploadQueue::~CUploadQueue()
 257 {
 258         wxASSERT(m_waitinglist.empty());
 259         wxASSERT(m_uploadinglist.empty());
 260 }
 261
 262
 263 bool CUploadQueue::IsOnUploadQueue(const CUpDownClient* client) const
 264 {
 265         return std::find(m_waitinglist.begin(), m_waitinglist.end(), client)
 266                 != m_waitinglist.end();
 267 }
 268
 269
 270 bool CUploadQueue::IsDownloading(CUpDownClient* client) const
 271 {
 272         return std::find(m_uploadinglist.begin(), m_uploadinglist.end(), client)
 273                 != m_uploadinglist.end();
 274 }
 275
 276
 277 CUpDownClient* CUploadQueue::GetWaitingClientByIP_UDP(uint32 dwIP, uint16 nUDPPort, bool bIgnorePortOnUniqueIP, bool* pbMultipleIPs)
 278 {
 279         CUpDownClient* pMatchingIPClient = NULL;
 280
 281         int cMatches = 0;
 282
 283         CClientPtrList::iterator it = m_waitinglist.begin();
 284         for (; it != m_waitinglist.end(); ++it) {
 285                 CUpDownClient* cur_client = *it;
 286
 287                 if ((dwIP == cur_client->GetIP()) && (nUDPPort == cur_client->GetUDPPort())) {
 288                         return cur_client;
 289                 } else if ((dwIP == cur_client->GetIP()) && bIgnorePortOnUniqueIP) {
 290                         pMatchingIPClient = cur_client;
 291                         cMatches++;
 292                 }
 293         }
 294
 295         if (pbMultipleIPs) {
 296                 *pbMultipleIPs = cMatches > 1;
 297         }
 298
 299         if (pMatchingIPClient && cMatches == 1) {
 300                 return pMatchingIPClient;
 301         } else {
 302                 return NULL;
 303         }
 304 }
 305
 306
 307 void CUploadQueue::AddClientToQueue(CUpDownClient* client)
 308 {
 309         if (theApp->serverconnect->IsConnected() && theApp->serverconnect->IsLowID() && !theApp->serverconnect->IsLocalServer(client->GetServerIP(),client->GetServerPort()) && client->GetDownloadState() == DS_NONE && !client->IsFriend() && theStats::GetWaitingUserCount() > 50) {
 310                 // Well, all that issues finish in the same: don't allow to add to the queue
 311                 return;
 312         }
 313
 314         if ( client->IsBanned() ) {
 315                 return;
 316         }
 317
 318         client->AddAskedCount();
 319         client->SetLastUpRequest();
 320
 321         // Find all clients with the same user-hash
 322         CClientList::SourceList found = theApp->clientlist->GetClientsByHash( client->GetUserHash() );
 323
 324         CClientList::SourceList::iterator it = found.begin();
 325         while (it != found.end()) {
 326                 CUpDownClient* cur_client = *it++;
 327
 328                 if ( IsOnUploadQueue( cur_client ) ) {
 329                         if ( cur_client == client ) {
 330                                 if ( client->m_bAddNextConnect && ( ( m_uploadinglist.size() < thePrefs::GetMaxUpload() ) || ( thePrefs::GetMaxUpload() == UNLIMITED ) ) ) {
 331                                         if (lastupslotHighID) {
 332                                                 client->m_bAddNextConnect = false;
 333                                                 RemoveFromWaitingQueue(client, true);
 334                                                 AddUpNextClient(client);
 335                                                 lastupslotHighID = false; // LowID alternate
 336                                                 return;
 337                                         }
 338                                 }
 339
 340                                 client->SendRankingInfo();
 341                                 Notify_QlistRefreshClient(client);
 342                                 return;
 343
 344                         } else {
 345                                 // Hash-clash, remove unidentified clients (possibly both)
 346
 347                                 if ( !cur_client->IsIdentified() ) {
 348                                         // Cur_client isn't identifed, remove it
 349                                         theApp->clientlist->AddTrackClient( cur_client );
 350
 351                                         RemoveFromWaitingQueue( cur_client );
 352                                         if ( !cur_client->GetSocket() ) {
 353                                                 if (cur_client->Disconnected( wxT("AddClientToQueue - same userhash") ) ) {
 354                                                         cur_client->Safe_Delete();
 355                                                 }
 356                                         }
 357                                 }
 358
 359                                 if ( !client->IsIdentified() ) {
 360                                         // New client isn't identified, remove it
 361                                         theApp->clientlist->AddTrackClient( client );
 362
 363                                         if ( !client->GetSocket() ) {
 364                                                 if ( client->Disconnected( wxT("AddClientToQueue - same userhash") ) ) {
 365                                                         client->Safe_Delete();
 366                                                 }
 367                                         }
 368
 369                                         return;
 370                                 }
 371                         }
 372                 }
 373         }
 374
 375         // Count the number of clients with the same IP-address
 376         found = theApp->clientlist->GetClientsByIP( client->GetIP() );
 377
 378         int ipCount = 0;
 379         for ( it = found.begin(); it != found.end(); it++ ) {
 380                 if ( ( *it == client ) || IsOnUploadQueue( *it ) ) {
 381                         ipCount++;
 382                 }
 383         }
 384
 385         // We do not accept more than 3 clients from the same IP
 386         if ( ipCount > 3 ) {
 387                 return;
 388         } else if ( theApp->clientlist->GetClientsFromIP(client->GetIP()) >= 3 ) {
 389                 return;
 390         }
 391
 392         // statistic values
 393         CKnownFile* reqfile = (CKnownFile*) client->GetUploadFile();
 394         if (reqfile) {
 395                 reqfile->statistic.AddRequest();
 396         }
 397
 398         // TODO find better ways to cap the list
 399         if (m_waitinglist.size() >= (thePrefs::GetQueueSize())) {
 400                 return;
 401         }
 402
 403         if (client->IsDownloading()) {
 404                 // he's already downloading and wants probably only another file
 405                 CPacket* packet = new CPacket(OP_ACCEPTUPLOADREQ, 0, OP_EDONKEYPROT);
 406                 theStats::AddUpOverheadFileRequest(packet->GetPacketSize());
 407                 AddDebugLogLineM( false, logLocalClient, wxT("Local Client: OP_ACCEPTUPLOADREQ to ") + client->GetFullIP() );
 408                 client->SendPacket(packet,true);
 409                 return;
 410         }
 411
 412         if (m_waitinglist.empty() && AcceptNewClient()) {
 413                 AddUpNextClient(client);
 414                 m_nLastStartUpload = ::GetTickCount();
 415         } else {
 416                 m_waitinglist.push_back(client);
 417                 theStats::AddWaitingClient();
 418                 client->ClearWaitStartTime();
 419                 client->ClearAskedCount();
 420                 client->SetUploadState(US_ONUPLOADQUEUE);
 421                 client->SendRankingInfo();
 422                 Notify_QlistAddClient(client);
 423                 Notify_ShowQueueCount(m_waitinglist.size());
 424         }
 425 }
 426
 427
 428 bool CUploadQueue::RemoveFromUploadQueue(CUpDownClient* client, bool updatewindow)
 429 {
 430         // Keep track of this client
 431         theApp->clientlist->AddTrackClient(client);
 432
 433         CClientPtrList::iterator it = std::find(m_uploadinglist.begin(),
 434                         m_uploadinglist.end(), client);
 435
 436         if (it != m_uploadinglist.end()) {
 437                 if (updatewindow) {
 438                         Notify_UploadCtrlRemoveClient(client);
 439                 }
 440                 m_uploadinglist.erase(it);
 441                 theStats::RemoveUploadingClient();
 442                 if( client->GetTransferredUp() ) {
 443                         theStats::AddSuccessfulUpload();
 444                         theStats::AddUploadTime(client->GetUpStartTimeDelay() / 1000);
 445                 } else {
 446                         theStats::AddFailedUpload();
 447                 }
 448                 client->SetUploadState(US_NONE);
 449                 client->ClearUploadBlockRequests();
 450                 return true;
 451         }
 452
 453         return false;
 454 }
 455
 456
 457 bool CUploadQueue::CheckForTimeOver(CUpDownClient* client)
 458 {
 459         if (thePrefs::TransferFullChunks()) {
 460                 if( client->GetUpStartTimeDelay() > 3600000 ) { // Try to keep the clients from downloading forever.
 461                         return true;
 462                 }
 463                 // For some reason, some clients can continue to download after a chunk size.
 464                 // Are they redownloading the same chunk over and over????
 465                 if( client->GetSessionUp() > 10485760 ) {
 466                         return true;
 467                 }
 468         } else {
 469                 CClientPtrList::iterator it = m_waitinglist.begin();
 470                 for (; it != m_waitinglist.end(); ++it ) {
 471                         if (client->GetScore(true,true) < (*it)->GetScore(true,false)) {
 472                                 return true;
 473                         }
 474                 }
 475         }
 476
 477         return false;
 478 }
 479
 480
 481 uint16 CUploadQueue::GetWaitingPosition(const CUpDownClient *client) const
 482 {
 483         if ( !IsOnUploadQueue(client) ) {
 484                 return 0;
 485         }
 486
 487         uint16 rank = 1;
 488         const uint32 myscore = client->GetScore(false);
 489         CClientPtrList::const_iterator it = m_waitinglist.begin();
 490         for (; it != m_waitinglist.end(); ++it) {
 491                 if ((*it)->GetScore(false) > myscore) {
 492                         rank++;
 493                 }
 494         }
 495
 496         return rank;
 497 }
 498
 499
 500 /*
 501  * This function removes a file indicated by filehash from suspended_uploads_list.
 502  */
 503 void CUploadQueue::ResumeUpload( const CMD4Hash& filehash )
 504 {
 505         //Find the position of the filehash in the list and remove it.
 506         suspendlist::iterator it = std::find( suspended_uploads_list.begin(),
 507                                                       suspended_uploads_list.end(),
 508                                                       filehash );
 509         if ( it != suspended_uploads_list.end() )
 510                 suspended_uploads_list.erase( it );
 511
 512         AddLogLineM( false, CFormat( _("Resuming uploads of file: %s" ) )
 513                                 % filehash.Encode() );
 514 }
 515
 516 /*
 517  * This function adds a file indicated by filehash to suspended_uploads_list
 518  */
 519 void CUploadQueue::SuspendUpload( const CMD4Hash& filehash )
 520 {
 521         AddLogLineM( false, CFormat( _("Suspending upload of file: %s" ) )
 522                                 % filehash.Encode() );
 523
 524         //Append the filehash to the list.
 525         suspended_uploads_list.push_back(filehash);
 526         wxString base16hash = filehash.Encode();
 527
 528         CClientPtrList::iterator it = m_uploadinglist.begin();
 529         while (it != m_uploadinglist.end()) {
 530                 CUpDownClient *potential = *it++;
 531                 //check if the client is uploading the file we need to suspend
 532                 if(potential->GetUploadFileID() == filehash) {
 533                         //remove the unlucky client from the upload queue and add to the waiting queue
 534                         RemoveFromUploadQueue(potential, true);
 535
 536                         m_waitinglist.push_back(potential);
 537                         theStats::AddWaitingClient();
 538                         potential->SetUploadState(US_ONUPLOADQUEUE);
 539                         potential->SendRankingInfo();
 540                         Notify_QlistRefreshClient(potential);
 541                         Notify_ShowQueueCount(m_waitinglist.size());
 542                 }
 543         }
 544 }
 545
 546 bool CUploadQueue::RemoveFromWaitingQueue(CUpDownClient* client, bool updatewindow)
 547 {
 548         CClientPtrList::iterator it = std::find(m_waitinglist.begin(),
 549                         m_waitinglist.end(), client);
 550
 551         if (it != m_waitinglist.end()) {
 552                 RemoveFromWaitingQueue(it);
 553                 if (updatewindow) {
 554                         Notify_ShowQueueCount(m_waitinglist.size());
 555                 }
 556                 return true;
 557         } else {
 558                 return false;
 559         }
 560 }
 561
 562
 563 void CUploadQueue::RemoveFromWaitingQueue(CClientPtrList::iterator pos)
 564 {
 565         CUpDownClient* todelete = *pos;
 566         m_waitinglist.erase(pos);
 567         theStats::RemoveWaitingClient();
 568         if( todelete->IsBanned() ) {
 569                 todelete->UnBan();
 570         }
 571         Notify_QlistRemoveClient(todelete);
 572         todelete->SetUploadState(US_NONE);
 573 }
 574
 575 // File_checked_for_headers