DHTAbstractNodeLookupTask.cc 5.3 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158
  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 "DHTAbstractNodeLookupTask.h"
  36. #include "DHTRoutingTable.h"
  37. #include "DHTMessageDispatcher.h"
  38. #include "DHTMessageFactory.h"
  39. #include "DHTMessage.h"
  40. #include "DHTNode.h"
  41. #include "DHTNodeLookupEntry.h"
  42. #include "DHTMessageCallbackImpl.h"
  43. #include "DHTBucket.h"
  44. #include "LogFactory.h"
  45. #include "Logger.h"
  46. #include "Util.h"
  47. #include "DHTIDCloser.h"
  48. #include <cstring>
  49. #include <algorithm>
  50. namespace aria2 {
  51. DHTAbstractNodeLookupTask::DHTAbstractNodeLookupTask(const unsigned char* targetID):
  52. _inFlightMessage(0)
  53. {
  54. memcpy(_targetID, targetID, DHT_ID_LENGTH);
  55. }
  56. void DHTAbstractNodeLookupTask::onReceived(const SharedHandle<DHTMessage>& message)
  57. {
  58. --_inFlightMessage;
  59. onReceivedInternal(message);
  60. std::deque<SharedHandle<DHTNodeLookupEntry> > newEntries = toEntries(getNodesFromMessage(message));
  61. size_t count = 0;
  62. for(std::deque<SharedHandle<DHTNodeLookupEntry> >::const_iterator i = newEntries.begin();
  63. i != newEntries.end(); ++i) {
  64. if(memcmp(_localNode->getID(), (*i)->_node->getID(), DHT_ID_LENGTH) != 0) {
  65. _entries.push_front(*i);
  66. ++count;
  67. }
  68. }
  69. _logger->debug("%u node lookup entries added.", count);
  70. std::stable_sort(_entries.begin(), _entries.end(), DHTIDCloser(_targetID));
  71. _entries.erase(std::unique(_entries.begin(), _entries.end()), _entries.end());
  72. _logger->debug("%u node lookup entries are unique.", _entries.size());
  73. if(_entries.size() > DHTBucket::K) {
  74. _entries.erase(_entries.begin()+DHTBucket::K, _entries.end());
  75. }
  76. if(needsAdditionalOutgoingMessage()) {
  77. sendMessage();
  78. }
  79. if(_inFlightMessage == 0) {
  80. _logger->debug("Finished node_lookup for node ID %s",
  81. Util::toHex(_targetID, DHT_ID_LENGTH).c_str());
  82. onFinish();
  83. updateBucket();
  84. _finished = true;
  85. }
  86. }
  87. void DHTAbstractNodeLookupTask::onTimeout(const SharedHandle<DHTNode>& node)
  88. {
  89. _logger->debug("node lookup message timeout for node ID=%s",
  90. Util::toHex(node->getID(), DHT_ID_LENGTH).c_str());
  91. --_inFlightMessage;
  92. for(std::deque<SharedHandle<DHTNodeLookupEntry> >::iterator i = _entries.begin(); i != _entries.end(); ++i) {
  93. if((*i)->_node == node) {
  94. _entries.erase(i);
  95. break;
  96. }
  97. }
  98. if(needsAdditionalOutgoingMessage()) {
  99. sendMessage();
  100. }
  101. if(_inFlightMessage == 0) {
  102. _logger->debug("Finished node_lookup for node ID %s",
  103. Util::toHex(_targetID, DHT_ID_LENGTH).c_str());
  104. onFinish();
  105. updateBucket();
  106. _finished = true;
  107. }
  108. }
  109. void DHTAbstractNodeLookupTask::sendMessage()
  110. {
  111. for(std::deque<SharedHandle<DHTNodeLookupEntry> >::iterator i = _entries.begin(); i != _entries.end() && _inFlightMessage < ALPHA; ++i) {
  112. if((*i)->_used == false) {
  113. ++_inFlightMessage;
  114. (*i)->_used = true;
  115. SharedHandle<DHTMessage> m = createMessage((*i)->_node);
  116. _dispatcher->addMessageToQueue(m, new DHTMessageCallbackImpl(this));
  117. }
  118. }
  119. }
  120. void DHTAbstractNodeLookupTask::updateBucket()
  121. {
  122. // TODO we have to something here?
  123. }
  124. void DHTAbstractNodeLookupTask::startup()
  125. {
  126. _entries = toEntries(_routingTable->getClosestKNodes(_targetID));
  127. if(_entries.empty()) {
  128. _finished = true;
  129. } else {
  130. // TODO use RTT here
  131. _inFlightMessage = 0;
  132. sendMessage();
  133. if(_inFlightMessage == 0) {
  134. _logger->debug("No message was sent in this lookup stage. Finished.");
  135. _finished = true;
  136. }
  137. }
  138. }
  139. std::deque<SharedHandle<DHTNodeLookupEntry> >
  140. DHTAbstractNodeLookupTask::toEntries(const std::deque<SharedHandle<DHTNode> >& nodes) const
  141. {
  142. std::deque<SharedHandle<DHTNodeLookupEntry> > entries;
  143. for(std::deque<SharedHandle<DHTNode> >::const_iterator i = nodes.begin(); i != nodes.end(); ++i) {
  144. SharedHandle<DHTNodeLookupEntry> e = new DHTNodeLookupEntry(*i);
  145. entries.push_back(e);
  146. }
  147. return entries;
  148. }
  149. } // namespace aria2