github
/
aria2
mirror of https://github.com/aria2/aria2
1
0
Fork 0
Code Issues Releases Wiki Activity

2010-06-12 Tatsuhiro Tsujikawa <t-tujikawa@users.sourceforge.net> 

Made protected member variable private. Added accessor funcs.
	* src/DHTAbstractNodeLookupTask.cc
	* src/DHTAbstractNodeLookupTask.h
	* src/DHTAbstractTask.h
	* src/DHTBucketRefreshTask.cc
	* src/DHTNodeLookupTask.cc
	* src/DHTNodeLookupTask.h
	* src/DHTPeerLookupTask.cc
	* src/DHTPeerLookupTask.h
	* src/DHTPingTask.cc
	* src/DHTPingTask.h
	* src/DHTReplaceNodeTask.cc
	* src/DHTReplaceNodeTask.h
pull/1/head
Tatsuhiro Tsujikawa 2010-06-12 02:54:39 +00:00
parent f7001132bc
commit ded040666e
13 changed files with 161 additions and 79 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

16
ChangeLog
View File

@ -1,3 +1,19 @@
2010-06-12 Tatsuhiro Tsujikawa <t-tujikawa@users.sourceforge.net>
Made protected member variable private. Added accessor funcs.
* src/DHTAbstractNodeLookupTask.cc
* src/DHTAbstractNodeLookupTask.h
* src/DHTAbstractTask.h
* src/DHTBucketRefreshTask.cc
* src/DHTNodeLookupTask.cc
* src/DHTNodeLookupTask.h
* src/DHTPeerLookupTask.cc
* src/DHTPeerLookupTask.h
* src/DHTPingTask.cc
* src/DHTPingTask.h
* src/DHTReplaceNodeTask.cc
* src/DHTReplaceNodeTask.h
2010-06-12 Tatsuhiro Tsujikawa <t-tujikawa@users.sourceforge.net>
Made protected member variable private. Added accessor funcs.

63
src/DHTAbstractNodeLookupTask.cc
View File

