Pathfinder.c 17 KB

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