DefaultBtInteractive.cc 22 KB

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