SubnodePathfinder.c 22 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597
  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/PingResponder.h"
  20. #include "subnode/BoilerplateResponder.h"
  21. #include "subnode/ReachabilityCollector.h"
  22. #include "crypto/AddressCalc.h"
  23. #include "dht/Address.h"
  24. #include "wire/DataHeader.h"
  25. #include "wire/RouteHeader.h"
  26. #include "dht/dhtcore/ReplySerializer.h"
  27. #include "util/AddrTools.h"
  28. #include "util/events/Timeout.h"
  29. #include "net/SwitchPinger.h"
  30. #include "switch/LabelSplicer.h"
  31. #include "wire/Error.h"
  32. #include "wire/PFChan.h"
  33. #include "wire/DataHeader.h"
  34. #include "util/CString.h"
  35. #include "wire/Metric.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* myPeerAddrs;
  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. Er_assert(Message_eshift(msg, PFChan_Node_SIZE));
  89. nodeForAddress((struct PFChan_Node*) msg->msgbytes, addr, metric);
  90. if (addr->path == UINT64_MAX) {
  91. ((struct PFChan_Node*) msg->msgbytes)->path_be = 0;
  92. }
  93. Er_assert(Message_epush32be(msg, msgType));
  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. Bits_memset(addrOut, 0, sizeof(struct Address));
  105. struct PFChan_Node node;
  106. Er_assert(Message_epop(msg, &node, PFChan_Node_SIZE));
  107. Assert_true(!Message_getLength(msg));
  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 [%d]",
  126. pathStr, 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 void queryRs(struct SubnodePathfinder_pvt* pf, uint8_t addr[16], uint8_t printedAddr[40])
  167. {
  168. if (!pf->pub.snh || !pf->pub.snh->snodeAddr.path) {
  169. Log_debug(pf->log, "Search for [%s] impossible because we have no snode", printedAddr);
  170. return;
  171. }
  172. struct Query q = { .routeFrom = { 0 } };
  173. Bits_memcpy(&q.target, &pf->pub.snh->snodeAddr, sizeof(struct Address));
  174. Bits_memcpy(q.routeFrom, pf->myAddress->ip6.bytes, 16);
  175. Bits_memcpy(q.routeFrom, addr, 16);
  176. if (Map_OfPromiseByQuery_indexForKey(&q, &pf->queryMap) > -1) {
  177. Log_debug(pf->log, "Search for [%s] skipped because one is outstanding", printedAddr);
  178. return;
  179. }
  180. struct MsgCore_Promise* qp = MsgCore_createQuery(pf->msgCore, 0, pf->alloc);
  181. struct SnodeQuery* snq = Allocator_calloc(qp->alloc, sizeof(struct SnodeQuery), 1);
  182. Identity_set(snq);
  183. snq->pf = pf;
  184. Dict* dict = qp->msg = Dict_new(qp->alloc);
  185. qp->cb = getRouteReply;
  186. qp->userData = snq;
  187. Assert_true(AddressCalc_validAddress(pf->pub.snh->snodeAddr.ip6.bytes));
  188. qp->target = &pf->pub.snh->snodeAddr;
  189. Log_debug(pf->log, "Sending getRoute to snode [%s] for [%s]",
  190. Address_toString(qp->target, qp->alloc)->bytes,
  191. printedAddr);
  192. Dict_putStringCC(dict, "sq", "gr", qp->alloc);
  193. String* src = String_newBinary(pf->myAddress->ip6.bytes, 16, qp->alloc);
  194. Dict_putStringC(dict, "src", src, qp->alloc);
  195. String* target = String_newBinary(addr, 16, qp->alloc);
  196. Dict_putStringC(dict, "tar", target, qp->alloc);
  197. int index = Map_OfPromiseByQuery_put(&q, &qp, &pf->queryMap);
  198. snq->mapHandle = pf->queryMap.handles[index];
  199. }
  200. static Iface_DEFUN searchReq(struct Message* msg, struct SubnodePathfinder_pvt* pf)
  201. {
  202. uint8_t addr[16];
  203. Er_assert(Message_epop(msg, addr, 16));
  204. Er_assert(Message_epop32be(msg));
  205. uint32_t version = Er_assert(Message_epop32be(msg));
  206. if (version && version < 20) { return Error(msg, "UNHANDLED"); }
  207. Assert_true(!Message_getLength(msg));
  208. uint8_t printedAddr[40];
  209. AddrTools_printIp(printedAddr, addr);
  210. // We're not going to query for a direct peer because it slows down the process significantly
  211. // and right now, Metric.h always prefers a reported peer to the advice of the snode, which
  212. for (int i = 0; i < pf->myPeerAddrs->length; ++i) {
  213. struct Address* myPeer = AddrSet_get(pf->myPeerAddrs, i);
  214. if (Bits_memcmp(myPeer->ip6.bytes, addr, 16)) { continue; }
  215. Log_debug(pf->log, "Skip for [%s] is our peer, provide that immediately", printedAddr);
  216. // warning: msg is nolonger usable, it has disappeared into sendNode
  217. queryRs(pf, addr, printedAddr);
  218. return sendNode(msg, myPeer, Metric_PF_PEER, PFChan_Pathfinder_NODE, pf);
  219. }
  220. if (pf->pub.snh && !Bits_memcmp(pf->pub.snh->snodeAddr.ip6.bytes, addr, 16)) {
  221. // Querying for a path TO our snode, we can return the path we know right now but also
  222. // make the query...
  223. Log_debug(pf->log, "Skip for [%s] is our snode, provide that immediately", printedAddr);
  224. queryRs(pf, addr, printedAddr);
  225. return sendNode(msg, &pf->pub.snh->snodeAddr, Metric_SNODE, PFChan_Pathfinder_NODE, pf);
  226. }
  227. queryRs(pf, addr, printedAddr);
  228. return NULL;
  229. }
  230. static void rcChange(struct ReachabilityCollector* rc,
  231. struct Address* nodeAddr,
  232. struct ReachabilityCollector_PeerInfo* pi)
  233. {
  234. struct SubnodePathfinder_pvt* pf = Identity_check((struct SubnodePathfinder_pvt*) rc->userData);
  235. ReachabilityAnnouncer_updatePeer(pf->ra, nodeAddr, pi);
  236. }
  237. struct Ping {
  238. struct SubnodePathfinder_pvt* pf;
  239. uint32_t mapHandle;
  240. Identity
  241. };
  242. static void pingReply(Dict* msg, struct Address* src, struct MsgCore_Promise* prom)
  243. {
  244. struct Ping* usp = Identity_check((struct Ping*) prom->userData);
  245. struct SubnodePathfinder_pvt* pf = Identity_check(usp->pf);
  246. int index = Map_OfPromiseByQuery_indexForHandle(usp->mapHandle, &pf->queryMap);
  247. Assert_true(index > -1);
  248. Map_OfPromiseByQuery_remove(index, &pf->queryMap);
  249. if (!src) {
  250. Log_debug(pf->log, "Ping timeout pinging [%s]",
  251. Address_toString(prom->target, prom->alloc)->bytes);
  252. return;
  253. }
  254. Log_debug(pf->log, "Ping reply from [%s]", Address_toString(src, prom->alloc)->bytes);
  255. struct Message* msgToCore = Message_new(0, 512, prom->alloc);
  256. Iface_CALL(sendNode, msgToCore, src, Metric_PING_REPLY, PFChan_Pathfinder_NODE, pf);
  257. }
  258. static void pingNode(struct SubnodePathfinder_pvt* pf, struct Address* addr)
  259. {
  260. struct Query q = { .routeFrom = { 0 } };
  261. Bits_memcpy(&q.target, addr, sizeof(struct Address));
  262. if (Map_OfPromiseByQuery_indexForKey(&q, &pf->queryMap) > -1) {
  263. Log_debug(pf->log, "Skipping ping because one is already outstanding");
  264. return;
  265. }
  266. // We have a path to the node but the session is not setup, lets ping them...
  267. struct MsgCore_Promise* qp = MsgCore_createQuery(pf->msgCore, 0, pf->alloc);
  268. struct Ping* usp = Allocator_calloc(qp->alloc, sizeof(struct Ping), 1);
  269. Identity_set(usp);
  270. usp->pf = pf;
  271. Dict* dict = qp->msg = Dict_new(qp->alloc);
  272. qp->cb = pingReply;
  273. qp->userData = usp;
  274. Assert_true(AddressCalc_validAddress(addr->ip6.bytes));
  275. Assert_true(addr->path);
  276. qp->target = Address_clone(addr, qp->alloc);
  277. Log_debug(pf->log, "Sending ping to [%s]",
  278. Address_toString(qp->target, qp->alloc)->bytes);
  279. Dict_putStringCC(dict, "q", "pn", qp->alloc);
  280. BoilerplateResponder_addBoilerplate(pf->br, dict, addr, qp->alloc);
  281. int index = Map_OfPromiseByQuery_put(&q, &qp, &pf->queryMap);
  282. usp->mapHandle = pf->queryMap.handles[index];
  283. }
  284. static Iface_DEFUN peer(struct Message* msg, struct SubnodePathfinder_pvt* pf)
  285. {
  286. struct Address addr = {0};
  287. uint32_t metric = addressForNode(&addr, msg);
  288. String* str = Address_toString(&addr, Message_getAlloc(msg));
  289. int index = AddrSet_indexOf(pf->myPeerAddrs, &addr, AddrSet_Match_BOTH);
  290. if (index > -1) {
  291. struct Address* myPeer = AddrSet_get(pf->myPeerAddrs, index);
  292. // Just in case there is a difference
  293. myPeer->protocolVersion = addr.protocolVersion;
  294. } else {
  295. Log_debug(pf->log, "Peer [%s]", str->bytes);
  296. AddrSet_add(pf->myPeerAddrs, &addr, AddrSet_Match_BOTH);
  297. // immediately ping them
  298. pingNode(pf, &addr);
  299. }
  300. ReachabilityCollector_change(pf->pub.rc, &addr);
  301. if ((metric & ~Metric_IC_PEER_MASK) != Metric_IC_PEER) {
  302. } else if ((metric & Metric_IC_PEER_MASK) == Metric_IC_PEER_MASK) {
  303. } else {
  304. ReachabilityCollector_lagSample(pf->pub.rc, addr.path, (metric & Metric_IC_PEER_MASK));
  305. }
  306. // We send this as Metric_PF_PEER because if it happens that this is a really crappy
  307. // peering link, we would prefer to take the route server's advice and bounce the connection
  308. // over a link which makes more sense to use.
  309. return sendNode(msg, &addr, Metric_PF_PEER, PFChan_Pathfinder_NODE, pf);
  310. }
  311. static Iface_DEFUN peerGone(struct Message* msg, struct SubnodePathfinder_pvt* pf)
  312. {
  313. struct Address addr = {0};
  314. addressForNode(&addr, msg);
  315. AddrSet_remove(pf->myPeerAddrs, &addr, AddrSet_Match_BOTH);
  316. ReachabilityCollector_unreachable(pf->pub.rc, &addr);
  317. return sendNode(msg, &addr, Metric_DEAD_LINK, PFChan_Pathfinder_NODE, pf);
  318. }
  319. static Iface_DEFUN session(struct Message* msg, struct SubnodePathfinder_pvt* pf)
  320. {
  321. struct Address addr = {0};
  322. addressForNode(&addr, msg);
  323. String* str = Address_toString(&addr, Message_getAlloc(msg));
  324. Log_debug(pf->log, "Session [%s]", str->bytes);
  325. //if (addr.protocolVersion) { NodeCache_discoverNode(pf->nc, &addr); }
  326. return NULL;
  327. }
  328. static Iface_DEFUN sessionEnded(struct Message* msg, struct SubnodePathfinder_pvt* pf)
  329. {
  330. struct Address addr = {0};
  331. addressForNode(&addr, msg);
  332. String* str = Address_toString(&addr, Message_getAlloc(msg));
  333. Log_debug(pf->log, "Session ended [%s]", str->bytes);
  334. //NodeCache_forgetNode(pf->nc, &addr);
  335. return NULL;
  336. }
  337. static Iface_DEFUN discoveredPath(struct Message* msg, struct SubnodePathfinder_pvt* pf)
  338. {
  339. //struct Address addr = {0};
  340. //addressForNode(&addr, msg);
  341. //Log_debug(pf->log, "discoveredPath(%s)", Address_toString(&addr, Message_getAlloc(msg))->bytes);
  342. //if (addr.protocolVersion) { NodeCache_discoverNode(pf->nc, &addr); }
  343. return NULL;
  344. }
  345. static Iface_DEFUN handlePing(struct Message* msg, struct SubnodePathfinder_pvt* pf)
  346. {
  347. //Log_debug(pf->log, "Received ping");
  348. Er_assert(Message_epush32be(msg, PFChan_Pathfinder_PONG));
  349. return Iface_next(&pf->pub.eventIf, msg);
  350. }
  351. static Iface_DEFUN handlePong(struct Message* msg, struct SubnodePathfinder_pvt* pf)
  352. {
  353. //Log_debug(pf->log, "Received pong");
  354. return NULL;
  355. }
  356. static Iface_DEFUN ctrlMsgFromSwitchPinger(struct Message* msg, struct Iface* iface)
  357. {
  358. struct SubnodePathfinder_pvt* pf =
  359. Identity_containerOf(iface, struct SubnodePathfinder_pvt, switchPingerIf);
  360. Er_assert(Message_epush32be(msg, PFChan_Pathfinder_CTRL_SENDMSG));
  361. return Iface_next(&pf->pub.eventIf, msg);
  362. }
  363. static Iface_DEFUN ctrlMsg(struct Message* msg, struct SubnodePathfinder_pvt* pf)
  364. {
  365. return Iface_next(&pf->switchPingerIf, msg);
  366. }
  367. static Iface_DEFUN unsetupSession(struct Message* msg, struct SubnodePathfinder_pvt* pf)
  368. {
  369. struct PFChan_Node node;
  370. Er_assert(Message_epop(msg, &node, PFChan_Node_SIZE));
  371. Assert_true(!Message_getLength(msg));
  372. struct Address addr = {
  373. .protocolVersion = Endian_bigEndianToHost32(node.version_be),
  374. .path = Endian_bigEndianToHost64(node.path_be),
  375. };
  376. Bits_memcpy(addr.ip6.bytes, node.ip6, 16);
  377. Bits_memcpy(addr.key, node.publicKey, 32);
  378. pingNode(pf, &addr);
  379. return NULL;
  380. }
  381. static Iface_DEFUN incomingMsg(struct Message* msg, struct SubnodePathfinder_pvt* pf)
  382. {
  383. return Iface_next(&pf->msgCoreIf, msg);
  384. }
  385. static Iface_DEFUN linkState(struct Message* msg, struct SubnodePathfinder_pvt* pf)
  386. {
  387. while (Message_getLength(msg)) {
  388. struct PFChan_LinkState_Entry lse;
  389. Er_assert(Message_epop(msg, &lse, PFChan_LinkState_Entry_SIZE));
  390. ReachabilityCollector_updateBandwidthAndDrops(
  391. pf->pub.rc,
  392. lse.peerLabel,
  393. lse.sumOfPackets,
  394. lse.sumOfDrops,
  395. lse.sumOfKb);
  396. }
  397. return NULL;
  398. }
  399. static Iface_DEFUN incomingFromMsgCore(struct Message* msg, struct Iface* iface)
  400. {
  401. struct SubnodePathfinder_pvt* pf =
  402. Identity_containerOf(iface, struct SubnodePathfinder_pvt, msgCoreIf);
  403. Assert_true(Message_getLength(msg) >= (RouteHeader_SIZE + DataHeader_SIZE));
  404. struct RouteHeader* rh = (struct RouteHeader*) msg->msgbytes;
  405. struct DataHeader* dh = (struct DataHeader*) &rh[1];
  406. Assert_true(DataHeader_getContentType(dh) == ContentType_CJDHT);
  407. Assert_true(!Bits_isZero(rh->publicKey, 32));
  408. Assert_true(rh->version_be);
  409. Assert_true(rh->sh.label_be);
  410. Er_assert(Message_epush32be(msg, PFChan_Pathfinder_SENDMSG));
  411. return Iface_next(&pf->pub.eventIf, msg);
  412. }
  413. static Iface_DEFUN incomingFromEventIf(struct Message* msg, struct Iface* eventIf)
  414. {
  415. struct SubnodePathfinder_pvt* pf =
  416. Identity_containerOf(eventIf, struct SubnodePathfinder_pvt, pub.eventIf);
  417. enum PFChan_Core ev = Er_assert(Message_epop32be(msg));
  418. if (SubnodePathfinder_pvt_state_INITIALIZING == pf->state) {
  419. Assert_true(ev == PFChan_Core_CONNECT);
  420. return connected(pf, msg);
  421. }
  422. switch (ev) {
  423. case PFChan_Core_SWITCH_ERR: return switchErr(msg, pf);
  424. case PFChan_Core_SEARCH_REQ: return searchReq(msg, pf);
  425. case PFChan_Core_PEER: return peer(msg, pf);
  426. case PFChan_Core_PEER_GONE: return peerGone(msg, pf);
  427. case PFChan_Core_SESSION: return session(msg, pf);
  428. case PFChan_Core_SESSION_ENDED: return sessionEnded(msg, pf);
  429. case PFChan_Core_DISCOVERED_PATH: return discoveredPath(msg, pf);
  430. case PFChan_Core_MSG: return incomingMsg(msg, pf);
  431. case PFChan_Core_PING: return handlePing(msg, pf);
  432. case PFChan_Core_PONG: return handlePong(msg, pf);
  433. case PFChan_Core_CTRL_MSG: return ctrlMsg(msg, pf);
  434. case PFChan_Core_UNSETUP_SESSION: return unsetupSession(msg, pf);
  435. case PFChan_Core_LINK_STATE: return linkState(msg, pf);
  436. default:;
  437. }
  438. Assert_failure("unexpected event [%d]", ev);
  439. }
  440. static void sendEvent(struct SubnodePathfinder_pvt* pf,
  441. enum PFChan_Pathfinder ev,
  442. void* data,
  443. int size)
  444. {
  445. struct Allocator* alloc = Allocator_child(pf->alloc);
  446. struct Message* msg = Message_new(0, 512+size, alloc);
  447. Er_assert(Message_epush(msg, data, size));
  448. Er_assert(Message_epush32be(msg, ev));
  449. Iface_send(&pf->pub.eventIf, msg);
  450. Allocator_free(alloc);
  451. }
  452. void SubnodePathfinder_start(struct SubnodePathfinder* sp)
  453. {
  454. struct SubnodePathfinder_pvt* pf = Identity_check((struct SubnodePathfinder_pvt*) sp);
  455. struct MsgCore* msgCore = pf->msgCore =
  456. MsgCore_new(pf->base, pf->rand, pf->alloc, pf->log, pf->myScheme);
  457. Iface_plumb(&pf->msgCoreIf, &msgCore->interRouterIf);
  458. PingResponder_new(pf->alloc, pf->log, msgCore, pf->br, pf->myScheme);
  459. struct ReachabilityCollector* rc = pf->pub.rc = ReachabilityCollector_new(
  460. pf->alloc, msgCore, pf->log, pf->base, pf->br, pf->myAddress, pf->myScheme, pf->sp);
  461. rc->userData = pf;
  462. rc->onChange = rcChange;
  463. struct SupernodeHunter* snh = pf->pub.snh = SupernodeHunter_new(
  464. pf->alloc, pf->log, pf->base, pf->sp, pf->myPeerAddrs, msgCore, pf->myAddress, rc);
  465. pf->ra = ReachabilityAnnouncer_new(
  466. pf->alloc, pf->log, pf->base, pf->rand, msgCore, snh, pf->privateKey, pf->myScheme, rc);
  467. struct PFChan_Pathfinder_Connect conn = {
  468. .superiority_be = Endian_hostToBigEndian32(1),
  469. .version_be = Endian_hostToBigEndian32(Version_CURRENT_PROTOCOL)
  470. };
  471. CString_safeStrncpy(conn.userAgent, "Cjdns subnode pathfinder", 64);
  472. sendEvent(pf, PFChan_Pathfinder_CONNECT, &conn, PFChan_Pathfinder_Connect_SIZE);
  473. }
  474. static void sendCurrentSupernode(void* vsp)
  475. {
  476. struct SubnodePathfinder_pvt* pf = Identity_check((struct SubnodePathfinder_pvt*) vsp);
  477. struct Allocator* alloc = Allocator_child(pf->alloc);
  478. struct Message* msgToCore = Message_new(0, 512, alloc);
  479. Iface_CALL(sendNode, msgToCore, &pf->pub.snh->snodeAddr, 0, PFChan_Pathfinder_SNODE, pf);
  480. Allocator_free(alloc);
  481. }
  482. struct SubnodePathfinder* SubnodePathfinder_new(struct Allocator* allocator,
  483. struct Log* log,
  484. struct EventBase* base,
  485. struct Random* rand,
  486. struct Address* myAddress,
  487. uint8_t* privateKey,
  488. struct EncodingScheme* myScheme)
  489. {
  490. struct Allocator* alloc = Allocator_child(allocator);
  491. struct SubnodePathfinder_pvt* pf =
  492. Allocator_calloc(alloc, sizeof(struct SubnodePathfinder_pvt), 1);
  493. Identity_set(pf);
  494. pf->alloc = alloc;
  495. pf->log = log;
  496. pf->base = base;
  497. pf->rand = rand;
  498. pf->myAddress = myAddress;
  499. pf->myPeerAddrs = AddrSet_new(alloc);
  500. pf->pub.eventIf.send = incomingFromEventIf;
  501. pf->msgCoreIf.send = incomingFromMsgCore;
  502. pf->privateKey = privateKey;
  503. pf->queryMap.allocator = Allocator_child(alloc);
  504. pf->myScheme = myScheme;
  505. pf->br = BoilerplateResponder_new(myScheme, alloc);
  506. pf->sp = SwitchPinger_new(base, rand, log, myAddress, alloc);
  507. pf->switchPingerIf.send = ctrlMsgFromSwitchPinger;
  508. Iface_plumb(&pf->switchPingerIf, &pf->sp->controlHandlerIf);
  509. Timeout_setInterval(sendCurrentSupernode, pf, 3000, base, alloc);
  510. return &pf->pub;
  511. }