DefaultBtInteractive.cc 21 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661
  1. /* <!-- copyright */
  2. /*
  3. * aria2 - The high speed download utility
  4. *
  5. * Copyright (C) 2006 Tatsuhiro Tsujikawa
  6. *
  7. * This program is free software; you can redistribute it and/or modify
  8. * it under the terms of the GNU General Public License as published by
  9. * the Free Software Foundation; either version 2 of the License, or
  10. * (at your option) any later version.
  11. *
  12. * This program is distributed in the hope that it will be useful,
  13. * but WITHOUT ANY WARRANTY; without even the implied warranty of
  14. * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
  15. * GNU General Public License for more details.
  16. *
  17. * You should have received a copy of the GNU General Public License
  18. * along with this program; if not, write to the Free Software
  19. * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
  20. *
  21. * In addition, as a special exception, the copyright holders give
  22. * permission to link the code of portions of this program with the
  23. * OpenSSL library under certain conditions as described in each
  24. * individual source file, and distribute linked combinations
  25. * including the two.
  26. * You must obey the GNU General Public License in all respects
  27. * for all of the code used other than OpenSSL. If you modify
  28. * file(s) with this exception, you may extend this exception to your
  29. * version of the file(s), but you are not obligated to do so. If you
  30. * do not wish to do so, delete this exception statement from your
  31. * version. If you delete this exception statement from all source
  32. * files in the program, then also delete it here.
  33. */
  34. /* copyright --> */
  35. #include "DefaultBtInteractive.h"
  36. #include <cstring>
  37. #include <vector>
  38. #include "prefs.h"
  39. #include "message.h"
  40. #include "BtHandshakeMessage.h"
  41. #include "util.h"
  42. #include "BtKeepAliveMessage.h"
  43. #include "BtChokeMessage.h"
  44. #include "BtUnchokeMessage.h"
  45. #include "BtRequestMessage.h"
  46. #include "BtPieceMessage.h"
  47. #include "DlAbortEx.h"
  48. #include "BtExtendedMessage.h"
  49. #include "HandshakeExtensionMessage.h"
  50. #include "UTPexExtensionMessage.h"
  51. #include "DefaultExtensionMessageFactory.h"
  52. #include "ExtensionMessageRegistry.h"
  53. #include "DHTNode.h"
  54. #include "Peer.h"
  55. #include "Piece.h"
  56. #include "DownloadContext.h"
  57. #include "PieceStorage.h"
  58. #include "PeerStorage.h"
  59. #include "BtRuntime.h"
  60. #include "BtMessageReceiver.h"
  61. #include "BtMessageDispatcher.h"
  62. #include "BtMessageFactory.h"
  63. #include "BtRequestFactory.h"
  64. #include "PeerConnection.h"
  65. #include "Logger.h"
  66. #include "LogFactory.h"
  67. #include "fmt.h"
  68. #include "RequestGroup.h"
  69. #include "RequestGroupMan.h"
  70. #include "bittorrent_helper.h"
  71. #include "UTMetadataRequestFactory.h"
  72. #include "UTMetadataRequestTracker.h"
  73. #include "wallclock.h"
  74. namespace aria2 {
  75. DefaultBtInteractive::DefaultBtInteractive
  76. (const std::shared_ptr<DownloadContext>& downloadContext,
  77. const std::shared_ptr<Peer>& peer)
  78. : cuid_(0),
  79. downloadContext_(downloadContext),
  80. peer_(peer),
  81. metadataGetMode_(false),
  82. localNode_(0),
  83. allowedFastSetSize_(10),
  84. haveTimer_(global::wallclock()),
  85. keepAliveTimer_(global::wallclock()),
  86. floodingTimer_(global::wallclock()),
  87. inactiveTimer_(global::wallclock()),
  88. pexTimer_(global::wallclock()),
  89. perSecTimer_(global::wallclock()),
  90. keepAliveInterval_(120),
  91. utPexEnabled_(false),
  92. dhtEnabled_(false),
  93. numReceivedMessage_(0),
  94. maxOutstandingRequest_(DEFAULT_MAX_OUTSTANDING_REQUEST),
  95. requestGroupMan_(0),
  96. tcpPort_(0),
  97. haveLastSent_(global::wallclock())
  98. {}
  99. DefaultBtInteractive::~DefaultBtInteractive() {}
  100. void DefaultBtInteractive::initiateHandshake() {
  101. std::shared_ptr<BtMessage> message =
  102. messageFactory_->createHandshakeMessage
  103. (bittorrent::getInfoHash(downloadContext_), bittorrent::getStaticPeerId());
  104. dispatcher_->addMessageToQueue(message);
  105. dispatcher_->sendMessages();
  106. }
  107. std::shared_ptr<BtMessage> DefaultBtInteractive::receiveHandshake(bool quickReply) {
  108. std::shared_ptr<BtHandshakeMessage> message =
  109. btMessageReceiver_->receiveHandshake(quickReply);
  110. if(!message) {
  111. return std::shared_ptr<BtMessage>();
  112. }
  113. if(memcmp(message->getPeerId(), bittorrent::getStaticPeerId(),
  114. PEER_ID_LENGTH) == 0) {
  115. throw DL_ABORT_EX
  116. (fmt("CUID#%" PRId64 " - Drop connection from the same Peer ID",
  117. cuid_));
  118. }
  119. const PeerSet& usedPeers = peerStorage_->getUsedPeers();
  120. for(PeerSet::const_iterator i = usedPeers.begin(), eoi = usedPeers.end();
  121. i != eoi; ++i) {
  122. if((*i)->isActive() &&
  123. memcmp((*i)->getPeerId(), message->getPeerId(), PEER_ID_LENGTH) == 0) {
  124. throw DL_ABORT_EX
  125. (fmt("CUID#%" PRId64 " - Same Peer ID has been already seen.",
  126. cuid_));
  127. }
  128. }
  129. peer_->setPeerId(message->getPeerId());
  130. if(message->isFastExtensionSupported()) {
  131. peer_->setFastExtensionEnabled(true);
  132. A2_LOG_INFO(fmt(MSG_FAST_EXTENSION_ENABLED, cuid_));
  133. }
  134. if(message->isExtendedMessagingEnabled()) {
  135. peer_->setExtendedMessagingEnabled(true);
  136. if(!utPexEnabled_) {
  137. extensionMessageRegistry_->removeExtension
  138. (ExtensionMessageRegistry::UT_PEX);
  139. }
  140. A2_LOG_INFO(fmt(MSG_EXTENDED_MESSAGING_ENABLED, cuid_));
  141. }
  142. if(message->isDHTEnabled()) {
  143. peer_->setDHTEnabled(true);
  144. A2_LOG_INFO(fmt(MSG_DHT_ENABLED_PEER, cuid_));
  145. }
  146. A2_LOG_INFO(fmt(MSG_RECEIVE_PEER_MESSAGE, cuid_,
  147. peer_->getIPAddress().c_str(), peer_->getPort(),
  148. message->toString().c_str()));
  149. return message;
  150. }
  151. std::shared_ptr<BtMessage> DefaultBtInteractive::receiveAndSendHandshake() {
  152. return receiveHandshake(true);
  153. }
  154. void DefaultBtInteractive::doPostHandshakeProcessing() {
  155. // Set time 0 to haveTimer to cache http/ftp download piece completion
  156. haveTimer_.reset(0);
  157. keepAliveTimer_ = global::wallclock();
  158. floodingTimer_ = global::wallclock();
  159. pexTimer_.reset(0);
  160. if(peer_->isExtendedMessagingEnabled()) {
  161. addHandshakeExtendedMessageToQueue();
  162. }
  163. if(!metadataGetMode_) {
  164. addBitfieldMessageToQueue();
  165. }
  166. if(peer_->isDHTEnabled() && dhtEnabled_) {
  167. addPortMessageToQueue();
  168. }
  169. if(!metadataGetMode_) {
  170. addAllowedFastMessageToQueue();
  171. }
  172. sendPendingMessage();
  173. }
  174. void DefaultBtInteractive::addPortMessageToQueue()
  175. {
  176. dispatcher_->addMessageToQueue
  177. (messageFactory_->createPortMessage(localNode_->getPort()));
  178. }
  179. void DefaultBtInteractive::addHandshakeExtendedMessageToQueue()
  180. {
  181. std::shared_ptr<HandshakeExtensionMessage> m(new HandshakeExtensionMessage());
  182. m->setClientVersion("aria2/" PACKAGE_VERSION);
  183. m->setTCPPort(tcpPort_);
  184. m->setExtensions(extensionMessageRegistry_->getExtensions());
  185. auto attrs = bittorrent::getTorrentAttrs(downloadContext_);
  186. if(!attrs->metadata.empty()) {
  187. m->setMetadataSize(attrs->metadataSize);
  188. }
  189. std::shared_ptr<BtMessage> msg = messageFactory_->createBtExtendedMessage(m);
  190. dispatcher_->addMessageToQueue(msg);
  191. }
  192. void DefaultBtInteractive::addBitfieldMessageToQueue() {
  193. if(peer_->isFastExtensionEnabled()) {
  194. if(pieceStorage_->allDownloadFinished()) {
  195. dispatcher_->addMessageToQueue(messageFactory_->createHaveAllMessage());
  196. } else if(pieceStorage_->getCompletedLength() > 0) {
  197. dispatcher_->addMessageToQueue(messageFactory_->createBitfieldMessage());
  198. } else {
  199. dispatcher_->addMessageToQueue(messageFactory_->createHaveNoneMessage());
  200. }
  201. } else {
  202. if(pieceStorage_->getCompletedLength() > 0) {
  203. dispatcher_->addMessageToQueue(messageFactory_->createBitfieldMessage());
  204. }
  205. }
  206. }
  207. void DefaultBtInteractive::addAllowedFastMessageToQueue() {
  208. if(peer_->isFastExtensionEnabled()) {
  209. std::vector<size_t> fastSet;
  210. bittorrent::computeFastSet(fastSet, peer_->getIPAddress(),
  211. downloadContext_->getNumPieces(),
  212. bittorrent::getInfoHash(downloadContext_),
  213. allowedFastSetSize_);
  214. for(std::vector<size_t>::const_iterator itr = fastSet.begin(),
  215. eoi = fastSet.end(); itr != eoi; ++itr) {
  216. dispatcher_->addMessageToQueue
  217. (messageFactory_->createAllowedFastMessage(*itr));
  218. }
  219. }
  220. }
  221. void DefaultBtInteractive::decideChoking() {
  222. if(peer_->shouldBeChoking()) {
  223. if(!peer_->amChoking()) {
  224. dispatcher_->addMessageToQueue(messageFactory_->createChokeMessage());
  225. }
  226. } else {
  227. if(peer_->amChoking()) {
  228. dispatcher_->addMessageToQueue(messageFactory_->createUnchokeMessage());
  229. }
  230. }
  231. }
  232. void DefaultBtInteractive::checkHave() {
  233. const size_t MIN_HAVE_PACK_SIZE = 20;
  234. const time_t MAX_HAVE_DELAY_SEC = 10;
  235. pieceStorage_->getAdvertisedPieceIndexes(haveIndexes_, cuid_, haveTimer_);
  236. haveTimer_ = global::wallclock();
  237. if(haveIndexes_.size() >= MIN_HAVE_PACK_SIZE) {
  238. if(peer_->isFastExtensionEnabled() &&
  239. pieceStorage_->allDownloadFinished()) {
  240. dispatcher_->addMessageToQueue(messageFactory_->createHaveAllMessage());
  241. } else {
  242. dispatcher_->addMessageToQueue(messageFactory_->createBitfieldMessage());
  243. }
  244. haveIndexes_.clear();
  245. } else {
  246. if(haveIndexes_.size() >= MIN_HAVE_PACK_SIZE ||
  247. haveLastSent_.difference(global::wallclock()) >= MAX_HAVE_DELAY_SEC) {
  248. haveLastSent_ = global::wallclock();
  249. for(std::vector<size_t>::const_iterator itr = haveIndexes_.begin(),
  250. eoi = haveIndexes_.end(); itr != eoi; ++itr) {
  251. dispatcher_->addMessageToQueue(messageFactory_->
  252. createHaveMessage(*itr));
  253. }
  254. haveIndexes_.clear();
  255. }
  256. }
  257. }
  258. void DefaultBtInteractive::sendKeepAlive() {
  259. if(keepAliveTimer_.difference(global::wallclock()) >= keepAliveInterval_) {
  260. dispatcher_->addMessageToQueue(messageFactory_->createKeepAliveMessage());
  261. dispatcher_->sendMessages();
  262. keepAliveTimer_ = global::wallclock();
  263. }
  264. }
  265. size_t DefaultBtInteractive::receiveMessages() {
  266. size_t countOldOutstandingRequest = dispatcher_->countOutstandingRequest();
  267. size_t msgcount = 0;
  268. while(1) {
  269. if(requestGroupMan_->doesOverallDownloadSpeedExceed() ||
  270. downloadContext_->getOwnerRequestGroup()->doesDownloadSpeedExceed()) {
  271. break;
  272. }
  273. std::shared_ptr<BtMessage> message = btMessageReceiver_->receiveMessage();
  274. if(!message) {
  275. break;
  276. }
  277. ++msgcount;
  278. A2_LOG_INFO(fmt(MSG_RECEIVE_PEER_MESSAGE,
  279. cuid_,
  280. peer_->getIPAddress().c_str(), peer_->getPort(),
  281. message->toString().c_str()));
  282. message->doReceivedAction();
  283. switch(message->getId()) {
  284. case BtKeepAliveMessage::ID:
  285. floodingStat_.incKeepAliveCount();
  286. break;
  287. case BtChokeMessage::ID:
  288. if(!peer_->peerChoking()) {
  289. floodingStat_.incChokeUnchokeCount();
  290. }
  291. break;
  292. case BtUnchokeMessage::ID:
  293. if(peer_->peerChoking()) {
  294. floodingStat_.incChokeUnchokeCount();
  295. }
  296. break;
  297. case BtPieceMessage::ID:
  298. case BtRequestMessage::ID:
  299. inactiveTimer_ = global::wallclock();
  300. break;
  301. }
  302. }
  303. if(!pieceStorage_->isEndGame() &&
  304. countOldOutstandingRequest > dispatcher_->countOutstandingRequest() &&
  305. (countOldOutstandingRequest - dispatcher_->countOutstandingRequest())*4 >=
  306. maxOutstandingRequest_) {
  307. maxOutstandingRequest_ =
  308. std::min((size_t)UB_MAX_OUTSTANDING_REQUEST,
  309. maxOutstandingRequest_*2);
  310. }
  311. return msgcount;
  312. }
  313. void DefaultBtInteractive::decideInterest() {
  314. if(pieceStorage_->hasMissingPiece(peer_)) {
  315. if(!peer_->amInterested()) {
  316. A2_LOG_DEBUG(fmt(MSG_PEER_INTERESTED, cuid_));
  317. dispatcher_->
  318. addMessageToQueue(messageFactory_->createInterestedMessage());
  319. }
  320. } else {
  321. if(peer_->amInterested()) {
  322. A2_LOG_DEBUG(fmt(MSG_PEER_NOT_INTERESTED, cuid_));
  323. dispatcher_->
  324. addMessageToQueue(messageFactory_->createNotInterestedMessage());
  325. }
  326. }
  327. }
  328. void DefaultBtInteractive::fillPiece(size_t maxMissingBlock) {
  329. if(pieceStorage_->hasMissingPiece(peer_)) {
  330. size_t numMissingBlock = btRequestFactory_->countMissingBlock();
  331. if(numMissingBlock >= maxMissingBlock) {
  332. return;
  333. }
  334. size_t diffMissingBlock = maxMissingBlock-numMissingBlock;
  335. std::vector<std::shared_ptr<Piece> > pieces;
  336. if(peer_->peerChoking()) {
  337. if(peer_->isFastExtensionEnabled()) {
  338. if(pieceStorage_->isEndGame()) {
  339. std::vector<size_t> excludedIndexes;
  340. excludedIndexes.reserve(btRequestFactory_->countTargetPiece());
  341. btRequestFactory_->getTargetPieceIndexes(excludedIndexes);
  342. pieceStorage_->getMissingFastPiece
  343. (pieces, diffMissingBlock, peer_, excludedIndexes, cuid_);
  344. } else {
  345. pieces.reserve(diffMissingBlock);
  346. pieceStorage_->getMissingFastPiece
  347. (pieces, diffMissingBlock, peer_, cuid_);
  348. }
  349. }
  350. } else {
  351. if(pieceStorage_->isEndGame()) {
  352. std::vector<size_t> excludedIndexes;
  353. excludedIndexes.reserve(btRequestFactory_->countTargetPiece());
  354. btRequestFactory_->getTargetPieceIndexes(excludedIndexes);
  355. pieceStorage_->getMissingPiece
  356. (pieces, diffMissingBlock, peer_, excludedIndexes, cuid_);
  357. } else {
  358. pieces.reserve(diffMissingBlock);
  359. pieceStorage_->getMissingPiece(pieces, diffMissingBlock, peer_, cuid_);
  360. }
  361. }
  362. for(std::vector<std::shared_ptr<Piece> >::const_iterator i =
  363. pieces.begin(), eoi = pieces.end(); i != eoi; ++i) {
  364. btRequestFactory_->addTargetPiece(*i);
  365. }
  366. }
  367. }
  368. void DefaultBtInteractive::addRequests() {
  369. if(!pieceStorage_->isEndGame() && !pieceStorage_->hasMissingUnusedPiece()) {
  370. pieceStorage_->enterEndGame();
  371. }
  372. fillPiece(maxOutstandingRequest_);
  373. size_t reqNumToCreate =
  374. maxOutstandingRequest_ <= dispatcher_->countOutstandingRequest() ?
  375. 0 : maxOutstandingRequest_-dispatcher_->countOutstandingRequest();
  376. if(reqNumToCreate > 0) {
  377. std::vector<std::shared_ptr<BtMessage> > requests;
  378. requests.reserve(reqNumToCreate);
  379. if(pieceStorage_->isEndGame()) {
  380. btRequestFactory_->createRequestMessagesOnEndGame(requests,reqNumToCreate);
  381. } else {
  382. btRequestFactory_->createRequestMessages(requests, reqNumToCreate);
  383. }
  384. dispatcher_->addMessageToQueue(requests);
  385. }
  386. }
  387. void DefaultBtInteractive::cancelAllPiece() {
  388. btRequestFactory_->removeAllTargetPiece();
  389. if(metadataGetMode_ && downloadContext_->getTotalLength() > 0) {
  390. std::vector<size_t> metadataRequests =
  391. utMetadataRequestTracker_->getAllTrackedIndex();
  392. for(std::vector<size_t>::const_iterator i = metadataRequests.begin(),
  393. eoi = metadataRequests.end(); i != eoi; ++i) {
  394. A2_LOG_DEBUG(fmt("Cancel metadata: piece=%lu",
  395. static_cast<unsigned long>(*i)));
  396. pieceStorage_->cancelPiece(pieceStorage_->getPiece(*i), cuid_);
  397. }
  398. }
  399. }
  400. void DefaultBtInteractive::sendPendingMessage() {
  401. dispatcher_->sendMessages();
  402. }
  403. void DefaultBtInteractive::detectMessageFlooding() {
  404. if(floodingTimer_.
  405. difference(global::wallclock()) >= FLOODING_CHECK_INTERVAL) {
  406. if(floodingStat_.getChokeUnchokeCount() >= 2 ||
  407. floodingStat_.getKeepAliveCount() >= 2) {
  408. throw DL_ABORT_EX(EX_FLOODING_DETECTED);
  409. } else {
  410. floodingStat_.reset();
  411. }
  412. floodingTimer_ = global::wallclock();
  413. }
  414. }
  415. void DefaultBtInteractive::checkActiveInteraction()
  416. {
  417. time_t inactiveTime = inactiveTimer_.difference(global::wallclock());
  418. // To allow aria2 to accept mutially interested peer, disconnect unintersted
  419. // peer.
  420. {
  421. const time_t interval = 30;
  422. if(!peer_->amInterested() && !peer_->peerInterested() &&
  423. inactiveTime >= interval) {
  424. peer_->setDisconnectedGracefully(true);
  425. // TODO change the message
  426. throw DL_ABORT_EX
  427. (fmt("Disconnect peer because we are not interested each other"
  428. " after %ld second(s).",
  429. static_cast<long int>(interval)));
  430. }
  431. }
  432. // Since the peers which are *just* connected and do nothing to improve
  433. // mutual download progress are completely waste of resources, those peers
  434. // are disconnected in a certain time period.
  435. {
  436. const time_t interval = 60;
  437. if(inactiveTime >= interval) {
  438. peer_->setDisconnectedGracefully(true);
  439. throw DL_ABORT_EX
  440. (fmt(EX_DROP_INACTIVE_CONNECTION,
  441. static_cast<long int>(interval)));
  442. }
  443. }
  444. // If both of us are seeders, drop connection.
  445. if(peer_->isSeeder() && pieceStorage_->downloadFinished()) {
  446. throw DL_ABORT_EX(MSG_GOOD_BYE_SEEDER);
  447. }
  448. }
  449. void DefaultBtInteractive::addPeerExchangeMessage()
  450. {
  451. if(pexTimer_.
  452. difference(global::wallclock()) >= UTPexExtensionMessage::DEFAULT_INTERVAL) {
  453. std::shared_ptr<UTPexExtensionMessage> m
  454. (new UTPexExtensionMessage(peer_->getExtensionMessageID
  455. (ExtensionMessageRegistry::UT_PEX)));
  456. const PeerSet& usedPeers = peerStorage_->getUsedPeers();
  457. for(PeerSet::const_iterator i = usedPeers.begin(), eoi = usedPeers.end();
  458. i != eoi && !m->freshPeersAreFull(); ++i) {
  459. if((*i)->isActive() && peer_->getIPAddress() != (*i)->getIPAddress()) {
  460. m->addFreshPeer(*i);
  461. }
  462. }
  463. const std::deque<std::shared_ptr<Peer> >& droppedPeers =
  464. peerStorage_->getDroppedPeers();
  465. for(std::deque<std::shared_ptr<Peer> >::const_iterator i =
  466. droppedPeers.begin(), eoi = droppedPeers.end();
  467. i != eoi && !m->droppedPeersAreFull();
  468. ++i) {
  469. if(peer_->getIPAddress() != (*i)->getIPAddress()) {
  470. m->addDroppedPeer(*i);
  471. }
  472. }
  473. std::shared_ptr<BtMessage> msg = messageFactory_->createBtExtendedMessage(m);
  474. dispatcher_->addMessageToQueue(msg);
  475. pexTimer_ = global::wallclock();
  476. }
  477. }
  478. void DefaultBtInteractive::doInteractionProcessing() {
  479. if(metadataGetMode_) {
  480. sendKeepAlive();
  481. numReceivedMessage_ = receiveMessages();
  482. // PieceStorage is re-initialized with metadata_size in
  483. // HandshakeExtensionMessage::doReceivedAction().
  484. pieceStorage_ =
  485. downloadContext_->getOwnerRequestGroup()->getPieceStorage();
  486. if(peer_->getExtensionMessageID(ExtensionMessageRegistry::UT_METADATA) &&
  487. downloadContext_->getTotalLength() > 0) {
  488. size_t num = utMetadataRequestTracker_->avail();
  489. if(num > 0) {
  490. std::vector<std::shared_ptr<BtMessage> > requests;
  491. utMetadataRequestFactory_->create(requests, num, pieceStorage_);
  492. dispatcher_->addMessageToQueue(requests);
  493. }
  494. if(perSecTimer_.difference(global::wallclock()) >= 1) {
  495. perSecTimer_ = global::wallclock();
  496. // Drop timeout request after queuing message to give a chance
  497. // to other connection to request piece.
  498. std::vector<size_t> indexes =
  499. utMetadataRequestTracker_->removeTimeoutEntry();
  500. for(std::vector<size_t>::const_iterator i = indexes.begin(),
  501. eoi = indexes.end(); i != eoi; ++i) {
  502. pieceStorage_->cancelPiece(pieceStorage_->getPiece(*i), cuid_);
  503. }
  504. }
  505. if(pieceStorage_->downloadFinished()) {
  506. downloadContext_->getOwnerRequestGroup()->setForceHaltRequested
  507. (true, RequestGroup::NONE);
  508. }
  509. }
  510. } else {
  511. checkActiveInteraction();
  512. decideChoking();
  513. detectMessageFlooding();
  514. if(perSecTimer_.difference(global::wallclock()) >= 1) {
  515. perSecTimer_ = global::wallclock();
  516. dispatcher_->checkRequestSlotAndDoNecessaryThing();
  517. }
  518. checkHave();
  519. sendKeepAlive();
  520. numReceivedMessage_ = receiveMessages();
  521. btRequestFactory_->removeCompletedPiece();
  522. decideInterest();
  523. if(!pieceStorage_->downloadFinished()) {
  524. addRequests();
  525. }
  526. }
  527. if(peer_->getExtensionMessageID(ExtensionMessageRegistry::UT_PEX) &&
  528. utPexEnabled_) {
  529. addPeerExchangeMessage();
  530. }
  531. sendPendingMessage();
  532. }
  533. void DefaultBtInteractive::setLocalNode(DHTNode* node)
  534. {
  535. localNode_ = node;
  536. }
  537. size_t DefaultBtInteractive::countPendingMessage()
  538. {
  539. return dispatcher_->countMessageInQueue();
  540. }
  541. bool DefaultBtInteractive::isSendingMessageInProgress()
  542. {
  543. return dispatcher_->isSendingInProgress();
  544. }
  545. size_t DefaultBtInteractive::countReceivedMessageInIteration() const
  546. {
  547. return numReceivedMessage_;
  548. }
  549. size_t DefaultBtInteractive::countOutstandingRequest()
  550. {
  551. if(metadataGetMode_) {
  552. return utMetadataRequestTracker_->count();
  553. } else {
  554. return dispatcher_->countOutstandingRequest();
  555. }
  556. }
  557. void DefaultBtInteractive::setBtRuntime
  558. (const std::shared_ptr<BtRuntime>& btRuntime)
  559. {
  560. btRuntime_ = btRuntime;
  561. }
  562. void DefaultBtInteractive::setPieceStorage
  563. (const std::shared_ptr<PieceStorage>& pieceStorage)
  564. {
  565. pieceStorage_ = pieceStorage;
  566. }
  567. void DefaultBtInteractive::setPeerStorage
  568. (const std::shared_ptr<PeerStorage>& peerStorage)
  569. {
  570. peerStorage_ = peerStorage;
  571. }
  572. void DefaultBtInteractive::setPeer(const std::shared_ptr<Peer>& peer)
  573. {
  574. peer_ = peer;
  575. }
  576. void DefaultBtInteractive::setBtMessageReceiver
  577. (const std::shared_ptr<BtMessageReceiver>& receiver)
  578. {
  579. btMessageReceiver_ = receiver;
  580. }
  581. void DefaultBtInteractive::setDispatcher
  582. (const std::shared_ptr<BtMessageDispatcher>& dispatcher)
  583. {
  584. dispatcher_ = dispatcher;
  585. }
  586. void DefaultBtInteractive::setBtRequestFactory
  587. (const std::shared_ptr<BtRequestFactory>& factory)
  588. {
  589. btRequestFactory_ = factory;
  590. }
  591. void DefaultBtInteractive::setPeerConnection
  592. (const std::shared_ptr<PeerConnection>& peerConnection)
  593. {
  594. peerConnection_ = peerConnection;
  595. }
  596. void DefaultBtInteractive::setExtensionMessageFactory
  597. (const std::shared_ptr<ExtensionMessageFactory>& factory)
  598. {
  599. extensionMessageFactory_ = factory;
  600. }
  601. void DefaultBtInteractive::setBtMessageFactory
  602. (const std::shared_ptr<BtMessageFactory>& factory)
  603. {
  604. messageFactory_ = factory;
  605. }
  606. void DefaultBtInteractive::setRequestGroupMan(RequestGroupMan* rgman)
  607. {
  608. requestGroupMan_ = rgman;
  609. }
  610. } // namespace aria2