| … | … | @@ -1,20 +1,19 @@ |
| 1 | | // swarmd-probe — the smallest real client, and the start of the C++ client |
| 2 | | // library SWARMD-DESIGN.md §7 item 7 asks for. |
| 1 | // swarmd-probe — the smallest real client, and now a demonstration that the |
| 2 | // client library carries one. |
| 3 | 3 | // |
| 4 | 4 | // swarmd-probe --socket <path> --topic-hex <64 hex> [--protocol <name>] |
| 5 | 5 | // [--send <text>] [--seconds N] |
| 6 | 6 | // [--serve-file <path>] [--fetch-hex <64hex> --dest <path>] |
| 7 | 7 | // |
| 8 | | // The content flags exercise the content plane: --serve-file registers a file by |
| 9 | | // its overlay content hash and prints that hash; --fetch-hex asks the swarm for |
| 10 | | // one and waits for on-content-done. Bytes never cross the socket either way — |
| 11 | | // the daemon writes the file and prints where it put it. |
| 8 | // ⛔ IT USED TO BE 250 LINES OF WIRE, AND TWO OF THEM WERE WRONG. It framed, |
| 9 | // encoded, decoded and matched replies by hand; its send branch claimed any |
| 10 | // reply with an id at or above 4, which swept up the content calls and reported |
| 11 | // a fetch's answer as a send result. All of that is now swarmd::Client's, which |
| 12 | // is the entire argument for having a client library: a program that wants to |
| 13 | // use the daemon should not also be implementing it. |
| 12 | 14 | // |
| 13 | | // ⛔ IT USES THE GENERATED ENCODERS. The first version of the channel gate |
| 14 | | // hand-rolled compact-encoding in Python and the daemon answered "undecodable |
| 15 | | // message" to every frame — the test was wrong, not the daemon. Hand-writing |
| 16 | | // the wire in a client is exactly what generating it was supposed to stop, and |
| 17 | | // a test client is still a client. |
| 15 | // What remains here is what a probe is actually for — say what happened, in |
| 16 | // words another program can read. |
| 18 | 17 | |
| 19 | 18 | #include <cstdio> |
| 20 | 19 | #include <cstdlib> |
| … | … | @@ -22,39 +21,13 @@ |
| 22 | 21 | #include <string> |
| 23 | 22 | #include <vector> |
| 24 | 23 | |
| 25 | | #include <sys/socket.h> |
| 26 | | #include <sys/un.h> |
| 27 | 24 | #include <time.h> |
| 28 | 25 | #include <unistd.h> |
| 29 | 26 | |
| 30 | | extern "C" { |
| 31 | | #include <compact.h> |
| 32 | | #include <rpc.h> |
| 33 | | #include "swarmd_hrpc.h" |
| 34 | | #include "swarmd_schema.h" |
| 35 | | } |
| 36 | | |
| 37 | | static int dial(const char *path) { |
| 38 | | int fd = socket(AF_UNIX, SOCK_STREAM, 0); |
| 39 | | struct sockaddr_un a; |
| 40 | | memset(&a, 0, sizeof(a)); |
| 41 | | a.sun_family = AF_UNIX; |
| 42 | | strncpy(a.sun_path, path, sizeof(a.sun_path) - 1); |
| 43 | | if (connect(fd, (struct sockaddr *)&a, sizeof(a)) != 0) { close(fd); return -1; } |
| 44 | | return fd; |
| 45 | | } |
| 46 | | |
| 47 | | static void send_frame(int fd, uint8_t *buf, size_t len) { |
| 48 | | const uint32_t n = (uint32_t)len; |
| 49 | | uint8_t hdr[4] = { (uint8_t)(n & 0xff), (uint8_t)((n >> 8) & 0xff), |
| 50 | | (uint8_t)((n >> 16) & 0xff), (uint8_t)((n >> 24) & 0xff) }; |
| 51 | | (void)!write(fd, hdr, 4); |
| 52 | | (void)!write(fd, buf, len); |
| 53 | | free(buf); // ⛔ the caller frees what *_encode_* allocated |
| 54 | | } |
| 27 | #include "client.h" |
| 55 | 28 | |
| 56 | 29 | static bool hex32(const char *s, uint8_t *out) { |
| 57 | | if (strlen(s) != 64) return false; |
| 30 | if (s == nullptr || strlen(s) != 64) return false; |
| 58 | 31 | for (int i = 0; i < 32; i++) { |
| 59 | 32 | char b[3] = { s[i * 2], s[i * 2 + 1], 0 }; |
| 60 | 33 | out[i] = (uint8_t)strtoul(b, nullptr, 16); |
| … | … | @@ -62,10 +35,17 @@ static bool hex32(const char *s, uint8_t *out) { |
| 62 | 35 | return true; |
| 63 | 36 | } |
| 64 | 37 | |
| 38 | static std::string to_hex(const uint8_t *h) { |
| 39 | static const char *d = "0123456789abcdef"; |
| 40 | std::string out(64, '0'); |
| 41 | for (int i = 0; i < 32; i++) { out[i * 2] = d[h[i] >> 4]; out[i * 2 + 1] = d[h[i] & 15]; } |
| 42 | return out; |
| 43 | } |
| 44 | |
| 65 | 45 | int main(int argc, char **argv) { |
| 66 | 46 | const char *sock = nullptr, *topic_hex = nullptr, *proto = "swarmd-probe/v1"; |
| 67 | | const char *send_text = nullptr; |
| 68 | | const char *serve_file = nullptr, *fetch_hex = nullptr, *dest = nullptr; |
| 47 | const char *send_text = nullptr, *serve_file = nullptr; |
| 48 | const char *fetch_hex = nullptr, *dest = nullptr; |
| 69 | 49 | int seconds = 30; |
| 70 | 50 | for (int i = 1; i < argc; i++) { |
| 71 | 51 | if (!strcmp(argv[i], "--socket") && i + 1 < argc) sock = argv[++i]; |
| … | … | @@ -78,221 +58,124 @@ int main(int argc, char **argv) { |
| 78 | 58 | else if (!strcmp(argv[i], "--seconds") && i + 1 < argc) seconds = atoi(argv[++i]); |
| 79 | 59 | } |
| 80 | 60 | uint8_t topic[32]; |
| 81 | | if (!sock || !topic_hex || !hex32(topic_hex, topic)) { |
| 61 | if (!sock || !hex32(topic_hex, topic)) { |
| 82 | 62 | fprintf(stderr, "usage: swarmd-probe --socket <p> --topic-hex <64hex> " |
| 83 | | "[--protocol n] [--send text] [--seconds N]\n"); |
| 63 | "[--protocol n] [--send text] [--serve-file p] " |
| 64 | "[--fetch-hex h --dest p] [--seconds N]\n"); |
| 84 | 65 | return 2; |
| 85 | 66 | } |
| 86 | 67 | |
| 87 | | int fd = dial(sock); |
| 88 | | if (fd < 0) { fprintf(stderr, "cannot reach %s\n", sock); return 2; } |
| 89 | | |
| 90 | | uint8_t *buf = nullptr; size_t len = 0; |
| 91 | | |
| 92 | | swarmd_hello_request_t hello; |
| 93 | | memset(&hello, 0, sizeof(hello)); |
| 94 | | hello.name.data = (utf8_t *)"probe"; |
| 95 | | hello.name.len = 5; |
| 96 | | memset(hello.app_identity, 0x11, 32); |
| 97 | | hello.version = 1; |
| 98 | | if (swarmd_encode_hello(1, &hello, &buf, &len) >= 0) send_frame(fd, buf, len); |
| 99 | | |
| 100 | | swarmd_swarm_join_t sj; |
| 101 | | memset(&sj, 0, sizeof(sj)); |
| 102 | | memcpy(sj.topic, topic, 32); |
| 103 | | sj.server = true; sj.client = true; |
| 104 | | if (swarmd_encode_swarm_join(2, &sj, &buf, &len) >= 0) send_frame(fd, buf, len); |
| 105 | | |
| 106 | | swarmd_channel_open_t co; |
| 107 | | memset(&co, 0, sizeof(co)); |
| 108 | | co.protocol.data = (utf8_t *)proto; |
| 109 | | co.protocol.len = strlen(proto); |
| 110 | | co.id = nullptr; co.id_len = 0; |
| 111 | | co.slots = 2; |
| 112 | | if (swarmd_encode_channel_open(3, &co, &buf, &len) >= 0) send_frame(fd, buf, len); |
| 68 | swarmd::Client c; |
| 69 | |
| 70 | bool received = false, fetched = false, failed = false; |
| 71 | std::string received_text, fetched_path, fail_reason; |
| 72 | |
| 73 | c.on.on_message = [&](uint64_t, uint64_t, uint64_t, uint64_t, |
| 74 | const uint8_t *bytes, size_t len) { |
| 75 | received_text.assign((const char *)bytes, len); |
| 76 | received = true; |
| 77 | }; |
| 78 | c.on.on_content_done = [&](const uint8_t *, const std::string &path) { |
| 79 | fetched_path = path; |
| 80 | fetched = true; |
| 81 | }; |
| 82 | c.on.on_content_failed = [&](const uint8_t *, const std::string &why) { |
| 83 | fail_reason = why; |
| 84 | failed = true; |
| 85 | }; |
| 86 | c.on.on_disconnect = [&](const std::string &why) { |
| 87 | fprintf(stderr, "swarmd went away: %s\n", why.c_str()); |
| 88 | }; |
| 89 | |
| 90 | if (!c.connect(sock, "probe")) { |
| 91 | fprintf(stderr, "cannot reach %s: %s\n", sock, c.last_error().c_str()); |
| 92 | return 2; |
| 93 | } |
| 94 | if (!c.join(topic)) { fprintf(stderr, "join failed: %s\n", c.last_error().c_str()); return 2; } |
| 113 | 95 | |
| 96 | const uint64_t ch = c.open_channel(proto, {}, 2); |
| 97 | if (ch == 0) { fprintf(stderr, "channel failed: %s\n", c.last_error().c_str()); return 2; } |
| 114 | 98 | printf("joined topic, opened %s\n", proto); |
| 115 | 99 | fflush(stdout); |
| 116 | 100 | |
| 117 | 101 | // ---- content ---- |
| 118 | | // A serve is two calls: ask the daemon for the file's overlay hash, then |
| 102 | // A serve is two calls: ask the DAEMON for the file's overlay hash, then |
| 119 | 103 | // register it. Asking the daemon rather than computing it here is the point — |
| 120 | | // a hash from anywhere else is a hash the overlay will reject. |
| 104 | // a hash from anywhere else is one the overlay will reject. |
| 121 | 105 | if (serve_file) { |
| 122 | | swarmd_path_t pr; |
| 123 | | memset(&pr, 0, sizeof(pr)); |
| 124 | | pr.path.data = (utf8_t *)serve_file; |
| 125 | | pr.path.len = strlen(serve_file); |
| 126 | | if (swarmd_encode_content_hash(10, &pr, &buf, &len) >= 0) send_frame(fd, buf, len); |
| 106 | uint8_t h[32]; |
| 107 | if (!c.content_hash(serve_file, h)) { |
| 108 | fprintf(stderr, "cannot hash %s: %s\n", serve_file, c.last_error().c_str()); |
| 109 | return 2; |
| 110 | } |
| 111 | if (!c.content_serve(h, serve_file)) { |
| 112 | fprintf(stderr, "serve refused: %s\n", c.last_error().c_str()); |
| 113 | return 2; |
| 114 | } |
| 115 | printf("SERVING %s %s\n", to_hex(h).c_str(), serve_file); |
| 116 | fflush(stdout); |
| 127 | 117 | } |
| 128 | 118 | if (fetch_hex && dest) { |
| 129 | 119 | uint8_t h[32]; |
| 130 | 120 | if (!hex32(fetch_hex, h)) { fprintf(stderr, "bad --fetch-hex\n"); return 2; } |
| 131 | | swarmd_content_fetch_t cf; |
| 132 | | memset(&cf, 0, sizeof(cf)); |
| 133 | | memcpy(cf.hash, h, 32); |
| 134 | | cf.dest_path.data = (utf8_t *)dest; |
| 135 | | cf.dest_path.len = strlen(dest); |
| 136 | | if (swarmd_encode_content_fetch(12, &cf, &buf, &len) >= 0) send_frame(fd, buf, len); |
| 121 | if (!c.content_fetch(h, dest)) { |
| 122 | fprintf(stderr, "fetch refused: %s\n", c.last_error().c_str()); |
| 123 | return 2; |
| 124 | } |
| 137 | 125 | printf("fetching %s -> %s\n", fetch_hex, dest); |
| 138 | 126 | fflush(stdout); |
| 139 | 127 | } |
| 140 | 128 | |
| 141 | | // ⛔ DO NOT SEND ON A TIMER. The first version fired once, eight seconds in, |
| 142 | | // whether or not a peer existed — and a send onto a channel the remote has |
| 143 | | // not opened yet is discarded by protomux. It worked on a local testnet, |
| 144 | | // where the mesh forms in under a second, and would have been a coin toss on |
| 145 | | // the public DHT. The daemon now answers `accepted` truthfully, so the right |
| 146 | | // trigger is that answer: keep offering the frame until it is taken. |
| 147 | | uint64_t socket_bytes_in = 0; |
| 129 | // ⛔ OFFER THE FRAME UNTIL IT IS TAKEN, do not send on a timer. A frame |
| 130 | // offered before the remote opens the channel is discarded by protomux, and |
| 131 | // the daemon now says so — `accepted` is the trigger, not the clock. |
| 148 | 132 | const time_t deadline = time(nullptr) + seconds; |
| 149 | | // ⛔ SENDS GET THEIR OWN ID RANGE. Request ids are how a reply is matched to |
| 150 | | // the call that asked for it, and "anything at or above 4" swept up the |
| 151 | | // content replies too — the fetch's answer was decoded as a send-result and |
| 152 | | // reported as "sent (accepted): (null)". Ids below 100 are the one-shot |
| 153 | | // setup calls; sends count from 100. |
| 154 | | const uint64_t FIRST_SEND_ID = 100; |
| 155 | | uint64_t next_send_id = FIRST_SEND_ID; |
| 156 | | bool send_accepted = false; |
| 157 | | time_t last_attempt = 0; |
| 158 | | std::vector<uint8_t> in; |
| 159 | | while (time(nullptr) < deadline) { |
| 160 | | if (send_text && !send_accepted && time(nullptr) != last_attempt) { |
| 161 | | last_attempt = time(nullptr); |
| 162 | | swarmd_send_request_t sr; |
| 163 | | memset(&sr, 0, sizeof(sr)); |
| 164 | | sr.channel = 1; sr.slot = 0; sr.target = 0; |
| 165 | | sr.bytes = (uint8_t *)send_text; |
| 166 | | sr.bytes_len = strlen(send_text); |
| 167 | | if (swarmd_encode_send(next_send_id++, &sr, &buf, &len) >= 0) { |
| 168 | | send_frame(fd, buf, len); |
| 133 | bool sent = false; |
| 134 | int attempts = 0; |
| 135 | time_t last_try = 0; |
| 136 | |
| 137 | while (time(nullptr) < deadline && !received && !fetched && !failed) { |
| 138 | if (send_text && !sent && time(nullptr) != last_try) { |
| 139 | last_try = time(nullptr); |
| 140 | attempts++; |
| 141 | if (c.send(ch, 0, 0, (const uint8_t *)send_text, strlen(send_text))) { |
| 142 | sent = true; |
| 143 | printf("sent (accepted after %d attempt%s): %s\n", |
| 144 | attempts, attempts == 1 ? "" : "s", send_text); |
| 145 | fflush(stdout); |
| 169 | 146 | } |
| 170 | 147 | } |
| 171 | | |
| 172 | | uint8_t tmp[65536]; |
| 173 | | struct timeval tv = { 1, 0 }; |
| 174 | | fd_set rf; FD_ZERO(&rf); FD_SET(fd, &rf); |
| 175 | | if (select(fd + 1, &rf, nullptr, nullptr, &tv) <= 0) continue; |
| 176 | | ssize_t r = read(fd, tmp, sizeof(tmp)); |
| 177 | | if (r <= 0) break; |
| 178 | | socket_bytes_in += (uint64_t)r; |
| 179 | | in.insert(in.end(), tmp, tmp + r); |
| 180 | | |
| 181 | | for (;;) { |
| 182 | | if (in.size() < 4) break; |
| 183 | | const uint32_t n = (uint32_t)in[0] | ((uint32_t)in[1] << 8) | |
| 184 | | ((uint32_t)in[2] << 16) | ((uint32_t)in[3] << 24); |
| 185 | | if (in.size() < 4 + n) break; |
| 186 | | compact_state_t st = { 0, n, in.data() + 4 }; |
| 187 | | rpc_message_t msg; |
| 188 | | if (rpc_decode_message(&st, &msg) >= 0) { |
| 189 | | if (msg.type == rpc_response && msg.id == 10) { |
| 190 | | // The daemon's answer to CONTENT_HASH. Register it, and print |
| 191 | | // the hash so a caller can hand it to the other end. |
| 192 | | swarmd_hash_t hr; |
| 193 | | memset(&hr, 0, sizeof(hr)); |
| 194 | | compact_state_t body = { 0, msg.len, msg.data }; |
| 195 | | if (swarmd_hash_decode(&body, &hr) >= 0 && serve_file) { |
| 196 | | char hex[65]; |
| 197 | | for (int i = 0; i < 32; i++) sprintf(hex + i * 2, "%02x", hr.hash[i]); |
| 198 | | hex[64] = 0; |
| 199 | | swarmd_content_serve_t cs; |
| 200 | | memset(&cs, 0, sizeof(cs)); |
| 201 | | memcpy(cs.hash, hr.hash, 32); |
| 202 | | cs.path.data = (utf8_t *)serve_file; |
| 203 | | cs.path.len = strlen(serve_file); |
| 204 | | uint8_t *b2 = nullptr; size_t l2 = 0; |
| 205 | | if (swarmd_encode_content_serve(11, &cs, &b2, &l2) >= 0) |
| 206 | | send_frame(fd, b2, l2); |
| 207 | | printf("SERVING %s %s\n", hex, serve_file); |
| 208 | | fflush(stdout); |
| 209 | | } |
| 210 | | } else if (msg.type == rpc_response && msg.id >= FIRST_SEND_ID) { |
| 211 | | // The answer to one of our send attempts. `accepted` is the |
| 212 | | // daemon saying it had an open channel to put the frame on. |
| 213 | | swarmd_send_result_t res; |
| 214 | | memset(&res, 0, sizeof(res)); |
| 215 | | compact_state_t body = { 0, msg.len, msg.data }; |
| 216 | | if (swarmd_send_result_decode(&body, &res) >= 0 && res.accepted && |
| 217 | | !send_accepted) { |
| 218 | | send_accepted = true; |
| 219 | | printf("sent (accepted after %d attempt%s): %s\n", |
| 220 | | (int)(msg.id - FIRST_SEND_ID + 1), |
| 221 | | msg.id == FIRST_SEND_ID ? "" : "s", send_text); |
| 222 | | fflush(stdout); |
| 223 | | } |
| 224 | | } else if (msg.type == rpc_request) { |
| 225 | | // An event: no reply, and dispatching it tells us which one. |
| 226 | | swarmd_hrpc_handlers_t h; |
| 227 | | memset(&h, 0, sizeof(h)); |
| 228 | | static std::string got; |
| 229 | | h.ctx = &got; |
| 230 | | h.on_on_message = [](void *ctx, const swarmd_message_t *rm) { |
| 231 | | ((std::string *)ctx)->assign((const char *)rm->bytes, rm->bytes_len); |
| 232 | | }; |
| 233 | | static std::string done_path, fail_reason; |
| 234 | | static bool content_done = false, content_failed = false; |
| 235 | | h.on_on_content_done = [](void *, const swarmd_content_done_t *d) { |
| 236 | | done_path.assign((const char *)d->path.data, d->path.len); |
| 237 | | content_done = true; |
| 238 | | }; |
| 239 | | h.on_on_content_failed = [](void *, const swarmd_content_failed_t *f) { |
| 240 | | fail_reason.assign((const char *)f->reason.data, f->reason.len); |
| 241 | | content_failed = true; |
| 242 | | }; |
| 243 | | uint8_t *rep = nullptr; size_t rl = 0; |
| 244 | | swarmd_hrpc_dispatch(&h, &msg, &rep, &rl); |
| 245 | | if (rep) free(rep); |
| 246 | | if (content_done) { |
| 247 | | // §6: bulk bytes never cross this socket. This is the |
| 248 | | // number that says so — everything the daemon told us, |
| 249 | | // for a file of any size. |
| 250 | | printf("SOCKET-BYTES %llu\n", |
| 251 | | (unsigned long long)socket_bytes_in); |
| 252 | | printf("FETCHED %s\n", done_path.c_str()); |
| 253 | | fflush(stdout); |
| 254 | | close(fd); |
| 255 | | return 0; |
| 256 | | } |
| 257 | | if (content_failed) { |
| 258 | | printf("FETCH FAILED: %s\n", fail_reason.c_str()); |
| 259 | | fflush(stdout); |
| 260 | | close(fd); |
| 261 | | return 1; |
| 262 | | } |
| 263 | | if (!got.empty()) { |
| 264 | | printf("RECEIVED: %s\n", got.c_str()); |
| 265 | | fflush(stdout); |
| 266 | | close(fd); |
| 267 | | return 0; |
| 268 | | } |
| 269 | | } |
| 270 | | } |
| 271 | | in.erase(in.begin(), in.begin() + 4 + n); |
| 272 | | } |
| 148 | if (!c.poll(200)) break; // the daemon closed |
| 273 | 149 | } |
| 274 | 150 | |
| 275 | | if (fetch_hex) { |
| 276 | | printf("the fetch never completed inside the window\n"); |
| 277 | | close(fd); |
| 151 | if (received) { |
| 152 | printf("RECEIVED: %s\n", received_text.c_str()); |
| 153 | return 0; |
| 154 | } |
| 155 | if (fetched) { |
| 156 | // §6: bulk bytes never cross this socket. This is the number that says |
| 157 | // so — everything the daemon told us, for a file of any size. |
| 158 | printf("SOCKET-BYTES %llu\n", (unsigned long long)c.bytes_in()); |
| 159 | printf("FETCHED %s\n", fetched_path.c_str()); |
| 160 | return 0; |
| 161 | } |
| 162 | if (failed) { |
| 163 | printf("FETCH FAILED: %s\n", fail_reason.c_str()); |
| 278 | 164 | return 1; |
| 279 | 165 | } |
| 166 | if (fetch_hex) { printf("the fetch never completed inside the window\n"); return 1; } |
| 280 | 167 | if (serve_file) { |
| 281 | | // A holder answers requests; it has nothing to wait for. Reaching the end |
| 282 | | // of the window is what success looks like. |
| 168 | // A holder answers requests; it has nothing to wait for. Reaching the |
| 169 | // end of the window is what success looks like. |
| 283 | 170 | printf("served for %ds\n", seconds); |
| 284 | | close(fd); |
| 285 | 171 | return 0; |
| 286 | 172 | } |
| 287 | | // A pure sender has nothing to receive, so say what it actually knows. |
| 288 | 173 | if (send_text) { |
| 289 | | printf(send_accepted ? "sent, never answered\n" |
| 290 | | : "the daemon never accepted the frame — no channel was " |
| 291 | | "open on any peer\n"); |
| 292 | | close(fd); |
| 293 | | return send_accepted ? 0 : 1; |
| 174 | printf(sent ? "sent, never answered\n" |
| 175 | : "the daemon never accepted the frame — no channel was open " |
| 176 | "on any peer\n"); |
| 177 | return sent ? 0 : 1; |
| 294 | 178 | } |
| 295 | 179 | printf("no message received\n"); |
| 296 | | close(fd); |
| 297 | 180 | return 1; |
| 298 | 181 | } |