@ -51,13 +51,15 @@
namespace aria2 {
DHTAbstractNodeLookupTask::DHTAbstractNodeLookupTask(const unsigned char* targetID):
DHTAbstractNodeLookupTask::DHTAbstractNodeLookupTask
(const unsigned char* targetID):
_inFlightMessage(0)
{
memcpy(_targetID, targetID, DHT_ID_LENGTH);
}
void DHTAbstractNodeLookupTask::onReceived(const SharedHandle<DHTMessage>& message)
void DHTAbstractNodeLookupTask::onReceived
(const SharedHandle<DHTMessage>& message)
{
--_inFlightMessage;
onReceivedInternal(message);
@ -69,23 +71,25 @@ void DHTAbstractNodeLookupTask::onReceived(const SharedHandle<DHTMessage>& messa
size_t count = 0;
for(std::vector<SharedHandle<DHTNodeLookupEntry> >::const_iterator i =
newEntries.begin(), eoi = newEntries.end(); i != eoi; ++i) {
if(memcmp(_localNode->getID(), (*i)->_node->getID(), DHT_ID_LENGTH) != 0) {
if(memcmp(getLocalNode()->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(getLogger()->debug()) {
getLogger()->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);
if(getLogger()->debug()) {
getLogger()->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(getLogger()->debug()) {
getLogger()->debug("%u node lookup entries are unique.", _entries.size());
}
if(_entries.size() > DHTBucket::K) {
_entries.erase(_entries.begin()+DHTBucket::K, _entries.end());
@ -95,9 +99,9 @@ void DHTAbstractNodeLookupTask::onReceived(const SharedHandle<DHTMessage>& messa
void DHTAbstractNodeLookupTask::onTimeout(const SharedHandle<DHTNode>& node)
{
if(_logger->debug()) {
_logger->debug("node lookup message timeout for node ID=%s",
util::toHex(node->getID(), DHT_ID_LENGTH).c_str());
if(getLogger()->debug()) {
getLogger()->debug("node lookup message timeout for node ID=%s",
util::toHex(node->getID(), DHT_ID_LENGTH).c_str());
}
--_inFlightMessage;
for(std::deque<SharedHandle<DHTNodeLookupEntry> >::iterator i =
@ -116,18 +120,18 @@ void DHTAbstractNodeLookupTask::sendMessageAndCheckFinish()
sendMessage();
}
if(_inFlightMessage == 0) {
if(_logger->debug()) {
_logger->debug("Finished node_lookup for node ID %s",
util::toHex(_targetID, DHT_ID_LENGTH).c_str());
if(getLogger()->debug()) {
getLogger()->debug("Finished node_lookup for node ID %s",
util::toHex(_targetID, DHT_ID_LENGTH).c_str());
}
onFinish();
updateBucket();
_finished = true;
setFinished(true);
} else {
if(_logger->debug()) {
_logger->debug("%d in flight message for node ID %s",
_inFlightMessage,
util::toHex(_targetID, DHT_ID_LENGTH).c_str());
if(getLogger()->debug()) {
getLogger()->debug("%d in flight message for node ID %s",
_inFlightMessage,
util::toHex(_targetID, DHT_ID_LENGTH).c_str());
}
}
}
@ -142,8 +146,9 @@ void DHTAbstractNodeLookupTask::sendMessage()
(*i)->_used = true;
SharedHandle<DHTMessage> m = createMessage((*i)->_node);
WeakHandle<DHTMessageCallbackListener> listener(this);
SharedHandle<DHTMessageCallback> callback(new DHTMessageCallbackImpl(listener));
_dispatcher->addMessageToQueue(m, callback);
SharedHandle<DHTMessageCallback> callback
(new DHTMessageCallbackImpl(listener));
getMessageDispatcher()->addMessageToQueue(m, callback);
}
}
}
@ -156,20 +161,20 @@ void DHTAbstractNodeLookupTask::updateBucket()
void DHTAbstractNodeLookupTask::startup()
{
std::vector<SharedHandle<DHTNode> > nodes;
_routingTable->getClosestKNodes(nodes, _targetID);
getRoutingTable()->getClosestKNodes(nodes, _targetID);
_entries.clear();
toEntries(_entries, nodes);
if(_entries.empty()) {
_finished = true;
setFinished(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.");
if(getLogger()->debug()) {
getLogger()->debug("No message was sent in this lookup stage. Finished.");
}
_finished = true;
setFinished(true);
}
}
}

18
src/DHTAbstractNodeLookupTask.h
View File

@ -49,8 +49,9 @@ namespace aria2 {
class DHTNode;
class DHTMessage;
class DHTAbstractNodeLookupTask:public DHTAbstractTask, public DHTMessageCallbackListener {
protected:
class DHTAbstractNodeLookupTask:public DHTAbstractTask,
public DHTMessageCallbackListener {
private:
unsigned char _targetID[DHT_ID_LENGTH];
std::deque<SharedHandle<DHTNodeLookupEntry> > _entries;
@ -73,6 +74,16 @@ protected:
void updateBucket();
void sendMessageAndCheckFinish();
protected:
const unsigned char* getTargetID() const
{
return _targetID;
}
const std::deque<SharedHandle<DHTNodeLookupEntry> >& getEntries() const
{
return _entries;
}
public:
DHTAbstractNodeLookupTask(const unsigned char* targetID);
@ -93,7 +104,8 @@ public:
virtual void onFinish() {}
virtual SharedHandle<DHTMessage> createMessage(const SharedHandle<DHTNode>& remoteNode) = 0;
virtual SharedHandle<DHTMessage> createMessage
(const SharedHandle<DHTNode>& remoteNode) = 0;
};
} // namespace aria2

37
src/DHTAbstractTask.h
View File

@ -51,7 +51,7 @@ class DHTTaskQueue;
class Logger;
class DHTAbstractTask:public DHTTask {
protected:
private:
bool _finished;
Logger* _logger;
@ -65,19 +65,54 @@ protected:
WeakHandle<DHTMessageFactory> _factory;
WeakHandle<DHTTaskQueue> _taskQueue;
protected:
void setFinished(bool f)
{
_finished = f;
}
Logger* getLogger() const
{
return _logger;
}
public:
DHTAbstractTask();
virtual bool finished();
const WeakHandle<DHTRoutingTable>& getRoutingTable() const
{
return _routingTable;
}
void setRoutingTable(const WeakHandle<DHTRoutingTable> routingTable);
const WeakHandle<DHTMessageDispatcher>& getMessageDispatcher() const
{
return _dispatcher;
}
void setMessageDispatcher(const WeakHandle<DHTMessageDispatcher> dispatcher);
const WeakHandle<DHTMessageFactory>& getMessageFactory() const
{
return _factory;
}
void setMessageFactory(const WeakHandle<DHTMessageFactory> factory);
const WeakHandle<DHTTaskQueue>& getTaskQueue() const
{
return _taskQueue;
}
void setTaskQueue(const WeakHandle<DHTTaskQueue> taskQueue);
const SharedHandle<DHTNode>& getLocalNode() const
{
return _localNode;
}
void setLocalNode(const SharedHandle<DHTNode>& localNode);
};

19
src/DHTBucketRefreshTask.cc
View File

@ -52,7 +52,7 @@ DHTBucketRefreshTask::~DHTBucketRefreshTask() {}
void DHTBucketRefreshTask::startup()
{
std::vector<SharedHandle<DHTBucket> > buckets;
_routingTable->getBuckets(buckets);
getRoutingTable()->getBuckets(buckets);
for(std::vector<SharedHandle<DHTBucket> >::iterator i = buckets.begin(),
eoi = buckets.end(); i != eoi; ++i) {
if(_forceRefresh || (*i)->needsRefresh()) {
@ -60,17 +60,18 @@ void DHTBucketRefreshTask::startup()
unsigned char targetID[DHT_ID_LENGTH];
(*i)->getRandomNodeID(targetID);
SharedHandle<DHTNodeLookupTask> task(new DHTNodeLookupTask(targetID));
task->setRoutingTable(_routingTable);
task->setMessageDispatcher(_dispatcher);
task->setMessageFactory(_factory);
task->setTaskQueue(_taskQueue);
task->setLocalNode(_localNode);
task->setRoutingTable(getRoutingTable());
task->setMessageDispatcher(getMessageDispatcher());
task->setMessageFactory(getMessageFactory());
task->setTaskQueue(getTaskQueue());
task->setLocalNode(getLocalNode());
_logger->info("Dispating bucket refresh. targetID=%s", util::toHex(targetID, DHT_ID_LENGTH).c_str());
_taskQueue->addPeriodicTask1(task);
getLogger()->info("Dispating bucket refresh. targetID=%s",
util::toHex(targetID, DHT_ID_LENGTH).c_str());
getTaskQueue()->addPeriodicTask1(task);
}
}
_finished = true;
setFinished(true);
}
void DHTBucketRefreshTask::setForceRefresh(bool forceRefresh)

2
src/DHTNodeLookupTask.cc
View File

@ -62,7 +62,7 @@ DHTNodeLookupTask::getNodesFromMessage
SharedHandle<DHTMessage>
DHTNodeLookupTask::createMessage(const SharedHandle<DHTNode>& remoteNode)
{
return _factory->createFindNodeMessage(remoteNode, _targetID);
return getMessageFactory()->createFindNodeMessage(remoteNode, getTargetID());
}
} // namespace aria2

3
src/DHTNodeLookupTask.h
View File

@ -46,7 +46,8 @@ public:
virtual void getNodesFromMessage(std::vector<SharedHandle<DHTNode> >& nodes,
const SharedHandle<DHTMessage>& message);
virtual SharedHandle<DHTMessage> createMessage(const SharedHandle<DHTNode>& remoteNode);
virtual SharedHandle<DHTMessage> createMessage
(const SharedHandle<DHTNode>& remoteNode);
};
} // namespace aria2

24
src/DHTPeerLookupTask.cc
View File

@ -66,23 +66,25 @@ DHTPeerLookupTask::getNodesFromMessage
}
}
void DHTPeerLookupTask::onReceivedInternal(const SharedHandle<DHTMessage>& message)
void DHTPeerLookupTask::onReceivedInternal
(const SharedHandle<DHTMessage>& message)
{
SharedHandle<DHTGetPeersReplyMessage> m(dynamic_pointer_cast<DHTGetPeersReplyMessage>(message));
SharedHandle<DHTGetPeersReplyMessage> m
(dynamic_pointer_cast<DHTGetPeersReplyMessage>(message));
if(m.isNull()) {
return;
}
SharedHandle<DHTNode> remoteNode = m->getRemoteNode();
_tokenStorage[util::toHex(remoteNode->getID(), DHT_ID_LENGTH)] = m->getToken();
_tokenStorage[util::toHex(remoteNode->getID(), DHT_ID_LENGTH)] =
m->getToken();
_peerStorage->addPeer(m->getValues());
_peers.insert(_peers.end(), m->getValues().begin(), m->getValues().end());
_logger->info("Received %u peers.", m->getValues().size());
getLogger()->info("Received %u peers.", m->getValues().size());
}
SharedHandle<DHTMessage> DHTPeerLookupTask::createMessage(const SharedHandle<DHTNode>& remoteNode)
{
return _factory->createGetPeersMessage(remoteNode, _targetID);
return getMessageFactory()->createGetPeersMessage(remoteNode, getTargetID());
}
void DHTPeerLookupTask::onFinish()
@ -90,17 +92,17 @@ void DHTPeerLookupTask::onFinish()
// send announce_peer message to K closest nodes
size_t num = DHTBucket::K;
for(std::deque<SharedHandle<DHTNodeLookupEntry> >::const_iterator i =
_entries.begin(), eoi = _entries.end(); i != eoi && num > 0; ++i,
--num) {
getEntries().begin(), eoi = getEntries().end();
i != eoi && num > 0; ++i, --num) {
if((*i)->_used) {
const SharedHandle<DHTNode>& node = (*i)->_node;
SharedHandle<DHTMessage> m =
_factory->createAnnouncePeerMessage
getMessageFactory()->createAnnouncePeerMessage
(node,
_targetID, // this is infoHash
getTargetID(), // this is infoHash
_btRuntime->getListenPort(),
_tokenStorage[util::toHex(node->getID(), DHT_ID_LENGTH)]);
_dispatcher->addMessageToQueue(m);
getMessageDispatcher()->addMessageToQueue(m);
}
}
}

3
src/DHTPeerLookupTask.h
View File

@ -62,7 +62,8 @@ public:
virtual void onReceivedInternal(const SharedHandle<DHTMessage>& message);
virtual SharedHandle<DHTMessage> createMessage(const SharedHandle<DHTNode>& remoteNode);
virtual SharedHandle<DHTMessage> createMessage
(const SharedHandle<DHTNode>& remoteNode);
virtual void onFinish();

23
src/DHTPingTask.cc
View File

@ -42,7 +42,8 @@
namespace aria2 {
DHTPingTask::DHTPingTask(const SharedHandle<DHTNode>& remoteNode, size_t numMaxRetry):
DHTPingTask::DHTPingTask
(const SharedHandle<DHTNode>& remoteNode, size_t numMaxRetry):
_remoteNode(remoteNode),
_numMaxRetry(numMaxRetry),
_numRetry(0),
@ -54,16 +55,18 @@ DHTPingTask::~DHTPingTask() {}
void DHTPingTask::startup()
{
SharedHandle<DHTMessage> m = _factory->createPingMessage(_remoteNode);
SharedHandle<DHTMessage> m =
getMessageFactory()->createPingMessage(_remoteNode);
WeakHandle<DHTMessageCallbackListener> listener(this);
SharedHandle<DHTMessageCallback> callback(new DHTMessageCallbackImpl(listener));
_dispatcher->addMessageToQueue(m, _timeout, callback);
SharedHandle<DHTMessageCallback> callback
(new DHTMessageCallbackImpl(listener));
getMessageDispatcher()->addMessageToQueue(m, _timeout, callback);
}
void DHTPingTask::onReceived(const SharedHandle<DHTMessage>& message)
{
_pingSuccessful = true;
_finished = true;
setFinished(true);
}
void DHTPingTask::onTimeout(const SharedHandle<DHTNode>& node)
@ -71,12 +74,14 @@ void DHTPingTask::onTimeout(const SharedHandle<DHTNode>& node)
++_numRetry;
if(_numRetry >= _numMaxRetry) {
_pingSuccessful = false;
_finished = true;
setFinished(true);
} else {
SharedHandle<DHTMessage> m = _factory->createPingMessage(_remoteNode);
SharedHandle<DHTMessage> m =
getMessageFactory()->createPingMessage(_remoteNode);
WeakHandle<DHTMessageCallbackListener> listener(this);
SharedHandle<DHTMessageCallback> callback(new DHTMessageCallbackImpl(listener));
_dispatcher->addMessageToQueue(m, _timeout, callback);
SharedHandle<DHTMessageCallback> callback
(new DHTMessageCallbackImpl(listener));
getMessageDispatcher()->addMessageToQueue(m, _timeout, callback);
}
}

2
src/DHTPingTask.h
View File

@ -42,7 +42,7 @@
namespace aria2 {
class DHTPingTask:public DHTAbstractTask, public DHTMessageCallbackListener {
protected:
private:
SharedHandle<DHTNode> _remoteNode;
size_t _numMaxRetry;

27
src/DHTReplaceNodeTask.cc
View File

@ -62,34 +62,37 @@ void DHTReplaceNodeTask::sendMessage()
{
SharedHandle<DHTNode> questionableNode = _bucket->getLRUQuestionableNode();
if(questionableNode.isNull()) {
_finished = true;
setFinished(true);
} else {
SharedHandle<DHTMessage> m = _factory->createPingMessage(questionableNode);
SharedHandle<DHTMessage> m =
getMessageFactory()->createPingMessage(questionableNode);
WeakHandle<DHTMessageCallbackListener> listener(this);
SharedHandle<DHTMessageCallback> callback(new DHTMessageCallbackImpl(listener));
_dispatcher->addMessageToQueue(m, _timeout, callback);
SharedHandle<DHTMessageCallback> callback
(new DHTMessageCallbackImpl(listener));
getMessageDispatcher()->addMessageToQueue(m, _timeout, callback);
}
}
void DHTReplaceNodeTask::onReceived(const SharedHandle<DHTMessage>& message)
{
_logger->info("ReplaceNode: Ping reply received from %s.",
message->getRemoteNode()->toString().c_str());
_finished = true;
getLogger()->info("ReplaceNode: Ping reply received from %s.",
message->getRemoteNode()->toString().c_str());
setFinished(true);
}
void DHTReplaceNodeTask::onTimeout(const SharedHandle<DHTNode>& node)
{
++_numRetry;
if(_numRetry >= MAX_RETRY) {
_logger->info("ReplaceNode: Ping failed %u times. Replace %s with %s.",
_numRetry, node->toString().c_str(), _newNode->toString().c_str());
getLogger()->info("ReplaceNode: Ping failed %u times. Replace %s with %s.",
_numRetry, node->toString().c_str(),
_newNode->toString().c_str());
node->markBad();
_bucket->addNode(_newNode);
_finished = true;
setFinished(true);
} else {
_logger->info("ReplaceNode: Ping reply timeout from %s. Try once more.",
node->toString().c_str());
getLogger()->info("ReplaceNode: Ping reply timeout from %s. Try once more.",
node->toString().c_str());
sendMessage();
}
}

3
src/DHTReplaceNodeTask.h
View File

@ -43,7 +43,8 @@ namespace aria2 {
class DHTBucket;
class DHTReplaceNodeTask:public DHTAbstractTask, public DHTMessageCallbackListener {
class DHTReplaceNodeTask:public DHTAbstractTask,
public DHTMessageCallbackListener {
private:
SharedHandle<DHTBucket> _bucket;