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         uint32_t bestscore = 0;
  77         uint32_t bestlowscore = 0;
  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                         uint32_t 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 void CUploadQueue::Process()
 188 {
 189         if (AcceptNewClient() && !m_waitinglist.empty()) {
 190                 m_nLastStartUpload = ::GetTickCount();
 191                 AddUpNextClient();
 192         }
 193
 194         // The loop that feeds the upload slots with data.
 195         CClientPtrList::iterator it = m_uploadinglist.begin();
 196         while (it != m_uploadinglist.end()) {
 197                 // Get the client. Note! Also updates pos as a side effect.
 198                 CUpDownClient* cur_client = *it++;
 199
 200                 // It seems chatting or friend slots can get stuck at times in upload.. This needs looked into..
 201                 if (!cur_client->GetSocket()) {
 202                         RemoveFromUploadQueue(cur_client, true);
 203                         if(cur_client->Disconnected(_T("CUploadQueue::Process"))){
 204                                 cur_client->Safe_Delete();
 205                         }
 206                 } else {
 207                         cur_client->SendBlockData();
 208                 }
 209         }
 210
 211         // Save used bandwidth for speed calculations
 212         uint64 sentBytes = theApp->uploadBandwidthThrottler->GetNumberOfSentBytesSinceLastCallAndReset();
 213         (void)theApp->uploadBandwidthThrottler->GetNumberOfSentBytesOverheadSinceLastCallAndReset();
 214
 215         // Update statistics
 216         if (sentBytes) {
 217                 theStats::AddSentBytes(sentBytes);
 218         }
 219 }
 220
 221
 222 bool CUploadQueue::AcceptNewClient()
 223 {
 224         // check if we can allow a new client to start downloading from us
 225         if (::GetTickCount() - m_nLastStartUpload < 1000 || m_uploadinglist.size() >= MAX_UP_CLIENTS_ALLOWED) {
 226                 return false;
 227         }
 228
 229         float kBpsUpPerClient = (float)thePrefs::GetSlotAllocation();
 230         float kBpsUp = theStats::GetUploadRate() / 1024.0f;
 231         if (thePrefs::GetMaxUpload() == UNLIMITED) {
 232                 if (m_uploadinglist.size() < ((uint32)((kBpsUp)/kBpsUpPerClient)+2)) {
 233                         return true;
 234                 }
 235         } else {
 236                 uint16 nMaxSlots = 0;
 237                 if (thePrefs::GetMaxUpload() >= 10) {
 238                         nMaxSlots = (uint16)floor((float)thePrefs::GetMaxUpload() / kBpsUpPerClient + 0.5);
 239                                 // floor(x + 0.5) is a way of doing round(x) that works with gcc < 3 ...
 240                         if (nMaxSlots < MIN_UP_CLIENTS_ALLOWED) {
 241                                 nMaxSlots=MIN_UP_CLIENTS_ALLOWED;
 242                         }
 243                 } else {
 244                         nMaxSlots = MIN_UP_CLIENTS_ALLOWED;
 245                 }
 246
 247                 if (m_uploadinglist.size() < nMaxSlots) {
 248                         return true;
 249                 }
 250         }
 251         return false;
 252 }
 253
 254
 255 CUploadQueue::~CUploadQueue()
 256 {
 257         wxASSERT(m_waitinglist.empty());
 258         wxASSERT(m_uploadinglist.empty());
 259 }
 260
 261
 262 bool CUploadQueue::IsOnUploadQueue(const CUpDownClient* client) const
 263 {
 264         return std::find(m_waitinglist.begin(), m_waitinglist.end(), client)
 265                 != m_waitinglist.end();
 266 }
 267
 268
 269 bool CUploadQueue::IsDownloading(CUpDownClient* client) const
 270 {
 271         return std::find(m_uploadinglist.begin(), m_uploadinglist.end(), client)
 272                 != m_uploadinglist.end();
 273 }
 274
 275
 276 CUpDownClient* CUploadQueue::GetWaitingClientByIP_UDP(uint32 dwIP, uint16 nUDPPort, bool bIgnorePortOnUniqueIP, bool* pbMultipleIPs)
 277 {
 278         CUpDownClient* pMatchingIPClient = NULL;
 279
 280         int cMatches = 0;
 281
 282         CClientPtrList::iterator it = m_waitinglist.begin();
 283         for (; it != m_waitinglist.end(); ++it) {
 284                 CUpDownClient* cur_client = *it;
 285
 286                 if ((dwIP == cur_client->GetIP()) && (nUDPPort == cur_client->GetUDPPort())) {
 287                         return cur_client;
 288                 } else if ((dwIP == cur_client->GetIP()) && bIgnorePortOnUniqueIP) {
 289                         pMatchingIPClient = cur_client;
 290                         cMatches++;
 291                 }
 292         }
 293
 294         if (pbMultipleIPs) {
 295                 *pbMultipleIPs = cMatches > 1;
 296         }
 297
 298         if (pMatchingIPClient && cMatches == 1) {
 299                 return pMatchingIPClient;
 300         } else {
 301                 return NULL;
 302         }
 303 }
 304
 305
 306 void CUploadQueue::AddClientToQueue(CUpDownClient* client)
 307 {
 308         if (theApp->serverconnect->IsConnected() && theApp->serverconnect->IsLowID() && !theApp->serverconnect->IsLocalServer(client->GetServerIP(),client->GetServerPort()) && client->GetDownloadState() == DS_NONE && !client->IsFriend() && theStats::GetWaitingUserCount() > 50) {
 309                 // Well, all that issues finish in the same: don't allow to add to the queue
 310                 return;
 311         }
 312
 313         if ( client->IsBanned() ) {
 314                 return;
 315         }
 316
 317         client->AddAskedCount();
 318         client->SetLastUpRequest();
 319
 320         // Find all clients with the same user-hash
 321         CClientList::SourceList found = theApp->clientlist->GetClientsByHash( client->GetUserHash() );
 322
 323         CClientList::SourceList::iterator it = found.begin();
 324         while (it != found.end()) {
 325                 CUpDownClient* cur_client = *it++;
 326
 327                 if ( IsOnUploadQueue( cur_client ) ) {
 328                         if ( cur_client == client ) {
 329                                 if ( client->m_bAddNextConnect && ( ( m_uploadinglist.size() < thePrefs::GetMaxUpload() ) || ( thePrefs::GetMaxUpload() == UNLIMITED ) ) ) {
 330                                         if (lastupslotHighID) {
 331                                                 client->m_bAddNextConnect = false;
 332                                                 RemoveFromWaitingQueue(client, true);
 333                                                 AddUpNextClient(client);
 334                                                 lastupslotHighID = false; // LowID alternate
 335                                                 return;
 336                                         }
 337                                 }
 338
 339                                 client->SendRankingInfo();
 340                                 Notify_QlistRefreshClient(client);
 341                                 return;
 342
 343                         } else {
 344                                 // Hash-clash, remove unidentified clients (possibly both)
 345
 346                                 if ( !cur_client->IsIdentified() ) {
 347                                         // Cur_client isn't identifed, remove it
 348                                         theApp->clientlist->AddTrackClient( cur_client );
 349
 350                                         RemoveFromWaitingQueue( cur_client );
 351                                         if ( !cur_client->GetSocket() ) {
 352                                                 if (cur_client->Disconnected( wxT("AddClientToQueue - same userhash") ) ) {
 353                                                         cur_client->Safe_Delete();
 354                                                 }
 355                                         }
 356                                 }
 357
 358                                 if ( !client->IsIdentified() ) {
 359                                         // New client isn't identified, remove it
 360                                         theApp->clientlist->AddTrackClient( client );
 361
 362                                         if ( !client->GetSocket() ) {
 363                                                 if ( client->Disconnected( wxT("AddClientToQueue - same userhash") ) ) {
 364                                                         client->Safe_Delete();
 365                                                 }
 366                                         }
 367
 368                                         return;
 369                                 }
 370                         }
 371                 }
 372         }
 373
 374         // Count the number of clients with the same IP-address
 375         found = theApp->clientlist->GetClientsByIP( client->GetIP() );
 376
 377         int ipCount = 0;
 378         for ( it = found.begin(); it != found.end(); it++ ) {
 379                 if ( ( *it == client ) || IsOnUploadQueue( *it ) ) {
 380                         ipCount++;
 381                 }
 382         }
 383
 384         // We do not accept more than 3 clients from the same IP
 385         if ( ipCount > 3 ) {
 386                 return;
 387         } else if ( theApp->clientlist->GetClientsFromIP(client->GetIP()) >= 3 ) {
 388                 return;
 389         }
 390
 391         // statistic values
 392         CKnownFile* reqfile = (CKnownFile*) client->GetUploadFile();
 393         if (reqfile) {
 394                 reqfile->statistic.AddRequest();
 395         }
 396
 397         // TODO find better ways to cap the list
 398         if (m_waitinglist.size() >= (thePrefs::GetQueueSize())) {
 399                 return;
 400         }
 401
 402         if (client->IsDownloading()) {
 403                 // he's already downloading and wants probably only another file
 404                 CPacket* packet = new CPacket(OP_ACCEPTUPLOADREQ, 0, OP_EDONKEYPROT);
 405                 theStats::AddUpOverheadFileRequest(packet->GetPacketSize());
 406                 AddDebugLogLineM( false, logLocalClient, wxT("Local Client: OP_ACCEPTUPLOADREQ to ") + client->GetFullIP() );
 407                 client->SendPacket(packet,true);
 408                 return;
 409         }
 410
 411         if (m_waitinglist.empty() && AcceptNewClient()) {
 412                 AddUpNextClient(client);
 413                 m_nLastStartUpload = ::GetTickCount();
 414         } else {
 415                 m_waitinglist.push_back(client);
 416                 theStats::AddWaitingClient();
 417                 client->ClearWaitStartTime();
 418                 client->ClearAskedCount();
 419                 client->SetUploadState(US_ONUPLOADQUEUE);
 420                 client->SendRankingInfo();
 421                 Notify_QlistAddClient(client);
 422                 Notify_ShowQueueCount(m_waitinglist.size());
 423         }
 424 }
 425
 426
 427 bool CUploadQueue::RemoveFromUploadQueue(CUpDownClient* client, bool updatewindow)
 428 {
 429         // Keep track of this client
 430         theApp->clientlist->AddTrackClient(client);
 431
 432         CClientPtrList::iterator it = std::find(m_uploadinglist.begin(),
 433                         m_uploadinglist.end(), client);
 434
 435         if (it != m_uploadinglist.end()) {
 436                 if (updatewindow) {
 437                         Notify_UploadCtrlRemoveClient(client);
 438                 }
 439                 m_uploadinglist.erase(it);
 440                 theStats::RemoveUploadingClient();
 441                 if( client->GetTransferredUp() ) {
 442                         theStats::AddSuccessfulUpload();
 443                         theStats::AddUploadTime(client->GetUpStartTimeDelay() / 1000);
 444                 } else {
 445                         theStats::AddFailedUpload();
 446                 }
 447                 client->SetUploadState(US_NONE);
 448                 client->ClearUploadBlockRequests();
 449                 return true;
 450         }
 451
 452         return false;
 453 }
 454
 455
 456 bool CUploadQueue::CheckForTimeOver(CUpDownClient* client)
 457 {
 458         if (thePrefs::TransferFullChunks()) {
 459                 if( client->GetUpStartTimeDelay() > 3600000 ) { // Try to keep the clients from downloading forever.
 460                         return true;
 461                 }
 462                 // For some reason, some clients can continue to download after a chunk size.
 463                 // Are they redownloading the same chunk over and over????
 464                 if( client->GetSessionUp() > 10485760 ) {
 465                         return true;
 466                 }
 467         } else {
 468                 CClientPtrList::iterator it = m_waitinglist.begin();
 469                 for (; it != m_waitinglist.end(); ++it ) {
 470                         if (client->GetScore(true,true) < (*it)->GetScore(true,false)) {
 471                                 return true;
 472                         }
 473                 }
 474         }
 475
 476         return false;
 477 }
 478
 479
 480 uint16 CUploadQueue::GetWaitingPosition(const CUpDownClient *client) const
 481 {
 482         if ( !IsOnUploadQueue(client) ) {
 483                 return 0;
 484         }
 485
 486         uint16 rank = 1;
 487         const uint32 myscore = client->GetScore(false);
 488         CClientPtrList::const_iterator it = m_waitinglist.begin();
 489         for (; it != m_waitinglist.end(); ++it) {
 490                 if ((*it)->GetScore(false) > myscore) {
 491                         rank++;
 492                 }
 493         }
 494
 495         return rank;
 496 }
 497
 498
 499 /*
 500  * This function removes a file indicated by filehash from suspended_uploads_list.
 501  */
 502 void CUploadQueue::ResumeUpload( const CMD4Hash& filehash )
 503 {
 504         //Find the position of the filehash in the list and remove it.
 505         suspendlist::iterator it = std::find( suspended_uploads_list.begin(),
 506                                                       suspended_uploads_list.end(),
 507                                                       filehash );
 508         if ( it != suspended_uploads_list.end() )
 509                 suspended_uploads_list.erase( it );
 510
 511         AddLogLineM( false, CFormat( _("Resuming uploads of file: %s" ) )
 512                                 % filehash.Encode() );
 513 }
 514
 515 /*
 516  * This function adds a file indicated by filehash to suspended_uploads_list
 517  */
 518 void CUploadQueue::SuspendUpload( const CMD4Hash& filehash )
 519 {
 520         AddLogLineM( false, CFormat( _("Suspending upload of file: %s" ) )
 521                                 % filehash.Encode() );
 522
 523         //Append the filehash to the list.
 524         suspended_uploads_list.push_back(filehash);
 525         wxString base16hash = filehash.Encode();
 526
 527         CClientPtrList::iterator it = m_uploadinglist.begin();
 528         while (it != m_uploadinglist.end()) {
 529                 CUpDownClient *potential = *it++;
 530                 //check if the client is uploading the file we need to suspend
 531                 if(potential->GetUploadFileID() == filehash) {
 532                         //remove the unlucky client from the upload queue and add to the waiting queue
 533                         RemoveFromUploadQueue(potential, true);
 534
 535                         m_waitinglist.push_back(potential);
 536                         theStats::AddWaitingClient();
 537                         potential->SetUploadState(US_ONUPLOADQUEUE);
 538                         potential->SendRankingInfo();
 539                         Notify_QlistRefreshClient(potential);
 540                         Notify_ShowQueueCount(m_waitinglist.size());
 541                 }
 542         }
 543 }
 544
 545 bool CUploadQueue::RemoveFromWaitingQueue(CUpDownClient* client, bool updatewindow)
 546 {
 547         CClientPtrList::iterator it = std::find(m_waitinglist.begin(),
 548                         m_waitinglist.end(), client);
 549
 550         if (it != m_waitinglist.end()) {
 551                 RemoveFromWaitingQueue(it);
 552                 if (updatewindow) {
 553                         Notify_ShowQueueCount(m_waitinglist.size());
 554                 }
 555                 return true;
 556         } else {
 557                 return false;
 558         }
 559 }
 560
 561
 562 void CUploadQueue::RemoveFromWaitingQueue(CClientPtrList::iterator pos)
 563 {
 564         CUpDownClient* todelete = *pos;
 565         m_waitinglist.erase(pos);
 566         theStats::RemoveWaitingClient();
 567         if( todelete->IsBanned() ) {
 568                 todelete->UnBan();
 569         }
 570         Notify_QlistRemoveClient(todelete);
 571         todelete->SetUploadState(US_NONE);
 572 }
 573
 574 // File_checked_for_headers