Pathfinder.c 17 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472
  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 "dht/Pathfinder.h"
  16. #include "dht/DHTModule.h"
  17. #include "dht/Address.h"
  18. #include "wire/DataHeader.h"
  19. #include "wire/RouteHeader.h"
  20. #include "dht/ReplyModule.h"
  21. #include "dht/EncodingSchemeModule.h"
  22. #include "dht/SerializationModule.h"
  23. #include "dht/dhtcore/RouterModule.h"
  24. #include "dht/dhtcore/RouterModule_admin.h"
  25. #include "dht/dhtcore/RumorMill.h"
  26. #include "dht/dhtcore/SearchRunner.h"
  27. #include "dht/dhtcore/SearchRunner_admin.h"
  28. #include "dht/dhtcore/NodeStore_admin.h"
  29. #include "dht/dhtcore/Janitor_admin.h"
  30. #include "dht/dhtcore/Janitor.h"
  31. #include "dht/dhtcore/Router_new.h"
  32. #include "util/AddrTools.h"
  33. #include "util/events/Timeout.h"
  34. #include "wire/Error.h"
  35. #include "util/CString.h"
  36. ///////////////////// [ Address ][ content... ]
  37. #define RUMORMILL_CAPACITY 64
  38. struct Pathfinder_pvt
  39. {
  40. struct Pathfinder pub;
  41. struct Iface eventIf;
  42. struct DHTModule dhtModule;
  43. struct Allocator* alloc;
  44. struct Log* log;
  45. struct EventBase* base;
  46. struct Random* rand;
  47. struct Admin* admin;
  48. struct EventEmitter* ee;
  49. // hack
  50. struct Node_Two asyncNode;
  51. struct Timeout* asyncTo;
  52. #define Pathfinder_pvt_state_INITIALIZING 0
  53. #define Pathfinder_pvt_state_RUNNING 1
  54. int state;
  55. // After begin connected, these fields will be filled.
  56. struct Address myAddr;
  57. struct DHTModuleRegistry* registry;
  58. struct NodeStore* nodeStore;
  59. struct Router* router;
  60. struct SearchRunner* searchRunner;
  61. struct RumorMill* rumorMill;
  62. struct Janitor* janitor;
  63. Identity
  64. };
  65. struct NodeStore* Pathfinder_getNodeStore(struct Pathfinder* pathfinder)
  66. {
  67. struct Pathfinder_pvt* pf = Identity_check((struct Pathfinder_pvt*) pathfinder);
  68. return pf->nodeStore;
  69. }
  70. static int incomingFromDHT(struct DHTMessage* dmessage, void* vpf)
  71. {
  72. struct Pathfinder_pvt* pf = Identity_check((struct Pathfinder_pvt*) vpf);
  73. struct Message* msg = dmessage->binMessage;
  74. struct Address* addr = dmessage->address;
  75. // Sanity check (make sure the addr was actually calculated)
  76. Assert_true(addr->ip6.bytes[0] == 0xfc);
  77. Message_shift(msg, PFChan_Msg_MIN_SIZE, NULL);
  78. struct PFChan_Msg* emsg = (struct PFChan_Msg*) msg->bytes;
  79. Bits_memset(emsg, 0, PFChan_Msg_MIN_SIZE);
  80. DataHeader_setVersion(&emsg->data, DataHeader_CURRENT_VERSION);
  81. DataHeader_setContentType(&emsg->data, ContentType_CJDHT);
  82. Bits_memcpyConst(emsg->route.ip6, addr->ip6.bytes, 16);
  83. emsg->route.version_be = Endian_hostToBigEndian32(addr->protocolVersion);
  84. emsg->route.sh.label_be = Endian_hostToBigEndian64(addr->path);
  85. Bits_memcpyConst(emsg->route.publicKey, addr->key, 32);
  86. Message_push32(msg, PFChan_Pathfinder_SENDMSG, NULL);
  87. if (dmessage->replyTo) {
  88. // see incomingMsg
  89. dmessage->replyTo->pleaseRespond = true;
  90. //Log_debug(pf->log, "send DHT reply");
  91. return 0;
  92. }
  93. //Log_debug(pf->log, "send DHT request");
  94. Iface_send(&pf->eventIf, msg);
  95. return 0;
  96. }
  97. static void nodeForAddress(struct PFChan_Node* nodeOut, struct Address* addr, uint32_t metric)
  98. {
  99. Bits_memset(nodeOut, 0, PFChan_Node_SIZE);
  100. nodeOut->version_be = Endian_hostToBigEndian32(addr->protocolVersion);
  101. nodeOut->metric_be = Endian_hostToBigEndian32(metric);
  102. nodeOut->path_be = Endian_hostToBigEndian64(addr->path);
  103. Bits_memcpyConst(nodeOut->publicKey, addr->key, 32);
  104. Bits_memcpyConst(nodeOut->ip6, addr->ip6.bytes, 16);
  105. }
  106. static Iface_DEFUN sendNode(struct Message* msg,
  107. struct Address* addr,
  108. uint32_t metric,
  109. struct Pathfinder_pvt* pf)
  110. {
  111. Message_reset(msg);
  112. Message_shift(msg, PFChan_Node_SIZE, NULL);
  113. nodeForAddress((struct PFChan_Node*) msg->bytes, addr, metric);
  114. if (addr->path == UINT64_MAX) {
  115. ((struct PFChan_Node*) msg->bytes)->path_be = 0;
  116. }
  117. Message_push32(msg, PFChan_Pathfinder_NODE, NULL);
  118. return Iface_next(&pf->eventIf, msg);
  119. }
  120. static void onBestPathChange(void* vPathfinder, struct Node_Two* node)
  121. {
  122. struct Pathfinder_pvt* pf = Identity_check((struct Pathfinder_pvt*) vPathfinder);
  123. struct Allocator* alloc = Allocator_child(pf->alloc);
  124. struct Message* msg = Message_new(0, 256, alloc);
  125. Iface_CALL(sendNode, msg, &node->address, 0xffffffffu - Node_getReach(node), pf);
  126. Allocator_free(alloc);
  127. }
  128. static Iface_DEFUN connected(struct Pathfinder_pvt* pf, struct Message* msg)
  129. {
  130. Log_debug(pf->log, "INIT");
  131. struct PFChan_Core_Connect conn;
  132. Message_pop(msg, &conn, PFChan_Core_Connect_SIZE, NULL);
  133. Assert_true(!msg->length);
  134. Bits_memcpyConst(pf->myAddr.key, conn.publicKey, 32);
  135. Address_getPrefix(&pf->myAddr);
  136. pf->myAddr.path = 1;
  137. // begin
  138. pf->registry = DHTModuleRegistry_new(pf->alloc);
  139. ReplyModule_register(pf->registry, pf->alloc);
  140. pf->rumorMill = RumorMill_new(pf->alloc, &pf->myAddr, RUMORMILL_CAPACITY, pf->log, "extern");
  141. pf->nodeStore = NodeStore_new(&pf->myAddr, pf->alloc, pf->base, pf->log, pf->rumorMill);
  142. pf->nodeStore->onBestPathChange = onBestPathChange;
  143. pf->nodeStore->onBestPathChangeCtx = pf;
  144. struct RouterModule* routerModule = RouterModule_register(pf->registry,
  145. pf->alloc,
  146. pf->myAddr.key,
  147. pf->base,
  148. pf->log,
  149. pf->rand,
  150. pf->nodeStore);
  151. pf->searchRunner = SearchRunner_new(pf->nodeStore,
  152. pf->log,
  153. pf->base,
  154. routerModule,
  155. pf->myAddr.ip6.bytes,
  156. pf->rumorMill,
  157. pf->alloc);
  158. pf->janitor = Janitor_new(routerModule,
  159. pf->nodeStore,
  160. pf->searchRunner,
  161. pf->rumorMill,
  162. pf->log,
  163. pf->alloc,
  164. pf->base,
  165. pf->rand);
  166. EncodingSchemeModule_register(pf->registry, pf->log, pf->alloc);
  167. SerializationModule_register(pf->registry, pf->log, pf->alloc);
  168. DHTModuleRegistry_register(&pf->dhtModule, pf->registry);
  169. pf->router = Router_new(routerModule, pf->nodeStore, pf->searchRunner, pf->alloc);
  170. // Now the admin stuff...
  171. if (pf->admin) {
  172. NodeStore_admin_register(pf->nodeStore, pf->admin, pf->alloc);
  173. RouterModule_admin_register(routerModule, pf->router, pf->admin, pf->alloc);
  174. SearchRunner_admin_register(pf->searchRunner, pf->admin, pf->alloc);
  175. Janitor_admin_register(pf->janitor, pf->admin, pf->alloc);
  176. }
  177. pf->state = Pathfinder_pvt_state_RUNNING;
  178. return NULL;
  179. }
  180. static void addressForNode(struct Address* addrOut, struct Message* msg)
  181. {
  182. struct PFChan_Node node;
  183. Message_pop(msg, &node, PFChan_Node_SIZE, NULL);
  184. Assert_true(!msg->length);
  185. addrOut->protocolVersion = Endian_bigEndianToHost32(node.version_be);
  186. addrOut->path = Endian_bigEndianToHost64(node.path_be);
  187. Bits_memcpyConst(addrOut->key, node.publicKey, 32);
  188. Bits_memcpyConst(addrOut->ip6.bytes, node.ip6, 16);
  189. }
  190. static Iface_DEFUN switchErr(struct Message* msg, struct Pathfinder_pvt* pf)
  191. {
  192. struct PFChan_Core_SwitchErr switchErr;
  193. Message_pop(msg, &switchErr, PFChan_Core_SwitchErr_MIN_SIZE, NULL);
  194. uint64_t path = Endian_bigEndianToHost64(switchErr.sh.label_be);
  195. uint64_t pathAtErrorHop = Endian_bigEndianToHost64(switchErr.ctrlErr.cause.label_be);
  196. uint8_t pathStr[20];
  197. AddrTools_printPath(pathStr, path);
  198. int err = Endian_bigEndianToHost32(switchErr.ctrlErr.errorType_be);
  199. Log_debug(pf->log, "switch err from [%s] type [%s][%d]", pathStr, Error_strerror(err), err);
  200. struct Node_Link* link = NodeStore_linkForPath(pf->nodeStore, path);
  201. uint8_t nodeAddr[16];
  202. if (link) {
  203. Bits_memcpyConst(nodeAddr, link->child->address.ip6.bytes, 16);
  204. }
  205. NodeStore_brokenLink(pf->nodeStore, path, pathAtErrorHop);
  206. if (link) {
  207. // Don't touch the node again, it might be a dangling pointer
  208. SearchRunner_search(nodeAddr, 20, 3, pf->searchRunner, pf->alloc);
  209. }
  210. return NULL;
  211. }
  212. static void asyncRespond(void* vPathfinder)
  213. {
  214. struct Pathfinder_pvt* pf = Identity_check((struct Pathfinder_pvt*) vPathfinder);
  215. pf->asyncTo = NULL;
  216. onBestPathChange(pf, &pf->asyncNode);
  217. }
  218. static Iface_DEFUN searchReq(struct Message* msg, struct Pathfinder_pvt* pf)
  219. {
  220. uint8_t addr[16];
  221. Message_pop(msg, addr, 16, NULL);
  222. Assert_true(!msg->length);
  223. uint8_t printedAddr[40];
  224. AddrTools_printIp(printedAddr, addr);
  225. Log_debug(pf->log, "Search req [%s]", printedAddr);
  226. struct Node_Two* node = NodeStore_nodeForAddr(pf->nodeStore, addr);
  227. if (node && !pf->asyncTo) {
  228. Bits_memcpyConst(&pf->asyncNode, node, sizeof(struct Node_Two));
  229. pf->asyncTo = Timeout_setTimeout(asyncRespond, pf, 0, pf->base, pf->alloc);
  230. } else {
  231. SearchRunner_search(addr, 20, 3, pf->searchRunner, pf->alloc);
  232. }
  233. return NULL;
  234. }
  235. static Iface_DEFUN peer(struct Message* msg, struct Pathfinder_pvt* pf)
  236. {
  237. struct Address addr;
  238. addressForNode(&addr, msg);
  239. String* str = Address_toString(&addr, msg->alloc);
  240. Log_debug(pf->log, "Peer [%s]", str->bytes);
  241. struct Node_Link* link = NodeStore_linkForPath(pf->nodeStore, addr.path);
  242. // It exists, it's parent is the self-node, and it's label is equal to the switchLabel.
  243. if (link
  244. && Node_getBestParent(link->child)
  245. && Node_getBestParent(link->child)->parent->address.path == 1
  246. && Node_getBestParent(link->child)->cannonicalLabel == addr.path)
  247. {
  248. return NULL;
  249. }
  250. //RumorMill_addNode(pf->rumorMill, &addr);
  251. Router_sendGetPeers(pf->router, &addr, 0, 0, pf->alloc);
  252. return sendNode(msg, &addr, 0xffffff00, pf);
  253. }
  254. static Iface_DEFUN peerGone(struct Message* msg, struct Pathfinder_pvt* pf)
  255. {
  256. struct Address addr;
  257. addressForNode(&addr, msg);
  258. String* str = Address_toString(&addr, msg->alloc);
  259. Log_debug(pf->log, "Peer gone [%s]", str->bytes);
  260. NodeStore_disconnectedPeer(pf->nodeStore, addr.path);
  261. // We notify about the node but with max metric so it will be removed soon.
  262. return sendNode(msg, &addr, 0xffffffff, pf);
  263. }
  264. static Iface_DEFUN session(struct Message* msg, struct Pathfinder_pvt* pf)
  265. {
  266. struct Address addr;
  267. addressForNode(&addr, msg);
  268. String* str = Address_toString(&addr, msg->alloc);
  269. Log_debug(pf->log, "Session [%s]", str->bytes);
  270. struct Node_Two* node = NodeStore_nodeForAddr(pf->nodeStore, addr.ip6.bytes);
  271. if (!node) {
  272. SearchRunner_search(addr.ip6.bytes, 20, 3, pf->searchRunner, pf->alloc);
  273. }
  274. return NULL;
  275. }
  276. static Iface_DEFUN sessionEnded(struct Message* msg, struct Pathfinder_pvt* pf)
  277. {
  278. struct Address addr;
  279. addressForNode(&addr, msg);
  280. String* str = Address_toString(&addr, msg->alloc);
  281. Log_debug(pf->log, "Session ended [%s]", str->bytes);
  282. return NULL;
  283. }
  284. static Iface_DEFUN discoveredPath(struct Message* msg, struct Pathfinder_pvt* pf)
  285. {
  286. struct Address addr;
  287. addressForNode(&addr, msg);
  288. // We're somehow aware of this path (even if it's unused)
  289. if (NodeStore_linkForPath(pf->nodeStore, addr.path)) { return NULL; }
  290. // If we don't already care about the destination, then don't do anything.
  291. struct Node_Two* nn = NodeStore_nodeForAddr(pf->nodeStore, addr.ip6.bytes);
  292. if (!nn) { return NULL; }
  293. // Our best path is "shorter" (label bits which is somewhat representitive of hop count)
  294. // basically this is just to dampen the flood to the RM because otherwise it prevents Janitor
  295. // from getting any actual work done.
  296. if (nn->address.path < addr.path) { return NULL; }
  297. Log_debug(pf->log, "Discovered path [%s]", Address_toString(&addr, msg->alloc)->bytes);
  298. RumorMill_addNode(pf->rumorMill, &addr);
  299. return NULL;
  300. }
  301. static Iface_DEFUN handlePing(struct Message* msg, struct Pathfinder_pvt* pf)
  302. {
  303. Log_debug(pf->log, "Received ping");
  304. Message_push32(msg, PFChan_Pathfinder_PONG, NULL);
  305. return Iface_next(&pf->eventIf, msg);
  306. }
  307. static Iface_DEFUN handlePong(struct Message* msg, struct Pathfinder_pvt* pf)
  308. {
  309. Log_debug(pf->log, "Received pong");
  310. return NULL;
  311. }
  312. static Iface_DEFUN incomingMsg(struct Message* msg, struct Pathfinder_pvt* pf)
  313. {
  314. struct Address addr;
  315. struct RouteHeader* hdr = (struct RouteHeader*) msg->bytes;
  316. Message_shift(msg, -(RouteHeader_SIZE + DataHeader_SIZE), NULL);
  317. Bits_memcpyConst(addr.ip6.bytes, hdr->ip6, 16);
  318. Bits_memcpyConst(addr.key, hdr->publicKey, 32);
  319. int version = addr.protocolVersion = Endian_bigEndianToHost32(hdr->version_be);
  320. addr.padding = 0;
  321. addr.path = Endian_bigEndianToHost64(hdr->sh.label_be);
  322. //Log_debug(pf->log, "Incoming DHT");
  323. struct DHTMessage dht = {
  324. .address = &addr,
  325. .binMessage = msg,
  326. .allocator = msg->alloc
  327. };
  328. DHTModuleRegistry_handleIncoming(&dht, pf->registry);
  329. if (dht.pleaseRespond) {
  330. // what a beautiful hack, see incomingFromDHT
  331. return Iface_next(&pf->eventIf, msg);
  332. } else if (!version && addr.protocolVersion) {
  333. return sendNode(msg, &addr, 0xfffffff0, pf);
  334. }
  335. return NULL;
  336. }
  337. static Iface_DEFUN incomingFromEventIf(struct Message* msg, struct Iface* eventIf)
  338. {
  339. struct Pathfinder_pvt* pf = Identity_containerOf(eventIf, struct Pathfinder_pvt, eventIf);
  340. enum PFChan_Core ev = Message_pop32(msg, NULL);
  341. if (Pathfinder_pvt_state_INITIALIZING == pf->state) {
  342. Assert_true(ev == PFChan_Core_CONNECT);
  343. return connected(pf, msg);
  344. }
  345. switch (ev) {
  346. case PFChan_Core_SWITCH_ERR: return switchErr(msg, pf);
  347. case PFChan_Core_SEARCH_REQ: return searchReq(msg, pf);
  348. case PFChan_Core_PEER: return peer(msg, pf);
  349. case PFChan_Core_PEER_GONE: return peerGone(msg, pf);
  350. case PFChan_Core_SESSION: return session(msg, pf);
  351. case PFChan_Core_SESSION_ENDED: return sessionEnded(msg, pf);
  352. case PFChan_Core_DISCOVERED_PATH: return discoveredPath(msg, pf);
  353. case PFChan_Core_MSG: return incomingMsg(msg, pf);
  354. case PFChan_Core_PING: return handlePing(msg, pf);
  355. case PFChan_Core_PONG: return handlePong(msg, pf);
  356. default:;
  357. }
  358. Assert_failure("unexpected event [%d]", ev);
  359. }
  360. static void sendEvent(struct Pathfinder_pvt* pf, enum PFChan_Pathfinder ev, void* data, int size)
  361. {
  362. struct Allocator* alloc = Allocator_child(pf->alloc);
  363. struct Message* msg = Message_new(0, 512+size, alloc);
  364. Message_push(msg, data, size, NULL);
  365. Message_push32(msg, ev, NULL);
  366. Iface_send(&pf->eventIf, msg);
  367. Allocator_free(alloc);
  368. }
  369. struct Pathfinder* Pathfinder_register(struct Allocator* alloc,
  370. struct Log* log,
  371. struct EventBase* base,
  372. struct Random* rand,
  373. struct Admin* admin,
  374. struct EventEmitter* ee)
  375. {
  376. struct Pathfinder_pvt* pf = Allocator_calloc(alloc, sizeof(struct Pathfinder_pvt), 1);
  377. pf->alloc = alloc;
  378. pf->log = log;
  379. pf->base = base;
  380. pf->rand = rand;
  381. pf->admin = admin;
  382. pf->ee = ee;
  383. Identity_set(pf);
  384. pf->eventIf.send = incomingFromEventIf;
  385. EventEmitter_regPathfinderIface(ee, &pf->eventIf);
  386. pf->dhtModule.context = pf;
  387. pf->dhtModule.handleOutgoing = incomingFromDHT;
  388. struct PFChan_Pathfinder_Connect conn = {
  389. .superiority_be = Endian_hostToBigEndian32(1),
  390. .version_be = Endian_hostToBigEndian32(Version_CURRENT_PROTOCOL)
  391. };
  392. CString_strncpy(conn.userAgent, "Cjdns internal pathfinder", 64);
  393. sendEvent(pf, PFChan_Pathfinder_CONNECT, &conn, PFChan_Pathfinder_Connect_SIZE);
  394. return &pf->pub;
  395. }