|
| 1 | +// Kademlia DHT: two TCP nodes form a DHT (a server, and a client that lists the |
| 2 | +// server as its bootstrap node), then round-trip a value and a provider record |
| 3 | +// over the wire. The generated bindings are asynchronous, so every call is |
| 4 | +// wrapped with the blocking helpers in common.h. |
| 5 | +// |
| 6 | +// The client learns the server through its `bootstrapNodes` config: at |
| 7 | +// construction the DHT inserts each bootstrap peer (with its addresses) into |
| 8 | +// the routing table, so the client can reach the server without any prior |
| 9 | +// lookup. All DHT queries here originate at the client and travel to the |
| 10 | +// server, which is why only the client needs to know the server. |
| 11 | +#include "common.h" |
| 12 | + |
| 13 | +// TransportType / MuxerType ordinals, mirrored from libp2p_ffi/config.nim. |
| 14 | +static const int64_t TransportTcp = 1; |
| 15 | +static const int64_t MuxerMplex = 0; |
| 16 | + |
| 17 | +// A key is just bytes for the DHT; the value is stored and fetched verbatim. |
| 18 | +static const char *ValueKey = "/cbind/kad/demo-key"; |
| 19 | +static const char *ValueData = "hello kademlia"; |
| 20 | + |
| 21 | +// libp2p_ctx_create_cid returns the CID string, so it needs its own waiter. |
| 22 | +typedef struct { |
| 23 | + atomic_int done; |
| 24 | + int err_code; |
| 25 | + char err[256]; |
| 26 | + char cid[256]; |
| 27 | +} CidWaiter; |
| 28 | + |
| 29 | +static void on_cid(int ec, const NimFfiStr *reply, const char *em, void *ud) { |
| 30 | + CidWaiter *w = (CidWaiter *)ud; |
| 31 | + w->err_code = ec; |
| 32 | + if (reply && reply->data) |
| 33 | + snprintf(w->cid, sizeof(w->cid), "%.*s", (int)reply->len, reply->data); |
| 34 | + if (em) |
| 35 | + snprintf(w->err, sizeof(w->err), "%s", em); |
| 36 | + atomic_store(&w->done, 1); |
| 37 | +} |
| 38 | + |
| 39 | +// kad_get_value replies with a ReadResponse carrying the stored bytes. |
| 40 | +typedef struct { |
| 41 | + atomic_int done; |
| 42 | + int err_code; |
| 43 | + char err[256]; |
| 44 | + uint8_t data[512]; |
| 45 | + size_t len; |
| 46 | +} ValueWaiter; |
| 47 | + |
| 48 | +static void on_value(int ec, const ReadResponse *reply, const char *em, |
| 49 | + void *ud) { |
| 50 | + ValueWaiter *w = (ValueWaiter *)ud; |
| 51 | + w->err_code = ec; |
| 52 | + if (reply && reply->data.data) { |
| 53 | + w->len = |
| 54 | + reply->data.len < sizeof(w->data) ? reply->data.len : sizeof(w->data); |
| 55 | + memcpy(w->data, reply->data.data, w->len); |
| 56 | + } |
| 57 | + if (em) |
| 58 | + snprintf(w->err, sizeof(w->err), "%s", em); |
| 59 | + atomic_store(&w->done, 1); |
| 60 | +} |
| 61 | + |
| 62 | +// kad_get_providers replies with the peers that announced the CID. |
| 63 | +#define MAX_PROVIDERS 8 |
| 64 | +typedef struct { |
| 65 | + atomic_int done; |
| 66 | + int err_code; |
| 67 | + char err[256]; |
| 68 | + char peerIds[MAX_PROVIDERS][128]; |
| 69 | + size_t n; |
| 70 | +} ProvidersWaiter; |
| 71 | + |
| 72 | +static void on_providers(int ec, const ProvidersResponse *reply, const char *em, |
| 73 | + void *ud) { |
| 74 | + ProvidersWaiter *w = (ProvidersWaiter *)ud; |
| 75 | + w->err_code = ec; |
| 76 | + if (reply) { |
| 77 | + w->n = reply->providers.len < MAX_PROVIDERS ? reply->providers.len |
| 78 | + : MAX_PROVIDERS; |
| 79 | + for (size_t i = 0; i < w->n; i++) |
| 80 | + if (reply->providers.data[i].peerId.data) |
| 81 | + snprintf(w->peerIds[i], sizeof(w->peerIds[i]), "%s", |
| 82 | + reply->providers.data[i].peerId.data); |
| 83 | + } |
| 84 | + if (em) |
| 85 | + snprintf(w->err, sizeof(w->err), "%s", em); |
| 86 | + atomic_store(&w->done, 1); |
| 87 | +} |
| 88 | + |
| 89 | +// Builds a kad-dht node. `boot` is NULL for the server (no bootstrap peers) or |
| 90 | +// the server's PeerInfo for the client, whose DHT then knows how to reach it. |
| 91 | +static LibP2PCtx *kadNode(const char *listenAddr, const char *label, |
| 92 | + const PeerInfoWaiter *boot) { |
| 93 | + NimFfiStr addrSlot = nimffi_str(listenAddr); |
| 94 | + Libp2pConfig cfg; |
| 95 | + memset(&cfg, 0, sizeof(cfg)); |
| 96 | + cfg.mountKad = true; |
| 97 | + cfg.addrs.data = &addrSlot; |
| 98 | + cfg.addrs.len = 1; |
| 99 | + cfg.muxer = MuxerMplex; |
| 100 | + cfg.transport = TransportTcp; |
| 101 | + |
| 102 | + NimFfiStr bootAddrs[MAX_ADDRS]; |
| 103 | + BootstrapNode bootNode; |
| 104 | + if (boot) { |
| 105 | + for (size_t i = 0; i < boot->naddrs; i++) |
| 106 | + bootAddrs[i] = nimffi_str(boot->addrs[i]); |
| 107 | + bootNode.peerId = nimffi_str(boot->peerId); |
| 108 | + bootNode.multiaddrs.data = bootAddrs; |
| 109 | + bootNode.multiaddrs.len = boot->naddrs; |
| 110 | + cfg.bootstrapNodes.data = &bootNode; |
| 111 | + cfg.bootstrapNodes.len = 1; |
| 112 | + } |
| 113 | + // await_create only reads cfg while encoding, so the stack-local bootstrap |
| 114 | + // views above stay valid for the whole call. |
| 115 | + return await_create(&cfg, label); |
| 116 | +} |
| 117 | + |
| 118 | +// Dials `to` from `from` so identify runs and a live connection backs the DHT |
| 119 | +// RPCs that follow. |
| 120 | +static bool dial(LibP2PCtx *from, const PeerInfoWaiter *to) { |
| 121 | + NimFfiStr connAddrs[MAX_ADDRS]; |
| 122 | + for (size_t i = 0; i < to->naddrs; i++) |
| 123 | + connAddrs[i] = nimffi_str(to->addrs[i]); |
| 124 | + ConnectRequest req = {nimffi_str(to->peerId), {connAddrs, to->naddrs}, 0}; |
| 125 | + BoolWaiter bw; |
| 126 | + return AWAIT_BOOL(bw, libp2p_ctx_connect(from, &req, on_bool, &bw), |
| 127 | + "connect"); |
| 128 | +} |
| 129 | + |
| 130 | +int main(void) { |
| 131 | + int status = 1; |
| 132 | + BoolWaiter bw; |
| 133 | + PeerInfoWaiter serverInfo; |
| 134 | + |
| 135 | + LibP2PCtx *server = kadNode("/ip4/127.0.0.1/tcp/5031", "server", NULL); |
| 136 | + if (!server) |
| 137 | + return 1; |
| 138 | + if (!AWAIT_BOOL(bw, libp2p_ctx_start(server, on_bool, &bw), "start server")) |
| 139 | + goto cleanup_server; |
| 140 | + if (!await_peerinfo(server, &serverInfo, "server peerinfo")) |
| 141 | + goto cleanup_server; |
| 142 | + printf("Server: %s\n", serverInfo.peerId); |
| 143 | + |
| 144 | + LibP2PCtx *client = kadNode("/ip4/127.0.0.1/tcp/5032", "client", &serverInfo); |
| 145 | + if (!client) |
| 146 | + goto cleanup_server; |
| 147 | + if (!AWAIT_BOOL(bw, libp2p_ctx_start(client, on_bool, &bw), "start client") || |
| 148 | + !dial(client, &serverInfo)) |
| 149 | + goto cleanup_client; |
| 150 | + |
| 151 | + // ── Value round-trip: server stores, client fetches over the DHT ────────── |
| 152 | + NimFfiBytes key = {(uint8_t *)ValueKey, strlen(ValueKey)}; |
| 153 | + NimFfiBytes value = {(uint8_t *)ValueData, strlen(ValueData)}; |
| 154 | + KadPutValueRequest putReq = {key, value}; |
| 155 | + if (!AWAIT_BOOL(bw, libp2p_ctx_kad_put_value(server, &putReq, on_bool, &bw), |
| 156 | + "put_value")) |
| 157 | + goto cleanup_client; |
| 158 | + printf("Server stored %zu bytes under '%s'\n", value.len, ValueKey); |
| 159 | + |
| 160 | + ValueWaiter vw; |
| 161 | + memset(&vw, 0, sizeof(vw)); |
| 162 | + // quorum 1: this two-node DHT has a single holder (the server), so one valid |
| 163 | + // response is enough. A negative value would ask for the default quorum (5), |
| 164 | + // which this topology can never reach; 0 is rejected by the binding. |
| 165 | + KadGetValueRequest getReq = {key, 1}; |
| 166 | + libp2p_ctx_kad_get_value(client, &getReq, on_value, &vw); |
| 167 | + if (!wait_done(&vw.done) || vw.err_code != 0) { |
| 168 | + fprintf(stderr, "get_value: %s\n", vw.err[0] ? vw.err : "unknown"); |
| 169 | + goto cleanup_client; |
| 170 | + } |
| 171 | + printf("Client fetched: %.*s\n", (int)vw.len, vw.data); |
| 172 | + if (vw.len != value.len || memcmp(vw.data, ValueData, vw.len) != 0) { |
| 173 | + fprintf(stderr, "Error: fetched value does not match\n"); |
| 174 | + goto cleanup_client; |
| 175 | + } |
| 176 | + |
| 177 | + // ── Provider round-trip: server advertises a CID, client discovers it ───── |
| 178 | + CidWaiter cw; |
| 179 | + memset(&cw, 0, sizeof(cw)); |
| 180 | + const char *cidData = "cbind-kad"; |
| 181 | + CreateCidRequest cidReq = {1, |
| 182 | + nimffi_str("raw"), |
| 183 | + nimffi_str("sha2-256"), |
| 184 | + {(uint8_t *)cidData, strlen(cidData)}}; |
| 185 | + libp2p_ctx_create_cid(client, &cidReq, on_cid, &cw); |
| 186 | + if (!wait_done(&cw.done) || cw.err_code != 0) { |
| 187 | + fprintf(stderr, "create_cid: %s\n", cw.err[0] ? cw.err : "unknown"); |
| 188 | + goto cleanup_client; |
| 189 | + } |
| 190 | + printf("CID: %s\n", cw.cid); |
| 191 | + |
| 192 | + if (!AWAIT_BOOL(bw, |
| 193 | + libp2p_ctx_kad_start_providing(server, nimffi_str(cw.cid), |
| 194 | + on_bool, &bw), |
| 195 | + "start_providing")) |
| 196 | + goto cleanup_client; |
| 197 | + printf("Server is providing %s\n", cw.cid); |
| 198 | + |
| 199 | + ProvidersWaiter pw; |
| 200 | + memset(&pw, 0, sizeof(pw)); |
| 201 | + libp2p_ctx_kad_get_providers(client, nimffi_str(cw.cid), on_providers, &pw); |
| 202 | + if (!wait_done(&pw.done) || pw.err_code != 0) { |
| 203 | + fprintf(stderr, "get_providers: %s\n", pw.err[0] ? pw.err : "unknown"); |
| 204 | + goto cleanup_client; |
| 205 | + } |
| 206 | + printf("Client found %zu provider(s)\n", pw.n); |
| 207 | + |
| 208 | + bool serverIsProvider = false; |
| 209 | + for (size_t i = 0; i < pw.n; i++) { |
| 210 | + printf(" provider: %s\n", pw.peerIds[i]); |
| 211 | + if (strcmp(pw.peerIds[i], serverInfo.peerId) == 0) |
| 212 | + serverIsProvider = true; |
| 213 | + } |
| 214 | + if (serverIsProvider) |
| 215 | + status = 0; |
| 216 | + else |
| 217 | + fprintf(stderr, "Error: server not found among providers\n"); |
| 218 | + |
| 219 | +cleanup_client: |
| 220 | + AWAIT_BOOL(bw, libp2p_ctx_stop(client, on_bool, &bw), "stop client"); |
| 221 | + libp2p_ctx_destroy(client); |
| 222 | +cleanup_server: |
| 223 | + AWAIT_BOOL(bw, libp2p_ctx_stop(server, on_bool, &bw), "stop server"); |
| 224 | + libp2p_ctx_destroy(server); |
| 225 | + return status; |
| 226 | +} |
0 commit comments