1 /****************************************************************************

     2 **

     3 ** Copyright (C) 2009 Nokia Corporation and/or its subsidiary(-ies).

     4 ** All rights reserved.

     5 ** Contact: Nokia Corporation (qt-info@nokia.com)

     6 **

     7 ** This file is part of the examples of the Qt Toolkit.

     8 **

     9 ** $QT_BEGIN_LICENSE:LGPL$

    10 ** No Commercial Usage

    11 ** This file contains pre-release code and may not be distributed.

    12 ** You may use this file in accordance with the terms and conditions

    13 ** contained in the Technology Preview License Agreement accompanying

    14 ** this package.

    15 **

    16 ** GNU Lesser General Public License Usage

    17 ** Alternatively, this file may be used under the terms of the GNU Lesser

    18 ** General Public License version 2.1 as published by the Free Software

    19 ** Foundation and appearing in the file LICENSE.LGPL included in the

    20 ** packaging of this file.  Please review the following information to

    21 ** ensure the GNU Lesser General Public License version 2.1 requirements

    22 ** will be met: http://www.gnu.org/licenses/old-licenses/lgpl-2.1.html.

    23 **

    24 ** In addition, as a special exception, Nokia gives you certain additional

    25 ** rights.  These rights are described in the Nokia Qt LGPL Exception

    26 ** version 1.1, included in the file LGPL_EXCEPTION.txt in this package.

    27 **

    28 ** If you have questions regarding the use of this file, please contact

    29 ** Nokia at qt-info@nokia.com.

    30 **

    31 **

    32 **

    33 **

    34 **

    35 **

    36 **

    37 **

    38 ** $QT_END_LICENSE$

    39 **

    40 ****************************************************************************/

    41 

    42 #include "connectionmanager.h"

    43 #include "filemanager.h"

    44 #include "metainfo.h"

    45 #include "torrentclient.h"

    46 #include "torrentserver.h"

    47 #include "trackerclient.h"

    48 #include "peerwireclient.h"

    49 #include "ratecontroller.h"

    50 

    51 #include <QtCore>

    52 #include <QNetworkInterface>

    53 

    54 // These constants could also be configurable by the user.

    55 static const int ServerMinPort = 6881;

    56 static const int ServerMaxPort = /* 6889 */ 7000;

    57 static const int BlockSize = 16384;

    58 static const int MaxBlocksInProgress = 5;

    59 static const int MaxBlocksInMultiMode = 2;

    60 static const int MaxConnectionPerPeer = 1;

    61 static const int RateControlWindowLength = 10;

    62 static const int RateControlTimerDelay = 1000;

    63 static const int MinimumTimeBeforeRevisit = 30;

    64 static const int MaxUploads = 4;

    65 static const int UploadScheduleInterval = 10000;

    66 static const int EndGamePieces = 5;

    67 

    68 class TorrentPiece {

    69 public:

    70     int index;

    71     int length;

    72     QBitArray completedBlocks;

    73     QBitArray requestedBlocks;

    74     bool inProgress;

    75 };

    76 

    77 class TorrentClientPrivate

    78 {

    79 public:

    80     TorrentClientPrivate(TorrentClient *qq);

    81 

    82     // State / error

    83     void setError(TorrentClient::Error error);

    84     void setState(TorrentClient::State state);

    85     TorrentClient::Error error;

    86     TorrentClient::State state;

    87     QString errorString;

    88     QString stateString;

    89 

    90     // Where to save data

    91     QString destinationFolder;

    92     MetaInfo metaInfo;

    93 

    94     // Announce tracker and file manager

    95     QByteArray peerId;

    96     QByteArray infoHash;

    97     TrackerClient trackerClient;

    98     FileManager fileManager;

    99 

   100     // Connections

   101     QList<PeerWireClient *> connections;

   102     QList<TorrentPeer *> peers;

   103     bool schedulerCalled;

   104     void callScheduler();

   105     bool connectingToClients;

   106     void callPeerConnector();

   107     int uploadScheduleTimer;

   108 

   109     // Pieces

   110     QMap<int, PeerWireClient *> readIds;

   111     QMultiMap<PeerWireClient *, TorrentPiece *> payloads;

   112     QMap<int, TorrentPiece *> pendingPieces;

   113     QBitArray completedPieces;

   114     QBitArray incompletePieces;

   115     int pieceCount;

   116 

   117     // Progress

   118     int lastProgressValue;

   119     qint64 downloadedBytes;

   120     qint64 uploadedBytes;

   121     int downloadRate[RateControlWindowLength];

   122     int uploadRate[RateControlWindowLength];

   123     int transferRateTimer;

   124 

   125     TorrentClient *q;

   126 };

   127 

   128 TorrentClientPrivate::TorrentClientPrivate(TorrentClient *qq)

   129     : trackerClient(qq), q(qq)

   130 {

   131     error = TorrentClient::UnknownError;

   132     state = TorrentClient::Idle;

   133     errorString = QT_TRANSLATE_NOOP(TorrentClient, "Unknown error");

   134     stateString = QT_TRANSLATE_NOOP(TorrentClient, "Idle");

   135     schedulerCalled = false;

   136     connectingToClients = false;

   137     uploadScheduleTimer = 0;

   138     lastProgressValue = -1;

   139     pieceCount = 0;

   140     downloadedBytes = 0;

   141     uploadedBytes = 0;

   142     memset(downloadRate, 0, sizeof(downloadRate));

   143     memset(uploadRate, 0, sizeof(uploadRate));

   144     transferRateTimer = 0;

   145 }

   146 

   147 void TorrentClientPrivate::setError(TorrentClient::Error errorCode)

   148 {

   149     this->error = errorCode;

   150     switch (error) {

   151     case TorrentClient::UnknownError:

   152         errorString = QT_TRANSLATE_NOOP(TorrentClient, "Unknown error");

   153         break;

   154     case TorrentClient::TorrentParseError:

   155         errorString = QT_TRANSLATE_NOOP(TorrentClient, "Invalid torrent data");

   156         break;

   157     case TorrentClient::InvalidTrackerError:

   158         errorString = QT_TRANSLATE_NOOP(TorrentClient, "Unable to connect to tracker");

   159         break;

   160     case TorrentClient::FileError:

   161         errorString = QT_TRANSLATE_NOOP(TorrentClient, "File error");

   162         break;

   163     case TorrentClient::ServerError:

   164         errorString = QT_TRANSLATE_NOOP(TorrentClient, "Unable to initialize server");

   165         break;

   166     }

   167     emit q->error(errorCode);

   168 }

   169 

   170 void TorrentClientPrivate::setState(TorrentClient::State state)

   171 {

   172     this->state = state;

   173     switch (state) {

   174     case TorrentClient::Idle:

   175         stateString = QT_TRANSLATE_NOOP(TorrentClient, "Idle");

   176         break;

   177     case TorrentClient::Paused:

   178         stateString = QT_TRANSLATE_NOOP(TorrentClient, "Paused");

   179         break;

   180     case TorrentClient::Stopping:

   181         stateString = QT_TRANSLATE_NOOP(TorrentClient, "Stopping");

   182         break;

   183     case TorrentClient::Preparing:

   184         stateString = QT_TRANSLATE_NOOP(TorrentClient, "Preparing");

   185         break;

   186     case TorrentClient::Searching:

   187         stateString = QT_TRANSLATE_NOOP(TorrentClient, "Searching");

   188         break;

   189     case TorrentClient::Connecting:

   190         stateString = QT_TRANSLATE_NOOP(TorrentClient, "Connecting");

   191         break;

   192     case TorrentClient::WarmingUp:

   193         stateString = QT_TRANSLATE_NOOP(TorrentClient, "Warming up");

   194         break;

   195     case TorrentClient::Downloading:

   196         stateString = QT_TRANSLATE_NOOP(TorrentClient, "Downloading");

   197         break;

   198     case TorrentClient::Endgame:

   199         stateString = QT_TRANSLATE_NOOP(TorrentClient, "Finishing");

   200         break;

   201     case TorrentClient::Seeding:

   202         stateString = QT_TRANSLATE_NOOP(TorrentClient, "Seeding");

   203         break;

   204     }

   205     emit q->stateChanged(state);

   206 }

   207 

   208 void TorrentClientPrivate::callScheduler()

   209 {

   210     if (!schedulerCalled) {

   211         schedulerCalled = true;

   212         QMetaObject::invokeMethod(q, "scheduleDownloads", Qt::QueuedConnection);

   213     }

   214 }

   215 

   216 void TorrentClientPrivate::callPeerConnector()

   217 {

   218     if (!connectingToClients) {

   219         connectingToClients = true;

   220         QTimer::singleShot(10000, q, SLOT(connectToPeers()));

   221     }

   222 }

   223 

   224 TorrentClient::TorrentClient(QObject *parent)

   225     : QObject(parent), d(new TorrentClientPrivate(this))

   226 {

   227     // Connect the file manager

   228     connect(&d->fileManager, SIGNAL(dataRead(int, int, int, const QByteArray &)),

   229             this, SLOT(sendToPeer(int, int, int, const QByteArray &)));

   230     connect(&d->fileManager, SIGNAL(verificationProgress(int)),

   231             this, SLOT(updateProgress(int)));

   232     connect(&d->fileManager, SIGNAL(verificationDone()),

   233             this, SLOT(fullVerificationDone()));

   234     connect(&d->fileManager, SIGNAL(pieceVerified(int, bool)),

   235             this, SLOT(pieceVerified(int, bool)));

   236     connect(&d->fileManager, SIGNAL(error()),

   237             this, SLOT(handleFileError()));

   238 

   239     // Connect the tracker client

   240     connect(&d->trackerClient, SIGNAL(peerListUpdated(const QList<TorrentPeer> &)),

   241             this, SLOT(addToPeerList(const QList<TorrentPeer> &)));

   242     connect(&d->trackerClient, SIGNAL(stopped()),

   243             this, SIGNAL(stopped()));

   244 }

   245 

   246 TorrentClient::~TorrentClient()

   247 {

   248     qDeleteAll(d->peers);

   249     qDeleteAll(d->pendingPieces);

   250     delete d;

   251 }

   252 

   253 bool TorrentClient::setTorrent(const QString &fileName)

   254 {

   255     QFile file(fileName);

   256     if (!file.open(QIODevice::ReadOnly) || !setTorrent(file.readAll())) {

   257         d->setError(TorrentParseError);

   258         return false;

   259     }

   260     return true;

   261 }

   262 

   263 bool TorrentClient::setTorrent(const QByteArray &torrentData)

   264 {

   265     if (!d->metaInfo.parse(torrentData)) {

   266         d->setError(TorrentParseError);

   267         return false;

   268     }

   269 

   270     // Calculate SHA1 hash of the "info" section in the torrent

   271     QByteArray infoValue = d->metaInfo.infoValue();

   272     d->infoHash = QCryptographicHash::hash(infoValue, QCryptographicHash::Sha1);

   273 

   274     return true;

   275 }

   276 

   277 MetaInfo TorrentClient::metaInfo() const

   278 {

   279     return d->metaInfo;

   280 }

   281 

   282 void TorrentClient::setDestinationFolder(const QString &directory)

   283 {

   284     d->destinationFolder = directory;

   285 }

   286 

   287 QString TorrentClient::destinationFolder() const

   288 {

   289     return d->destinationFolder;

   290 }

   291 

   292 void TorrentClient::setDumpedState(const QByteArray &dumpedState)

   293 {

   294     // Recover partially completed pieces

   295     QDataStream stream(dumpedState);

   296 

   297     quint16 version = 0;

   298     stream >> version;

   299     if (version != 2)

   300         return;

   301 

   302     stream >> d->completedPieces;

   303 

   304     while (!stream.atEnd()) {

   305         int index;

   306         int length;

   307         QBitArray completed;

   308         stream >> index >> length >> completed;

   309         if (stream.status() != QDataStream::Ok) {

   310             d->completedPieces.clear();

   311             break;

   312         }

   313 

   314         TorrentPiece *piece = new TorrentPiece;

   315         piece->index = index;

   316         piece->length = length;

   317         piece->completedBlocks = completed;

   318         piece->requestedBlocks.resize(completed.size());

   319         piece->inProgress = false;

   320         d->pendingPieces[index] = piece;

   321     }

   322 }

   323 

   324 QByteArray TorrentClient::dumpedState() const

   325 {

   326     QByteArray partials;

   327     QDataStream stream(&partials, QIODevice::WriteOnly);

   328 

   329     stream << quint16(2);

   330     stream << d->completedPieces;

   331 

   332     // Save the state of all partially downloaded pieces into a format

   333     // suitable for storing in settings.

   334     QMap<int, TorrentPiece *>::ConstIterator it = d->pendingPieces.constBegin();

   335     while (it != d->pendingPieces.constEnd()) {

   336         TorrentPiece *piece = it.value();

   337         if (blocksLeftForPiece(piece) > 0 && blocksLeftForPiece(piece) < piece->completedBlocks.size()) {

   338             stream << piece->index;

   339             stream << piece->length;

   340             stream << piece->completedBlocks;

   341         }

   342         ++it;

   343     }

   344 

   345     return partials;

   346 }

   347 

   348 qint64 TorrentClient::progress() const

   349 {

   350     return d->lastProgressValue;

   351 }

   352 

   353 void TorrentClient::setDownloadedBytes(qint64 bytes)

   354 {

   355     d->downloadedBytes = bytes;

   356 }

   357 

   358 qint64 TorrentClient::downloadedBytes() const

   359 {

   360     return d->downloadedBytes;

   361 }

   362 

   363 void TorrentClient::setUploadedBytes(qint64 bytes)

   364 {

   365     d->uploadedBytes = bytes;

   366 }

   367 

   368 qint64 TorrentClient::uploadedBytes() const

   369 {

   370     return d->uploadedBytes;

   371 }

   372 

   373 int TorrentClient::connectedPeerCount() const

   374 {

   375     int tmp = 0;

   376     foreach (PeerWireClient *client, d->connections) {

   377         if (client->state() == QAbstractSocket::ConnectedState)

   378             ++tmp;

   379     }

   380     return tmp;

   381 }

   382 

   383 int TorrentClient::seedCount() const

   384 {

   385     int tmp = 0;

   386     foreach (PeerWireClient *client, d->connections) {

   387         if (client->availablePieces().count(true) == d->pieceCount)

   388             ++tmp;

   389     }

   390     return tmp;

   391 }

   392 

   393 TorrentClient::State TorrentClient::state() const

   394 {

   395     return d->state;

   396 }

   397 

   398 QString TorrentClient::stateString() const

   399 {

   400     return d->stateString;

   401 }

   402 

   403 TorrentClient::Error TorrentClient::error() const

   404 {

   405     return d->error;

   406 }

   407 

   408 QString TorrentClient::errorString() const

   409 {

   410     return d->errorString;

   411 }

   412 

   413 QByteArray TorrentClient::peerId() const

   414 {

   415     return d->peerId;

   416 }

   417 

   418 QByteArray TorrentClient::infoHash() const

   419 {

   420     return d->infoHash;

   421 }

   422 

   423 void TorrentClient::start()

   424 {

   425     if (d->state != Idle)

   426         return;

   427 

   428     TorrentServer::instance()->addClient(this);

   429 

   430     // Initialize the file manager

   431     d->setState(Preparing);

   432     d->fileManager.setMetaInfo(d->metaInfo);

   433     d->fileManager.setDestinationFolder(d->destinationFolder);

   434     d->fileManager.setCompletedPieces(d->completedPieces);

   435     d->fileManager.start(QThread::LowestPriority);

   436     d->fileManager.startDataVerification();

   437 }

   438 

   439 void TorrentClient::stop()

   440 {

   441     if (d->state == Stopping)

   442         return;

   443 

   444     TorrentServer::instance()->removeClient(this);

   445 

   446     // Update the state

   447     State oldState = d->state;

   448     d->setState(Stopping);

   449 

   450     // Stop the timer

   451     if (d->transferRateTimer) {

   452         killTimer(d->transferRateTimer);

   453         d->transferRateTimer = 0;

   454     }

   455 

   456     // Abort all existing connections

   457     foreach (PeerWireClient *client, d->connections) {

   458         RateController::instance()->removeSocket(client);

   459         ConnectionManager::instance()->removeConnection(client);

   460         client->abort();

   461     }

   462     d->connections.clear();

   463 

   464     // Perhaps stop the tracker

   465     if (oldState > Preparing) {

   466         d->trackerClient.stop();

   467     } else {

   468         d->setState(Idle);

   469         emit stopped();

   470     }

   471 }

   472 

   473 void TorrentClient::setPaused(bool paused)

   474 {

   475     if (paused) {

   476         // Abort all connections, and set the max number of

   477         // connections to 0. Keep the list of peers, so we can quickly

   478         // resume later.

   479         d->setState(Paused);

   480         foreach (PeerWireClient *client, d->connections)

   481             client->abort();

   482         d->connections.clear();

   483         TorrentServer::instance()->removeClient(this);

   484     } else {

   485         // Restore the max number of connections, and start the peer

   486         // connector. We should also quickly start receiving incoming

   487         // connections.

   488         d->setState(d->completedPieces.count(true) == d->fileManager.pieceCount()

   489                     ? Seeding : Searching);

   490         connectToPeers();

   491         TorrentServer::instance()->addClient(this);

   492     }

   493 }

   494 

   495 void TorrentClient::timerEvent(QTimerEvent *event)

   496 {

   497     if (event->timerId() == d->uploadScheduleTimer) {

   498         // Update the state of who's choked and who's not

   499         scheduleUploads();

   500         return;

   501     }

   502 

   503     if (event->timerId() != d->transferRateTimer) {

   504         QObject::timerEvent(event);

   505         return;

   506     }

   507 

   508     // Calculate average upload/download rate

   509     qint64 uploadBytesPerSecond = 0;

   510     qint64 downloadBytesPerSecond = 0;

   511     for (int i = 0; i < RateControlWindowLength; ++i) {

   512         uploadBytesPerSecond += d->uploadRate[i];

   513         downloadBytesPerSecond += d->downloadRate[i];

   514     }

   515     uploadBytesPerSecond /= qint64(RateControlWindowLength);

   516     downloadBytesPerSecond /= qint64(RateControlWindowLength);

   517     for (int i = RateControlWindowLength - 2; i >= 0; --i) {

   518         d->uploadRate[i + 1] = d->uploadRate[i];

   519         d->downloadRate[i + 1] = d->downloadRate[i];

   520     }

   521     d->uploadRate[0] = 0;

   522     d->downloadRate[0] = 0;

   523     emit uploadRateUpdated(int(uploadBytesPerSecond));

   524     emit downloadRateUpdated(int(downloadBytesPerSecond));

   525 

   526     // Stop the timer if there is no activity.

   527     if (downloadBytesPerSecond == 0 && uploadBytesPerSecond == 0) {

   528         killTimer(d->transferRateTimer);

   529         d->transferRateTimer = 0;

   530     }

   531 }

   532 

   533 void TorrentClient::sendToPeer(int readId, int pieceIndex, int begin, const QByteArray &data)

   534 {

   535     // Send the requested block to the peer if the client connection

   536     // still exists; otherwise do nothing. This slot is called by the

   537     // file manager after it has read a block of data.

   538     PeerWireClient *client = d->readIds.value(readId);

   539     if (client) {

   540         if ((client->peerWireState() & PeerWireClient::ChokingPeer) == 0)

   541             client->sendBlock(pieceIndex, begin, data);

   542     }

   543     d->readIds.remove(readId);

   544 }

   545 

   546 void TorrentClient::fullVerificationDone()

   547 {

   548     // Update our list of completed and incomplete pieces.

   549     d->completedPieces = d->fileManager.completedPieces();

   550     d->incompletePieces.resize(d->completedPieces.size());

   551     d->pieceCount = d->completedPieces.size();

   552     for (int i = 0; i < d->fileManager.pieceCount(); ++i) {

   553         if (!d->completedPieces.testBit(i))

   554             d->incompletePieces.setBit(i);

   555     }

   556 

   557     updateProgress();

   558 

   559     // If the checksums show that what the dumped state thought was

   560     // partial was in fact complete, then we trust the checksums.

   561     QMap<int, TorrentPiece *>::Iterator it = d->pendingPieces.begin();

   562     while (it != d->pendingPieces.end()) {

   563         if (d->completedPieces.testBit(it.key()))

   564             it = d->pendingPieces.erase(it);

   565         else

   566             ++it;

   567     }

   568 

   569     d->uploadScheduleTimer = startTimer(UploadScheduleInterval);

   570 

   571     // Start the server

   572     TorrentServer *server = TorrentServer::instance();

   573     if (!server->isListening()) {

   574         // Set up the peer wire server

   575         for (int i = ServerMinPort; i <= ServerMaxPort; ++i) {

   576             if (server->listen(QHostAddress::Any, i))

   577                 break;

   578         }

   579         if (!server->isListening()) {

   580             d->setError(ServerError);

   581             return;

   582         }

   583     }

   584 

   585     d->setState(d->completedPieces.count(true) == d->pieceCount ? Seeding : Searching);

   586 

   587     // Start the tracker client

   588     d->trackerClient.start(d->metaInfo);

   589 }

   590 

   591 void TorrentClient::pieceVerified(int pieceIndex, bool ok)

   592 {

   593     TorrentPiece *piece = d->pendingPieces.value(pieceIndex);

   594 

   595     // Remove this piece from all payloads

   596     QMultiMap<PeerWireClient *, TorrentPiece *>::Iterator it = d->payloads.begin();

   597     while (it != d->payloads.end()) {

   598         if (it.value()->index == pieceIndex)

   599             it = d->payloads.erase(it);

   600         else

   601             ++it;

   602     }

   603 

   604     if (!ok) {

   605         // If a piece did not pass the SHA1 check, we'll simply clear

   606         // its state, and the scheduler will re-request it

   607         piece->inProgress = false;

   608         piece->completedBlocks.fill(false);

   609         piece->requestedBlocks.fill(false);

   610         d->callScheduler();

   611         return;

   612     }

   613 

   614     // Update the peer list so we know who's still interesting.

   615     foreach (TorrentPeer *peer, d->peers) {

   616         if (!peer->interesting)

   617             continue;

   618         bool interesting = false;

   619         for (int i = 0; i < d->pieceCount; ++i) {

   620             if (peer->pieces.testBit(i) && d->incompletePieces.testBit(i)) {

   621                 interesting = true;

   622                 break;

   623             }

   624         }

   625         peer->interesting = interesting;

   626     }

   627 

   628     // Delete the piece and update our structures.

   629     delete piece;

   630     d->pendingPieces.remove(pieceIndex);

   631     d->completedPieces.setBit(pieceIndex);

   632     d->incompletePieces.clearBit(pieceIndex);

   633 

   634     // Notify connected peers.

   635     foreach (PeerWireClient *client, d->connections) {

   636         if (client->state() == QAbstractSocket::ConnectedState

   637             && !client->availablePieces().testBit(pieceIndex)) {

   638             client->sendPieceNotification(pieceIndex);

   639         }

   640     }

   641 

   642     // Notify the tracker if we've entered Seeding status; otherwise

   643     // call the scheduler.

   644     int completed = d->completedPieces.count(true);

   645     if (completed == d->pieceCount) {

   646         if (d->state != Seeding) {

   647             d->setState(Seeding);

   648             d->trackerClient.startSeeding();

   649         }

   650     } else {

   651         if (completed == 1)

   652             d->setState(Downloading);

   653         else if (d->incompletePieces.count(true) < 5 && d->pendingPieces.size() > d->incompletePieces.count(true))

   654             d->setState(Endgame);

   655         d->callScheduler();

   656     }

   657 

   658     updateProgress();

   659 }

   660 

   661 void TorrentClient::handleFileError()

   662 {

   663     if (d->state == Paused)

   664         return;

   665     setPaused(true);

   666     emit error(FileError);

   667 }

   668 

   669 void TorrentClient::connectToPeers()

   670 {

   671     d->connectingToClients = false;

   672 

   673     if (d->state == Stopping || d->state == Idle || d->state == Paused)

   674         return;

   675 

   676     if (d->state == Searching)

   677         d->setState(Connecting);

   678 

   679     // Find the list of peers we are not currently connected to, where

   680     // the more interesting peers are listed more than once.

   681     QList<TorrentPeer *> weighedPeers = weighedFreePeers();

   682 

   683     // Start as many connections as we can

   684     while (!weighedPeers.isEmpty() && ConnectionManager::instance()->canAddConnection()

   685            && (qrand() % (ConnectionManager::instance()->maxConnections() / 2))) {

   686         PeerWireClient *client = new PeerWireClient(ConnectionManager::instance()->clientId(), this);

   687         RateController::instance()->addSocket(client);

   688         ConnectionManager::instance()->addConnection(client);

   689 

   690         initializeConnection(client);

   691         d->connections << client;

   692 

   693         // Pick a random peer from the list of weighed peers.

   694         TorrentPeer *peer = weighedPeers.takeAt(qrand() % weighedPeers.size());

   695         weighedPeers.removeAll(peer);

   696         peer->connectStart = QDateTime::currentDateTime().toTime_t();

   697         peer->lastVisited = peer->connectStart;

   698 

   699         // Connect to the peer.

   700         client->setPeer(peer);

   701         client->connectToHost(peer->address, peer->port);

   702     }

   703 }

   704 

   705 QList<TorrentPeer *> TorrentClient::weighedFreePeers() const

   706 {

   707     QList<TorrentPeer *> weighedPeers;

   708 

   709     // Generate a list of peers that we want to connect to.

   710     uint now = QDateTime::currentDateTime().toTime_t();

   711     QList<TorrentPeer *> freePeers;

   712     QMap<QString, int> connectionsPerPeer;

   713     foreach (TorrentPeer *peer, d->peers) {

   714         bool busy = false;

   715         foreach (PeerWireClient *client, d->connections) {

   716             if (client->state() == PeerWireClient::ConnectedState

   717                 && client->peerAddress() == peer->address

   718                 && client->peerPort() == peer->port) {

   719                 if (++connectionsPerPeer[peer->address.toString()] >= MaxConnectionPerPeer) {

   720                     busy = true;

   721                     break;

   722                 }

   723             }

   724         }

   725         if (!busy && (now - peer->lastVisited) > uint(MinimumTimeBeforeRevisit))

   726             freePeers << peer;

   727     }

   728 

   729     // Nothing to connect to

   730     if (freePeers.isEmpty())

   731         return weighedPeers;

   732 

   733     // Assign points based on connection speed and pieces available.

   734     QList<QPair<int, TorrentPeer *> > points;

   735     foreach (TorrentPeer *peer, freePeers) {

   736         int tmp = 0;

   737         if (peer->interesting) {

   738             tmp += peer->numCompletedPieces;

   739             if (d->state == Seeding)

   740                 tmp = d->pieceCount - tmp;

   741             if (!peer->connectStart) // An unknown peer is as interesting as a seed

   742                 tmp += d->pieceCount;

   743 

   744             // 1/5 of the total score for each second below 5 it takes to

   745             // connect.

   746             if (peer->connectTime < 5)

   747                 tmp += (d->pieceCount / 10) * (5 - peer->connectTime);

   748         }

   749         points << QPair<int, TorrentPeer *>(tmp, peer);

   750     }

   751     qSort(points);

   752 

   753     // Minimize the list so the point difference is never more than 1.

   754     typedef QPair<int,TorrentPeer*> PointPair;

   755     QMultiMap<int, TorrentPeer *> pointMap;

   756     int lowestScore = 0;

   757     int lastIndex = 0;

   758     foreach (PointPair point, points) {

   759         if (point.first > lowestScore) {

   760             lowestScore = point.first;

   761             ++lastIndex;

   762         }

   763         pointMap.insert(lastIndex, point.second);

   764     }

   765 

   766     // Now make up a list of peers where the ones with more points are

   767     // listed many times.

   768     QMultiMap<int, TorrentPeer *>::ConstIterator it = pointMap.constBegin();

   769     while (it != pointMap.constEnd()) {

   770         for (int i = 0; i < it.key() + 1; ++i)

   771             weighedPeers << it.value();

   772         ++it;

   773     }

   774 

   775     return weighedPeers;

   776 }

   777 

   778 void TorrentClient::setupIncomingConnection(PeerWireClient *client)

   779 {

   780     // Connect signals

   781     initializeConnection(client);

   782 

   783     // Initialize this client

   784     RateController::instance()->addSocket(client);

   785     d->connections << client;

   786 

   787     client->initialize(d->infoHash, d->pieceCount);

   788     client->sendPieceList(d->completedPieces);

   789 

   790     emit peerInfoUpdated();

   791 

   792     if (d->state == Searching || d->state == Connecting) {

   793         int completed = d->completedPieces.count(true);

   794         if (completed == 0)

   795             d->setState(WarmingUp);

   796         else if (d->incompletePieces.count(true) < 5 && d->pendingPieces.size() > d->incompletePieces.count(true))

   797             d->setState(Endgame);

   798     }

   799 

   800     if (d->connections.isEmpty())

   801         scheduleUploads();

   802 }

   803 

   804 void TorrentClient::setupOutgoingConnection()

   805 {

   806     PeerWireClient *client = qobject_cast<PeerWireClient *>(sender());

   807 

   808     // Update connection statistics.

   809     foreach (TorrentPeer *peer, d->peers) {

   810         if (peer->port == client->peerPort() && peer->address == client->peerAddress()) {

   811             peer->connectTime = peer->lastVisited - peer->connectStart;

   812             break;

   813         }

   814     }

   815 

   816     // Send handshake and piece list

   817     client->initialize(d->infoHash, d->pieceCount);

   818     client->sendPieceList(d->completedPieces);

   819 

   820     emit peerInfoUpdated();

   821 

   822     if (d->state == Searching || d->state == Connecting) {

   823         int completed = d->completedPieces.count(true);

   824         if (completed == 0)

   825             d->setState(WarmingUp);

   826         else if (d->incompletePieces.count(true) < 5 && d->pendingPieces.size() > d->incompletePieces.count(true))

   827             d->setState(Endgame);

   828     }

   829 }

   830 

   831 void TorrentClient::initializeConnection(PeerWireClient *client)

   832 {

   833     connect(client, SIGNAL(connected()),

   834             this, SLOT(setupOutgoingConnection()));

   835     connect(client, SIGNAL(disconnected()),

   836             this, SLOT(removeClient()));

   837     connect(client, SIGNAL(error(QAbstractSocket::SocketError)),

   838             this, SLOT(removeClient()));

   839     connect(client, SIGNAL(piecesAvailable(const QBitArray &)),

   840             this, SLOT(peerPiecesAvailable(const QBitArray &)));

   841     connect(client, SIGNAL(blockRequested(int, int, int)),

   842             this, SLOT(peerRequestsBlock(int, int, int)));

   843     connect(client, SIGNAL(blockReceived(int, int, const QByteArray &)),

   844             this, SLOT(blockReceived(int, int, const QByteArray &)));

   845     connect(client, SIGNAL(choked()),

   846             this, SLOT(peerChoked()));

   847     connect(client, SIGNAL(unchoked()),

   848             this, SLOT(peerUnchoked()));

   849     connect(client, SIGNAL(bytesWritten(qint64)),

   850             this, SLOT(peerWireBytesWritten(qint64)));

   851     connect(client, SIGNAL(bytesReceived(qint64)),

   852             this, SLOT(peerWireBytesReceived(qint64)));

   853 }

   854 

   855 void TorrentClient::removeClient()

   856 {

   857     PeerWireClient *client = static_cast<PeerWireClient *>(sender());

   858 

   859     // Remove the host from our list of known peers if the connection

   860     // failed.

   861     if (client->peer() && client->error() == QAbstractSocket::ConnectionRefusedError)

   862         d->peers.removeAll(client->peer());

   863 

   864     // Remove the client from RateController and all structures.

   865     RateController::instance()->removeSocket(client);

   866     d->connections.removeAll(client);

   867     QMultiMap<PeerWireClient *, TorrentPiece *>::Iterator it = d->payloads.find(client);

   868     while (it != d->payloads.end() && it.key() == client) {

   869         TorrentPiece *piece = it.value();

   870         piece->inProgress = false;

   871         piece->requestedBlocks.fill(false);

   872         it = d->payloads.erase(it);

   873     }

   874 

   875     // Remove pending read requests.

   876     QMapIterator<int, PeerWireClient *> it2(d->readIds);

   877     while (it2.findNext(client))

   878         d->readIds.remove(it2.key());

   879 

   880     // Delete the client later.

   881     disconnect(client, SIGNAL(disconnected()), this, SLOT(removeClient()));

   882     client->deleteLater();

   883     ConnectionManager::instance()->removeConnection(client);

   884 

   885     emit peerInfoUpdated();

   886     d->callPeerConnector();

   887 }

   888 

   889 void TorrentClient::peerPiecesAvailable(const QBitArray &pieces)

   890 {

   891     PeerWireClient *client = qobject_cast<PeerWireClient *>(sender());

   892 

   893     // Find the peer in our list of announced peers. If it's there,

   894     // then we can use the piece list into to gather statistics that

   895     // help us decide what peers to connect to.

   896     TorrentPeer *peer = 0;

   897     QList<TorrentPeer *>::Iterator it = d->peers.begin();

   898     while (it != d->peers.end()) {

   899         if ((*it)->address == client->peerAddress() && (*it)->port == client->peerPort()) {

   900             peer = *it;

   901             break;

   902         }

   903         ++it;

   904     }

   905 

   906     // If the peer is a seed, and we are in seeding mode, then the

   907     // peer is uninteresting.

   908     if (pieces.count(true) == d->pieceCount) {

   909         if (peer)

   910             peer->seed = true;

   911         emit peerInfoUpdated();

   912         if (d->state == Seeding) {

   913             client->abort();

   914             return;

   915         } else {

   916             if (peer)

   917                 peer->interesting = true;

   918             if ((client->peerWireState() & PeerWireClient::InterestedInPeer) == 0)

   919                 client->sendInterested();

   920             d->callScheduler();

   921             return;

   922         }

   923     }

   924 

   925     // Update our list of available pieces.

   926     if (peer) {

   927         peer->pieces = pieces;

   928         peer->numCompletedPieces = pieces.count(true);

   929     }

   930 

   931     // Check for interesting pieces, and tell the peer whether we are

   932     // interested or not.

   933     bool interested = false;

   934     int piecesSize = pieces.size();

   935     for (int pieceIndex = 0; pieceIndex < piecesSize; ++pieceIndex) {

   936         if (!pieces.testBit(pieceIndex))

   937             continue;

   938         if (!d->completedPieces.testBit(pieceIndex)) {

   939             interested = true;

   940             if ((client->peerWireState() & PeerWireClient::InterestedInPeer) == 0) {

   941                 if (peer)

   942                     peer->interesting = true;

   943                 client->sendInterested();

   944             }

   945 

   946             QMultiMap<PeerWireClient *, TorrentPiece *>::Iterator it = d->payloads.find(client);

   947             int inProgress = 0;

   948             while (it != d->payloads.end() && it.key() == client) {

   949                 if (it.value()->inProgress)

   950                     inProgress += it.value()->requestedBlocks.count(true);

   951                 ++it;

   952             }

   953             if (!inProgress)

   954                 d->callScheduler();

   955             break;

   956         }

   957     }

   958     if (!interested && (client->peerWireState() & PeerWireClient::InterestedInPeer)) {

   959         if (peer)

   960             peer->interesting = false;

   961         client->sendNotInterested();

   962     }

   963 }

   964 

   965 void TorrentClient::peerRequestsBlock(int pieceIndex, int begin, int length)

   966 {

   967     PeerWireClient *client = qobject_cast<PeerWireClient *>(sender());

   968 

   969     // Silently ignore requests from choked peers

   970     if (client->peerWireState() & PeerWireClient::ChokingPeer)

   971         return;

   972 

   973     // Silently ignore requests for pieces we don't have.

   974     if (!d->completedPieces.testBit(pieceIndex))

   975         return;

   976 

   977     // Request the block from the file manager

   978     d->readIds.insert(d->fileManager.read(pieceIndex, begin, length),

   979                       qobject_cast<PeerWireClient *>(sender()));

   980 }

   981 

   982 void TorrentClient::blockReceived(int pieceIndex, int begin, const QByteArray &data)

   983 {

   984     PeerWireClient *client = qobject_cast<PeerWireClient *>(sender());

   985     if (data.size() == 0) {

   986         client->abort();

   987         return;

   988     }

   989 

   990     // Ignore it if we already have this block.

   991     int blockBit = begin / BlockSize;

   992     TorrentPiece *piece = d->pendingPieces.value(pieceIndex);

   993     if (!piece || piece->completedBlocks.testBit(blockBit)) {

   994         // Discard blocks that we already have, and fill up the pipeline.

   995         requestMore(client);

   996         return;

   997     }

   998 

   999     // If we are in warmup or endgame mode, cancel all duplicate

  1000     // requests for this block.

  1001     if (d->state == WarmingUp || d->state == Endgame) {

  1002         QMultiMap<PeerWireClient *, TorrentPiece *>::Iterator it = d->payloads.begin();

  1003         while (it != d->payloads.end()) {

  1004             PeerWireClient *otherClient = it.key();

  1005             if (otherClient != client && it.value()->index == pieceIndex) {

  1006                 if (otherClient->incomingBlocks().contains(TorrentBlock(pieceIndex, begin, data.size())))

  1007                     it.key()->cancelRequest(pieceIndex, begin, data.size());

  1008             }

  1009             ++it;

  1010         }

  1011     }

  1012 

  1013     if (d->state != Downloading && d->state != Endgame && d->completedPieces.count(true) > 0)

  1014         d->setState(Downloading);

  1015 

  1016     // Store this block

  1017     d->fileManager.write(pieceIndex, begin, data);

  1018     piece->completedBlocks.setBit(blockBit);

  1019     piece->requestedBlocks.clearBit(blockBit);

  1020 

  1021     if (blocksLeftForPiece(piece) == 0) {

  1022         // Ask the file manager to verify the newly downloaded piece

  1023         d->fileManager.verifyPiece(piece->index);

  1024         

  1025         // Remove this piece from all payloads

  1026         QMultiMap<PeerWireClient *, TorrentPiece *>::Iterator it = d->payloads.begin();

  1027         while (it != d->payloads.end()) {

  1028             if (!it.value() || it.value()->index == piece->index)

  1029                 it = d->payloads.erase(it);

  1030             else

  1031                 ++it;

  1032         }

  1033     }

  1034 

  1035     // Fill up the pipeline.

  1036     requestMore(client);

  1037 }

  1038 

  1039 void TorrentClient::peerWireBytesWritten(qint64 size)

  1040 {

  1041     if (!d->transferRateTimer)

  1042         d->transferRateTimer = startTimer(RateControlTimerDelay);

  1043 

  1044     d->uploadRate[0] += size;

  1045     d->uploadedBytes += size;

  1046     emit dataSent(size);

  1047 }

  1048 

  1049 void TorrentClient::peerWireBytesReceived(qint64 size)

  1050 {

  1051     if (!d->transferRateTimer)

  1052         d->transferRateTimer = startTimer(RateControlTimerDelay);

  1053 

  1054     d->downloadRate[0] += size;

  1055     d->downloadedBytes += size;

  1056     emit dataSent(size);

  1057 }

  1058 

  1059 int TorrentClient::blocksLeftForPiece(const TorrentPiece *piece) const

  1060 {

  1061     int blocksLeft = 0;

  1062     int completedBlocksSize = piece->completedBlocks.size();

  1063     for (int i = 0; i < completedBlocksSize; ++i) {

  1064         if (!piece->completedBlocks.testBit(i))

  1065             ++blocksLeft;

  1066     }

  1067     return blocksLeft;

  1068 }

  1069 

  1070 void TorrentClient::scheduleUploads()

  1071 {

  1072     // Generate a list of clients sorted by their transfer

  1073     // speeds.  When leeching, we sort by download speed, and when

  1074     // seeding, we sort by upload speed. Seeds are left out; there's

  1075     // no use in unchoking them.

  1076     QList<PeerWireClient *> allClients = d->connections;

  1077     QMultiMap<int, PeerWireClient *> transferSpeeds;

  1078     foreach (PeerWireClient *client, allClients) {

  1079         if (client->state() == QAbstractSocket::ConnectedState

  1080             && client->availablePieces().count(true) != d->pieceCount) {

  1081             if (d->state == Seeding) {

  1082                 transferSpeeds.insert(client->uploadSpeed(), client);

  1083             } else {

  1084                 transferSpeeds.insert(client->downloadSpeed(), client);

  1085             }

  1086         }

  1087     }

  1088 

  1089     // Unchoke the top 'MaxUploads' downloaders (peers that we are

  1090     // uploading to) and choke all others.

  1091     int maxUploaders = MaxUploads;

  1092     QMapIterator<int, PeerWireClient *> it(transferSpeeds);

  1093     it.toBack();

  1094     while (it.hasPrevious()) {

  1095         PeerWireClient *client = it.previous().value();

  1096         bool interested = (client->peerWireState() & PeerWireClient::PeerIsInterested);

  1097 

  1098         if (maxUploaders) {

  1099             allClients.removeAll(client);

  1100             if (client->peerWireState() & PeerWireClient::ChokingPeer)

  1101                 client->unchokePeer();

  1102             --maxUploaders;

  1103             continue;

  1104         }

  1105 

  1106         if ((client->peerWireState() & PeerWireClient::ChokingPeer) == 0) {

  1107             if ((qrand() % 10) == 0) 

  1108                 client->abort();

  1109             else

  1110                 client->chokePeer();

  1111             allClients.removeAll(client);

  1112         }

  1113         if (!interested)

  1114             allClients.removeAll(client);

  1115     }

  1116 

  1117     // Only interested peers are left in allClients. Unchoke one

  1118     // random peer to allow it to compete for a position among the

  1119     // downloaders.  (This is known as an "optimistic unchoke".)

  1120     if (!allClients.isEmpty()) {

  1121         PeerWireClient *client = allClients[qrand() % allClients.size()];

  1122         if (client->peerWireState() & PeerWireClient::ChokingPeer)

  1123             client->unchokePeer();

  1124     }

  1125 }

  1126 

  1127 void TorrentClient::scheduleDownloads()

  1128 {

  1129     d->schedulerCalled = false;

  1130 

  1131     if (d->state == Stopping || d->state == Paused || d->state == Idle)

  1132         return;

  1133 

  1134     // Check what each client is doing, and assign payloads to those