RouterModule.c 28 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706
  1. /* vim: set expandtab ts=4 sw=4: */
  2. /*
  3. * You may redistribute this program and/or modify it under the terms of
  4. * the GNU General Public License as published by the Free Software Foundation,
  5. * either version 3 of the License, or (at your option) any later version.
  6. *
  7. * This program is distributed in the hope that it will be useful,
  8. * but WITHOUT ANY WARRANTY; without even the implied warranty of
  9. * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
  10. * GNU General Public License for more details.
  11. *
  12. * You should have received a copy of the GNU General Public License
  13. * along with this program. If not, see <http://www.gnu.org/licenses/>.
  14. */
  15. #include "benc/String.h"
  16. #include "dht/Address.h"
  17. #include "dht/dhtcore/RouterModule.h"
  18. #include "dht/dhtcore/RouterModule_pvt.h"
  19. #include "dht/dhtcore/Node.h"
  20. #include "dht/dhtcore/NodeList.h"
  21. #include "dht/dhtcore/NodeStore.h"
  22. #include "dht/dhtcore/VersionList.h"
  23. #include "dht/CJDHTConstants.h"
  24. #include "dht/DHTMessage.h"
  25. #include "dht/DHTModule.h"
  26. #include "dht/DHTModuleRegistry.h"
  27. #include "util/log/Log.h"
  28. #include "memory/Allocator.h"
  29. #include "switch/LabelSplicer.h"
  30. #include "switch/NumberCompress.h"
  31. #include "util/events/EventBase.h"
  32. #include "util/AverageRoller.h"
  33. #include "util/Bits.h"
  34. #include "util/Hex.h"
  35. #include "util/Endian.h"
  36. #include "util/Pinger.h"
  37. #include "util/events/Time.h"
  38. #include "util/events/Timeout.h"
  39. #include "util/version/Version.h"
  40. #include "wire/Message.h"
  41. /*
  42. * The router module is the central part of the DHT engine.
  43. * It's job is to maintain a routing table which is updated by all incoming packets.
  44. * When it gets an incoming query, its job is to add nodes to the reply so that the asking node
  45. * can find other nodes which are closer to its target than us.
  46. *
  47. * This implementation does not split nodes explicitly into buckets not does it explicitly try to
  48. * distinguish between "good" and "bad" nodes. Instead it tries to determine which node will help
  49. * get to the requested record the fastest. Instead of periodicly pinging a random node in each
  50. * "bucket", this implementation periodically searches for a random[1] hash. When a node is sent a
  51. * query, the the distance[2] between it and the first node is divided by the amount of time it
  52. * takes the node to respond, for each successful search, this number is added to an attribute of
  53. * the node called "reach".
  54. *
  55. * Visually representing a node as an area whose location is defined by the node id and its size is
  56. * defined by the node reach, you can see that there is a possibility for a record to be closer in
  57. * key space to node2 while is is still further inside of node1's reach thus node1 is a better
  58. * choice for the next node to ask.
  59. *
  60. * |<--------- Node 1 ---------->|
  61. * |<--- Node 2 ---->|
  62. * ^----- Desired record location.
  63. *
  64. * New nodes are inserted into the table but with a reach of 0. It is up to the search client to
  65. * send search requests to them so they can prove their validity and have their reach number
  66. * updated.
  67. *
  68. * Reach of a node is incremented by 2 every time the node responds to a query and incremented by 1
  69. * every time a node sends a query of its own. This has almost no effect except that it means a
  70. * node which has recently sent data will be preferred over one which has not.
  71. *
  72. * When a search is carried out, the next K returned nodes are not necessarily the closest known
  73. * nodes to the id of the record. The nodes returned will be the nodes with the lowest
  74. * distance:reach ratio. The distance:reach ratio is calculated by dividing the distance between
  75. * the node and the record by the node's reach number. Actually it is done by multiplying
  76. * UINT32_MAX minus the distance by the reach so that it does not need to use slower divison.
  77. * See: NodeCollector.h
  78. *
  79. * Since information about a node becomes stale over time, all reach numbers are decreased by
  80. * the constant REACH_DECREASE_PER_SECOND times the number of seconds since the last cycle,
  81. * this operation is performed periodicly every LOCAL_MAINTENANCE_SEARCH_MILLISECONDS unless
  82. * a local maintainence search is being run which is not often once the network is stable.
  83. *
  84. * TODO ---
  85. * In order to have the nodes with least distance:reach ratio ready to handle any incoming search,
  86. * we precompute the borders where the "best next node" changes. This computation is best understood
  87. * by graphing the nodes with their location in keyspace on the X axis and their reach on the Y
  88. * axis. The border between two nodes, nodeA and nodeB is the location where a line drawn from the
  89. * X axis up to either node location would be the same angle.
  90. *
  91. * ^ ^
  92. * | nodeA | nodeA
  93. * | |\ | |\__
  94. * | | \ | | \__
  95. * | | \ nodeB | | \nodeB
  96. * | | \ /| | | \__
  97. * | | \ / | | | | \__
  98. * | | \/ | | | | \__
  99. * +---------------------------------------> +--------------------------------------->
  100. * ^-- border ^-- border2
  101. *
  102. * Everything to the left of the border and everything to the right of border2 is to be serviced by
  103. * nodeA. Everything between the two borders is serviced by nodeB. Border2 is found by
  104. * drawing a line from the point given for nodeA to through the point given for nodeB and finding
  105. * the intersection of that line with the Y axis. border and border2 are shown on different graphs
  106. * only to limit clutter, they are the same nodeA and nodeB.
  107. *
  108. * When resolving a search, this implementation will lookup the location of the searched for record
  109. * and return the nodes which belong to the insides of the nearest K borders, this guarantees return
  110. * of the nodes whose distance:reach ratio is the lowest for that location.
  111. * ---
  112. *
  113. * This implementation must never respond to a search by sending any node who's id is not closer
  114. * to the target than its own. Such an event would lead to the possibility of "routing loops" and
  115. * must be prevented. Searches for which this node has the lowest distance:reach ratio will be
  116. * replied to with nodes which have 0 reach but are closer than this node or, if there are no such
  117. * nodes, no nodes at all.
  118. *
  119. * The search consumer in this routing module tries to minimize the amount of traffic sent when
  120. * doing a lookup. To achieve this, it sends a request only to the last node in the search response
  121. * packet, after the global mean response time has passed without it getting a response, it sends
  122. * requests to the second to last and so forth, working backward. Nodes which fail to respond in
  123. * time have their reach immedietly set to zero.
  124. *
  125. * The global mean response time is the average amount of time it takes a node to respond to a
  126. * search query. It is a rolling average over the past 256 seconds.
  127. *
  128. * To maximize the quality of service offered by this node this implementation will repeat
  129. * searches which it handles every number of seconds given by the constant:
  130. * GLOBAL_MAINTENANCE_SEARCH_MILLISECONDS.
  131. * Any incoming search with a get_peers request is eligable to be repeated.
  132. *
  133. * [1] The implementation runs periodic searches for random hashes but unless the search target
  134. * is closer in keyspace to this node than it is to any node with non-zero reach, the search
  135. * is not performed. This means that the node will send out lots of searches when it doesn't
  136. * know many reliable nodes but it will taper off like a governer as it becomes more
  137. * integrated in the network. These searches are run every number of milliseconds given
  138. * by the constant LOCAL_MAINTENANCE_SEARCH_MILLISECONDS.
  139. *
  140. * [2] If a response "overshoots" the record requested then it is calculated as if it had undershot
  141. * by the same amount so as not to provide arbitrage advantage to nodes who return results which
  142. * are very far away yet very inaccurate. If it overshoots by more than the distance between the
  143. * node and the searched for location (this should never happen), it is considered to be 0.
  144. */
  145. /*--------------------Constants--------------------*/
  146. /** The number of seconds of time overwhich to calculate the global mean response time. */
  147. #define GMRT_SECONDS 256
  148. /**
  149. * The number to initialize the global mean response time averager with so that it will
  150. * return sane results early on, this number can be much higher than the expected average.
  151. */
  152. #define GMRT_INITAL_MILLISECONDS 5000
  153. /** The number of nodes which we will keep track of. */
  154. #define NODE_STORE_SIZE 8192
  155. /** The number of milliseconds between attempting local maintenance searches. */
  156. #define LOCAL_MAINTENANCE_SEARCH_MILLISECONDS 1000
  157. /**
  158. * The number of milliseconds to pass between global maintainence searches.
  159. * These are searches for random targets which are used to discover new nodes.
  160. */
  161. #define GLOBAL_MAINTENANCE_SEARCH_MILLISECONDS 30000
  162. #define SEARCH_REPEAT_MILLISECONDS 7500
  163. /** The number of times the GMRT before pings should be timed out. */
  164. #define PING_TIMEOUT_GMRT_MULTIPLIER 100
  165. /** The minimum amount of time before a ping should timeout. */
  166. #define PING_TIMEOUT_MINIMUM 3000
  167. /** You are not expected to understand this. */
  168. #define LINK_STATE_MULTIPLIER 536870
  169. /** All searches will be killed after this amount of time nomatter how bad the GMRT is. */
  170. #define MAX_TIMEOUT 10000
  171. /** Never allow a search to be timed out in less than this number of milliseconds. */
  172. #define MIN_TIMEOUT 10
  173. /**
  174. * Used to keep reach a weighted rolling average of recent ping/search times.
  175. * The smaller this value, the more significant recent pings/searches are to reach.
  176. */
  177. #define REACH_WINDOW 8
  178. /*--------------------Prototypes--------------------*/
  179. static int handleIncoming(struct DHTMessage* message, void* vcontext);
  180. static int handleOutgoing(struct DHTMessage* message, void* vcontext);
  181. /*--------------------Interface--------------------*/
  182. /**
  183. * Register a new RouterModule.
  184. *
  185. * @param registry the DHT module registry for signal handling.
  186. * @param allocator a means to allocate memory.
  187. * @param myAddress the address for this DHT node.
  188. * @param nodeStore the place to put the nodes
  189. * @return the RouterModule.
  190. */
  191. struct RouterModule* RouterModule_register(struct DHTModuleRegistry* registry,
  192. struct Allocator* allocator,
  193. const uint8_t myAddress[Address_KEY_SIZE],
  194. struct EventBase* eventBase,
  195. struct Log* logger,
  196. struct Random* rand,
  197. struct NodeStore* nodeStore)
  198. {
  199. struct RouterModule* const out = Allocator_calloc(allocator, sizeof(struct RouterModule), 1);
  200. struct DHTModule* dm = Allocator_clone(allocator, (&(struct DHTModule) {
  201. .name = "RouterModule",
  202. .context = out,
  203. .handleIncoming = handleIncoming,
  204. .handleOutgoing = handleOutgoing
  205. }));
  206. DHTModuleRegistry_register(dm, registry);
  207. Address_forKey(&out->address, myAddress);
  208. out->gmrtRoller = AverageRoller_new(GMRT_SECONDS, eventBase, allocator);
  209. AverageRoller_update(out->gmrtRoller, GMRT_INITAL_MILLISECONDS);
  210. out->nodeStore = nodeStore;
  211. out->registry = registry;
  212. out->eventBase = eventBase;
  213. out->logger = logger;
  214. out->allocator = allocator;
  215. out->rand = rand;
  216. out->pinger = Pinger_new(eventBase, rand, logger, allocator);
  217. out->encodingScheme = NumberCompress_defineScheme(allocator);
  218. Identity_set(out);
  219. return out;
  220. }
  221. /**
  222. * The amount of time to wait before skipping over the first node and trying another in a search.
  223. * Any node which can't beat this time will have its reach set to 0.
  224. *
  225. * @param module this module.
  226. * @return the timeout time.
  227. */
  228. uint64_t RouterModule_searchTimeoutMilliseconds(struct RouterModule* module)
  229. {
  230. uint64_t x = (((uint64_t) AverageRoller_getAverage(module->gmrtRoller)) * 4);
  231. x = x + (Random_uint32(module->rand) % (x | 1)) / 2;
  232. return (x > MAX_TIMEOUT) ? MAX_TIMEOUT : (x < MIN_TIMEOUT) ? MIN_TIMEOUT : x;
  233. }
  234. static uint32_t reachAfterDecay(const uint32_t oldReach)
  235. {
  236. return (oldReach - (oldReach / REACH_WINDOW));
  237. }
  238. static uint32_t reachAfterTimeout(const uint32_t oldReach)
  239. {
  240. return (oldReach / 2);
  241. }
  242. static uint32_t nextReach(const uint32_t oldReach, const uint32_t millisecondsLag)
  243. {
  244. int64_t out = reachAfterDecay(millisecondsLag) +
  245. ((UINT32_MAX / REACH_WINDOW) / millisecondsLag);
  246. // TODO: is this safe?
  247. Assert_true(out < (UINT32_MAX - 1024) && out > 0);
  248. return out;
  249. }
  250. static inline int sendNodes(struct NodeList* nodeList,
  251. struct DHTMessage* message,
  252. struct RouterModule* module,
  253. uint32_t askerVersion)
  254. {
  255. struct DHTMessage* query = message->replyTo;
  256. String* nodes =
  257. String_newBinary(NULL, nodeList->size * Address_SERIALIZED_SIZE, message->allocator);
  258. struct VersionList* versions = VersionList_new(nodeList->size, message->allocator);
  259. int i = 0;
  260. for (; i < (int)nodeList->size; i++) {
  261. // We have to modify the reply in case this node uses a longer label discriminator
  262. // in our switch than its target address, the target address *must* have the same
  263. // length or longer.
  264. struct Address addr;
  265. Bits_memcpyConst(&addr, &nodeList->nodes[i]->address, sizeof(struct Address));
  266. addr.path = LabelSplicer_getLabelFor(addr.path, query->address->path);
  267. #ifdef PARANOIA
  268. int encIdx = EncodingScheme_getFormNum(module->encodingScheme, query->address->path);
  269. Assert_true(encIdx != EncodingScheme_getFormNum_INVALID);
  270. int resEncIdx = EncodingScheme_getFormNum(module->encodingScheme,
  271. nodeList->nodes[i]->address.path);
  272. if (nodeList->nodes[i]->address.path != 1 && resEncIdx < encIdx) {
  273. uint64_t converted = EncodingScheme_convertLabel(module->encodingScheme,
  274. nodeList->nodes[i]->address.path,
  275. encIdx);
  276. if (converted == UINT64_MAX && (addr.path >> 59) != 0) {
  277. } else {
  278. Assert_true(converted == addr.path);
  279. }
  280. }
  281. #endif
  282. Address_serialize(&nodes->bytes[i * Address_SERIALIZED_SIZE], &addr);
  283. versions->versions[i] = nodeList->nodes[i]->address.protocolVersion;
  284. Assert_true(!Bits_isZero(&nodes->bytes[i * Address_SERIALIZED_SIZE],
  285. Address_SERIALIZED_SIZE));
  286. }
  287. nodes->len = i * Address_SERIALIZED_SIZE;
  288. versions->length = i;
  289. if (i > 0) {
  290. Dict_putString(message->asDict, CJDHTConstants_NODES, nodes, message->allocator);
  291. String* versionsStr = VersionList_stringify(versions, message->allocator);
  292. Dict_putString(message->asDict,
  293. CJDHTConstants_NODE_PROTOCOLS,
  294. versionsStr,
  295. message->allocator);
  296. }
  297. return 0;
  298. }
  299. /**
  300. * Handle an incoming search query.
  301. * This is setup to handle the outgoing *response* to the query, it should
  302. * be called from handleOutgoing() and populate the response with nodes.
  303. *
  304. * @param message the empty response message to populate.
  305. * @param replyArgs the arguments dictionary in the response (to be populated).
  306. * @param module the routing module context.
  307. * @return 0 as long as the packet should not be stopped (at this point always 0).
  308. */
  309. static inline int handleQuery(struct DHTMessage* message,
  310. struct RouterModule* module)
  311. {
  312. struct DHTMessage* query = message->replyTo;
  313. int64_t* versionPtr = Dict_getInt(query->asDict, CJDHTConstants_PROTOCOL);
  314. uint32_t version = (versionPtr && *versionPtr <= UINT32_MAX) ? *versionPtr : 0;
  315. struct NodeList* nodeList = NULL;
  316. String* queryType = Dict_getString(query->asDict, CJDHTConstants_QUERY);
  317. if (String_equals(queryType, CJDHTConstants_QUERY_FN)) {
  318. // get the target
  319. String* target = Dict_getString(query->asDict, CJDHTConstants_TARGET);
  320. if (target == NULL || target->len != Address_SEARCH_TARGET_SIZE) {
  321. return 0;
  322. }
  323. struct Address targetAddr = { .path = 0 };
  324. Bits_memcpyConst(targetAddr.ip6.bytes, target->bytes, Address_SEARCH_TARGET_SIZE);
  325. // send the closest nodes
  326. nodeList = NodeStore_getClosestNodes(module->nodeStore,
  327. &targetAddr,
  328. RouterModule_K,
  329. version,
  330. message->allocator);
  331. } else if (String_equals(queryType, CJDHTConstants_QUERY_GP)) {
  332. // get the target
  333. String* target = Dict_getString(query->asDict, CJDHTConstants_TARGET);
  334. if (target == NULL || target->len != 8) {
  335. return 0;
  336. }
  337. uint64_t targetPath;
  338. Bits_memcpyConst(&targetPath, target->bytes, 8);
  339. targetPath = Endian_bigEndianToHost64(targetPath);
  340. nodeList =
  341. NodeStore_getPeers(targetPath, RouterModule_K, message->allocator, module->nodeStore);
  342. }
  343. return (nodeList) ? sendNodes(nodeList, message, module, version) : 0;
  344. }
  345. /**
  346. * We handle 2 kinds of packets on the outgoing.
  347. * 1. our requests
  348. * 2. our replies to others' requests.
  349. * Everything is tagged with our address, replies to requests which are not ping requests
  350. * will also be given a list of nodes.
  351. */
  352. static int handleOutgoing(struct DHTMessage* message, void* vcontext)
  353. {
  354. struct RouterModule* module = Identity_check((struct RouterModule*) vcontext);
  355. Dict_putInt(message->asDict,
  356. CJDHTConstants_PROTOCOL,
  357. Version_CURRENT_PROTOCOL,
  358. message->allocator);
  359. if (message->replyTo != NULL) {
  360. return handleQuery(message, module);
  361. }
  362. return 0;
  363. }
  364. struct PingContext
  365. {
  366. struct RouterModule_Promise pub;
  367. /** nonNull if this ping is part of a search. */
  368. struct SearchContext* search;
  369. struct RouterModule* router;
  370. struct Address address;
  371. /** The internal ping structure */
  372. struct Pinger_Ping* pp;
  373. /** A template of the message to be sent. */
  374. Dict* messageDict;
  375. Identity
  376. };
  377. static void sendMsg(String* txid, void* vpingContext)
  378. {
  379. struct PingContext* pc = Identity_check((struct PingContext*) vpingContext);
  380. // "t":"1234"
  381. Dict_putString(pc->messageDict, CJDHTConstants_TXID, txid, pc->pp->pingAlloc);
  382. struct Allocator* temp = Allocator_child(pc->pp->pingAlloc);
  383. struct Message* msg = Message_new(0, DHTMessage_MAX_SIZE + 512, temp);
  384. struct DHTMessage* dmesg = Allocator_calloc(temp, sizeof(struct DHTMessage), 1);
  385. dmesg->binMessage = msg;
  386. dmesg->address = &pc->address;
  387. dmesg->asDict = pc->messageDict;
  388. dmesg->allocator = temp;
  389. DHTModuleRegistry_handleOutgoing(dmesg, pc->router->registry);
  390. }
  391. static void onTimeout(uint32_t milliseconds, struct PingContext* pctx)
  392. {
  393. struct Node_Two* n = NodeStore_closestNode(pctx->router->nodeStore, pctx->address.path);
  394. // Ping timeout -> decrease reach
  395. if (n && !Bits_memcmp(pctx->address.key, n->address.key, 32)) {
  396. uint32_t newReach = reachAfterTimeout(n->pathQuality);
  397. #ifdef Log_DEBUG
  398. uint8_t addr[60];
  399. Address_print(addr, &n->address);
  400. Log_debug(pctx->router->logger,
  401. "Ping timeout for %s, after %lums. changing reach from %u to %u\n",
  402. addr,
  403. (unsigned long)milliseconds,
  404. n->pathQuality,
  405. (unsigned int)newReach);
  406. #endif
  407. NodeStore_updateReach(pctx->router->nodeStore, n, newReach);
  408. }
  409. if (pctx->pub.callback) {
  410. pctx->pub.callback(&pctx->pub, milliseconds, NULL, NULL);
  411. }
  412. }
  413. static uint64_t pingTimeoutMilliseconds(struct RouterModule* module)
  414. {
  415. uint64_t out = AverageRoller_getAverage(module->gmrtRoller) * PING_TIMEOUT_GMRT_MULTIPLIER;
  416. return (out < PING_TIMEOUT_MINIMUM) ? PING_TIMEOUT_MINIMUM : out;
  417. }
  418. /**
  419. * The only type of message we handle on the incoming side is
  420. * a response to one of our queries.
  421. */
  422. static int handleIncoming(struct DHTMessage* message, void* vcontext)
  423. {
  424. String* txid = Dict_getString(message->asDict, CJDHTConstants_TXID);
  425. String* query = Dict_getString(message->asDict, CJDHTConstants_QUERY);
  426. if (query || !txid) {
  427. return 0;
  428. }
  429. struct RouterModule* module = Identity_check((struct RouterModule*) vcontext);
  430. // This is retreived below by onResponseOrTimeout()
  431. module->currentMessage = message;
  432. Pinger_pongReceived(txid, module->pinger);
  433. module->currentMessage = NULL;
  434. return 0;
  435. }
  436. // ping or search response came in
  437. static void onResponseOrTimeout(String* data, uint32_t milliseconds, void* vping)
  438. {
  439. struct PingContext* pctx = Identity_check((struct PingContext*) vping);
  440. if (data == NULL) {
  441. // This is how Pinger signals a timeout.
  442. onTimeout(milliseconds, pctx);
  443. return;
  444. }
  445. struct RouterModule* module = pctx->router;
  446. // Grab out the message which was put here in handleIncoming()
  447. struct DHTMessage* message = module->currentMessage;
  448. module->currentMessage = NULL;
  449. // This should never happen
  450. if (!Address_isSameIp(&pctx->address, message->address)) {
  451. #ifdef Log_WARN
  452. uint8_t expectedAddr[60];
  453. Address_print(expectedAddr, &pctx->address);
  454. uint8_t receivedAddr[60];
  455. Address_print(receivedAddr, message->address);
  456. Log_warn(module->logger,
  457. "Got return packet from different address than search was sent!\n"
  458. "Expected:%s\n"
  459. " Got:%s\n",
  460. expectedAddr,
  461. receivedAddr);
  462. #endif
  463. return;
  464. }
  465. // update the GMRT
  466. AverageRoller_update(pctx->router->gmrtRoller, milliseconds);
  467. /*
  468. Log_debug(pctx->router->logger,
  469. "Received response in %u milliseconds, gmrt now %u\n",
  470. milliseconds,
  471. AverageRoller_getAverage(pctx->router->gmrtRoller));
  472. */
  473. // prevent division by zero
  474. if (milliseconds == 0) { milliseconds++; }
  475. struct Node_Two* node = NodeStore_closestNode(module->nodeStore, message->address->path);
  476. if (node && !Bits_memcmp(node->address.key, message->address->key, 32)) {
  477. // This path is already known
  478. NodeStore_updateReach(module->nodeStore, node, nextReach(0, milliseconds));
  479. } else {
  480. struct Node_Link* link = NodeStore_discoverNode(module->nodeStore,
  481. message->address,
  482. message->encodingScheme,
  483. message->encIndex,
  484. nextReach(0, milliseconds));
  485. node = (link) ? link->child : NULL;
  486. }
  487. // EncodingSchemeModule should have added this node to the store, check it.
  488. if (!node) {
  489. #ifdef Log_DEBUG
  490. uint8_t printedAddr[60];
  491. Address_print(printedAddr, message->address);
  492. Log_info(module->logger, "Got message from nonexistant node! [%s]\n", printedAddr);
  493. #endif
  494. return;
  495. }
  496. #ifdef Log_DEBUG
  497. String* versionBin = Dict_getString(message->asDict, CJDHTConstants_VERSION);
  498. if (versionBin && versionBin->len == 20) {
  499. uint8_t printedAddr[60];
  500. Address_print(printedAddr, message->address);
  501. uint8_t versionStr[41];
  502. Hex_encode(versionStr, 41, (uint8_t*) versionBin->bytes, 20);
  503. Log_debug(module->logger, "Got pong! [%s] ver[%s]\n", printedAddr, versionStr);
  504. }
  505. #endif
  506. if (pctx->pub.callback) {
  507. pctx->pub.callback(&pctx->pub, milliseconds, message->address, message->asDict);
  508. }
  509. }
  510. struct RouterModule_Promise* RouterModule_newMessage(struct Address* addr,
  511. uint32_t timeoutMilliseconds,
  512. struct RouterModule* module,
  513. struct Allocator* alloc)
  514. {
  515. // sending yourself a ping?
  516. // Assert_true(Bits_memcmp(addr->key, module->address.key, 32));
  517. Assert_true(addr->path ==
  518. EncodingScheme_convertLabel(module->nodeStore->selfNode->encodingScheme,
  519. addr->path,
  520. EncodingScheme_convertLabel_convertTo_CANNONICAL));
  521. if (timeoutMilliseconds == 0) {
  522. timeoutMilliseconds = pingTimeoutMilliseconds(module);
  523. }
  524. struct Pinger_Ping* pp = Pinger_newPing(NULL,
  525. onResponseOrTimeout,
  526. sendMsg,
  527. timeoutMilliseconds,
  528. alloc,
  529. module->pinger);
  530. struct PingContext* pctx = Allocator_clone(pp->pingAlloc, (&(struct PingContext) {
  531. .pub = {
  532. .alloc = pp->pingAlloc
  533. },
  534. .router = module,
  535. .pp = pp
  536. }));
  537. Identity_set(pctx);
  538. Bits_memcpyConst(&pctx->address, addr, sizeof(struct Address));
  539. pp->context = pctx;
  540. return &pctx->pub;
  541. }
  542. void RouterModule_sendMessage(struct RouterModule_Promise* promise, Dict* request)
  543. {
  544. struct PingContext* pctx = Identity_check((struct PingContext*)promise);
  545. pctx->messageDict = request;
  546. // actual send is triggered asynchronously
  547. }
  548. struct RouterModule_Promise* RouterModule_pingNode(struct Address* addr,
  549. uint32_t timeoutMilliseconds,
  550. struct RouterModule* module,
  551. struct Allocator* alloc)
  552. {
  553. struct RouterModule_Promise* promise =
  554. RouterModule_newMessage(addr, timeoutMilliseconds, module, alloc);
  555. Dict* d = Dict_new(promise->alloc);
  556. Dict_putString(d, CJDHTConstants_QUERY, CJDHTConstants_QUERY_PING, promise->alloc);
  557. RouterModule_sendMessage(promise, d);
  558. #ifdef Log_DEBUG
  559. uint8_t buff[60];
  560. Address_print(buff, addr);
  561. Log_debug(module->logger, "Sending ping [%u] to [%s]",
  562. ((struct PingContext*)promise)->pp->handle, buff);
  563. #endif
  564. Assert_true(addr->path != 0);
  565. return promise;
  566. }
  567. struct RouterModule_Promise* RouterModule_getPeers(struct Address* addr,
  568. uint64_t nearbyLabel,
  569. uint32_t timeoutMilliseconds,
  570. struct RouterModule* module,
  571. struct Allocator* alloc)
  572. {
  573. struct RouterModule_Promise* promise =
  574. RouterModule_newMessage(addr, timeoutMilliseconds, module, alloc);
  575. Dict* d = Dict_new(promise->alloc);
  576. Dict_putString(d, CJDHTConstants_QUERY, CJDHTConstants_QUERY_GP, promise->alloc);
  577. uint64_t nearbyLabel_be = Endian_hostToBigEndian64(nearbyLabel);
  578. String* target = String_newBinary((char*)&nearbyLabel_be, 8, promise->alloc);
  579. Dict_putString(d, CJDHTConstants_TARGET, target, promise->alloc);
  580. RouterModule_sendMessage(promise, d);
  581. return promise;
  582. }
  583. struct Node_Two* RouterModule_lookup(uint8_t targetAddr[Address_SEARCH_TARGET_SIZE],
  584. struct RouterModule* module)
  585. {
  586. struct Address addr = { .path = 0 };
  587. Bits_memcpyConst(addr.ip6.bytes, targetAddr, Address_SEARCH_TARGET_SIZE);
  588. return NodeStore_getBest(&addr, module->nodeStore);
  589. }
  590. struct Node_Two* RouterModule_nodeForPath(uint64_t path, struct RouterModule* module)
  591. {
  592. struct Node_Link* l = NodeStore_linkForPath(module->nodeStore, path);
  593. return l ? l->child : NULL;
  594. }
  595. void RouterModule_brokenPath(const uint64_t path, struct RouterModule* module)
  596. {
  597. NodeStore_brokenPath(path, module->nodeStore);
  598. }
  599. uint32_t RouterModule_globalMeanResponseTime(struct RouterModule* module)
  600. {
  601. return (uint32_t) AverageRoller_getAverage(module->gmrtRoller);
  602. }