SubnodePathfinder.c 21 KB

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