/* */ #include "DHTAbstractNodeLookupTask.h" #include #include #include "DHTRoutingTable.h" #include "DHTMessageDispatcher.h" #include "DHTMessageFactory.h" #include "DHTMessage.h" #include "DHTNode.h" #include "DHTMessageCallbackImpl.h" #include "DHTBucket.h" #include "LogFactory.h" #include "Logger.h" #include "util.h" #include "DHTIDCloser.h" namespace aria2 { DHTAbstractNodeLookupTask::DHTAbstractNodeLookupTask(const unsigned char* targetID): _inFlightMessage(0) { memcpy(_targetID, targetID, DHT_ID_LENGTH); } void DHTAbstractNodeLookupTask::onReceived(const SharedHandle& message) { --_inFlightMessage; onReceivedInternal(message); std::vector > nodes; getNodesFromMessage(nodes, message); std::vector > newEntries; toEntries(newEntries, nodes); size_t count = 0; for(std::vector >::const_iterator i = newEntries.begin(), eoi = newEntries.end(); i != eoi; ++i) { if(memcmp(_localNode->getID(), (*i)->_node->getID(), DHT_ID_LENGTH) != 0) { _entries.push_front(*i); ++count; if(_logger->debug()) { _logger->debug("Received nodes: id=%s, ip=%s", util::toHex((*i)->_node->getID(), DHT_ID_LENGTH).c_str(), (*i)->_node->getIPAddress().c_str()); } } } if(_logger->debug()) { _logger->debug("%u node lookup entries added.", count); } std::stable_sort(_entries.begin(), _entries.end(), DHTIDCloser(_targetID)); _entries.erase(std::unique(_entries.begin(), _entries.end()), _entries.end()); if(_logger->debug()) { _logger->debug("%u node lookup entries are unique.", _entries.size()); } if(_entries.size() > DHTBucket::K) { _entries.erase(_entries.begin()+DHTBucket::K, _entries.end()); } sendMessageAndCheckFinish(); } void DHTAbstractNodeLookupTask::onTimeout(const SharedHandle& node) { if(_logger->debug()) { _logger->debug("node lookup message timeout for node ID=%s", util::toHex(node->getID(), DHT_ID_LENGTH).c_str()); } --_inFlightMessage; for(std::deque >::iterator i = _entries.begin(), eoi = _entries.end(); i != eoi; ++i) { if((*i)->_node == node) { _entries.erase(i); break; } } sendMessageAndCheckFinish(); } void DHTAbstractNodeLookupTask::sendMessageAndCheckFinish() { if(needsAdditionalOutgoingMessage()) { sendMessage(); } if(_inFlightMessage == 0) { if(_logger->debug()) { _logger->debug("Finished node_lookup for node ID %s", util::toHex(_targetID, DHT_ID_LENGTH).c_str()); } onFinish(); updateBucket(); _finished = true; } else { if(_logger->debug()) { _logger->debug("%d in flight message for node ID %s", _inFlightMessage, util::toHex(_targetID, DHT_ID_LENGTH).c_str()); } } } void DHTAbstractNodeLookupTask::sendMessage() { for(std::deque >::iterator i = _entries.begin(), eoi = _entries.end(); i != eoi && _inFlightMessage < ALPHA; ++i) { if((*i)->_used == false) { ++_inFlightMessage; (*i)->_used = true; SharedHandle m = createMessage((*i)->_node); WeakHandle listener(this); SharedHandle callback(new DHTMessageCallbackImpl(listener)); _dispatcher->addMessageToQueue(m, callback); } } } void DHTAbstractNodeLookupTask::updateBucket() { // TODO we have to something here? } void DHTAbstractNodeLookupTask::startup() { std::vector > nodes; _routingTable->getClosestKNodes(nodes, _targetID); _entries.clear(); toEntries(_entries, nodes); if(_entries.empty()) { _finished = true; } else { // TODO use RTT here _inFlightMessage = 0; sendMessage(); if(_inFlightMessage == 0) { if(_logger->debug()) { _logger->debug("No message was sent in this lookup stage. Finished."); } _finished = true; } } } } // namespace aria2