From 63cc9554f2cd93e46d0413a21cbce05014e2f2b3 Mon Sep 17 00:00:00 2001 From: eunhhu Date: Mon, 28 Sep 2026 19:41:08 +0100 Subject: Measure a link at rest: 132 bytes against a QUIC connection's 38,106 (094) M1 of design 011. A Pure node allows 1,000 to 1,000,000 senders; each sends one order and, once told it is stored, rests. At rest a link costs the node 132 bytes (the sender allowed, 73; its number heard, about 60) and no transport state; with a million the node runs in 139 MB. A quinn 0.11 service holding the same exchange's clients connected at rest costs 38,106 bytes a connection: 289 times as much, past the design's 1/100. A link at rest takes its next record in 0.56 ms to stored, one X25519 on the node included; a new QUIC connection takes 1.06 ms and 2.8 times the receiver's CPU, without storing. research/experiments/019-idle-links: links.c, many senders written from serve/link.h's packet format alone (a second implementation), and the quinn service and clients (quic/, built offline); run.py makes both measurements. --- .../2026-09-28-links-linux-aarch64.jsonl | 4 + .../2026-09-28-quinn-linux-aarch64.jsonl | 3 + research/experiments/019-idle-links/links.c | 624 ++++++++++++++++++++ .../experiments/019-idle-links/quic/Cargo.lock | 634 +++++++++++++++++++++ .../experiments/019-idle-links/quic/Cargo.toml | 18 + .../experiments/019-idle-links/quic/src/main.rs | 329 +++++++++++ research/experiments/019-idle-links/run.py | 362 ++++++++++++ 7 files changed, 1974 insertions(+) create mode 100644 research/experiments/019-idle-links/2026-09-28-links-linux-aarch64.jsonl create mode 100644 research/experiments/019-idle-links/2026-09-28-quinn-linux-aarch64.jsonl create mode 100644 research/experiments/019-idle-links/links.c create mode 100644 research/experiments/019-idle-links/quic/Cargo.lock create mode 100644 research/experiments/019-idle-links/quic/Cargo.toml create mode 100644 research/experiments/019-idle-links/quic/src/main.rs create mode 100644 research/experiments/019-idle-links/run.py (limited to 'research/experiments/019-idle-links') diff --git a/research/experiments/019-idle-links/2026-09-28-links-linux-aarch64.jsonl b/research/experiments/019-idle-links/2026-09-28-links-linux-aarch64.jsonl new file mode 100644 index 0000000..eccdf16 --- /dev/null +++ b/research/experiments/019-idle-links/2026-09-28-links-linux-aarch64.jsonl @@ -0,0 +1,4 @@ +{"senders": 1000, "idle_ms": 1000, "start_empty_seconds": 0.006, "empty": {"rss_kb": 7840, "hwm_kb": 7840, "anonymous_kb": 1920}, "start_allowed_seconds": 0.005, "allowed": {"rss_kb": 7872, "hwm_kb": 7872, "anonymous_kb": 1952}, "send": {"devices": 1000, "stored": 1000, "failed": 0, "resent": 0, "seconds": 0.28, "per_second": 3567, "stored_ms_p50": 70.335, "stored_ms_p99": 73.269, "stored_ms_max": 142.015}, "node_cpu_ms_per_send": 0.27, "sent": {"rss_kb": 10576, "hwm_kb": 10576, "anonymous_kb": 4576}, "after_idle": {"rss_kb": 10576, "hwm_kb": 10576, "anonymous_kb": 4576}, "wake": {"devices": 1000, "stored": 1000, "failed": 0, "resent": 0, "seconds": 0.567, "per_second": 1763, "stored_ms_p50": 0.558, "stored_ms_p99": 0.803, "stored_ms_max": 2.22}, "node_cpu_ms_per_wake": 0.29, "listener": {"listener_packets": 4000, "listener_received": 2000, "listener_deliveries": 2000, "listener_woken": 2000, "listener_swept": 1000, "listener_active": 1000}, "store_bytes": 9249128, "allowed_bytes_per_sender": 32.8, "after_idle_bytes_per_sender": 2719.7, "active_bytes_per_sender": 0.0} +{"senders": 10000, "idle_ms": 1000, "start_empty_seconds": 0.005, "empty": {"rss_kb": 7792, "hwm_kb": 7792, "anonymous_kb": 1872}, "start_allowed_seconds": 0.01, "allowed": {"rss_kb": 8656, "hwm_kb": 8656, "anonymous_kb": 2752}, "send": {"devices": 10000, "stored": 10000, "failed": 0, "resent": 0, "seconds": 2.706, "per_second": 3695, "stored_ms_p50": 67.724, "stored_ms_p99": 92.981, "stored_ms_max": 164.554}, "node_cpu_ms_per_send": 0.272, "sent": {"rss_kb": 20256, "hwm_kb": 20256, "anonymous_kb": 14272}, "after_idle": {"rss_kb": 20144, "hwm_kb": 20144, "anonymous_kb": 14160}, "wake": {"devices": 1000, "stored": 1000, "failed": 0, "resent": 0, "seconds": 0.548, "per_second": 1825, "stored_ms_p50": 0.558, "stored_ms_p99": 0.641, "stored_ms_max": 1.658}, "node_cpu_ms_per_wake": 0.28, "listener": {"listener_packets": 22000, "listener_received": 11000, "listener_deliveries": 11000, "listener_woken": 11000, "listener_swept": 10000, "listener_active": 1000}, "store_bytes": 9512137, "allowed_bytes_per_sender": 90.1, "after_idle_bytes_per_sender": 1258.3, "active_bytes_per_sender": 11.5} +{"senders": 100000, "idle_ms": 1000, "start_empty_seconds": 0.005, "empty": {"rss_kb": 7808, "hwm_kb": 7808, "anonymous_kb": 1888}, "start_allowed_seconds": 0.056, "allowed": {"rss_kb": 15440, "hwm_kb": 15440, "anonymous_kb": 9520}, "send": {"devices": 100000, "stored": 100000, "failed": 0, "resent": 0, "seconds": 26.95, "per_second": 3711, "stored_ms_p50": 68.192, "stored_ms_p99": 84.642, "stored_ms_max": 156.787}, "node_cpu_ms_per_send": 0.2727, "sent": {"rss_kb": 32176, "hwm_kb": 32176, "anonymous_kb": 26160}, "after_idle": {"rss_kb": 32112, "hwm_kb": 32112, "anonymous_kb": 26096}, "wake": {"devices": 1000, "stored": 1000, "failed": 0, "resent": 0, "seconds": 0.556, "per_second": 1799, "stored_ms_p50": 0.56, "stored_ms_p99": 0.689, "stored_ms_max": 1.716}, "node_cpu_ms_per_wake": 0.28, "listener": {"listener_packets": 202000, "listener_received": 101000, "listener_deliveries": 101000, "listener_woken": 101000, "listener_swept": 100000, "listener_active": 1000}, "store_bytes": 9512137, "allowed_bytes_per_sender": 78.2, "after_idle_bytes_per_sender": 247.9, "active_bytes_per_sender": 0.7} +{"senders": 1000000, "idle_ms": 1000, "start_empty_seconds": 0.005, "empty": {"rss_kb": 7792, "hwm_kb": 7792, "anonymous_kb": 1888}, "start_allowed_seconds": 0.674, "allowed": {"rss_kb": 79728, "hwm_kb": 79728, "anonymous_kb": 73824}, "send": {"devices": 1000000, "stored": 1000000, "failed": 0, "resent": 0, "seconds": 272.342, "per_second": 3672, "stored_ms_p50": 67.898, "stored_ms_p99": 132.387, "stored_ms_max": 403.293}, "node_cpu_ms_per_send": 0.2742, "sent": {"rss_kb": 148016, "hwm_kb": 148016, "anonymous_kb": 142016}, "after_idle": {"rss_kb": 148016, "hwm_kb": 148016, "anonymous_kb": 142016}, "wake": {"devices": 1000, "stored": 1000, "failed": 0, "resent": 0, "seconds": 0.56, "per_second": 1787, "stored_ms_p50": 0.564, "stored_ms_p99": 0.688, "stored_ms_max": 1.736}, "node_cpu_ms_per_wake": 0.28, "listener": {"listener_packets": 2002000, "listener_received": 1001000, "listener_deliveries": 1001000, "listener_woken": 1001000, "listener_swept": 1000000, "listener_active": 1000}, "store_bytes": 68232393, "allowed_bytes_per_sender": 73.7, "after_idle_bytes_per_sender": 143.5, "active_bytes_per_sender": 0.0} diff --git a/research/experiments/019-idle-links/2026-09-28-quinn-linux-aarch64.jsonl b/research/experiments/019-idle-links/2026-09-28-quinn-linux-aarch64.jsonl new file mode 100644 index 0000000..ce041f5 --- /dev/null +++ b/research/experiments/019-idle-links/2026-09-28-quinn-linux-aarch64.jsonl @@ -0,0 +1,3 @@ +{"clients": 1000, "empty": {"rss_kb": 3856, "hwm_kb": 3856, "anonymous_kb": 432}, "connect": {"clients": 1000, "failed": 0, "seconds": 3.069, "per_second": 326, "stored_ms_p50": 34.112, "stored_ms_p99": 3042.646}, "service_cpu_ms_per_connect": 0.71, "connected": {"rss_kb": 42640, "hwm_kb": 42640, "anonymous_kb": 38960}, "again": {"again": 1000, "seconds": 0.092}, "service_cpu_ms_per_again": 0.07, "after_again": {"rss_kb": 42640, "hwm_kb": 42640, "anonymous_kb": 38960}, "fresh": {"clients": 1000, "failed": 0, "seconds": 1.119, "per_second": 893, "stored_ms_p50": 1.062, "stored_ms_p99": 1.869}, "service_cpu_ms_per_fresh": 0.78, "bytes_per_client": 39452.7} +{"clients": 5000, "empty": {"rss_kb": 3856, "hwm_kb": 3856, "anonymous_kb": 432}, "connect": {"clients": 5000, "failed": 0, "seconds": 7.312, "per_second": 684, "stored_ms_p50": 31.847, "stored_ms_p99": 3067.619}, "service_cpu_ms_per_connect": 0.744, "connected": {"rss_kb": 191168, "hwm_kb": 191168, "anonymous_kb": 187488}, "again": {"again": 5000, "seconds": 0.355}, "service_cpu_ms_per_again": 0.048, "after_again": {"rss_kb": 192288, "hwm_kb": 192288, "anonymous_kb": 188608}, "fresh": {"clients": 1000, "failed": 0, "seconds": 1.063, "per_second": 941, "stored_ms_p50": 1.052, "stored_ms_p99": 1.21}, "service_cpu_ms_per_fresh": 0.77, "bytes_per_client": 38309.1} +{"clients": 10000, "empty": {"rss_kb": 3856, "hwm_kb": 3856, "anonymous_kb": 432}, "connect": {"clients": 10000, "failed": 0, "seconds": 15.04, "per_second": 665, "stored_ms_p50": 31.097, "stored_ms_p99": 3071.455}, "service_cpu_ms_per_connect": 0.745, "connected": {"rss_kb": 377232, "hwm_kb": 377232, "anonymous_kb": 373552}, "again": {"again": 10000, "seconds": 0.738}, "service_cpu_ms_per_again": 0.049, "after_again": {"rss_kb": 379392, "hwm_kb": 379392, "anonymous_kb": 375712}, "fresh": {"clients": 1000, "failed": 0, "seconds": 1.122, "per_second": 892, "stored_ms_p50": 1.063, "stored_ms_p99": 1.627}, "service_cpu_ms_per_fresh": 0.78, "bytes_per_client": 38207.5} diff --git a/research/experiments/019-idle-links/links.c b/research/experiments/019-idle-links/links.c new file mode 100644 index 0000000..732429a --- /dev/null +++ b/research/experiments/019-idle-links/links.c @@ -0,0 +1,624 @@ +#define _GNU_SOURCE +/* Many senders to one Pure node over links without connections + * (docs/design/011-pure-transport.md, M1). Written from the packet format in + * serve/link.h alone, with OpenSSL: a second implementation of it. + * + * links keys N FILE + * N X25519 keys into FILE (each a private then a public half, 32 bytes + * each) and FILE.senders, a line each for pure-serve --senders: + * dNNNNNNN KEY + * links send FILE MODULE INPUT HOST PORT NODE_KEY FIRST COUNT SEQ [WINDOW] + * devices FIRST to FIRST + COUNT - 1 each send record SEQ (an order) + * to INPUT (whose shape MODULE says) and wait until the node says it + * stored it, WINDOW devices at a time, each on a socket of its own; + * then a line of JSON: how many, how long, and the time from each + * device's first packet to its record stored. + * links run FILE MODULE INPUT HOST PORT NODE_KEY FIRST COUNT PERIOD_MS SECONDS + * devices FIRST.. each make a record every PERIOD_MS for SECONDS and + * send what is not stored (again after 200 ms, doubling to a second, + * without an answer), however the node comes and goes; then wait until + * all is stored and write a line of JSON: records made and stored, + * the records stored in each 100 ms, and the time from a record made + * to it stored. */ +#include "module.h" +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +enum { KEY = 32, HEADER = 21, TAG = 16, DATAGRAM = 1200, NONCE = 12 }; + +static void put16(unsigned char *p, uint32_t v) { p[0] = (unsigned char)v, p[1] = (unsigned char)(v >> 8); } +static void put32(unsigned char *p, uint32_t v) { for (int i = 0; i < 4; ++i) p[i] = (unsigned char)(v >> 8 * i); } +static void put64(unsigned char *p, uint64_t v) { for (int i = 0; i < 8; ++i) p[i] = (unsigned char)(v >> 8 * i); } +static uint32_t get32(const unsigned char *p) { + return (uint32_t)p[0] | (uint32_t)p[1] << 8 | (uint32_t)p[2] << 16 | (uint32_t)p[3] << 24; +} +static uint64_t get64(const unsigned char *p) { return (uint64_t)get32(p) | (uint64_t)get32(p + 4) << 32; } + +static uint64_t now_ns(void) { + struct timespec t; + clock_gettime(CLOCK_MONOTONIC, &t); + return (uint64_t)t.tv_sec * 1000000000u + (uint64_t)t.tv_nsec; +} + +static uint32_t fnv(uint32_t h, const void *bytes, size_t size) { + const unsigned char *p = bytes; + for (size_t i = 0; i < size; ++i) h = (h ^ p[i]) * 16777619u; + return h; +} + +static void fail(const char *what) { + fprintf(stderr, "links: %s\n", what); + exit(1); +} + +/* ---- Keys */ + +static void make_key(unsigned char out[2 * KEY]) { + EVP_PKEY *key = EVP_PKEY_Q_keygen(NULL, NULL, "X25519"); + size_t a = KEY, b = KEY; + if (!key || EVP_PKEY_get_raw_private_key(key, out, &a) != 1 || EVP_PKEY_get_raw_public_key(key, out + KEY, &b) != 1) + fail("cannot make a key"); + EVP_PKEY_free(key); +} + +typedef struct { unsigned char *keys; size_t first, count; } Making; + +static void *make_keys(void *context) { + Making *m = context; + for (size_t i = 0; i < m->count; ++i) make_key(m->keys + (m->first + i) * 2 * KEY); + return NULL; +} + +static int keys(size_t n, const char *path) { + unsigned char *all = malloc(n * 2 * KEY); + if (!all) fail("out of memory"); + pthread_t threads[4]; + Making parts[4]; + for (int t = 0; t < 4; ++t) { + parts[t] = (Making){all, n * t / 4, n * (t + 1) / 4 - n * t / 4}; + pthread_create(&threads[t], NULL, make_keys, &parts[t]); + } + for (int t = 0; t < 4; ++t) pthread_join(threads[t], NULL); + FILE *file = fopen(path, "wb"); + if (!file || fwrite(all, 2 * KEY, n, file) != n || fclose(file)) fail("cannot write the keys"); + char senders[4096]; + snprintf(senders, sizeof(senders), "%s.senders", path); + FILE *lines = fopen(senders, "w"); + if (!lines) fail("cannot write the senders"); + for (size_t i = 0; i < n; ++i) { + fprintf(lines, "d%07zu ", i); + for (int b = 0; b < KEY; ++b) fprintf(lines, "%02x", all[i * 2 * KEY + KEY + b]); + fputc('\n', lines); + } + if (fclose(lines)) fail("cannot write the senders"); + free(all); + return 0; +} + +/* ---- Keys of a link (serve/link.h): X25519, HKDF-SHA256, AES-128-GCM */ + +static bool secret_of(const unsigned char own[2 * KEY], const unsigned char peer[KEY], unsigned char out[32]) { + EVP_PKEY *mine = EVP_PKEY_new_raw_private_key(EVP_PKEY_X25519, NULL, own, KEY); + EVP_PKEY *theirs = EVP_PKEY_new_raw_public_key(EVP_PKEY_X25519, NULL, peer, KEY); + EVP_PKEY_CTX *c = mine && theirs ? EVP_PKEY_CTX_new(mine, NULL) : NULL; + size_t length = 32; + bool ok = c && EVP_PKEY_derive_init(c) == 1 && EVP_PKEY_derive_set_peer(c, theirs) == 1 && + EVP_PKEY_derive(c, out, &length) == 1 && length == 32; + EVP_PKEY_CTX_free(c); + EVP_PKEY_free(mine); + EVP_PKEY_free(theirs); + return ok; +} + +static EVP_KDF *hkdf; + +typedef struct { EVP_CIPHER_CTX *seal, *open; unsigned char base[NONCE]; } Aead; + +static bool derive(const unsigned char secret[32], const unsigned char from[KEY], const unsigned char to[KEY], + uint64_t from_run, uint64_t to_run, Aead *out) { + unsigned char info[16 + 2 * KEY + 16], key[16 + NONCE]; + memcpy(info, "pure link 0 keys", 16); + memcpy(info + 16, from, KEY); + memcpy(info + 16 + KEY, to, KEY); + put64(info + 16 + 2 * KEY, from_run); + put64(info + 24 + 2 * KEY, to_run); + static const char salt[] = "pure link 0"; + EVP_KDF_CTX *c = EVP_KDF_CTX_new(hkdf); + OSSL_PARAM params[] = {OSSL_PARAM_construct_utf8_string(OSSL_KDF_PARAM_DIGEST, (char *)"SHA256", 0), + OSSL_PARAM_construct_octet_string(OSSL_KDF_PARAM_KEY, (void *)secret, 32), + OSSL_PARAM_construct_octet_string(OSSL_KDF_PARAM_SALT, (void *)salt, sizeof(salt) - 1), + OSSL_PARAM_construct_octet_string(OSSL_KDF_PARAM_INFO, info, sizeof(info)), + OSSL_PARAM_construct_end()}; + bool ok = c && EVP_KDF_derive(c, key, sizeof(key), params) == 1; + EVP_KDF_CTX_free(c); + if (!ok) return false; + memcpy(out->base, key + 16, NONCE); + out->seal = EVP_CIPHER_CTX_new(); + out->open = EVP_CIPHER_CTX_new(); + return out->seal && out->open && EVP_EncryptInit_ex(out->seal, EVP_aes_128_gcm(), NULL, key, NULL) == 1 && + EVP_DecryptInit_ex(out->open, EVP_aes_128_gcm(), NULL, key, NULL) == 1; +} + +static void aead_free(Aead *a) { + EVP_CIPHER_CTX_free(a->seal); + EVP_CIPHER_CTX_free(a->open); + *a = (Aead){0}; +} + +static void nonce_of(unsigned char out[NONCE], const unsigned char base[NONCE], uint64_t pn) { + memcpy(out, base, NONCE); + for (int i = 0; i < 8; ++i) out[4 + i] ^= (unsigned char)(pn >> (56 - 8 * i)); +} + +static size_t seal(Aead *a, uint32_t id, uint64_t run, uint64_t pn, const unsigned char *body, size_t size, + unsigned char *out) { + unsigned char nonce[NONCE]; + int n = 0; + out[0] = 0x70; + put32(out + 1, id); + put64(out + 5, run); + put64(out + 13, pn); + nonce_of(nonce, a->base, pn); + if (EVP_EncryptInit_ex(a->seal, NULL, NULL, NULL, nonce) != 1 || EVP_EncryptUpdate(a->seal, NULL, &n, out, HEADER) != 1 || + EVP_EncryptUpdate(a->seal, out + HEADER, &n, body, (int)size) != 1 || + EVP_EncryptFinal_ex(a->seal, out + HEADER + n, &n) != 1 || + EVP_CIPHER_CTX_ctrl(a->seal, EVP_CTRL_GCM_GET_TAG, TAG, out + HEADER + size) != 1) + return 0; + return HEADER + size + TAG; +} + +static bool open_packet(Aead *a, const unsigned char *packet, size_t n, unsigned char *plain) { + unsigned char nonce[NONCE], tag[TAG]; + int length = 0, rest = 0; + size_t size = n - HEADER - TAG; + memcpy(tag, packet + n - TAG, TAG); + nonce_of(nonce, a->base, get64(packet + 13)); + return EVP_DecryptInit_ex(a->open, NULL, NULL, NULL, nonce) == 1 && + EVP_DecryptUpdate(a->open, NULL, &length, packet, HEADER) == 1 && + EVP_DecryptUpdate(a->open, plain, &length, packet + HEADER, (int)size) == 1 && + EVP_CIPHER_CTX_ctrl(a->open, EVP_CTRL_GCM_SET_TAG, TAG, tag) == 1 && + EVP_DecryptFinal_ex(a->open, plain + length, &rest) == 1; +} + +/* ---- What is sent: an order as the receiving input stores it */ + +static PureModule module; + +static uint32_t shape_of(const PureModule *m, uint32_t input) { + const PureInputShape *s = &m->input[input]; + uint32_t h = fnv(2166136261u, &s->prefix, sizeof(s->prefix)); + for (uint32_t f = 0; f < s->field_count; ++f) { + const PureField *field = &m->field[s->field_start + f]; + unsigned char fixed[7] = {field->type, (unsigned char)field->rows, (unsigned char)(field->rows >> 8)}; + put32(fixed + 3, field->offset); + h = fnv(fnv(h, field->name, strlen(field->name) + 1), fixed, sizeof(fixed)); + } + return h; +} + +/* A message of input: each u32 field 1, each number 0, each text or bytes "pen". */ +static size_t order(const PureModule *m, uint32_t input, unsigned char *out) { + const PureInputShape *s = &m->input[input]; + memset(out, 0, s->prefix); + size_t at = s->prefix; + for (uint32_t f = 0; f < s->field_count; ++f) { + const PureField *field = &m->field[s->field_start + f]; + if (field->type == PURE_FIELD_U32) put32(out + field->offset, 1); + else if (field->type == PURE_FIELD_TEXT || field->type == PURE_FIELD_BYTES) { + if (field->offset != PURE_FIELD_REST) put32(out + field->offset, 3); + memcpy(out + at, "pen", 3); + at += 3; + } + } + return at; +} + +/* ---- Devices */ + +typedef struct { + size_t device; + int fd; + uint64_t run, pn, started, sent, in_run; + Aead out, in; + unsigned char secret[32]; + bool busy, in_ready; + unsigned tries; +} Slot; + +static unsigned char *all_keys, node_key[KEY], payload[512]; +static size_t payload_size; +static uint32_t shape, node_id; +static const char *input_name; +static uint64_t seq; +static double *waited; +static size_t waited_count, resent, failed; + +static uint64_t random64(void) { + uint64_t v = 0; + while (getrandom(&v, sizeof(v), 0) != sizeof(v) || !v) {} + return v; +} + +static void send_record(Slot *s) { + unsigned char body[DATAGRAM], packet[DATAGRAM]; + char source[16]; + snprintf(source, sizeof(source), "d%07zu", s->device); + size_t at = 0, a = strlen(source), b = strlen(input_name); + body[at++] = 1; /* STREAM */ + body[at++] = (unsigned char)a; + memcpy(body + at, source, a); + at += a; + body[at++] = (unsigned char)b; + memcpy(body + at, input_name, b); + at += b; + put32(body + at, shape); + at += 4; + body[at++] = 2; /* RECORD */ + put64(body + at, seq); + put32(body + at + 8, (uint32_t)payload_size); + put32(body + at + 12, 0); + put16(body + at + 16, (uint32_t)payload_size); + memcpy(body + at + 18, payload, payload_size); + at += 18 + payload_size; + const unsigned char *own = all_keys + s->device * 2 * KEY; + size_t n = seal(&s->out, fnv(2166136261u, own + KEY, KEY), s->run, s->pn++, body, at, packet); + if (!n || send(s->fd, packet, n, 0) < 0) { /* sent again at its time */ } + s->sent = now_ns(); +} + +static void start_device(Slot *s, size_t device) { + const unsigned char *own = all_keys + device * 2 * KEY; + s->device = device; + s->run = random64(); + s->pn = 0; + s->tries = 0; + s->busy = true; + s->in_ready = false; + s->started = now_ns(); + if (!secret_of(own, node_key, s->secret) || !derive(s->secret, own + KEY, node_key, s->run, 0, &s->out)) + fail("cannot make a device's keys"); + send_record(s); +} + +/* An answer: done once it says the record is stored. */ +static bool take_answer(Slot *s, const unsigned char *packet, size_t n) { + unsigned char plain[DATAGRAM]; + if (n < HEADER + TAG || packet[0] != 0x70 || get32(packet + 1) != node_id) return false; + const unsigned char *own = all_keys + s->device * 2 * KEY; + uint64_t run = get64(packet + 5); + if (!s->in_ready || s->in_run != run) { + aead_free(&s->in); + if (!derive(s->secret, node_key, own + KEY, run, s->run, &s->in)) return false; + s->in_ready = true; + s->in_run = run; + } + if (!open_packet(&s->in, packet, n, plain)) return false; + size_t size = n - HEADER - TAG; + for (size_t at = 0; at < size;) { + unsigned type = plain[at++]; + if (type == 3) { /* ACK */ + if (at + 13 > size) return false; + at += 13 + 16u * plain[at + 12]; + } else if (type == 4) { /* STORED */ + if (at + 1 > size) return false; + size_t a = plain[at], b = at + 1 + a < size ? plain[at + 1 + a] : 0; + if (at + 2 + a + b + 13 > size) return false; + uint64_t heard = get64(plain + at + 2 + a + b); + at += 2 + a + b + 13; + if (heard >= seq) return true; + } else + return false; + } + return false; +} + +static int by_value(const void *a, const void *b) { + double x = *(const double *)a, y = *(const double *)b; + return x < y ? -1 : x > y; +} + +static int send_all(int argc, char **argv) { + const char *path = argv[2], *module_path = argv[3]; + input_name = argv[4]; + size_t first = strtoull(argv[8], NULL, 10), count = strtoull(argv[9], NULL, 10), window = argc > 11 ? strtoull(argv[11], NULL, 10) : 256; + seq = strtoull(argv[10], NULL, 10); + char error[256]; + if (pure_module_read(module_path, &module, error, sizeof(error)) != PURE_OK) fail(error); + int input = pure_module_input(&module, input_name); + if (input < 0) fail("no such input"); + shape = shape_of(&module, (uint32_t)input); + payload_size = order(&module, (uint32_t)input, payload); + for (int i = 0; i < KEY; ++i) sscanf(argv[7] + 2 * i, "%2hhx", &node_key[i]); + node_id = fnv(2166136261u, node_key, KEY); + FILE *file = fopen(path, "rb"); + if (!file) fail("cannot read the keys"); + fseek(file, 0, SEEK_END); + size_t n = (size_t)ftell(file) / (2 * KEY); + if (first + count > n) fail("not so many keys"); + all_keys = malloc(n * 2 * KEY); + fseek(file, 0, SEEK_SET); + if (!all_keys || fread(all_keys, 2 * KEY, n, file) != n) fail("cannot read the keys"); + fclose(file); + hkdf = EVP_KDF_fetch(NULL, "HKDF", NULL); + struct addrinfo hints = {.ai_family = AF_UNSPEC, .ai_socktype = SOCK_DGRAM}, *found; + if (getaddrinfo(argv[5], argv[6], &hints, &found)) fail("no such host"); + if (window > count) window = count; + Slot *slots = calloc(window, sizeof(*slots)); + waited = malloc(count * sizeof(*waited)); + int poller = epoll_create1(0); + for (size_t k = 0; k < window; ++k) { + slots[k].fd = socket(found->ai_family, SOCK_DGRAM | SOCK_NONBLOCK, 0); + if (slots[k].fd < 0 || connect(slots[k].fd, found->ai_addr, found->ai_addrlen)) fail("no socket"); + struct epoll_event e = {.events = EPOLLIN, .data.u64 = k}; + epoll_ctl(poller, EPOLL_CTL_ADD, slots[k].fd, &e); + } + uint64_t began = now_ns(); + size_t next = 0, done = 0; + for (size_t k = 0; k < window; ++k) start_device(&slots[k], first + next++); + struct epoll_event events[256]; + while (done < count) { + int ready = epoll_wait(poller, events, 256, 20); + uint64_t now = now_ns(); + for (int e = 0; e < ready; ++e) { + Slot *s = &slots[events[e].data.u64]; + unsigned char packet[DATAGRAM + 64]; + for (ssize_t got; s->busy && (got = recv(s->fd, packet, sizeof(packet), 0)) > 0;) { + if (!take_answer(s, packet, (size_t)got)) continue; + waited[waited_count++] = (double)(now_ns() - s->started) / 1e6; + aead_free(&s->out); + aead_free(&s->in); + s->busy = false; + ++done; + if (next < count) start_device(s, first + next++); + } + } + for (size_t k = 0; k < window; ++k) { /* no answer: sent again, as a link does */ + Slot *s = &slots[k]; + if (!s->busy || now < s->sent + 200000000u) continue; + if (++s->tries > 50) { + ++failed, ++done; + aead_free(&s->out); + aead_free(&s->in); + s->busy = false; + if (next < count) start_device(s, first + next++); + continue; + } + ++resent; + send_record(s); + } + } + double seconds = (double)(now_ns() - began) / 1e9; + qsort(waited, waited_count, sizeof(*waited), by_value); + double p50 = waited_count ? waited[waited_count / 2] : 0, p99 = waited_count ? waited[waited_count * 99 / 100] : 0; + printf("{\"devices\":%zu,\"stored\":%zu,\"failed\":%zu,\"resent\":%zu,\"seconds\":%.3f,\"per_second\":%.0f," + "\"stored_ms_p50\":%.3f,\"stored_ms_p99\":%.3f,\"stored_ms_max\":%.3f}\n", + count, waited_count, failed, resent, seconds, (double)count / seconds, p50, p99, + waited_count ? waited[waited_count - 1] : 0); + freeaddrinfo(found); + return failed ? 1 : 0; +} + +/* ---- Devices that keep sending: an outbox each */ + +enum { OUTBOX = 64, BATCH = 8 }; +typedef struct { + size_t device; + int fd; + uint64_t run, pn, in_run, made, stored; /* records made and stored: numbers 1..made, 1..stored */ + uint64_t made_at[OUTBOX]; /* when each record not stored was made (ns), by number % OUTBOX */ + uint64_t next_make, sent, wait; /* when the next record is made; when last sent; the wait for an answer */ + Aead out, in; + unsigned char secret[32]; + bool in_ready; +} Device; + +static void send_outbox(Device *d, uint64_t now) { + unsigned char body[DATAGRAM], packet[DATAGRAM]; + char source[16]; + snprintf(source, sizeof(source), "d%07zu", d->device); + size_t at = 0, a = strlen(source), b = strlen(input_name); + body[at++] = 1; /* STREAM */ + body[at++] = (unsigned char)a; + memcpy(body + at, source, a); + at += a; + body[at++] = (unsigned char)b; + memcpy(body + at, input_name, b); + at += b; + put32(body + at, shape); + at += 4; + for (uint64_t seq = d->stored + 1; seq <= d->made && seq <= d->stored + BATCH; ++seq) { + body[at++] = 2; /* RECORD */ + put64(body + at, seq); + put32(body + at + 8, (uint32_t)payload_size); + put32(body + at + 12, 0); + put16(body + at + 16, (uint32_t)payload_size); + memcpy(body + at + 18, payload, payload_size); + at += 18 + payload_size; + } + const unsigned char *own = all_keys + d->device * 2 * KEY; + size_t n = seal(&d->out, fnv(2166136261u, own + KEY, KEY), d->run, d->pn++, body, at, packet); + if (!n || send(d->fd, packet, n, 0) < 0) { /* sent again at its time */ } + d->sent = now; +} + +/* An answer: how far the node stored (0 when it says nothing of it). */ +static uint64_t stored_in(Device *d, const unsigned char *packet, size_t n) { + unsigned char plain[DATAGRAM]; + if (n < HEADER + TAG || packet[0] != 0x70 || get32(packet + 1) != node_id) return 0; + const unsigned char *own = all_keys + d->device * 2 * KEY; + uint64_t run = get64(packet + 5), heard = 0; + if (!d->in_ready || d->in_run != run) { /* the node started again: its answers come in another run */ + Aead in = {0}; + if (!derive(d->secret, node_key, own + KEY, run, d->run, &in)) return 0; + if (!open_packet(&in, packet, n, plain)) { + aead_free(&in); + return 0; + } + aead_free(&d->in); + d->in = in; + d->in_ready = true; + d->in_run = run; + } else if (!open_packet(&d->in, packet, n, plain)) + return 0; + size_t size = n - HEADER - TAG; + for (size_t at = 0; at < size;) { + unsigned type = plain[at++]; + if (type == 3) { + if (at + 13 > size) break; + at += 13 + 16u * plain[at + 12]; + } else if (type == 4) { + if (at + 1 > size) break; + size_t a = plain[at], b = at + 1 + a < size ? plain[at + 1 + a] : 0; + if (at + 2 + a + b + 13 > size) break; + uint64_t said = get64(plain + at + 2 + a + b); + if (said > heard) heard = said; + at += 2 + a + b + 13; + } else + break; + } + return heard; +} + +static int run_all(char **argv) { + const char *path = argv[2], *module_path = argv[3]; + input_name = argv[4]; + size_t first = strtoull(argv[8], NULL, 10), count = strtoull(argv[9], NULL, 10); + uint64_t period = strtoull(argv[10], NULL, 10) * 1000000u, seconds = strtoull(argv[11], NULL, 10); + char error[256]; + if (pure_module_read(module_path, &module, error, sizeof(error)) != PURE_OK) fail(error); + int input = pure_module_input(&module, input_name); + if (input < 0) fail("no such input"); + shape = shape_of(&module, (uint32_t)input); + payload_size = order(&module, (uint32_t)input, payload); + for (int i = 0; i < KEY; ++i) sscanf(argv[7] + 2 * i, "%2hhx", &node_key[i]); + node_id = fnv(2166136261u, node_key, KEY); + FILE *file = fopen(path, "rb"); + if (!file) fail("cannot read the keys"); + fseek(file, 0, SEEK_END); + size_t n = (size_t)ftell(file) / (2 * KEY); + if (first + count > n) fail("not so many keys"); + all_keys = malloc(n * 2 * KEY); + fseek(file, 0, SEEK_SET); + if (!all_keys || fread(all_keys, 2 * KEY, n, file) != n) fail("cannot read the keys"); + fclose(file); + hkdf = EVP_KDF_fetch(NULL, "HKDF", NULL); + struct addrinfo hints = {.ai_family = AF_UNSPEC, .ai_socktype = SOCK_DGRAM}, *found; + if (getaddrinfo(argv[5], argv[6], &hints, &found)) fail("no such host"); + Device *devices = calloc(count, sizeof(*devices)); + int poller = epoll_create1(0); + uint64_t began = now_ns(), end = began + seconds * 1000000000u; + for (size_t k = 0; k < count; ++k) { + Device *d = &devices[k]; + const unsigned char *own = all_keys + (first + k) * 2 * KEY; + d->device = first + k; + d->run = random64(); + d->fd = socket(found->ai_family, SOCK_DGRAM | SOCK_NONBLOCK, 0); + if (d->fd < 0 || connect(d->fd, found->ai_addr, found->ai_addrlen)) fail("no socket"); + if (!secret_of(own, node_key, d->secret) || !derive(d->secret, own + KEY, node_key, d->run, 0, &d->out)) + fail("cannot make a device's keys"); + d->next_make = began + period * k / count; /* spread over the period */ + d->wait = 200000000u; + struct epoll_event e = {.events = EPOLLIN, .data.u64 = k}; + epoll_ctl(poller, EPOLL_CTL_ADD, d->fd, &e); + } + began = now_ns(); /* the keys took their time: the run starts now */ + end = began + seconds * 1000000000u; + for (size_t k = 0; k < count; ++k) devices[k].next_make = began + period * k / count; + size_t buckets = (size_t)(seconds + 60) * 10; + uint32_t *timeline = calloc(buckets, sizeof(*timeline)); + size_t latency_capacity = 1024, latency_count = 0; + double *latency = malloc(latency_capacity * sizeof(*latency)); + uint64_t made = 0, stored = 0, sends = 0, again = 0, last_report = began, last_scan = 0; + struct epoll_event events[256]; + for (;;) { + uint64_t now = now_ns(); + bool making = now < end; + if (!making && stored == made) break; + if (now > end + 60000000000u) break; /* a minute past the end: what is left was not stored */ + int ready = epoll_wait(poller, events, 256, 5); + now = now_ns(); + for (int e = 0; e < ready; ++e) { + Device *d = &devices[events[e].data.u64]; + unsigned char packet[DATAGRAM + 64]; + for (ssize_t got; (got = recv(d->fd, packet, sizeof(packet), 0)) > 0;) { + uint64_t heard = stored_in(d, packet, (size_t)got); + if (heard > d->made) heard = d->made; + for (uint64_t seq = d->stored + 1; seq <= heard; ++seq) { + size_t bucket = (size_t)((now - began) / 100000000u); + if (bucket < buckets) ++timeline[bucket]; + if (latency_count == latency_capacity) latency = realloc(latency, (latency_capacity *= 2) * sizeof(*latency)); + latency[latency_count++] = (double)(now - d->made_at[seq % OUTBOX]) / 1e6; + ++stored; + } + if (heard > d->stored) { + d->stored = heard; + d->wait = 200000000u; + } + } + } + if (now - last_scan < 5000000u) continue; /* the devices' clocks, every 5 ms */ + last_scan = now; + for (size_t k = 0; k < count; ++k) { + Device *d = &devices[k]; + if (making && now >= d->next_make && d->made - d->stored < OUTBOX) { + d->made_at[++d->made % OUTBOX] = now; + ++made; + d->next_make += period; + send_outbox(d, now); + ++sends; + } else if (d->stored < d->made && now >= d->sent + d->wait) { /* no answer: again, waiting longer */ + send_outbox(d, now); + ++sends, ++again; + d->wait = d->wait * 2 > 1000000000u ? 1000000000u : d->wait * 2; + } + } + if (now - last_report >= 1000000000u) { + last_report = now; + fprintf(stderr, "{\"t\":%.1f,\"made\":%" PRIu64 ",\"stored\":%" PRIu64 "}\n", (double)(now - began) / 1e9, made, stored); + } + } + qsort(latency, latency_count, sizeof(*latency), by_value); + /* The longest time without a record stored, from the first to the last one stored. */ + size_t last = 0, longest = 0, gap = 0, since = SIZE_MAX; + for (size_t b = 0; b < buckets; ++b) + if (timeline[b]) { + if (since != SIZE_MAX && b - since - 1 > longest) longest = b - since - 1, gap = since + 1; + since = b; + last = b; + } + printf("{\"devices\":%zu,\"made\":%" PRIu64 ",\"stored\":%" PRIu64 ",\"sends\":%" PRIu64 ",\"again\":%" PRIu64 + ",\"stored_ms_p50\":%.3f,\"stored_ms_p99\":%.3f,\"stored_ms_max\":%.3f,\"longest_gap_ms\":%zu,\"gap_from_ms\":%zu," + "\"timeline_100ms\":[", + count, made, stored, sends, again, latency_count ? latency[latency_count / 2] : 0, + latency_count ? latency[latency_count * 99 / 100] : 0, latency_count ? latency[latency_count - 1] : 0, + longest * 100, gap * 100); + for (size_t b = 0; b <= last; ++b) printf("%s%u", b ? "," : "", timeline[b]); + printf("]}\n"); + freeaddrinfo(found); + return stored == made ? 0 : 1; +} + +int main(int argc, char **argv) { + if (argc == 4 && !strcmp(argv[1], "keys")) return keys(strtoull(argv[2], NULL, 10), argv[3]); + if ((argc == 11 || argc == 12) && !strcmp(argv[1], "send")) return send_all(argc, argv); + if (argc == 12 && !strcmp(argv[1], "run")) return run_all(argv); + fputs("links keys N FILE | links send FILE MODULE INPUT HOST PORT NODE_KEY FIRST COUNT SEQ [WINDOW]\n" + " | links run FILE MODULE INPUT HOST PORT NODE_KEY FIRST COUNT PERIOD_MS SECONDS\n", stderr); + return 2; +} diff --git a/research/experiments/019-idle-links/quic/Cargo.lock b/research/experiments/019-idle-links/quic/Cargo.lock new file mode 100644 index 0000000..f31be94 --- /dev/null +++ b/research/experiments/019-idle-links/quic/Cargo.lock @@ -0,0 +1,634 @@ +# This file is automatically @generated by Cargo. +# It is not intended for manual editing. +version = 4 + +[[package]] +name = "bumpalo" +version = "3.20.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "72f5acc6cb2ba439de613abc23857ec3d78374d8ed5ac84e9d11336e87da8649" + +[[package]] +name = "bytes" +version = "1.12.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fc652a48c352aef3ea3aed32080501cf3ef6ed5da78602a020c991775b0aff04" + +[[package]] +name = "cc" +version = "1.4.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0ad534f4357a5264cce5019c989cf66a4f0dc4e0d1b1d15f8aacec0ff7360273" +dependencies = [ + "find-msvc-tools", + "shlex", +] + +[[package]] +name = "cfg-if" +version = "1.0.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9330f8b2ff13f34540b44e946ef35111825727b38d33286ef986142615121801" + +[[package]] +name = "cfg_aliases" +version = "0.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f079e83a288787bcd14a6aea84cee5c87a67c5a3e660c30f557a3d24761b3527" + +[[package]] +name = "chacha20" +version = "0.10.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "65c35e4b699c7e15ccbe7ee35c005e4fc0a278d22238a2857e6ce2dadeda1b06" +dependencies = [ + "cfg-if", + "cpufeatures", + "rand_core", +] + +[[package]] +name = "cpufeatures" +version = "0.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5ca28b0ae3115b884660db4118d803791fd6756b6e88f39c0f3f7859060d7566" +dependencies = [ + "libc", +] + +[[package]] +name = "find-msvc-tools" +version = "0.1.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d45db016d36b838f563236e9193d0ee6ce38f3f68b6c94e914b4929c96bbb890" + +[[package]] +name = "futures-core" +version = "0.3.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "92d699e522242e69e3003b94ecc1f960f3a5e015aa7c5d7486e65ad01dd94f5e" + +[[package]] +name = "futures-task" +version = "0.3.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cd417de3d1d015fc3bfd2b1ea46dfc7bab72ef86f1cc7cc9c78e728b34a6d1fd" + +[[package]] +name = "futures-util" +version = "0.3.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0d50a92467f8ba5dd6e3ee5d4bd04d73ab2e4e1c44474a0674821dfce14b79bc" +dependencies = [ + "futures-core", + "futures-task", + "pin-project-lite", + "slab", +] + +[[package]] +name = "getrandom" +version = "0.2.17" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ff2abc00be7fca6ebc474524697ae276ad847ad0a6b3faa4bcb027e9a4614ad0" +dependencies = [ + "cfg-if", + "js-sys", + "libc", + "wasi", + "wasm-bindgen", +] + +[[package]] +name = "getrandom" +version = "0.4.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "300e883d756b2e4ec94e02791f39b04b522276138852cfc41d9fb7e904106099" +dependencies = [ + "cfg-if", + "js-sys", + "libc", + "r-efi", + "rand_core", + "wasm-bindgen", +] + +[[package]] +name = "idle-quic" +version = "0.1.0" +dependencies = [ + "quinn", + "ring", + "rustls", + "tokio", +] + +[[package]] +name = "js-sys" +version = "0.3.104" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0e0c1080212aad755ea003d18543e8768dd432c48819efd73a7bf1e39b7a5a3a" +dependencies = [ + "cfg-if", + "futures-util", + "wasm-bindgen", +] + +[[package]] +name = "libc" +version = "0.2.189" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3eaf3ede3fee6db1a4c2ee091bf8a8b4dccdc6d17f656fb07896ee72867612f2" + +[[package]] +name = "lru-slab" +version = "0.1.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "112b39cec0b298b6c1999fee3e31427f74f676e4cb9879ed1a121b43661a4154" + +[[package]] +name = "mio" +version = "1.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "30d65c71f1ce40ab09135ce117d742b9f8a19ff91a41a8b57ed50bc2de59c427" +dependencies = [ + "libc", + "wasi", + "windows-sys 0.61.2", +] + +[[package]] +name = "once_cell" +version = "1.21.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9f7c3e4beb33f85d45ae3e3a1792185706c8e16d043238c593331cc7cd313b50" + +[[package]] +name = "pin-project-lite" +version = "0.2.17" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a89322df9ebe1c1578d689c92318e070967d1042b512afbe49518723f4e6d5cd" + +[[package]] +name = "proc-macro2" +version = "1.0.107" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "985e7ec9bb745e6ce6535b544d84d6cd6f7ad8bd711c398938ae983b91a766d9" +dependencies = [ + "unicode-ident", +] + +[[package]] +name = "quinn" +version = "0.11.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0c1a41e437b6bbd489372cd4971de128e85c855f56c57f283d20ff016cf7c0a8" +dependencies = [ + "bytes", + "cfg_aliases", + "pin-project-lite", + "quinn-proto", + "quinn-udp", + "rustc-hash", + "rustls", + "socket2", + "thiserror", + "tokio", + "tracing", + "web-time", +] + +[[package]] +name = "quinn-proto" +version = "0.11.17" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "04759210543be93709136e28212294a659ef5001836ff4eab4d663e4529bba83" +dependencies = [ + "bytes", + "getrandom 0.4.3", + "lru-slab", + "rand", + "rand_pcg", + "ring", + "rustc-hash", + "rustls", + "rustls-pki-types", + "slab", + "thiserror", + "tinyvec", + "tracing", + "web-time", +] + +[[package]] +name = "quinn-udp" +version = "0.5.15" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "35a133f956daabe89a61a685c2649f13d82d5aa4bd5d12d1277e1072a21c0694" +dependencies = [ + "cfg_aliases", + "libc", + "once_cell", + "socket2", + "tracing", + "windows-sys 0.61.2", +] + +[[package]] +name = "quote" +version = "1.0.47" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1fbf4db142a473a8d80c26bbf18454ed458bf8d26c8219c331daecfdbd079001" +dependencies = [ + "proc-macro2", +] + +[[package]] +name = "r-efi" +version = "6.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f8dcc9c7d52a811697d2151c701e0d08956f92b0e24136cf4cf27b57a6a0d9bf" + +[[package]] +name = "rand" +version = "0.10.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c7f5fa3a058cd35567ef9bfa5e75732bee0f9e4c55fa90477bef2dfcdbc4be80" +dependencies = [ + "chacha20", + "getrandom 0.4.3", + "rand_core", +] + +[[package]] +name = "rand_core" +version = "0.10.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "63b8176103e19a2643978565ca18b50549f6101881c443590420e4dc998a3c69" + +[[package]] +name = "rand_pcg" +version = "0.10.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "caa0f4137e1c0a72f4c651489402276c8e8e1cf081f3b0ba156d2cbeef09e86a" +dependencies = [ + "rand_core", +] + +[[package]] +name = "ring" +version = "0.17.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a4689e6c2294d81e88dc6261c768b63bc4fcdb852be6d1352498b114f61383b7" +dependencies = [ + "cc", + "cfg-if", + "getrandom 0.2.17", + "libc", + "untrusted", + "windows-sys 0.52.0", +] + +[[package]] +name = "rustc-hash" +version = "2.1.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6b1e7f9a428571be2dc5bc0505c13fb6bf936822b894ec87abf8a08a4e51742d" + +[[package]] +name = "rustls" +version = "0.23.43" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0283386ce02abc0151e1761d08802dfe86c173b0b494af5cbc086574e453da06" +dependencies = [ + "once_cell", + "ring", + "rustls-pki-types", + "rustls-webpki", + "subtle", + "zeroize", +] + +[[package]] +name = "rustls-pki-types" +version = "1.15.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2f4925028c7eb5d1fcdaf196971378ed9d2c1c4efc7dc5d011256f76c99c0a96" +dependencies = [ + "web-time", + "zeroize", +] + +[[package]] +name = "rustls-webpki" +version = "0.103.15" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f3c3cf1d8b1e7d4927e2d154c3fcb02979afb9939629c62cd9048d4f07b60ac2" +dependencies = [ + "ring", + "rustls-pki-types", + "untrusted", +] + +[[package]] +name = "rustversion" +version = "1.0.23" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cf54715a573b99ac80df0bc206da022bcd442c974952c7b9720069370852e21f" + +[[package]] +name = "shlex" +version = "2.0.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f8fadd59c855ef2080decdef8ff161eb6661b86933c9d82e5ba29dc602a55aba" + +[[package]] +name = "slab" +version = "0.4.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0c790de23124f9ab44544d7ac05d60440adc586479ce501c1d6d7da3cd8c9cf5" + +[[package]] +name = "socket2" +version = "0.6.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c3d1e2c7f27f8d4cb10542a02c49005dbd6e93095799d6f3be745fae9f8fedd4" +dependencies = [ + "libc", + "windows-sys 0.61.2", +] + +[[package]] +name = "subtle" +version = "2.6.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "13c2bddecc57b384dee18652358fb23172facb8a2c51ccc10d74c157bdea3292" + +[[package]] +name = "syn" +version = "2.0.119" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "872831b642d1a07999a962a351ed35b955ea2cfc8f3862091e2a240a84f17297" +dependencies = [ + "proc-macro2", + "quote", + "unicode-ident", +] + +[[package]] +name = "syn" +version = "3.0.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e6275cddf4610d1775e6d1fe9469b2e77d0f39fd98fb7450901b821e0c53649f" +dependencies = [ + "proc-macro2", + "quote", + "unicode-ident", +] + +[[package]] +name = "thiserror" +version = "2.0.20" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ec86235f5fcc2a73650310756d2ac5b138a5780bbbdfae3eeccec992c435ba4f" +dependencies = [ + "thiserror-impl", +] + +[[package]] +name = "thiserror-impl" +version = "2.0.20" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bc04cd3e1236dd4a98afca4569f2deb3f120e5422a4023be2cb683f8486292af" +dependencies = [ + "proc-macro2", + "quote", + "syn 3.0.4", +] + +[[package]] +name = "tinyvec" +version = "1.12.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bb4ebadaa0af04fab11ae01eb5f9fdb5f9c5b875506e210e71c07873528baa7f" +dependencies = [ + "tinyvec_macros", +] + +[[package]] +name = "tinyvec_macros" +version = "0.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1f3ccbac311fea05f86f61904b462b55fb3df8837a366dfc601a0161d0532f20" + +[[package]] +name = "tokio" +version = "1.53.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "202caea871b69668250d242070849eb495be178ed697a3e98aebce5bc81a0bed" +dependencies = [ + "bytes", + "libc", + "mio", + "pin-project-lite", + "socket2", + "tokio-macros", + "windows-sys 0.61.2", +] + +[[package]] +name = "tokio-macros" +version = "2.7.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "78773a2a397f451582ce068015985c33193cf6dea8b74d2a639fe457b2f07b0e" +dependencies = [ + "proc-macro2", + "quote", + "syn 3.0.4", +] + +[[package]] +name = "tracing" +version = "0.1.44" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "63e71662fa4b2a2c3a26f570f037eb95bb1f85397f3cd8076caed2f026a6d100" +dependencies = [ + "pin-project-lite", + "tracing-core", +] + +[[package]] +name = "tracing-core" +version = "0.1.36" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "db97caf9d906fbde555dd62fa95ddba9eecfd14cb388e4f491a66d74cd5fb79a" +dependencies = [ + "once_cell", +] + +[[package]] +name = "unicode-ident" +version = "1.0.24" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e6e4313cd5fcd3dad5cafa179702e2b244f760991f45397d14d4ebf38247da75" + +[[package]] +name = "untrusted" +version = "0.9.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8ecb6da28b8a351d773b68d5825ac39017e680750f980f3a1a85cd8dd28a47c1" + +[[package]] +name = "wasi" +version = "0.11.1+wasi-snapshot-preview1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ccf3ec651a847eb01de73ccad15eb7d99f80485de043efb2f370cd654f4ea44b" + +[[package]] +name = "wasm-bindgen" +version = "0.2.127" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1b70935747edd64d89de3efa29d73789b806c15798f8e7dca4d8ac356b50ce70" +dependencies = [ + "cfg-if", + "once_cell", + "rustversion", + "wasm-bindgen-macro", + "wasm-bindgen-shared", +] + +[[package]] +name = "wasm-bindgen-macro" +version = "0.2.127" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "77775f8f3f7217702089053b94958f8f54061a3f663417df76e19cbdcca29bc1" +dependencies = [ + "quote", + "wasm-bindgen-macro-support", +] + +[[package]] +name = "wasm-bindgen-macro-support" +version = "0.2.127" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e11d33f857dc2fb11b8bc75aee111aa9cbeb12cd9f25efd3d4c2a3dd4e235284" +dependencies = [ + "bumpalo", + "proc-macro2", + "quote", + "syn 2.0.119", + "wasm-bindgen-shared", +] + +[[package]] +name = "wasm-bindgen-shared" +version = "0.2.127" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7ef64dbcc55df09c7e5a46182d181c2cfa3e925f3da937ea764728b4bbb9dcbf" +dependencies = [ + "unicode-ident", +] + +[[package]] +name = "web-time" +version = "1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5a6580f308b1fad9207618087a65c04e7a10bc77e02c8e84e9b00dd4b12fa0bb" +dependencies = [ + "js-sys", + "wasm-bindgen", +] + +[[package]] +name = "windows-link" +version = "0.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f0805222e57f7521d6a62e36fa9163bc891acd422f971defe97d64e70d0a4fe5" + +[[package]] +name = "windows-sys" +version = "0.52.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "282be5f36a8ce781fad8c8ae18fa3f9beff57ec1b52cb3de0789201425d9a33d" +dependencies = [ + "windows-targets", +] + +[[package]] +name = "windows-sys" +version = "0.61.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ae137229bcbd6cdf0f7b80a31df61766145077ddf49416a728b02cb3921ff3fc" +dependencies = [ + "windows-link", +] + +[[package]] +name = "windows-targets" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9b724f72796e036ab90c1021d4780d4d3d648aca59e491e6b98e725b84e99973" +dependencies = [ + "windows_aarch64_gnullvm", + "windows_aarch64_msvc", + "windows_i686_gnu", + "windows_i686_gnullvm", + "windows_i686_msvc", + "windows_x86_64_gnu", + "windows_x86_64_gnullvm", + "windows_x86_64_msvc", +] + +[[package]] +name = "windows_aarch64_gnullvm" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "32a4622180e7a0ec044bb555404c800bc9fd9ec262ec147edd5989ccd0c02cd3" + +[[package]] +name = "windows_aarch64_msvc" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "09ec2a7bb152e2252b53fa7803150007879548bc709c039df7627cabbd05d469" + +[[package]] +name = "windows_i686_gnu" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8e9b5ad5ab802e97eb8e295ac6720e509ee4c243f69d781394014ebfe8bbfa0b" + +[[package]] +name = "windows_i686_gnullvm" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0eee52d38c090b3caa76c563b86c3a4bd71ef1a819287c19d586d7334ae8ed66" + +[[package]] +name = "windows_i686_msvc" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "240948bc05c5e7c6dabba28bf89d89ffce3e303022809e73deaefe4f6ec56c66" + +[[package]] +name = "windows_x86_64_gnu" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "147a5c80aabfbf0c7d901cb5895d1de30ef2907eb21fbbab29ca94c5b08b1a78" + +[[package]] +name = "windows_x86_64_gnullvm" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "24d5b23dc417412679681396f2b49f3de8c1473deb516bd34410872eff51ed0d" + +[[package]] +name = "windows_x86_64_msvc" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "589f6da84c646204747d1270a2a5661ea66ed1cced2631d546fdfb155959f9ec" + +[[package]] +name = "zeroize" +version = "1.9.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e13c156562582aa81c60cb29407084cdb54c4164760106ab78e6c5b0858cf64e" diff --git a/research/experiments/019-idle-links/quic/Cargo.toml b/research/experiments/019-idle-links/quic/Cargo.toml new file mode 100644 index 0000000..cc0f527 --- /dev/null +++ b/research/experiments/019-idle-links/quic/Cargo.toml @@ -0,0 +1,18 @@ +[package] +name = "idle-quic" +version = "0.1.0" +edition = "2021" +publish = false + +# The QUIC side of M1 (../run.py, docs/design/011-pure-transport.md): quinn +# 0.11 with rustls and ring, as a Rust service would use it. +[dependencies] +quinn = { version = "0.11", default-features = false, features = ["rustls-ring", "runtime-tokio"] } +rustls = { version = "0.23", default-features = false, features = ["ring", "std"] } +tokio = { version = "1", features = ["rt", "rt-multi-thread", "macros", "time", "net", "sync", "io-util", "io-std"] } +ring = "0.17" + +[profile.release] +debug = 1 + +[workspace] diff --git a/research/experiments/019-idle-links/quic/src/main.rs b/research/experiments/019-idle-links/quic/src/main.rs new file mode 100644 index 0000000..cdb0376 --- /dev/null +++ b/research/experiments/019-idle-links/quic/src/main.rs @@ -0,0 +1,329 @@ +//! A QUIC service holding many clients at rest, for M1 of +//! docs/design/011-pure-transport.md: the same exchange as a Pure link's +//! (a client sends one order, the service stores it, says so, and the two +//! stay connected, quiet), over quinn 0.11 with rustls (AES-128-GCM). +//! +//! idle-quic serve CERT.der KEY.der [PORT] +//! a service on 127.0.0.1 (the port on stdout as JSON), one thread; +//! a line of JSON on stdout at every 1,000th order stored. Its +//! stateless reset key comes from KEY.der, so one started again on +//! the same port resets the connections of the one before at once +//! idle-quic connect PORT FIRST COUNT WINDOW +//! clients FIRST.. each connect, send order 1 and wait until it is +//! stored, WINDOW at a time over one endpoint; a line of JSON; the +//! connections stay open until stdin closes; then COUNT orders more +//! (number 2) over them, and another line +//! idle-quic steady PORT FIRST COUNT PERIOD_MS SECONDS WINDOW +//! clients FIRST.. connect (WINDOW at a time), then each makes an +//! order every PERIOD_MS for SECONDS and sends it over its connection; +//! one that fails (no answer in a second, or the connection lost) +//! connects again and sends it again. Then a line of JSON: orders made +//! and stored, connections made again, the orders stored in each +//! 100 ms, the time from an order made to it stored +use std::collections::HashMap; +use std::net::SocketAddr; +use std::sync::{Arc, Mutex}; +use std::time::{Duration, Instant}; + +use quinn::crypto::rustls::QuicClientConfig; +use quinn::{ClientConfig, Endpoint, IdleTimeout, ServerConfig, TransportConfig, VarInt}; +use rustls::client::danger::{HandshakeSignatureValid, ServerCertVerified, ServerCertVerifier}; +use rustls::pki_types::{CertificateDer, PrivateKeyDer, PrivatePkcs8KeyDer, ServerName, UnixTime}; +use rustls::{DigitallySignedStruct, SignatureScheme}; +use tokio::io::AsyncReadExt; + +fn transport() -> Arc { + let mut t = TransportConfig::default(); + t.max_idle_timeout(Some(IdleTimeout::from(VarInt::from_u32(3_600_000)))); // an hour: connections at rest stay + t.keep_alive_interval(None); + Arc::new(t) +} + +/// The order: the client's number and the order's. +fn order(client: u64, seq: u64) -> [u8; 64] { + let mut bytes = [0u8; 64]; + bytes[..8].copy_from_slice(&client.to_le_bytes()); + bytes[8..16].copy_from_slice(&seq.to_le_bytes()); + bytes[16..19].copy_from_slice(b"pen"); + bytes +} + +fn serve(cert: Vec, key: Vec, port: u16) { + let runtime = tokio::runtime::Builder::new_current_thread().enable_all().build().unwrap(); + runtime.block_on(async move { + let reset = ring::hmac::Key::new(ring::hmac::HMAC_SHA256, ring::digest::digest(&ring::digest::SHA256, &key).as_ref()); + let mut config = ServerConfig::with_single_cert( + vec![CertificateDer::from(cert)], + PrivateKeyDer::Pkcs8(PrivatePkcs8KeyDer::from(key)), + ) + .unwrap(); + config.transport_config(transport()); + let socket = std::net::UdpSocket::bind(SocketAddr::from(([127, 0, 0, 1], port))).unwrap(); + let endpoint = Endpoint::new(quinn::EndpointConfig::new(Arc::new(reset)), Some(config), socket, + quinn::default_runtime().unwrap()).unwrap(); + println!("{{\"port\":{}}}", endpoint.local_addr().unwrap().port()); + // What the service keeps: each client's last order heard, and its connection. + let heard: Arc>> = Arc::default(); + let kept: Arc>> = Arc::default(); + let stored = Arc::new(std::sync::atomic::AtomicU64::new(0)); + while let Some(incoming) = endpoint.accept().await { + let (heard, kept, stored) = (heard.clone(), kept.clone(), stored.clone()); + tokio::spawn(async move { + let Ok(connection) = incoming.await else { return }; + kept.lock().unwrap().push(connection.clone()); + while let Ok((mut send, mut recv)) = connection.accept_bi().await { + let Ok(bytes) = recv.read_to_end(1024).await else { return }; + if bytes.len() < 16 { + return; + } + let client = u64::from_le_bytes(bytes[..8].try_into().unwrap()); + let seq = u64::from_le_bytes(bytes[8..16].try_into().unwrap()); + let fresh = { + let mut heard = heard.lock().unwrap(); + let last = heard.entry(client).or_insert(0); + let fresh = seq > *last; + *last = (*last).max(seq); + fresh + }; + let _ = send.write_all(&[1]).await; + let _ = send.finish(); + if fresh { + let n = stored.fetch_add(1, std::sync::atomic::Ordering::Relaxed) + 1; + if n % 1000 == 0 { + println!("{{\"stored\":{n}}}"); + } + } + } + }); + } + }); +} + +#[derive(Debug)] +struct Anyone; // the test's own service: its certificate is not checked + +impl ServerCertVerifier for Anyone { + fn verify_server_cert(&self, _: &CertificateDer<'_>, _: &[CertificateDer<'_>], _: &ServerName<'_>, _: &[u8], + _: UnixTime) -> Result { + Ok(ServerCertVerified::assertion()) + } + fn verify_tls12_signature(&self, message: &[u8], cert: &CertificateDer<'_>, dss: &DigitallySignedStruct) + -> Result { + rustls::crypto::verify_tls12_signature(message, cert, dss, + &rustls::crypto::ring::default_provider().signature_verification_algorithms) + } + fn verify_tls13_signature(&self, message: &[u8], cert: &CertificateDer<'_>, dss: &DigitallySignedStruct) + -> Result { + rustls::crypto::verify_tls13_signature(message, cert, dss, + &rustls::crypto::ring::default_provider().signature_verification_algorithms) + } + fn supported_verify_schemes(&self) -> Vec { + rustls::crypto::ring::default_provider().signature_verification_algorithms.supported_schemes() + } +} + +fn percentile(values: &mut [f64], p: f64) -> f64 { + if values.is_empty() { + return 0.0; + } + values.sort_by(|a, b| a.partial_cmp(b).unwrap()); + values[((values.len() as f64 * p) as usize).min(values.len() - 1)] +} + +fn client_config() -> ClientConfig { + let crypto = rustls::ClientConfig::builder_with_provider(Arc::new(rustls::crypto::ring::default_provider())) + .with_protocol_versions(&[&rustls::version::TLS13]) + .unwrap() + .dangerous() + .with_custom_certificate_verifier(Arc::new(Anyone)) + .with_no_client_auth(); + let mut config = ClientConfig::new(Arc::new(QuicClientConfig::try_from(crypto).unwrap())); + config.transport_config(transport()); + config +} + +async fn exchange(connection: &quinn::Connection, client: u64, seq: u64) -> bool { + let Ok((mut send, mut recv)) = connection.open_bi().await else { return false }; + if send.write_all(&order(client, seq)).await.is_err() || send.finish().is_err() { + return false; + } + matches!(recv.read_to_end(16).await, Ok(answer) if answer == [1]) +} + +fn connect(port: u16, first: u64, count: u64, window: usize) { + let runtime = tokio::runtime::Builder::new_current_thread().enable_all().build().unwrap(); + runtime.block_on(async move { + let mut endpoint = Endpoint::client(SocketAddr::from(([127, 0, 0, 1], 0))).unwrap(); + endpoint.set_default_client_config(client_config()); + let server = SocketAddr::from(([127, 0, 0, 1], port)); + let window = Arc::new(tokio::sync::Semaphore::new(window)); + let kept: Arc>> = Arc::default(); + let waited: Arc>> = Arc::default(); + let began = Instant::now(); + let mut tasks = Vec::new(); + for client in first..first + count { + let permit = window.clone().acquire_owned().await.unwrap(); + let (endpoint, kept, waited) = (endpoint.clone(), kept.clone(), waited.clone()); + tasks.push(tokio::spawn(async move { + let started = Instant::now(); + let Ok(connecting) = endpoint.connect(server, "localhost") else { return false }; + let Ok(connection) = connecting.await else { return false }; + let ok = exchange(&connection, client, 1).await; + waited.lock().unwrap().push(started.elapsed().as_secs_f64() * 1000.0); + kept.lock().unwrap().push(connection); + drop(permit); + ok + })); + } + let mut failed = 0; + for task in tasks { + if !task.await.unwrap_or(false) { + failed += 1; + } + } + let seconds = began.elapsed().as_secs_f64(); + let mut w = waited.lock().unwrap().clone(); + println!("{{\"clients\":{count},\"failed\":{failed},\"seconds\":{seconds:.3},\"per_second\":{:.0},\"stored_ms_p50\":{:.3},\"stored_ms_p99\":{:.3}}}", + count as f64 / seconds, percentile(&mut w, 0.5), percentile(&mut w, 0.99)); + // At rest until stdin closes; then one more order each, over the same connections. + let mut rest = Vec::new(); + let _ = tokio::io::stdin().read_to_end(&mut rest).await; + let connections = kept.lock().unwrap().clone(); + let began = Instant::now(); + let mut again = 0; + for (k, connection) in connections.iter().enumerate() { + if exchange(connection, first + k as u64, 2).await { + again += 1; + } + } + println!("{{\"again\":{again},\"seconds\":{:.3}}}", began.elapsed().as_secs_f64()); + tokio::time::sleep(Duration::from_millis(100)).await; + }); +} + +/// Clients that keep sending, connecting again when their connection fails. +fn steady(port: u16, first: u64, count: u64, period: Duration, seconds: u64, window: usize) { + let runtime = tokio::runtime::Builder::new_current_thread().enable_all().build().unwrap(); + runtime.block_on(async move { + let mut endpoint = Endpoint::client(SocketAddr::from(([127, 0, 0, 1], 0))).unwrap(); + endpoint.set_default_client_config(client_config()); + let server = SocketAddr::from(([127, 0, 0, 1], port)); + let connecting = Arc::new(tokio::sync::Semaphore::new(window)); + let began = Instant::now(); + let end = began + Duration::from_secs(seconds) + Duration::from_secs(2); + let buckets = ((seconds + 70) * 10) as usize; + let timeline: Arc> = + Arc::new((0..buckets).map(|_| std::sync::atomic::AtomicU32::new(0)).collect()); + let waited: Arc>> = Arc::default(); + let (made, again) = (Arc::new(std::sync::atomic::AtomicU64::new(0)), Arc::new(std::sync::atomic::AtomicU64::new(0))); + let mut tasks = Vec::new(); + for (k, client) in (first..first + count).enumerate() { + let (endpoint, connecting, timeline, waited, made, again) = + (endpoint.clone(), connecting.clone(), timeline.clone(), waited.clone(), made.clone(), again.clone()); + let start = began + Duration::from_secs(2) + period.mul_f64(k as f64 / count as f64); + tasks.push(tokio::spawn(async move { + let mut connection: Option = None; + let mut seq = 0u64; + let mut next = start; + loop { + if connection.is_none() { + let permit = connecting.acquire().await.unwrap(); + if let Ok(c) = endpoint.connect(server, "localhost") { + if let Ok(Ok(c)) = tokio::time::timeout(Duration::from_secs(2), c).await { + connection = Some(c); + } + } + drop(permit); + if connection.is_none() { + tokio::time::sleep(Duration::from_millis(200)).await; + continue; + } + } + if Instant::now() >= end { + return; + } + tokio::time::sleep_until(next.into()).await; + seq += 1; + made.fetch_add(1, std::sync::atomic::Ordering::Relaxed); + let order_made = Instant::now(); + next += period; + loop { + let c = connection.clone().unwrap(); + let done = tokio::time::timeout(Duration::from_secs(1), exchange(&c, client, seq)).await; + if matches!(done, Ok(true)) { + let now = Instant::now(); + let bucket = ((now - began).as_millis() / 100) as usize; + if bucket < timeline.len() { + timeline[bucket].fetch_add(1, std::sync::atomic::Ordering::Relaxed); + } + waited.lock().unwrap().push((now - order_made).as_secs_f64() * 1000.0); + break; + } + // Lost: a new connection (a handshake), and the order again. + again.fetch_add(1, std::sync::atomic::Ordering::Relaxed); + connection = None; + while connection.is_none() { + let permit = connecting.acquire().await.unwrap(); + if let Ok(c) = endpoint.connect(server, "localhost") { + if let Ok(Ok(c)) = tokio::time::timeout(Duration::from_secs(2), c).await { + connection = Some(c); + } + } + drop(permit); + if connection.is_none() { + tokio::time::sleep(Duration::from_millis(200)).await; + } + } + } + if Instant::now() >= end { + return; + } + } + })); + } + for task in tasks { + let _ = task.await; + } + let mut w = waited.lock().unwrap().clone(); + let stored = w.len(); + let timeline: Vec = timeline.iter().map(|b| b.load(std::sync::atomic::Ordering::Relaxed)).collect(); + let last = timeline.iter().rposition(|&b| b > 0).unwrap_or(0); + let (mut longest, mut gap, mut since) = (0usize, 0usize, None::); + for (b, &n) in timeline.iter().enumerate().take(last + 1) { + if n > 0 { + if let Some(s) = since { + if b - s - 1 > longest { + longest = b - s - 1; + gap = s + 1; + } + } + since = Some(b); + } + } + let series: Vec = timeline[..=last].iter().map(|n| n.to_string()).collect(); + println!("{{\"clients\":{count},\"made\":{},\"stored\":{stored},\"again\":{},\"stored_ms_p50\":{:.3},\"stored_ms_p99\":{:.3},\"stored_ms_max\":{:.3},\"longest_gap_ms\":{},\"gap_from_ms\":{},\"timeline_100ms\":[{}]}}", + made.load(std::sync::atomic::Ordering::Relaxed), again.load(std::sync::atomic::Ordering::Relaxed), + percentile(&mut w, 0.5), percentile(&mut w, 0.99), percentile(&mut w, 1.0), longest * 100, gap * 100, + series.join(",")); + }); +} + +fn main() { + let args: Vec = std::env::args().collect(); + match args.get(1).map(String::as_str) { + Some("serve") if args.len() == 4 || args.len() == 5 => serve(std::fs::read(&args[2]).unwrap(), + std::fs::read(&args[3]).unwrap(), + args.get(4).map_or(0, |p| p.parse().unwrap())), + Some("steady") if args.len() == 8 => steady(args[2].parse().unwrap(), args[3].parse().unwrap(), + args[4].parse().unwrap(), + Duration::from_millis(args[5].parse().unwrap()), + args[6].parse().unwrap(), args[7].parse().unwrap()), + Some("connect") if args.len() == 6 => connect(args[2].parse().unwrap(), args[3].parse().unwrap(), + args[4].parse().unwrap(), args[5].parse().unwrap()), + _ => { + eprintln!("idle-quic serve CERT.der KEY.der [PORT] | idle-quic connect PORT FIRST COUNT WINDOW | \ + idle-quic steady PORT FIRST COUNT PERIOD_MS SECONDS WINDOW"); + std::process::exit(2); + } + } +} diff --git a/research/experiments/019-idle-links/run.py b/research/experiments/019-idle-links/run.py new file mode 100644 index 0000000..1a179e8 --- /dev/null +++ b/research/experiments/019-idle-links/run.py @@ -0,0 +1,362 @@ +#!/usr/bin/env python3 +"""M1 of docs/design/011-pure-transport.md: what a sender at rest costs a +Pure node that takes records from it over a link without a connection. + +For each count N of senders: a node (pure-serve, the receiver of +tests/check.py's link checks) allows N senders (--senders) and serves its +--link-port; build/links (links.c) makes each sender send one order and +wait until the node stored it; then every sender is quiet. The node's +memory (VmRSS, and Pss of its anonymous memory) is read when it has +allowed them, right after they sent, and after it let their active state +go (--link-idle-ms); then 1,000 senders at rest send one more order each +(the wake: one X25519 each on the node). A line of JSON for each N. + +With --quic, the same for a quinn service (quic/, built offline): n +clients connect, send one order each and wait until it is stored, and stay +connected at rest; the service's memory is read then; then each sends one +more order over its connection. + + python3 research/experiments/019-idle-links/run.py [--counts 1000,10000] + [--store DIRECTORY] [--out FILE.jsonl] [--quic] +""" +import argparse +import json +import os +import random +import select +import socket +import subprocess +import sys +import tempfile +import time +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[3] +sys.path.insert(0, str(ROOT / 'tests')) +import check # noqa: E402 (LINK_RECEIVER, compile_module) + + +def memory(pid): + """The node's resident memory and the proportional share of its anonymous memory (kB).""" + status = dict(line.split(':', 1) for line in Path(f'/proc/{pid}/status').read_text().splitlines() if ':' in line) + rollup = {} + for line in Path(f'/proc/{pid}/smaps_rollup').read_text().splitlines()[1:]: + name, value = line.split(':', 1) + rollup[name] = int(value.split()[0]) + return {'rss_kb': int(status['VmRSS'].split()[0]), 'hwm_kb': int(status['VmHWM'].split()[0]), + 'anonymous_kb': rollup.get('Anonymous', 0)} + + +def cpu_ms(pid): + """The process's CPU time so far (user and system), in ms.""" + fields = Path(f'/proc/{pid}/stat').read_text().rsplit(')', 1)[1].split() + return (int(fields[11]) + int(fields[12])) * 1000 / os.sysconf('SC_CLK_TCK') + + +def free_port(kind): + with socket.socket(socket.AF_INET, kind) as probe: + probe.bind(('127.0.0.1', 0)) + return probe.getsockname()[1] + + +def start(args): + process = subprocess.Popen([str(a) for a in args], stdout=subprocess.PIPE, stderr=subprocess.PIPE) + ready, _, _ = select.select([process.stdout], [], [], 600) + line = process.stdout.readline() if ready else b'' + if not line: + process.kill() + raise SystemExit(f'the node did not start: {process.stderr.read(2000)}') + return process, json.loads(line) + + +def view(port, name): + import http.client + connection = http.client.HTTPConnection('127.0.0.1', port, timeout=30) + connection.request('GET', f'/views/{name}') + body = connection.getresponse().read() + connection.close() + return json.loads(body) + + +def settle(pid, seconds): + """Memory once the node is quiet: read until it stops moving.""" + time.sleep(seconds) + last = memory(pid) + for _ in range(20): + time.sleep(0.5) + now = memory(pid) + if abs(now['rss_kb'] - last['rss_kb']) < 64: + return now + last = now + return last + + +def one(n, work, keys, module, idle_ms, window): + senders = work / f'senders-{n}' + with open(f'{keys}.senders') as source, open(senders, 'w') as out: + for _, line in zip(range(n), source): + out.write(line) + store = work / f'store-{n}' + subprocess.run(['rm', '-rf', str(store)], check=True) + port, link_port = free_port(socket.SOCK_STREAM), free_port(socket.SOCK_DGRAM) + base = [ROOT / 'build/pure-serve', '--module', module, '--store', store, '--port', port, '--key', work / 'node.key', + '--link-port', link_port, '--link-idle-ms', idle_ms] + row = {'senders': n, 'idle_ms': idle_ms} + # The node alone, then with the senders allowed. + for name, args in (('empty', base), ('allowed', base + ['--senders', senders])): + subprocess.run(['rm', '-rf', str(store)], check=True) + started = time.monotonic() + node, info = start(args) + row[f'start_{name}_seconds'] = round(time.monotonic() - started, 3) + row[name] = settle(node.pid, 1.0) + if name == 'empty': + node.terminate() + node.communicate(timeout=60) + node_key = info['public_key'] + links = [ROOT / 'build/links', 'send', keys, module, 'order', '127.0.0.1', link_port, node_key] + cpu = cpu_ms(node.pid) + sent = subprocess.run([str(a) for a in links + [0, n, 1, window]], capture_output=True, text=True) + assert sent.returncode == 0, sent.stderr + row['send'] = json.loads(sent.stdout) + row['node_cpu_ms_per_send'] = round((cpu_ms(node.pid) - cpu) / n, 4) + row['sent'] = memory(node.pid) + row['after_idle'] = settle(node.pid, idle_ms / 1000 + 1.5) + assert view(port, 'count') == n, view(port, 'count') + wake = min(n, 1000) + first = random.Random(n).randrange(0, n - wake + 1) + cpu = cpu_ms(node.pid) + woken = subprocess.run([str(a) for a in links + [first, wake, 2, 1]], capture_output=True, text=True) + assert woken.returncode == 0, woken.stderr + row['wake'] = json.loads(woken.stdout) + row['node_cpu_ms_per_wake'] = round((cpu_ms(node.pid) - cpu) / wake, 4) + assert view(port, 'count') == n + wake + node.terminate() + _, error = node.communicate(timeout=120) + stats = [json.loads(line) for line in error.decode().splitlines() if line.startswith('{"active"')][-1] + row['listener'] = {k: v for k, v in stats.items() if k.startswith('listener_')} + row['store_bytes'] = sum(f.stat().st_size for f in store.rglob('*') if f.is_file()) + for name in ('allowed', 'after_idle'): + row[f'{name}_bytes_per_sender'] = round((row[name]['anonymous_kb'] - row['empty']['anonymous_kb']) * 1024 / n, 1) + row['active_bytes_per_sender'] = round((row['sent']['anonymous_kb'] - row['after_idle']['anonymous_kb']) * 1024 / n, 1) + return row + + +def quic_one(n, work, window): + """The same with quinn: a service holding n clients connected at rest.""" + binary = Path(__file__).parent / 'quic/target/release/idle-quic' + server, info = start([binary, 'serve', work / 'cert.der', work / 'key.der']) + row = {'clients': n, 'empty': settle(server.pid, 1.0)} + cpu = cpu_ms(server.pid) + client = subprocess.Popen([str(binary), 'connect', str(info['port']), '0', str(n), str(window)], + stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE) + ready, _, _ = select.select([client.stdout], [], [], 3600) + line = client.stdout.readline() if ready else b'' + assert line, client.stderr.read(2000) + row['connect'] = json.loads(line) + row['service_cpu_ms_per_connect'] = round((cpu_ms(server.pid) - cpu) / n, 4) + row['connected'] = settle(server.pid, 2.0) + cpu = cpu_ms(server.pid) + client.stdin.close() # one more order each, over the connections at rest + ready, _, _ = select.select([client.stdout], [], [], 3600) + row['again'] = json.loads(client.stdout.readline()) + row['service_cpu_ms_per_again'] = round((cpu_ms(server.pid) - cpu) / n, 4) + row['after_again'] = memory(server.pid) + client.wait(timeout=600) + # Connections not kept: each order on a new one, one at a time (a handshake each). + cpu = cpu_ms(server.pid) + fresh = subprocess.run([str(binary), 'connect', str(info['port']), str(n), '1000', '1'], input=b'', + capture_output=True, timeout=600) + row['fresh'] = json.loads(fresh.stdout.decode().splitlines()[0]) + row['service_cpu_ms_per_fresh'] = round((cpu_ms(server.pid) - cpu) / 1000, 4) + server.kill() + server.wait() + row['bytes_per_client'] = round((row['connected']['anonymous_kb'] - row['empty']['anonymous_kb']) * 1024 / n, 1) + return row + + +def make_certificate(work): + """A self-signed certificate (ECDSA P-256) and its key, as DER.""" + subprocess.run(['openssl', 'req', '-x509', '-newkey', 'ec', '-pkeyopt', 'ec_paramgen_curve:prime256v1', '-nodes', + '-keyout', str(work / 'key.pem'), '-out', str(work / 'cert.pem'), '-days', '2', '-subj', '/CN=localhost', + '-addext', 'subjectAltName=DNS:localhost'], check=True, capture_output=True) + subprocess.run(['openssl', 'x509', '-in', str(work / 'cert.pem'), '-outform', 'DER', '-out', str(work / 'cert.der')], + check=True) + subprocess.run(['openssl', 'pkcs8', '-topk8', '-nocrypt', '-in', str(work / 'key.pem'), '-outform', 'DER', '-out', + str(work / 'key.der')], check=True) + + +def quic(counts, work, window, out): + root = Path(__file__).parent / 'quic' + subprocess.run(['cargo', 'build', '--release', '--offline', '--quiet'], cwd=root, check=True) + make_certificate(work) + for n in counts: + line = json.dumps(quic_one(n, work, window)) + print(line, flush=True) + if out: + with open(out, 'a') as file: + file.write(line + '\n') + + +def recovery(timeline, gap_from_ms, gap_ms): + """From a timeline of orders stored each 100 ms: the steady rate before + the gap, and how long after the gap began until a second of storing was + back at 90 % of it with what waited cleared (the rate seen at least once + and the backlog gone: the cumulative stored caught up with the steady + line).""" + start = gap_from_ms // 100 + before = timeline[max(0, start - 50):start] + steady = sum(before) / len(before) if before else 0 + total_before = sum(timeline[:start]) + for b in range(start + gap_ms // 100, len(timeline)): + expected = total_before + steady * (b + 1 - start) + if sum(timeline[:b + 1]) >= expected * 0.99: + return {'steady_per_100ms': round(steady, 1), 'caught_up_ms': (b + 1 - start) * 100} + return {'steady_per_100ms': round(steady, 1), 'caught_up_ms': None} + + +def restart_pure(n, work, keys, module, period_ms, seconds, kill_at, memo): + """M2 for Pure: n devices send a record every period_ms; the node is + killed (-9) at kill_at seconds and started again at once.""" + senders = work / f'senders-{n}' + with open(f'{keys}.senders') as source, open(senders, 'w') as out: + for _, line in zip(range(n), source): + out.write(line) + store = work / f'restart-{n}' + subprocess.run(['rm', '-rf', str(store), str(work / 'node.key.links')], check=True) + port, link_port = free_port(socket.SOCK_STREAM), free_port(socket.SOCK_DGRAM) + args = [ROOT / 'build/pure-serve', '--module', module, '--store', store, '--port', port, '--key', work / 'node.key', + '--link-port', link_port, '--senders', senders] + node, info = start(args) + generator = subprocess.Popen([str(a) for a in [ROOT / 'build/links', 'run', keys, module, 'order', '127.0.0.1', + link_port, info['public_key'], 0, n, period_ms, seconds]], + stdout=subprocess.PIPE, stderr=subprocess.PIPE) + for line in generator.stderr: # progress, a line a second + if json.loads(line).get('t', 0) >= kill_at: + break + cpu_before = cpu_ms(node.pid) + time.sleep(1.0) + steady_cpu = cpu_ms(node.pid) - cpu_before # ms of CPU in the second before the kill + node.kill() + node.wait() + killed = time.monotonic() + if not memo: + (work / 'node.key.links').unlink() + node, _ = start(args) + row = {'devices': n, 'period_ms': period_ms, 'memo': memo, 'restart_ms': round((time.monotonic() - killed) * 1000, 1), + 'steady_cpu_ms_per_s': round(steady_cpu, 1)} + cpu_start = cpu_ms(node.pid) + time.sleep(10) + row['recovery_cpu_ms_first_10s'] = round(cpu_ms(node.pid) - cpu_start, 1) + out, error = generator.communicate(timeout=600) + result = json.loads(out) + row['generator'] = {k: v for k, v in result.items() if k != 'timeline_100ms'} + row['timeline_100ms'] = result['timeline_100ms'] + row['recovery'] = recovery(result['timeline_100ms'], result['gap_from_ms'], result['longest_gap_ms']) + count = view(port, 'count') + row['applied'] = count + assert count == result['made'] == result['stored'], (count, result['made'], result['stored']) + node.terminate() + _, error = node.communicate(timeout=120) + stats = [json.loads(line) for line in error.decode().splitlines() if line.startswith('{"active"')][-1] + row['listener'] = {k: v for k, v in stats.items() if k.startswith('listener_')} + return row + + +def restart_quic(n, work, period_ms, seconds, kill_at): + """M2 for quinn: the same load; the service is killed (-9) and started + again on its port, with the same key (so it resets the old connections).""" + binary = Path(__file__).parent / 'quic/target/release/idle-quic' + port = free_port(socket.SOCK_DGRAM) + args = [binary, 'serve', work / 'cert.der', work / 'key.der', port] + server, _ = start(args) + client = subprocess.Popen([str(binary), 'steady', str(port), '0', str(n), str(period_ms), str(seconds), '256'], + stdout=subprocess.PIPE, stderr=subprocess.PIPE) + began = time.monotonic() + time.sleep(kill_at) + cpu_before = cpu_ms(server.pid) + time.sleep(1.0) + steady_cpu = cpu_ms(server.pid) - cpu_before + server.kill() + server.wait() + killed = time.monotonic() + server, _ = start(args) + row = {'clients': n, 'period_ms': period_ms, 'restart_ms': round((time.monotonic() - killed) * 1000, 1), + 'killed_at_s': round(killed - began, 1), 'steady_cpu_ms_per_s': round(steady_cpu, 1)} + cpu_start = cpu_ms(server.pid) + time.sleep(10) + row['recovery_cpu_ms_first_10s'] = round(cpu_ms(server.pid) - cpu_start, 1) + out, _ = client.communicate(timeout=900) + result = json.loads(out) + row['client'] = {k: v for k, v in result.items() if k != 'timeline_100ms'} + row['timeline_100ms'] = result['timeline_100ms'] + row['recovery'] = recovery(result['timeline_100ms'], result['gap_from_ms'], result['longest_gap_ms']) + server.kill() + server.wait() + return row + + +def main(): + parser = argparse.ArgumentParser() + parser.add_argument('--counts', default='1000,10000,100000,1000000') + parser.add_argument('--store', type=Path, default=None, help='where the stores go (default: a temporary directory)') + parser.add_argument('--idle-ms', type=int, default=2000) + parser.add_argument('--window', type=int, default=256) + parser.add_argument('--out', type=Path, default=None) + parser.add_argument('--quic', action='store_true', help='measure the quinn service instead (quic/)') + parser.add_argument('--restart', action='store_true', help='M2: kill -9 the receiver under load (with --quic, quinn)') + parser.add_argument('--no-memo', action='store_true', help='M2: the node forgets the secrets it remembered') + parser.add_argument('--period-ms', type=int, default=5000) + parser.add_argument('--seconds', type=int, default=40) + parser.add_argument('--kill-at', type=int, default=20) + args = parser.parse_args() + counts = [int(c) for c in args.counts.split(',')] + if args.restart: + with tempfile.TemporaryDirectory(prefix='pure-m2-', dir=args.store) as directory: + work = Path(directory) + if args.quic: + subprocess.run(['cargo', 'build', '--release', '--offline', '--quiet'], cwd=Path(__file__).parent / 'quic', + check=True) + make_certificate(work) + else: + source = work / 'receiver.aup' + source.write_text(check.LINK_RECEIVER) + module = work / 'receiver.pmod' + check.compile_module(Path(os.environ.get('AUP', ROOT.parent / 'aup/build/aup')), source, module) + keys = work / 'keys' + subprocess.run([str(ROOT / 'build/links'), 'keys', str(max(counts)), str(keys)], check=True) + for n in counts: + row = restart_quic(n, work, args.period_ms, args.seconds, args.kill_at) if args.quic else \ + restart_pure(n, work, keys, module, args.period_ms, args.seconds, args.kill_at, not args.no_memo) + line = json.dumps(row) + print(line, flush=True) + if args.out: + with open(args.out, 'a') as out: + out.write(line + '\n') + return + if args.quic: + with tempfile.TemporaryDirectory(prefix='pure-m1-quic-', dir=args.store) as directory: + quic(counts, Path(directory), args.window, args.out) + return + subprocess.run(['make', '-s', '-C', str(ROOT), 'build/pure-serve', 'build/pure-crypto.so', 'build/libpure.a'], check=True) + subprocess.run(['cc', '-O2', '-g', '-std=c17', '-Wall', '-Wextra', '-Werror', f'-I{ROOT}/core', + str(Path(__file__).parent / 'links.c'), str(ROOT / 'build/libpure.a'), '-lcrypto', '-lm', '-pthread', + '-o', str(ROOT / 'build/links')], check=True) + with tempfile.TemporaryDirectory(prefix='pure-m1-', dir=args.store) as directory: + work = Path(directory) + source = work / 'receiver.aup' + source.write_text(check.LINK_RECEIVER) + module = work / 'receiver.pmod' + check.compile_module(Path(os.environ.get('AUP', ROOT.parent / 'aup/build/aup')), source, module) + keys = work / 'keys' + subprocess.run([str(ROOT / 'build/links'), 'keys', str(max(counts)), str(keys)], check=True) + for n in counts: + row = one(n, work, keys, module, args.idle_ms, args.window) + line = json.dumps(row) + print(line, flush=True) + if args.out: + with open(args.out, 'a') as out: + out.write(line + '\n') + + +if __name__ == '__main__': + main() -- cgit v1.2.3