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 #ifndef D_DHT_ABSTRACT_NODE_LOOKUP_TASK_H
36 #define D_DHT_ABSTRACT_NODE_LOOKUP_TASK_H
37 
38 #include "DHTAbstractTask.h"
39 
40 #include <cstring>
41 #include <algorithm>
42 #include <deque>
43 #include <vector>
44 
45 #include "DHTConstants.h"
46 #include "DHTNodeLookupEntry.h"
47 #include "DHTRoutingTable.h"
48 #include "DHTMessageDispatcher.h"
49 #include "DHTMessageFactory.h"
50 #include "DHTMessage.h"
51 #include "DHTNode.h"
52 #include "DHTBucket.h"
53 #include "LogFactory.h"
54 #include "Logger.h"
55 #include "util.h"
56 #include "DHTIDCloser.h"
57 #include "a2functional.h"
58 #include "fmt.h"
59 
60 namespace aria2 {
61 
62 class DHTNode;
63 class DHTMessage;
64 
65 template <class ResponseMessage>
66 class DHTAbstractNodeLookupTask : public DHTAbstractTask {
67 private:
68   unsigned char targetID_[DHT_ID_LENGTH];
69 
70   std::deque<std::unique_ptr<DHTNodeLookupEntry>> entries_;
71 
72   size_t inFlightMessage_;
73 
74   template <typename Container>
toEntries(Container & entries,const std::vector<std::shared_ptr<DHTNode>> & nodes)75   void toEntries(Container& entries,
76                  const std::vector<std::shared_ptr<DHTNode>>& nodes) const
77   {
78     for (auto& node : nodes) {
79       entries.push_back(make_unique<DHTNodeLookupEntry>(node));
80     }
81   }
82 
sendMessage()83   void sendMessage()
84   {
85     for (auto i = std::begin(entries_), eoi = std::end(entries_);
86          i != eoi && inFlightMessage_ < ALPHA; ++i) {
87       if ((*i)->used == false) {
88         ++inFlightMessage_;
89         (*i)->used = true;
90         getMessageDispatcher()->addMessageToQueue(createMessage((*i)->node),
91                                                   createCallback());
92       }
93     }
94   }
95 
sendMessageAndCheckFinish()96   void sendMessageAndCheckFinish()
97   {
98     if (needsAdditionalOutgoingMessage()) {
99       sendMessage();
100     }
101     if (inFlightMessage_ == 0) {
102       A2_LOG_DEBUG(fmt("Finished node_lookup for node ID %s",
103                        util::toHex(targetID_, DHT_ID_LENGTH).c_str()));
104       onFinish();
105       updateBucket();
106       setFinished(true);
107     }
108     else {
109       A2_LOG_DEBUG(fmt("%lu in flight message for node ID %s",
110                        static_cast<unsigned long>(inFlightMessage_),
111                        util::toHex(targetID_, DHT_ID_LENGTH).c_str()));
112     }
113   }
114 
updateBucket()115   void updateBucket() {}
116 
117 protected:
getTargetID()118   const unsigned char* getTargetID() const { return targetID_; }
119 
getEntries()120   const std::deque<std::unique_ptr<DHTNodeLookupEntry>>& getEntries() const
121   {
122     return entries_;
123   }
124 
125   virtual void getNodesFromMessage(std::vector<std::shared_ptr<DHTNode>>& nodes,
126                                    const ResponseMessage* message) = 0;
127 
onReceivedInternal(const ResponseMessage * message)128   virtual void onReceivedInternal(const ResponseMessage* message) {}
129 
needsAdditionalOutgoingMessage()130   virtual bool needsAdditionalOutgoingMessage() { return true; }
131 
onFinish()132   virtual void onFinish() {}
133 
134   virtual std::unique_ptr<DHTMessage>
135   createMessage(const std::shared_ptr<DHTNode>& remoteNode) = 0;
136 
137   virtual std::unique_ptr<DHTMessageCallback> createCallback() = 0;
138 
139 public:
DHTAbstractNodeLookupTask(const unsigned char * targetID)140   DHTAbstractNodeLookupTask(const unsigned char* targetID) : inFlightMessage_(0)
141   {
142     memcpy(targetID_, targetID, DHT_ID_LENGTH);
143   }
144 
145   static const size_t ALPHA = 3;
146 
startup()147   virtual void startup() CXX11_OVERRIDE
148   {
149     std::vector<std::shared_ptr<DHTNode>> nodes;
150     getRoutingTable()->getClosestKNodes(nodes, targetID_);
151     entries_.clear();
152     toEntries(entries_, nodes);
153     if (entries_.empty()) {
154       setFinished(true);
155     }
156     else {
157       // TODO use RTT here
158       inFlightMessage_ = 0;
159       sendMessage();
160       if (inFlightMessage_ == 0) {
161         A2_LOG_DEBUG("No message was sent in this lookup stage. Finished.");
162         setFinished(true);
163       }
164     }
165   }
166 
onReceived(const ResponseMessage * message)167   void onReceived(const ResponseMessage* message)
168   {
169     --inFlightMessage_;
170     // Replace old Node ID with new Node ID.
171     for (auto& entry : entries_) {
172       if (entry->node->getIPAddress() ==
173               message->getRemoteNode()->getIPAddress() &&
174           entry->node->getPort() == message->getRemoteNode()->getPort()) {
175         entry->node = message->getRemoteNode();
176       }
177     }
178     onReceivedInternal(message);
179     std::vector<std::shared_ptr<DHTNode>> nodes;
180     getNodesFromMessage(nodes, message);
181     std::vector<std::unique_ptr<DHTNodeLookupEntry>> newEntries;
182     toEntries(newEntries, nodes);
183 
184     size_t count = 0;
185     for (auto& ne : newEntries) {
186       if (memcmp(getLocalNode()->getID(), ne->node->getID(), DHT_ID_LENGTH) !=
187           0) {
188         A2_LOG_DEBUG(fmt("Received nodes: id=%s, ip=%s",
189                          util::toHex(ne->node->getID(), DHT_ID_LENGTH).c_str(),
190                          ne->node->getIPAddress().c_str()));
191         entries_.push_front(std::move(ne));
192         ++count;
193       }
194     }
195     A2_LOG_DEBUG(fmt("%lu node lookup entries added.",
196                      static_cast<unsigned long>(count)));
197     std::stable_sort(std::begin(entries_), std::end(entries_),
198                      DHTIDCloser(targetID_));
199     entries_.erase(
200         std::unique(std::begin(entries_), std::end(entries_),
201                     DerefEqualTo<std::unique_ptr<DHTNodeLookupEntry>>{}),
202         std::end(entries_));
203     A2_LOG_DEBUG(fmt("%lu node lookup entries are unique.",
204                      static_cast<unsigned long>(entries_.size())));
205     if (entries_.size() > DHTBucket::K) {
206       entries_.erase(std::begin(entries_) + DHTBucket::K, std::end(entries_));
207     }
208     sendMessageAndCheckFinish();
209   }
210 
onTimeout(const std::shared_ptr<DHTNode> & node)211   void onTimeout(const std::shared_ptr<DHTNode>& node)
212   {
213     A2_LOG_DEBUG(fmt("node lookup message timeout for node ID=%s",
214                      util::toHex(node->getID(), DHT_ID_LENGTH).c_str()));
215     --inFlightMessage_;
216     for (auto i = std::begin(entries_), eoi = std::end(entries_); i != eoi;
217          ++i) {
218       if (*(*i)->node == *node) {
219         entries_.erase(i);
220         break;
221       }
222     }
223     sendMessageAndCheckFinish();
224   }
225 };
226 
227 } // namespace aria2
228 
229 #endif // D_DHT_ABSTRACT_NODE_LOOKUP_TASK_H
230