src/main/torrentclient.cpp

   1 /***************************************************************************
   2  *                                                                         *
   3  *   Copyright (C) 2006 David Brodsky                                      *
   4  *                                                                         *
   5  *   This program is free software; you can redistribute it and/or         *
   6  *   modify it under the terms of the GNU General Public License as        *
   7  *   published by the Free Software Foundation and appearing               *
   8  *   in the file LICENSE.GPL included in the packaging of this file.       *
   9  *                                                                         *
  10  *   This program is distributed in the hope that it will be useful,       *
  11  *   but WITHOUT ANY WARRANTY; without even the implied warranty of        *
  12  *   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU      *
  13  *   General Public License for more details.                              *
  14  *                                                                         *
  15  ***************************************************************************/
  16
  17 #include <tairon/core/config.h>
  18 #include <tairon/core/log.h>
  19 #include <tairon/net/timer.h>
  20
  21 #include "torrentclient.h"
  22
  23 #include "core/bitfield.h"
  24 #include "connection.h"
  25 #include "storage.h"
  26 #include "torrentmanager.h"
  27 #include "trackerclient.h"
  28 #include "trackermanager.h"
  29
  30 namespace Tairent
  31 {
  32
  33 namespace Main
  34 {
  35
  36 /* {{{ TorrentClient::TorrentClient(TorrentStruct *, const String &, Storage *) */
  37 TorrentClient::TorrentClient(TorrentStruct *t, const String &ih, Storage *s) : downloaded(0), infoHash(ih), storage(s), torrent(t), uploaded(0)
  38 {
  39         if (!s)
  40                 storage = new Storage(getMetaInfo()["info"]);
  41
  42         storage->invalidDataSignal.connect(Tairon::Core::methodFunctor(this, &TorrentClient::invalidDataReceived));
  43         storage->pieceDownloadedSignal.connect(Tairon::Core::methodFunctor(this, &TorrentClient::pieceDownloadedAndChecked));
  44
  45         trackerClient = TrackerManager::self()->getTracker(this);
  46         trackerClient->getPeers();
  47
  48         maxConnections = atoi((*Tairon::Core::Config::self())["max-connections"]);
  49         maxUploads = atoi((*Tairon::Core::Config::self())["max-uploads"]);
  50         minUploads = atoi((*Tairon::Core::Config::self())["min-uploads"]);
  51
  52         timer = new Tairon::Net::Timer();
  53         timer->timeoutSignal.connect(Tairon::Core::threadMethodDFunctor(Tairon::Core::Thread::current(), this, &TorrentClient::rechoke));
  54         timer->start(10000);
  55 }
  56 /* }}} */
  57
  58 /* {{{ TorrentClient::~TorrentClient() */
  59 TorrentClient::~TorrentClient()
  60 {
  61         timer->destroy();
  62
  63         delete trackerClient;
  64
  65         // storage deletion is done by torrent manager
  66 }
  67 /* }}} */
  68
  69 /* {{{ TorrentClient::addPeers(const std::list<PeerStruct> &) */
  70 void TorrentClient::addPeers(const std::list<PeerStruct> &peers)
  71 {
  72         String clientID(TorrentManager::self()->getClientID());
  73
  74         for (std::list<PeerStruct>::const_iterator it = peers.begin(); it != peers.end(); ++it) {
  75                 if (connections.count(it->id)) // we're already connected to this peer
  76                         continue;
  77                 if (pendingConnections.count(it->id)) // we're trying to connect to this peer
  78                         continue;
  79                 if (it->id == clientID) // don't connect to ourselves
  80                         continue;
  81
  82                 Connection *c = new Connection(it->ip, it->port, it->id, this);
  83                 c->closedSignal.connect(Tairon::Core::methodFunctor(this, &TorrentClient::connectingFailed));
  84                 pendingConnections[it->id] = c;
  85
  86                 if (pendingConnections.size() + connections.size() >= maxConnections)
  87                         return;
  88         }
  89 }
  90 /* }}} */
  91
  92 /* {{{ TorrentClient::connectingFailed(Connection *) */
  93 void TorrentClient::connectingFailed(Connection *c)
  94 {
  95         pendingConnections.erase(c->getPeerID());
  96         delete c;
  97 }
  98 /* }}} */
  99
 100 /* {{{ TorrentClient::connectionClosed(Connection *) */
 101 void TorrentClient::connectionClosed(Connection *c)
 102 {
 103         storage->unmapPieces(c->getPieceWriter());
 104         c->destroyPieceWriters();
 105         storage->unmapPieces(c->getPieceReader());
 106         c->destroyPieceReaders();
 107
 108         storage->removeBitField(c->getBitField());
 109
 110         connections.erase(c->getPeerID());
 111         delete c;
 112 }
 113 /* }}} */
 114
 115 /* {{{ TorrentClient::getInfoHash() */
 116 const String &TorrentClient::getInfoHash()
 117 {
 118         return infoHash;
 119 }
 120 /* }}} */
 121
 122 /* {{{ TorrentClient::getMetaInfo() */
 123 const Tairent::Core::BEncode &TorrentClient::getMetaInfo()
 124 {
 125         return *torrent->metaInfo;
 126 }
 127 /* }}} */
 128
 129 /* {{{ TorrentClient::getRemainingSize() */
 130 uint64_t TorrentClient::getRemainingSize()
 131 {
 132         return storage->getRemainingSize();
 133 }
 134 /* }}} */
 135
 136 /* {{{ TorrentClient::getStorage() */
 137 Storage *TorrentClient::getStorage()
 138 {
 139         return storage;
 140 }
 141 /* }}} */
 142
 143 /* {{{ TorrentClient::gotBitField(Connection *) */
 144 void TorrentClient::gotBitField(Connection *c)
 145 {
 146         storage->addBitField(c->getBitField());
 147         if (storage->shouldBeInterested(c->getBitField()))
 148                 c->sendInterested();
 149 }
 150 /* }}} */
 151
 152 /* {{{ TorrentClient::gotChoke(Connection *) */
 153 void TorrentClient::gotChoke(Connection *c)
 154 {
 155         storage->reRequest(c->getRequested());
 156         c->clearRequested();
 157 }
 158 /* }}} */
 159
 160 /* {{{ TorrentClient::gotHave(uint32_t, Connection *) */
 161 void TorrentClient::gotHave(uint32_t index, Connection *c)
 162 {
 163         if (storage->gotHave(index)) {
 164                 c->sendInterested();
 165                 if (!c->isChoked() && (c->getRequestsCount() < 5)) // TODO: do it better
 166                         requestMore(c);
 167         }
 168 }
 169 /* }}} */
 170
 171 /* {{{ TorrentClient::gotInterestedChange(Connection *) */
 172 void TorrentClient::gotInterestedChange(Connection *c)
 173 {
 174         if (c->isPeerChoked())
 175                 return;
 176         rechoke();
 177 }
 178 /* }}} */
 179
 180 /* {{{ TorrentClient::gotPiece(uint32_t, uint32_t, Connection *) */
 181 void TorrentClient::gotPiece(uint32_t index, uint32_t start, Connection *c)
 182 {
 183         downloaded += c->getPieceLength();
 184         storage->gotPiece(index, start, c);
 185 }
 186 /* }}} */
 187
 188 /* {{{ TorrentClient::gotUnchoke(Connection *) */
 189 void TorrentClient::gotUnchoke(Connection *c)
 190 {
 191         if (c->isInterested())
 192                 requestMore(c);
 193 }
 194 /* }}} */
 195
 196 /* {{{ TorrentClient::invalidDataReceived(Connection *) */
 197 void TorrentClient::invalidDataReceived(Connection *c)
 198 {
 199         c->close();
 200         connectionClosed(c);
 201 }
 202 /* }}} */
 203
 204 /* {{{ TorrentClient::newConnection(Connection *) */
 205 void TorrentClient::newConnection(Connection *c)
 206 {
 207         if (pendingConnections.count(c->getPeerID()))
 208                 pendingConnections.erase(c->getPeerID());
 209
 210         if (pendingConnections.size() + connections.size() >= maxConnections) {
 211                 c->close();
 212                 delete c;
 213                 return;
 214         }
 215
 216         if (connections.count(c->getPeerID())) { // already connected peer
 217                 c->close();
 218                 delete c;
 219                 return;
 220         }
 221
 222         connections[c->getPeerID()] = c;
 223         c->closedSignal.connect(Tairon::Core::methodFunctor(this, &TorrentClient::connectionClosed));
 224 }
 225 /* }}} */
 226
 227 /* {{{ TorrentClient::pieceDownloaded(uint32_t, uint32_t, Connection *) */
 228 void TorrentClient::pieceDownloaded(uint32_t index, uint32_t start, Connection *c)
 229 {
 230         storage->pieceDownloaded(index, start);
 231         requestMore(c);
 232 }
 233 /* }}} */
 234
 235 /* {{{ TorrentClient::pieceDownloadedAndChecked(uint32_t) */
 236 void TorrentClient::pieceDownloadedAndChecked(uint32_t index)
 237 {
 238         std::map<String, Connection *> temp(connections);
 239         for (std::map<String, Connection *>::const_iterator it = temp.begin(); it != temp.end(); ++it)
 240                 it->second->sendHave(index);
 241 }
 242 /* }}} */
 243
 244 /* {{{ TorrentClient::pieceSent(uint32_t, uint32_t, Connection *) */
 245 void TorrentClient::pieceSent(uint32_t index, uint32_t length, Connection *c)
 246 {
 247         storage->pieceSent(index);
 248         uploaded += length;
 249 }
 250 /* }}} */
 251
 252 /* {{{ TorrentClient::rechoke() */
 253 void TorrentClient::rechoke()
 254 {
 255         DEBUG("rechoking");
 256
 257         // mapping connections sorted by their upload rate
 258         std::multimap<double, Connection *> preferred;
 259         for (std::map<String, Connection *>::iterator it = connections.begin(); it != connections.end(); ++it)
 260                 if (it->second->isPeerInterested() && !it->second->isSnubbed())
 261                         preferred.insert(std::pair<double, Connection *>(it->second->getIncomingRate(), it->second));
 262
 263         // number of preferred upload slots
 264         int prefCount = preferred.size();
 265         if (maxUploads < prefCount)
 266                 prefCount = 7;
 267
 268         // pick preferred connections
 269         std::set<Connection *> mask;
 270         for (std::multimap<double, Connection *>::reverse_iterator it = preferred.rbegin(); it != preferred.rend(); ++it) {
 271                 mask.insert(it->second);
 272                 if (!--prefCount)
 273                         break;
 274         }
 275
 276         // number of connections that are not preferred and we want them unchoked
 277         int count = 1;
 278         if (1 - prefCount > count)
 279                 count = 1 - prefCount;
 280
 281         // unchoke preferred connections, remaining just put in a list
 282         Connection **toUnchoke = new Connection *[connections.size() - mask.size()];
 283         int toUnchokeIndex = 0;
 284         std::map<String, Connection *> temp(connections);
 285         for (std::map<String, Connection *>::iterator it = temp.begin(); it != temp.end(); ++it)
 286                 if (mask.count(it->second))
 287                         it->second->sendUnchoke();
 288                 else
 289                         toUnchoke[toUnchokeIndex++] = it->second;
 290
 291         // shuffle the list
 292         for (int i = 0; i < toUnchokeIndex; ++i) {
 293                 int r = random() % toUnchokeIndex;
 294                 Connection *temp = toUnchoke[i];
 295                 toUnchoke[i] = toUnchoke[r];
 296                 toUnchoke[r] = temp;
 297         }
 298
 299         // unchoke remaining connections and choke the rest
 300         for (int i = 0; i < toUnchokeIndex; ++i)
 301                 if (count) {
 302                         if (toUnchoke[i]->isPeerInterested())
 303                                 --count;
 304                         toUnchoke[i]->sendUnchoke();
 305                 } else
 306                         toUnchoke[i]->sendChoke();
 307
 308         delete [] toUnchoke;
 309 }
 310 /* }}} */
 311
 312 /* {{{ TorrentClient::requestMore(Connection *) */
 313 void TorrentClient::requestMore(Connection *c)
 314 {
 315         uint32_t index;
 316         uint32_t start;
 317         uint32_t length;
 318
 319         if (storage->pickPiece(c->getBitField(), index, start, length))
 320                 c->sendRequest(index, start, length);
 321         else
 322                 c->sendNotInterested();
 323 }
 324 /* }}} */
 325
 326 /* {{{ TorrentClient::stop() */
 327 void TorrentClient::stop()
 328 {
 329         trackerClient->stop();
 330
 331         for (std::map<String, Connection *>::const_iterator it = connections.begin(); it != connections.end(); ++it) {
 332                 it->second->close();
 333                 delete it->second;
 334         }
 335         connections.clear();
 336
 337         for (std::map<String, Connection *>::const_iterator it = pendingConnections.begin(); it != pendingConnections.end(); ++it) {
 338                 it->second->close();
 339                 delete it->second;
 340         }
 341         pendingConnections.clear();
 342 }
 343 /* }}} */
 344
 345 }; // namespace Main
 346
 347 }; // namespace Tairent
 348
 349 // vim: ai sw=4 ts=4 noet fdm=marker