1
0

SubnodePathfinder.c 21 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591
  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 <https://www.gnu.org/licenses/>.
  14. */
  15. #include "subnode/SubnodePathfinder.h"
  16. #include "subnode/AddrSet.h"
  17. #include "subnode/MsgCore.h"
  18. #include "subnode/SupernodeHunter.h"
  19. #include "subnode/GetPeersResponder.h"
  20. #include "subnode/PingResponder.h"
  21. #include "subnode/BoilerplateResponder.h"
  22. #include "subnode/ReachabilityCollector.h"
  23. #include "crypto/AddressCalc.h"
  24. #include "dht/Address.h"
  25. #include "wire/DataHeader.h"
  26. #include "wire/RouteHeader.h"
  27. #include "dht/dhtcore/ReplySerializer.h"
  28. #include "util/AddrTools.h"
  29. #include "util/events/Timeout.h"
  30. #include "net/SwitchPinger.h"
  31. #include "switch/LabelSplicer.h"
  32. #include "wire/Error.h"
  33. #include "wire/PFChan.h"
  34. #include "wire/DataHeader.h"
  35. #include "util/CString.h"
  36. #include "subnode/ReachabilityAnnouncer.h"
  37. struct Query {
  38. struct Address target;
  39. uint8_t routeFrom[16];
  40. uint8_t routeTo[16];
  41. };
  42. #define Map_NAME OfPromiseByQuery
  43. #define Map_KEY_TYPE struct Query
  44. #define Map_VALUE_TYPE struct MsgCore_Promise*
  45. #define Map_ENABLE_HANDLES
  46. #include "util/Map.h"
  47. struct SubnodePathfinder_pvt
  48. {
  49. struct SubnodePathfinder pub;
  50. struct Iface msgCoreIf;
  51. struct Allocator* alloc;
  52. struct Log* log;
  53. struct EventBase* base;
  54. struct Random* rand;
  55. #define SubnodePathfinder_pvt_state_INITIALIZING 0
  56. #define SubnodePathfinder_pvt_state_RUNNING 1
  57. int state;
  58. struct Address* myAddress;
  59. struct AddrSet* myPeers;
  60. struct MsgCore* msgCore;
  61. struct Admin* admin;
  62. struct BoilerplateResponder* br;
  63. struct ReachabilityAnnouncer* ra;
  64. struct Map_OfPromiseByQuery queryMap;
  65. struct SwitchPinger* sp;
  66. struct Iface switchPingerIf;
  67. struct EncodingScheme* myScheme;
  68. uint8_t* privateKey;
  69. String* encodingSchemeStr;
  70. Identity
  71. };
  72. static void nodeForAddress(struct PFChan_Node* nodeOut, struct Address* addr, uint32_t metric)
  73. {
  74. Bits_memset(nodeOut, 0, PFChan_Node_SIZE);
  75. nodeOut->version_be = Endian_hostToBigEndian32(addr->protocolVersion);
  76. nodeOut->metric_be = Endian_hostToBigEndian32(metric);
  77. nodeOut->path_be = Endian_hostToBigEndian64(addr->path);
  78. Bits_memcpy(nodeOut->publicKey, addr->key, 32);
  79. Bits_memcpy(nodeOut->ip6, addr->ip6.bytes, 16);
  80. }
  81. static Iface_DEFUN sendNode(struct Message* msg,
  82. struct Address* addr,
  83. uint32_t metric,
  84. enum PFChan_Pathfinder msgType,
  85. struct SubnodePathfinder_pvt* pf)
  86. {
  87. Message_reset(msg);
  88. Message_shift(msg, PFChan_Node_SIZE, NULL);
  89. nodeForAddress((struct PFChan_Node*) msg->bytes, addr, metric);
  90. if (addr->path == UINT64_MAX) {
  91. ((struct PFChan_Node*) msg->bytes)->path_be = 0;
  92. }
  93. Message_push32(msg, msgType, NULL);
  94. return Iface_next(&pf->pub.eventIf, msg);
  95. }
  96. static Iface_DEFUN connected(struct SubnodePathfinder_pvt* pf, struct Message* msg)
  97. {
  98. Log_debug(pf->log, "INIT");
  99. pf->state = SubnodePathfinder_pvt_state_RUNNING;
  100. return NULL;
  101. }
  102. static uint32_t addressForNode(struct Address* addrOut, struct Message* msg)
  103. {
  104. struct PFChan_Node node;
  105. Message_pop(msg, &node, PFChan_Node_SIZE, NULL);
  106. Assert_true(!msg->length);
  107. addrOut->protocolVersion = Endian_bigEndianToHost32(node.version_be);
  108. addrOut->path = Endian_bigEndianToHost64(node.path_be);
  109. Bits_memcpy(addrOut->key, node.publicKey, 32);
  110. Bits_memcpy(addrOut->ip6.bytes, node.ip6, 16);
  111. return Endian_bigEndianToHost32(node.metric_be);
  112. }
  113. static Iface_DEFUN switchErr(struct Message* msg, struct SubnodePathfinder_pvt* pf)
  114. {
  115. struct PFChan_Core_SwitchErr switchErr;
  116. Message_pop(msg, &switchErr, PFChan_Core_SwitchErr_MIN_SIZE, NULL);
  117. uint64_t path = Endian_bigEndianToHost64(switchErr.sh.label_be);
  118. if (pf->pub.snh->snodeAddr.path &&
  119. pf->pub.snh->snodeAddr.path != path &&
  120. LabelSplicer_routesThrough(pf->pub.snh->snodeAddr.path, path)) {
  121. uint8_t pathStr[20];
  122. AddrTools_printPath(pathStr, path);
  123. int err = Endian_bigEndianToHost32(switchErr.ctrlErr.errorType_be);
  124. Log_debug(pf->log, "switch err from active snode [%s] type [%s][%d]",
  125. pathStr, Error_strerror(err), err);
  126. pf->pub.snh->snodeIsReachable = false;
  127. if (pf->pub.snh->onSnodeUnreachable) {
  128. pf->pub.snh->onSnodeUnreachable(pf->pub.snh, 0, 0);
  129. }
  130. }
  131. return NULL;
  132. }
  133. struct SnodeQuery {
  134. struct SubnodePathfinder_pvt* pf;
  135. uint32_t mapHandle;
  136. Identity
  137. };
  138. static void getRouteReply(Dict* msg, struct Address* src, struct MsgCore_Promise* prom)
  139. {
  140. struct SnodeQuery* snq = Identity_check((struct SnodeQuery*) prom->userData);
  141. struct SubnodePathfinder_pvt* pf = Identity_check(snq->pf);
  142. int index = Map_OfPromiseByQuery_indexForHandle(snq->mapHandle, &pf->queryMap);
  143. Assert_true(index > -1);
  144. Map_OfPromiseByQuery_remove(index, &pf->queryMap);
  145. if (!src) {
  146. Log_debug(pf->log, "GetRoute timeout");
  147. return;
  148. }
  149. Log_debug(pf->log, "Search reply!");
  150. struct Address_List* al = ReplySerializer_parse(src, msg, pf->log, false, prom->alloc);
  151. if (!al || al->length == 0) { return; }
  152. Log_debug(pf->log, "reply with[%s]", Address_toString(&al->elems[0], prom->alloc)->bytes);
  153. if (al->elems[0].protocolVersion < 20) {
  154. Log_debug(pf->log, "not sending [%s] because version is old",
  155. Address_toString(&al->elems[0], prom->alloc)->bytes);
  156. return;
  157. }
  158. //NodeCache_discoverNode(pf->nc, &al->elems[0]);
  159. struct Message* msgToCore = Message_new(0, 512, prom->alloc);
  160. Iface_CALL(sendNode, msgToCore, &al->elems[0], 0xfff00033, PFChan_Pathfinder_NODE, pf);
  161. }
  162. static Iface_DEFUN searchReq(struct Message* msg, struct SubnodePathfinder_pvt* pf)
  163. {
  164. uint8_t addr[16];
  165. Message_pop(msg, addr, 16, NULL);
  166. Message_pop32(msg, NULL);
  167. uint32_t version = Message_pop32(msg, NULL);
  168. if (version && version < 20) { return NULL; }
  169. Assert_true(!msg->length);
  170. uint8_t printedAddr[40];
  171. AddrTools_printIp(printedAddr, addr);
  172. Log_debug(pf->log, "Search req [%s]", printedAddr);
  173. for (int i = 0; i < pf->myPeers->length; ++i) {
  174. struct Address* myPeer = AddrSet_get(pf->myPeers, i);
  175. if (!Bits_memcmp(myPeer->ip6.bytes, addr, 16)) {
  176. return sendNode(msg, myPeer, 0xfff00000, PFChan_Pathfinder_NODE, pf);
  177. }
  178. }
  179. if (!pf->pub.snh || !pf->pub.snh->snodeAddr.path) { return NULL; }
  180. if (!Bits_memcmp(pf->pub.snh->snodeAddr.ip6.bytes, addr, 16)) {
  181. return sendNode(msg, &pf->pub.snh->snodeAddr, 0xfff00000, PFChan_Pathfinder_NODE, pf);
  182. }
  183. struct Query q = { .routeFrom = { 0 } };
  184. Bits_memcpy(&q.target, &pf->pub.snh->snodeAddr, sizeof(struct Address));
  185. Bits_memcpy(q.routeFrom, pf->myAddress->ip6.bytes, 16);
  186. Bits_memcpy(q.routeFrom, addr, 16);
  187. if (Map_OfPromiseByQuery_indexForKey(&q, &pf->queryMap) > -1) {
  188. Log_debug(pf->log, "Skipping snode query because one is outstanding");
  189. return NULL;
  190. }
  191. struct MsgCore_Promise* qp = MsgCore_createQuery(pf->msgCore, 0, pf->alloc);
  192. struct SnodeQuery* snq = Allocator_calloc(qp->alloc, sizeof(struct SnodeQuery), 1);
  193. Identity_set(snq);
  194. snq->pf = pf;
  195. Dict* dict = qp->msg = Dict_new(qp->alloc);
  196. qp->cb = getRouteReply;
  197. qp->userData = snq;
  198. Assert_true(AddressCalc_validAddress(pf->pub.snh->snodeAddr.ip6.bytes));
  199. qp->target = &pf->pub.snh->snodeAddr;
  200. Log_debug(pf->log, "Sending getRoute to snode %s",
  201. Address_toString(qp->target, qp->alloc)->bytes);
  202. Dict_putStringCC(dict, "sq", "gr", qp->alloc);
  203. String* src = String_newBinary(pf->myAddress->ip6.bytes, 16, qp->alloc);
  204. Dict_putStringC(dict, "src", src, qp->alloc);
  205. String* target = String_newBinary(addr, 16, qp->alloc);
  206. Dict_putStringC(dict, "tar", target, qp->alloc);
  207. int index = Map_OfPromiseByQuery_put(&q, &qp, &pf->queryMap);
  208. snq->mapHandle = pf->queryMap.handles[index];
  209. return NULL;
  210. }
  211. static void rcChange(struct ReachabilityCollector* rc,
  212. uint8_t nodeIpv6[16],
  213. struct ReachabilityCollector_PeerInfo* pi)
  214. {
  215. struct SubnodePathfinder_pvt* pf = Identity_check((struct SubnodePathfinder_pvt*) rc->userData);
  216. ReachabilityAnnouncer_updatePeer(pf->ra, nodeIpv6, pi);
  217. }
  218. static Iface_DEFUN peer(struct Message* msg, struct SubnodePathfinder_pvt* pf)
  219. {
  220. struct Address addr;
  221. uint32_t metric = addressForNode(&addr, msg);
  222. String* str = Address_toString(&addr, msg->alloc);
  223. Log_debug(pf->log, "Peer [%s]", str->bytes);
  224. int index = AddrSet_indexOf(pf->myPeers, &addr);
  225. if (index > -1) {
  226. struct Address* myPeer = AddrSet_get(pf->myPeers, index);
  227. if (myPeer->path == addr.path && myPeer->protocolVersion == addr.protocolVersion) {
  228. return NULL;
  229. }
  230. AddrSet_remove(pf->myPeers, myPeer);
  231. }
  232. AddrSet_add(pf->myPeers, &addr);
  233. //NodeCache_discoverNode(pf->nc, &addr);
  234. ReachabilityCollector_change(pf->pub.rc, &addr);
  235. if ((metric & 0xffff) < 0xffff) {
  236. ReachabilityCollector_lagSample(pf->pub.rc, addr.path, (metric & 0xffff));
  237. }
  238. return sendNode(msg, &addr, 0xfff00000, PFChan_Pathfinder_NODE, pf);
  239. }
  240. static Iface_DEFUN peerGone(struct Message* msg, struct SubnodePathfinder_pvt* pf)
  241. {
  242. struct Address addr;
  243. addressForNode(&addr, msg);
  244. for (int i = pf->myPeers->length - 1; i >= 0; i--) {
  245. struct Address* myPeer = AddrSet_get(pf->myPeers, i);
  246. if (myPeer->path == addr.path) {
  247. String* str = Address_toString(myPeer, msg->alloc);
  248. AddrSet_remove(pf->myPeers, myPeer);
  249. Log_debug(pf->log, "Peer gone [%s]", str->bytes);
  250. }
  251. }
  252. //NodeCache_forgetNode(pf->nc, &addr);
  253. struct Address zaddr;
  254. Bits_memcpy(&zaddr, &addr, Address_SIZE);
  255. zaddr.path = 0;
  256. ReachabilityCollector_change(pf->pub.rc, &zaddr);
  257. // We notify about the node but with max metric so it will be removed soon.
  258. return sendNode(msg, &addr, 0xffffffff, PFChan_Pathfinder_NODE, pf);
  259. }
  260. static Iface_DEFUN session(struct Message* msg, struct SubnodePathfinder_pvt* pf)
  261. {
  262. struct Address addr;
  263. addressForNode(&addr, msg);
  264. String* str = Address_toString(&addr, msg->alloc);
  265. Log_debug(pf->log, "Session [%s]", str->bytes);
  266. //if (addr.protocolVersion) { NodeCache_discoverNode(pf->nc, &addr); }
  267. return NULL;
  268. }
  269. static Iface_DEFUN sessionEnded(struct Message* msg, struct SubnodePathfinder_pvt* pf)
  270. {
  271. struct Address addr;
  272. addressForNode(&addr, msg);
  273. String* str = Address_toString(&addr, msg->alloc);
  274. Log_debug(pf->log, "Session ended [%s]", str->bytes);
  275. //NodeCache_forgetNode(pf->nc, &addr);
  276. return NULL;
  277. }
  278. static Iface_DEFUN discoveredPath(struct Message* msg, struct SubnodePathfinder_pvt* pf)
  279. {
  280. //struct Address addr;
  281. //addressForNode(&addr, msg);
  282. //Log_debug(pf->log, "discoveredPath(%s)", Address_toString(&addr, msg->alloc)->bytes);
  283. //if (addr.protocolVersion) { NodeCache_discoverNode(pf->nc, &addr); }
  284. return NULL;
  285. }
  286. static Iface_DEFUN handlePing(struct Message* msg, struct SubnodePathfinder_pvt* pf)
  287. {
  288. //Log_debug(pf->log, "Received ping");
  289. Message_push32(msg, PFChan_Pathfinder_PONG, NULL);
  290. return Iface_next(&pf->pub.eventIf, msg);
  291. }
  292. static Iface_DEFUN handlePong(struct Message* msg, struct SubnodePathfinder_pvt* pf)
  293. {
  294. //Log_debug(pf->log, "Received pong");
  295. return NULL;
  296. }
  297. static Iface_DEFUN ctrlMsgFromSwitchPinger(struct Message* msg, struct Iface* iface)
  298. {
  299. struct SubnodePathfinder_pvt* pf =
  300. Identity_containerOf(iface, struct SubnodePathfinder_pvt, switchPingerIf);
  301. Message_push32(msg, PFChan_Pathfinder_CTRL_SENDMSG, NULL);
  302. return Iface_next(&pf->pub.eventIf, msg);
  303. }
  304. static Iface_DEFUN ctrlMsg(struct Message* msg, struct SubnodePathfinder_pvt* pf)
  305. {
  306. return Iface_next(&pf->switchPingerIf, msg);
  307. }
  308. struct UnsetupSessionPing {
  309. struct SubnodePathfinder_pvt* pf;
  310. uint32_t mapHandle;
  311. Identity
  312. };
  313. static void unsetupSessionPingReply(Dict* msg, struct Address* src, struct MsgCore_Promise* prom)
  314. {
  315. struct UnsetupSessionPing* usp =
  316. Identity_check((struct UnsetupSessionPing*) prom->userData);
  317. struct SubnodePathfinder_pvt* pf = Identity_check(usp->pf);
  318. int index = Map_OfPromiseByQuery_indexForHandle(usp->mapHandle, &pf->queryMap);
  319. Assert_true(index > -1);
  320. Map_OfPromiseByQuery_remove(index, &pf->queryMap);
  321. if (!src) {
  322. //Log_debug(pf->log, "Ping timeout");
  323. return;
  324. }
  325. //Log_debug(pf->log, "\n\n\n\nPING reply from [%s]!\n\n\n\n",
  326. // Address_toString(src, prom->alloc)->bytes);
  327. struct Message* msgToCore = Message_new(0, 512, prom->alloc);
  328. Iface_CALL(sendNode, msgToCore, src, 0xfffffff0, PFChan_Pathfinder_NODE, pf);
  329. }
  330. static Iface_DEFUN unsetupSession(struct Message* msg, struct SubnodePathfinder_pvt* pf)
  331. {
  332. struct PFChan_Node node;
  333. Message_pop(msg, &node, PFChan_Node_SIZE, NULL);
  334. Assert_true(!msg->length);
  335. struct Query q = { .routeFrom = { 0 } };
  336. struct Address* addr = &q.target;
  337. Bits_memcpy(addr->ip6.bytes, node.ip6, 16);
  338. Bits_memcpy(addr->key, node.publicKey, 32);
  339. addr->protocolVersion = Endian_bigEndianToHost32(node.version_be);
  340. addr->path = Endian_bigEndianToHost64(node.path_be);
  341. if (Map_OfPromiseByQuery_indexForKey(&q, &pf->queryMap) > -1) {
  342. Log_debug(pf->log, "Skipping ping because one is already outstanding");
  343. return NULL;
  344. }
  345. // We have a path to the node but the session is not setup, lets ping them...
  346. struct MsgCore_Promise* qp = MsgCore_createQuery(pf->msgCore, 0, pf->alloc);
  347. struct UnsetupSessionPing* usp =
  348. Allocator_calloc(qp->alloc, sizeof(struct UnsetupSessionPing), 1);
  349. Identity_set(usp);
  350. usp->pf = pf;
  351. Dict* dict = qp->msg = Dict_new(qp->alloc);
  352. qp->cb = unsetupSessionPingReply;
  353. qp->userData = usp;
  354. Assert_true(AddressCalc_validAddress(addr->ip6.bytes));
  355. Assert_true(addr->path);
  356. qp->target = Address_clone(addr, qp->alloc);
  357. //Log_debug(pf->log, "unsetupSession sending ping to [%s]",
  358. // Address_toString(qp->target, qp->alloc)->bytes);
  359. Dict_putStringCC(dict, "q", "pn", qp->alloc);
  360. BoilerplateResponder_addBoilerplate(pf->br, dict, addr, qp->alloc);
  361. int index = Map_OfPromiseByQuery_put(&q, &qp, &pf->queryMap);
  362. usp->mapHandle = pf->queryMap.handles[index];
  363. return NULL;
  364. }
  365. static Iface_DEFUN incomingMsg(struct Message* msg, struct SubnodePathfinder_pvt* pf)
  366. {
  367. return Iface_next(&pf->msgCoreIf, msg);
  368. }
  369. static Iface_DEFUN linkState(struct Message* msg, struct SubnodePathfinder_pvt* pf)
  370. {
  371. while (msg->length) {
  372. struct PFChan_LinkState_Entry lse;
  373. Message_pop(msg, &lse, PFChan_LinkState_Entry_SIZE, NULL);
  374. ReachabilityCollector_updateBandwidthAndDrops(
  375. pf->pub.rc,
  376. Endian_bigEndianToHost32(lse.peerLabel_be),
  377. Endian_bigEndianToHost32(lse.sumOfPackets_be),
  378. Endian_bigEndianToHost32(lse.sumOfDrops_be),
  379. Endian_bigEndianToHost32(lse.sumOfKb_be)
  380. );
  381. }
  382. return NULL;
  383. }
  384. static Iface_DEFUN incomingFromMsgCore(struct Message* msg, struct Iface* iface)
  385. {
  386. struct SubnodePathfinder_pvt* pf =
  387. Identity_containerOf(iface, struct SubnodePathfinder_pvt, msgCoreIf);
  388. Assert_true(msg->length >= (RouteHeader_SIZE + DataHeader_SIZE));
  389. struct RouteHeader* rh = (struct RouteHeader*) msg->bytes;
  390. struct DataHeader* dh = (struct DataHeader*) &rh[1];
  391. Assert_true(DataHeader_getContentType(dh) == ContentType_CJDHT);
  392. Assert_true(!Bits_isZero(rh->publicKey, 32));
  393. Assert_true(rh->version_be);
  394. Assert_true(rh->sh.label_be);
  395. Message_push32(msg, PFChan_Pathfinder_SENDMSG, NULL);
  396. return Iface_next(&pf->pub.eventIf, msg);
  397. }
  398. static Iface_DEFUN incomingFromEventIf(struct Message* msg, struct Iface* eventIf)
  399. {
  400. struct SubnodePathfinder_pvt* pf =
  401. Identity_containerOf(eventIf, struct SubnodePathfinder_pvt, pub.eventIf);
  402. enum PFChan_Core ev = Message_pop32(msg, NULL);
  403. if (SubnodePathfinder_pvt_state_INITIALIZING == pf->state) {
  404. Assert_true(ev == PFChan_Core_CONNECT);
  405. return connected(pf, msg);
  406. }
  407. switch (ev) {
  408. case PFChan_Core_SWITCH_ERR: return switchErr(msg, pf);
  409. case PFChan_Core_SEARCH_REQ: return searchReq(msg, pf);
  410. case PFChan_Core_PEER: return peer(msg, pf);
  411. case PFChan_Core_PEER_GONE: return peerGone(msg, pf);
  412. case PFChan_Core_SESSION: return session(msg, pf);
  413. case PFChan_Core_SESSION_ENDED: return sessionEnded(msg, pf);
  414. case PFChan_Core_DISCOVERED_PATH: return discoveredPath(msg, pf);
  415. case PFChan_Core_MSG: return incomingMsg(msg, pf);
  416. case PFChan_Core_PING: return handlePing(msg, pf);
  417. case PFChan_Core_PONG: return handlePong(msg, pf);
  418. case PFChan_Core_CTRL_MSG: return ctrlMsg(msg, pf);
  419. case PFChan_Core_UNSETUP_SESSION: return unsetupSession(msg, pf);
  420. case PFChan_Core_LINK_STATE: return linkState(msg, pf);
  421. default:;
  422. }
  423. Assert_failure("unexpected event [%d]", ev);
  424. }
  425. static void sendEvent(struct SubnodePathfinder_pvt* pf,
  426. enum PFChan_Pathfinder ev,
  427. void* data,
  428. int size)
  429. {
  430. struct Allocator* alloc = Allocator_child(pf->alloc);
  431. struct Message* msg = Message_new(0, 512+size, alloc);
  432. Message_push(msg, data, size, NULL);
  433. Message_push32(msg, ev, NULL);
  434. Iface_send(&pf->pub.eventIf, msg);
  435. Allocator_free(alloc);
  436. }
  437. void SubnodePathfinder_start(struct SubnodePathfinder* sp)
  438. {
  439. struct SubnodePathfinder_pvt* pf = Identity_check((struct SubnodePathfinder_pvt*) sp);
  440. struct MsgCore* msgCore = pf->msgCore =
  441. MsgCore_new(pf->base, pf->rand, pf->alloc, pf->log, pf->myScheme);
  442. Iface_plumb(&pf->msgCoreIf, &msgCore->interRouterIf);
  443. PingResponder_new(pf->alloc, pf->log, msgCore, pf->br);
  444. GetPeersResponder_new(
  445. pf->alloc, pf->log, pf->myPeers, pf->myAddress, msgCore, pf->br, pf->myScheme);
  446. struct ReachabilityCollector* rc = pf->pub.rc = ReachabilityCollector_new(
  447. pf->alloc, msgCore, pf->log, pf->base, pf->br, pf->myAddress, pf->myScheme);
  448. rc->userData = pf;
  449. rc->onChange = rcChange;
  450. struct SupernodeHunter* snh = pf->pub.snh = SupernodeHunter_new(
  451. pf->alloc, pf->log, pf->base, pf->sp, pf->myPeers, msgCore, pf->myAddress, rc);
  452. pf->ra = ReachabilityAnnouncer_new(
  453. pf->alloc, pf->log, pf->base, pf->rand, msgCore, snh, pf->privateKey, pf->myScheme);
  454. struct PFChan_Pathfinder_Connect conn = {
  455. .superiority_be = Endian_hostToBigEndian32(1),
  456. .version_be = Endian_hostToBigEndian32(Version_CURRENT_PROTOCOL)
  457. };
  458. CString_strncpy(conn.userAgent, "Cjdns subnode pathfinder", 64);
  459. sendEvent(pf, PFChan_Pathfinder_CONNECT, &conn, PFChan_Pathfinder_Connect_SIZE);
  460. }
  461. static void sendCurrentSupernode(void* vsp)
  462. {
  463. struct SubnodePathfinder_pvt* pf = Identity_check((struct SubnodePathfinder_pvt*) vsp);
  464. struct Allocator* alloc = Allocator_child(pf->alloc);
  465. struct Message* msgToCore = Message_new(0, 512, alloc);
  466. Iface_CALL(sendNode, msgToCore, &pf->pub.snh->snodeAddr, 0, PFChan_Pathfinder_SNODE, pf);
  467. Allocator_free(alloc);
  468. }
  469. struct SubnodePathfinder* SubnodePathfinder_new(struct Allocator* allocator,
  470. struct Log* log,
  471. struct EventBase* base,
  472. struct Random* rand,
  473. struct Address* myAddress,
  474. uint8_t* privateKey,
  475. struct EncodingScheme* myScheme)
  476. {
  477. struct Allocator* alloc = Allocator_child(allocator);
  478. struct SubnodePathfinder_pvt* pf =
  479. Allocator_calloc(alloc, sizeof(struct SubnodePathfinder_pvt), 1);
  480. Identity_set(pf);
  481. pf->alloc = alloc;
  482. pf->log = log;
  483. pf->base = base;
  484. pf->rand = rand;
  485. pf->myAddress = myAddress;
  486. pf->myPeers = AddrSet_new(alloc);
  487. pf->pub.eventIf.send = incomingFromEventIf;
  488. pf->msgCoreIf.send = incomingFromMsgCore;
  489. pf->privateKey = privateKey;
  490. pf->queryMap.allocator = Allocator_child(alloc);
  491. pf->myScheme = myScheme;
  492. pf->br = BoilerplateResponder_new(myScheme, alloc);
  493. pf->sp = SwitchPinger_new(base, rand, log, myAddress, alloc);
  494. pf->switchPingerIf.send = ctrlMsgFromSwitchPinger;
  495. Iface_plumb(&pf->switchPingerIf, &pf->sp->controlHandlerIf);
  496. Timeout_setInterval(sendCurrentSupernode, pf, 3000, base, alloc);
  497. return &pf->pub;
  498. }