diff options
| -rw-r--r-- | Makefile | 12 | ||||
| -rw-r--r-- | link/link-protocol.c | 1494 | ||||
| -rw-r--r-- | link/link-protocol.h | 160 | ||||
| -rw-r--r-- | link/link.c | 1514 | ||||
| -rw-r--r-- | link/link.h | 106 | ||||
| -rw-r--r-- | link/test-link.c | 300 | ||||
| -rw-r--r-- | research/experiments/019-idle-links/transport.c | 6 | ||||
| -rw-r--r-- | serve/serve.c | 17 |
8 files changed, 2054 insertions, 1555 deletions
@@ -20,12 +20,12 @@ BUILD ?= build # surfaces link it); libpure-link.a is the links with the crypto loader. # prototypes/ keeps the earlier document runtime (000-077). # Unit tests sit by their part (core/test-module.c, store/test-tree-store.c, -# store/test-log.c, serve/test-message-json.c); +# store/test-log.c, link/test-link.c, serve/test-message-json.c); # tests/check.py checks the hosts end to end. INCLUDES = -Icore MODEL = core/module.c core/module-list.c core/module-save.c core/module-text.c core/module-store.c core/flow.c STORE = store/store-tree.c store/store-log.c -LINK = link/link.c link/crypto.c +LINK = link/link-protocol.c link/link.c link/crypto.c SERVE = serve/serve.c serve/deliver.c serve/message-json.c HEADERS = $(wildcard core/*.h) $(wildcard link/*.h) $(wildcard serve/*.h) $(wildcard store/*.h) OBJECTS = $(patsubst %.c,$(BUILD)/obj/%.o,$(MODEL) $(STORE)) @@ -75,6 +75,11 @@ $(BUILD)/test-module: core/test-module.c $(MODEL) $(STORE) $(HEADERS) @mkdir -p $(BUILD) $(CC) $(CPPFLAGS) $(CFLAGS) $(INCLUDES) -DPURE_STORE_TEST core/test-module.c $(MODEL) $(STORE) $(LDFLAGS) -pthread -lm $(LDLIBS) -o $@ +# A link's two sides through a network of the test's own (pure-crypto.so beside it). +$(BUILD)/test-link: link/test-link.c $(BUILD)/libpure-link.a $(BUILD)/pure-crypto.so $(wildcard link/*.h) + @mkdir -p $(BUILD) + $(CC) $(CPPFLAGS) $(CFLAGS) link/test-link.c $(BUILD)/libpure-link.a $(LDFLAGS) -pthread -ldl $(LDLIBS) -o $@ + $(BUILD)/test-message-json: serve/test-message-json.c serve/message-json.c $(BUILD)/libpure.a $(HEADERS) @mkdir -p $(BUILD) $(CC) $(CPPFLAGS) $(CFLAGS) $(INCLUDES) -Iserve serve/test-message-json.c serve/message-json.c $(BUILD)/libpure.a \ @@ -91,10 +96,11 @@ $(BUILD)/test-tree-store: store/test-tree-store.c $(STORE) $(HEADERS) # The HTTP host's checks build their apps with aup (the sibling checkout) and # compare documents with the earlier document server. AUP ?= ../aup -check: all $(BUILD)/test-module $(BUILD)/test-tree-store $(BUILD)/test-log $(BUILD)/test-message-json +check: all $(BUILD)/test-module $(BUILD)/test-tree-store $(BUILD)/test-log $(BUILD)/test-link $(BUILD)/test-message-json $(BUILD)/test-tree-store $(BUILD)/test-log $(BUILD)/test-module + $(BUILD)/test-link $(BUILD)/test-message-json $(MAKE) -C $(AUP) build/aup $(MAKE) -C prototypes/connections BUILD=build/tree STORE=store-tree.c build/tree/pure-connections diff --git a/link/link-protocol.c b/link/link-protocol.c new file mode 100644 index 0000000..94a965b --- /dev/null +++ b/link/link-protocol.c @@ -0,0 +1,1494 @@ +#define _GNU_SOURCE +/* Links without connections as state alone (link/link-protocol.h, + * docs/design/011-pure-transport.md). */ +#include "link-protocol.h" +#include <fcntl.h> +#include <malloc.h> +#include <stdio.h> +#include <stdlib.h> +#include <string.h> +#include <sys/mman.h> +#include <sys/random.h> +#include <sys/stat.h> +#include <unistd.h> + +enum { VERSION = 0x70, HEADER = 21, DATAGRAM = PURE_LINK_DATAGRAM, BODY = DATAGRAM - HEADER - PURE_AEAD_TAG, PIECE = 1000 }; +enum { F_PADDING = 0, F_STREAM = 1, F_RECORD = 2, F_ACK = 3, F_STORED = 4, F_PING = 5, F_RECORDS = 6, STORED_GAP = 1 }; +enum { RECORDS_HEAD = 10 }; /* type, the first number, the count */ +enum { RANGES = 32, RUNS = 2, UNITS = 64, CREDIT = 1024, BATCH = 256, RECORD_HEAD = 19 }; +enum { WINDOW_FIRST = 10 * DATAGRAM, WINDOW_LEAST = 2 * DATAGRAM }; +#define MS UINT64_C(1000000) +/* RFC 9002's values, but a first RTT of 100 ms (a LAN's, not the 333 ms of + * the Internet at large); a probe at least every second while a receiver is + * away; a record received but not stored after a second is sent again. */ +static const uint64_t FIRST_RTT = 100 * MS, ACK_DELAY_MOST = 25 * MS, GRANULARITY = 1 * MS, PTO_MOST = 1000 * MS, + STORED_WAIT = 1000 * MS, RECORD_MOST = UINT64_C(64) << 20, HELD_MOST = UINT64_C(16) << 20; + +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 get16(const unsigned char *p) { return (uint32_t)p[0] | (uint32_t)p[1] << 8; } +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; } +/* LEB128: seven bits a byte, low first. */ +static size_t put_varint(unsigned char *p, uint64_t v) { + size_t n = 0; + for (; v >= 0x80; v >>= 7) p[n++] = (unsigned char)(v | 0x80); + p[n++] = (unsigned char)v; + return n; +} +static size_t varint_size(uint64_t v) { + size_t n = 1; + for (; v >= 0x80; v >>= 7) ++n; + return n; +} +/* A varint from p (size bytes): how many bytes it took, or 0. */ +static size_t get_varint(const unsigned char *p, size_t size, uint64_t *v) { + *v = 0; + for (size_t n = 0; n < size && n < 10; ++n) { + *v |= (uint64_t)(p[n] & 0x7f) << (7 * n); + if (!(p[n] & 0x80)) return n + 1; + } + return 0; +} + +static uint64_t least(uint64_t a, uint64_t b) { return a < b ? a : b; } +static uint64_t most(uint64_t a, uint64_t b) { return a > b ? a : b; } + +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 uint32_t key_id(const unsigned char key[PURE_SEAL_KEY]) { return fnv(2166136261u, key, PURE_SEAL_KEY); } + +static uint64_t random64(void) { + uint64_t value = 0; + while (getrandom(&value, sizeof(value), 0) != sizeof(value) || !value) {} + return value; +} + +/* The key and nonce base of packets from `from` (in its run from_run) to + * `to`, answering to_run (0 for none). */ +static bool derive(const unsigned char secret[32], const unsigned char from[PURE_SEAL_KEY], + const unsigned char to[PURE_SEAL_KEY], uint64_t from_run, uint64_t to_run, + unsigned char key[PURE_AEAD_KEY], unsigned char base[PURE_AEAD_NONCE]) { + unsigned char info[16 + 2 * PURE_SEAL_KEY + 16], out[PURE_AEAD_KEY + PURE_AEAD_NONCE]; + memcpy(info, "pure link 0 keys", 16); + memcpy(info + 16, from, PURE_SEAL_KEY); + memcpy(info + 16 + PURE_SEAL_KEY, to, PURE_SEAL_KEY); + put64(info + 16 + 2 * PURE_SEAL_KEY, from_run); + put64(info + 24 + 2 * PURE_SEAL_KEY, to_run); + if (!pure_link_derive(secret, info, sizeof(info), out, sizeof(out))) return false; + memcpy(key, out, PURE_AEAD_KEY); + memcpy(base, out + PURE_AEAD_KEY, PURE_AEAD_NONCE); + explicit_bzero(out, sizeof(out)); + return true; +} + +/* The same as a context of its own, for a sender's one link. */ +static PureAead *aead_of(const unsigned char secret[32], const unsigned char from[PURE_SEAL_KEY], + const unsigned char to[PURE_SEAL_KEY], uint64_t from_run, uint64_t to_run, + unsigned char base[PURE_AEAD_NONCE]) { + unsigned char key[PURE_AEAD_KEY]; + PureAead *aead = derive(secret, from, to, from_run, to_run, key, base) ? pure_aead(key) : NULL; + explicit_bzero(key, sizeof(key)); + return aead; +} + +static void nonce_of(unsigned char out[PURE_AEAD_NONCE], const unsigned char base[PURE_AEAD_NONCE], uint64_t pn) { + memcpy(out, base, PURE_AEAD_NONCE); + for (int i = 0; i < 8; ++i) out[4 + i] ^= (unsigned char)(pn >> (56 - 8 * i)); +} + +/* A packet: the header, then body sealed (with key, or the context's own + * when NULL); its size, or 0. */ +static size_t seal_packet(PureAead *aead, const unsigned char *key, const unsigned char base[PURE_AEAD_NONCE], uint32_t id, + uint64_t run, uint64_t pn, const unsigned char *body, size_t size, unsigned char *out) { + unsigned char nonce[PURE_AEAD_NONCE]; + out[0] = VERSION; + put32(out + 1, id); + put64(out + 5, run); + put64(out + 13, pn); + nonce_of(nonce, base, pn); + return pure_aead_seal(aead, key, nonce, out, HEADER, body, size, out + HEADER) ? HEADER + size + PURE_AEAD_TAG : 0; +} + +static size_t put_names(unsigned char *p, const char *source, const char *input) { + size_t a = strlen(source), b = strlen(input); + p[0] = (unsigned char)a; + memcpy(p + 1, source, a); + p[1 + a] = (unsigned char)b; + memcpy(p + 2 + a, input, b); + return 2 + a + b; +} + +/* A source and an input's names from p (size bytes); how many it takes, or 0. */ +static size_t get_names(const unsigned char *p, size_t size, char source[PURE_LINK_NAME + 1], + char input[PURE_LINK_NAME + 1]) { + if (size < 1 || !p[0] || p[0] > PURE_LINK_NAME || size < 2u + p[0]) return 0; + size_t a = p[0], b = p[1 + a]; + if (!b || b > PURE_LINK_NAME || size < 2 + a + b) return 0; + memcpy(source, p + 1, a); + source[a] = 0; + memcpy(input, p + 2 + a, b); + input[b] = 0; + return 2 + a + b; +} + +/* ---- The sending side */ + +enum { UNSENT, FLYING, RECEIVED }; +typedef struct { uint64_t pn; uint8_t state; } Piece; +/* A record queued: its bytes at `at` in the link's byte ring, its pieces' + * states (one kept here, more beside). */ +typedef struct { uint64_t seq; size_t at; uint32_t size, pieces; Piece *more, one; } Record; +static Piece *piece_of(Record *r, uint32_t i) { return r->pieces == 1 ? &r->one : &r->more[i]; } +typedef struct { uint64_t seq; uint32_t piece; } Unit; +typedef struct { + uint64_t pn, sent; /* its number; when it went (ns) */ + uint32_t bytes; + bool eliciting, done; /* acknowledged or lost */ + uint16_t count; + Unit units[UNITS]; +} Flight; + +struct PureLinkSender { + uint32_t shape, own_id, peer_id; + char input[PURE_LINK_NAME + 1], source[PURE_LINK_NAME + 1]; + const PureSealKey *own; + unsigned char own_public[PURE_SEAL_KEY], peer[PURE_SEAL_KEY], secret[32]; + Record *records; /* the sends not yet stored, oldest first (a ring of consecutive numbers) */ + size_t head, count, capacity; + /* Their bytes, one after another as queued (a record's never cut by the + * ring's end): taken back in order as they are stored. */ + unsigned char *ring; + size_t ring_capacity; + uint64_t queued, stored; /* the last queued; saved by the receiver */ + uint64_t run, next_pn, in_run; + PureAead *out, *in; /* made once the receiver's key is known */ + unsigned char out_base[PURE_AEAD_NONCE], in_base[PURE_AEAD_NONCE]; + Flight *flights; /* one for each number sent, oldest first (a ring) */ + size_t flight_head, flight_count, flight_capacity; + uint64_t in_flight, window, threshold, recovery_start, eliciting; + uint64_t largest_acked, latest_rtt, smoothed_rtt, rtt_var, min_rtt; + bool acked_any, sampled; + uint64_t loss_time, last_eliciting, credit, stored_changed; + uint64_t received; /* pieces received and not yet stored */ + uint64_t unsent_from; /* no record before this number has a piece unsent */ + unsigned pto_count, probes; + PureLinkCounts counts; +}; + +static Record *record_at(PureLinkSender *l, size_t k) { return &l->records[(l->head + k) % l->capacity]; } + +/* The queued record numbered seq, or NULL (stored already, or not queued). */ +static Record *record_of(PureLinkSender *l, uint64_t seq) { + if (!l->count || seq < record_at(l, 0)->seq || seq - record_at(l, 0)->seq >= l->count) return NULL; + return record_at(l, (size_t)(seq - record_at(l, 0)->seq)); +} + +/* Every change of a piece's state goes through here: the link knows at once + * how many pieces wait to be stored, and where the unsent ones begin. */ +static void set_piece(PureLinkSender *l, Record *r, uint32_t i, uint8_t state) { + uint8_t was = (*piece_of(r, i)).state; + if (was == state) return; + if (was == RECEIVED) --l->received; + if (state == RECEIVED) ++l->received; + if (state == UNSENT && r->seq < l->unsent_from) l->unsent_from = r->seq; + (*piece_of(r, i)).state = state; +} + +/* Where size bytes go at the end of the byte ring (after the newest record, + * or from its start when that is free and the end is not), or SIZE_MAX. */ +static size_t ring_place(PureLinkSender *l, size_t size) { + if (!l->count) return size <= l->ring_capacity ? 0 : SIZE_MAX; + Record *oldest = record_at(l, 0), *newest = record_at(l, l->count - 1); + size_t head = oldest->at, tail = newest->at + newest->size; + if (newest->at >= head) /* the bytes are [head, tail): after them, or before them from the start */ + return l->ring_capacity - tail >= size ? tail : size <= head ? 0 : SIZE_MAX; + return head - tail >= size ? tail : SIZE_MAX; /* around the end: between the newest and the oldest */ +} + +/* A ring room for size bytes more: twice as large (or more), the records + * moved into it one after another. */ +static bool ring_grow(PureLinkSender *l, size_t size) { + size_t used = 0; + for (size_t k = 0; k < l->count; ++k) used += record_at(l, k)->size; + size_t capacity = l->ring_capacity ? 2 * l->ring_capacity : 4 << 10; + while (capacity < used + size + 1) capacity *= 2; + unsigned char *ring = malloc(capacity); + if (!ring) return false; + for (size_t k = 0, at = 0; k < l->count; ++k) { + Record *r = record_at(l, k); + if (r->size) memcpy(ring + at, l->ring + r->at, r->size); + r->at = at; + at += r->size; + } + free(l->ring); + l->ring = ring; + l->ring_capacity = capacity; + return true; +} + +bool pure_link_sender_queue(PureLinkSender *l, uint64_t seq, const unsigned char *bytes, size_t size) { + Record r = {.seq = seq, .size = (uint32_t)size}; + r.pieces = size ? (uint32_t)((size + PIECE - 1) / PIECE) : 1; + if (size > RECORD_MOST || (r.pieces > 1 && !(r.more = calloc(r.pieces, sizeof(Piece))))) return false; + bool ok = seq > l->queued && (!l->count || seq == l->queued + 1); + size_t at = ok ? ring_place(l, size) : SIZE_MAX; + if (ok && at == SIZE_MAX) ok = ring_grow(l, size) && (at = ring_place(l, size)) != SIZE_MAX; + if (ok) { + r.at = at; + if (size) memcpy(l->ring + at, bytes, size); + } + if (ok && l->count == l->capacity) { + size_t capacity = l->capacity ? 2 * l->capacity : 16; + Record *grown = malloc(capacity * sizeof(*grown)); + if (grown) { + for (size_t k = 0; k < l->count; ++k) grown[k] = *record_at(l, k); + free(l->records); + l->records = grown; + l->capacity = capacity; + l->head = 0; + } else + ok = false; + } + if (!ok) { + free(r.more); + return false; + } + *record_at(l, l->count++) = r; + l->queued = seq; + return true; +} + +static Flight *flight_of(PureLinkSender *l, uint64_t pn) { + if (!l->flight_count) return NULL; + uint64_t first = l->flights[l->flight_head].pn; + if (pn < first || pn - first >= l->flight_count) return NULL; + return &l->flights[(l->flight_head + (size_t)(pn - first)) % l->flight_capacity]; +} + +/* Room for one more flight: made before a packet is, so every number sent + * has its flight (they are found by number). */ +static bool flight_room(PureLinkSender *l) { + if (l->flight_count < l->flight_capacity) return true; + size_t capacity = l->flight_capacity ? 2 * l->flight_capacity : 8; + Flight *grown = malloc(capacity * sizeof(*grown)); + if (!grown) return false; + for (size_t k = 0; k < l->flight_count; ++k) grown[k] = l->flights[(l->flight_head + k) % l->flight_capacity]; + free(l->flights); + l->flights = grown; + l->flight_capacity = capacity; + l->flight_head = 0; + return true; +} + +static void push_flight(PureLinkSender *l, const Flight *f) { + l->flights[(l->flight_head + l->flight_count++) % l->flight_capacity] = *f; +} + +static void settle(PureLinkSender *l, Flight *f) { + f->done = true; + if (f->eliciting) { + --l->eliciting; + l->in_flight -= f->bytes; + } +} + +/* A packet lost: what it carried and was not received otherwise goes again. */ +static void lose(PureLinkSender *l, Flight *f, uint64_t now) { + settle(l, f); + ++l->counts.lost; + for (uint16_t u = 0; u < f->count; ++u) { + Record *r = record_of(l, f->units[u].seq); + Piece *p = r ? &(*piece_of(r, f->units[u].piece)) : NULL; + if (p && p->state == FLYING && p->pn == f->pn) { + set_piece(l, r, f->units[u].piece, UNSENT); + ++l->counts.resent; + } + } + if (f->sent > l->recovery_start) { /* one congestion event per round trip */ + l->recovery_start = now; + l->window = most(l->window / 2, WINDOW_LEAST); + l->threshold = l->window; + } +} + +static void detect_lost(PureLinkSender *l, uint64_t now) { + l->loss_time = 0; + if (!l->acked_any) return; + uint64_t rtt = most(l->latest_rtt, l->sampled ? l->smoothed_rtt : FIRST_RTT); + uint64_t delay = most(rtt * 9 / 8, GRANULARITY); + for (size_t k = 0; k < l->flight_count; ++k) { + Flight *f = &l->flights[(l->flight_head + k) % l->flight_capacity]; + if (f->pn >= l->largest_acked) break; + if (f->done) continue; + if (f->sent + delay <= now || l->largest_acked >= f->pn + 3) lose(l, f, now); + else if (!l->loss_time || f->sent + delay < l->loss_time) l->loss_time = f->sent + delay; + } +} + +static void sample_rtt(PureLinkSender *l, uint64_t latest, uint64_t ack_delay) { + l->latest_rtt = latest; + if (!l->sampled) { + l->min_rtt = l->smoothed_rtt = latest; + l->rtt_var = latest / 2; + l->sampled = true; + return; + } + l->min_rtt = least(l->min_rtt, latest); + ack_delay = least(ack_delay, ACK_DELAY_MOST); + uint64_t adjusted = latest >= l->min_rtt + ack_delay ? latest - ack_delay : latest; + uint64_t difference = l->smoothed_rtt > adjusted ? l->smoothed_rtt - adjusted : adjusted - l->smoothed_rtt; + l->rtt_var = (3 * l->rtt_var + difference) / 4; + l->smoothed_rtt = (7 * l->smoothed_rtt + adjusted) / 8; +} + +static void on_ack(PureLinkSender *l, const unsigned char *p, size_t size, uint64_t now) { + if (size < 13 || size < 13 + 16u * p[12]) return; + uint64_t largest = get64(p), delay = (uint64_t)get32(p + 8) * 1000; + Flight *top = NULL; + for (unsigned k = 0; k < p[12]; ++k) { + uint64_t first = get64(p + 13 + 16 * k), last = get64(p + 21 + 16 * k); + if (first > last || !l->flight_count) continue; + uint64_t oldest = l->flights[l->flight_head].pn, newest = oldest + l->flight_count - 1; + for (uint64_t pn = most(first, oldest); pn <= least(last, newest); ++pn) { + Flight *f = flight_of(l, pn); + if (!f || f->done) continue; + settle(l, f); + ++l->counts.acknowledged; + for (uint16_t u = 0; u < f->count; ++u) { + Record *r = record_of(l, f->units[u].seq); + if (r && (*piece_of(r, f->units[u].piece)).state != RECEIVED) { + set_piece(l, r, f->units[u].piece, RECEIVED); + l->stored_changed = now; /* waits for its save from now */ + } + } + if (f->eliciting && f->sent > l->recovery_start) + l->window += l->window < l->threshold ? f->bytes : (uint64_t)DATAGRAM * f->bytes / l->window; + if (pn == largest) top = f; + } + } + if (!l->acked_any || largest > l->largest_acked) l->largest_acked = largest; + l->acked_any = true; + if (top && top->eliciting) sample_rtt(l, now - top->sent, delay); + l->pto_count = 0; + detect_lost(l, now); + while (l->flight_count && l->flights[l->flight_head].done) { + l->flight_head = (l->flight_head + 1) % l->flight_capacity; + --l->flight_count; + } +} + +static void on_stored(PureLinkSender *l, const unsigned char *p, size_t size, uint64_t now) { + char source[PURE_LINK_NAME + 1], input[PURE_LINK_NAME + 1]; + size_t names = get_names(p, size, source, input); + if (!names || size < names + 13 || strcmp(source, l->source) || strcmp(input, l->input)) return; + uint64_t heard = get64(p + names); + unsigned flags = p[names + 8]; + uint32_t credit = get32(p + names + 9); + if (heard > l->stored) { + while (l->count && record_at(l, 0)->seq <= heard) { + Record *r = record_at(l, 0); + for (uint32_t i = 0; i < r->pieces; ++i) l->received -= (*piece_of(r, i)).state == RECEIVED; + free(r->more); /* its bytes go back with the ring's head */ + l->head = (l->head + 1) % l->capacity; + --l->count; + } + l->stored = heard; + l->stored_changed = now; + } + if (flags & STORED_GAP) /* it did not keep what came after a gap: send it again */ + for (size_t k = 0; k < l->count && l->received; ++k) { + Record *r = record_at(l, k); + for (uint32_t i = 0; i < r->pieces; ++i) + if ((*piece_of(r, i)).state == RECEIVED) set_piece(l, r, i, UNSENT); + } + l->credit = credit ? least(credit, 65536) : 1; +} + +void pure_link_sender_take(PureLinkSender *l, const unsigned char *packet, size_t n, uint64_t now) { + unsigned char plain[DATAGRAM], nonce[PURE_AEAD_NONCE]; + if (!l->out || n < HEADER + PURE_AEAD_TAG || n > DATAGRAM || packet[0] != VERSION || get32(packet + 1) != l->peer_id) + return; /* not ours, or an error from an earlier send (nobody listening) */ + uint64_t run = get64(packet + 5), pn = get64(packet + 13); + if (!l->in || l->in_run != run) { + pure_aead_free(l->in); + l->in = aead_of(l->secret, l->peer, l->own_public, run, l->run, l->in_base); + l->in_run = run; + if (!l->in) return; + } + nonce_of(nonce, l->in_base, pn); + size_t size = n - HEADER - PURE_AEAD_TAG; + if (!pure_aead_open(l->in, NULL, nonce, packet, HEADER, packet + HEADER, n - HEADER, plain)) return; + ++l->counts.received; + for (size_t at = 0; at < size;) { + unsigned type = plain[at++]; + if (type == F_ACK && at + 13 <= size && at + 13 + 16u * plain[at + 12] <= size) { + on_ack(l, plain + at, size - at, now); + at += 13 + 16u * plain[at + 12]; + } else if (type == F_STORED) { + char source[PURE_LINK_NAME + 1], input[PURE_LINK_NAME + 1]; + size_t names = get_names(plain + at, size - at, source, input); + if (!names || at + names + 13 > size) break; + on_stored(l, plain + at, size - at, now); + at += names + 13; + } else if (type != F_PING) + break; /* padding, or what this side does not know: the rest */ + } +} + +/* A packet of pieces not sent yet, in order, within the receiver's credit: + * its body's size, or 0 when there is none. */ +static size_t fill(PureLinkSender *l, unsigned char *body, Flight *f, bool pad) { + size_t at = 0; + body[at++] = F_STREAM; + at += put_names(body + at, l->source, l->input); + put32(body + at, l->shape); + at += 4; + /* The credit counts from the oldest record not stored: after a restart + * the sender does not know yet how far the receiver stored. */ + uint64_t base = l->count ? record_at(l, 0)->seq - 1 : l->stored; + /* From the first record that may have an unsent piece: the ones before + * it are sent, and every one found sent moves that mark on. */ + size_t k = l->count && l->unsent_from > base + 1 ? (size_t)(l->unsent_from - base - 1) : 0; + bool before = true; + /* A record none of whose pieces went yet goes whole when it fits what is + * left of the packet: whole records whose numbers follow on share a + * RECORDS frame (its count at run). Else its unsent pieces go, each in a + * RECORD frame. */ + size_t run = 0; + uint64_t next = 0; + for (; k < l->count && f->count < UNITS; ++k) { + Record *r = record_at(l, k); + if (r->seq > base + l->credit) break; + bool unsent = false, fresh = true, joins = run && r->seq == next; + for (uint32_t i = 0; i < r->pieces && fresh; ++i) fresh = piece_of(r, i)->state == UNSENT; + size_t whole = (joins ? 0 : RECORDS_HEAD) + varint_size(r->size) + r->size; + if (fresh && f->count + r->pieces <= UNITS && at + whole <= BODY) { + if (!joins) { + body[at] = F_RECORDS; + put64(body + at + 1, r->seq); + body[at + 9] = 0; + run = at + 9; + at += RECORDS_HEAD; + } + at += put_varint(body + at, r->size); + if (r->size) memcpy(body + at, l->ring + r->at, r->size); + at += r->size; + ++body[run]; + next = r->seq + 1; + for (uint32_t i = 0; i < r->pieces; ++i) { + set_piece(l, r, i, FLYING); + piece_of(r, i)->pn = f->pn; + f->units[f->count++] = (Unit){r->seq, i}; + } + ++l->counts.records; + } else if (fresh && r->pieces == 1) + unsent = true; /* too large for what is left: the next packet */ + else + for (uint32_t i = 0; i < r->pieces; ++i) { + if (piece_of(r, i)->state != UNSENT) continue; + uint32_t offset = i * PIECE, length = (uint32_t)least(PIECE, r->size - offset); + if (f->count == UNITS || at + RECORD_HEAD + length > BODY) { + unsent = true; + break; + } + body[at] = F_RECORD; + put64(body + at + 1, r->seq); + put32(body + at + 9, r->size); + put32(body + at + 13, offset); + put16(body + at + 17, length); + if (length) memcpy(body + at + RECORD_HEAD, l->ring + r->at + offset, length); + at += RECORD_HEAD + length; + run = 0; + set_piece(l, r, i, FLYING); + piece_of(r, i)->pn = f->pn; + f->units[f->count++] = (Unit){r->seq, i}; + ++l->counts.records; + } + if (before && !unsent) l->unsent_from = r->seq + 1; + before = before && !unsent; + if (unsent) { /* the packet is full: whole, so it can go in a burst */ + if (pad && f->count) { + memset(body + at, F_PADDING, BODY - at); + at = BODY; + } + break; + } + } + return f->count ? at : 0; +} + +/* Its number is spent all the same. */ +void pure_link_sender_refused(PureLinkSender *l, uint64_t pn) { + Flight *f = flight_of(l, pn); + if (!f || f->done) return; + for (uint16_t u = 0; u < f->count; ++u) { + Record *r = record_of(l, f->units[u].seq); + if (r && (*piece_of(r, f->units[u].piece)).pn == pn) set_piece(l, r, f->units[u].piece, UNSENT); + } + settle(l, f); + f->eliciting = false; + f->count = 0; + --l->counts.packets; +} + +/* Seal f with body into out, and count it sent: its size, or 0 (what it + * carried is then unsent again). Its number is spent either way. */ +static size_t transmit(PureLinkSender *l, Flight *f, const unsigned char *body, size_t size, unsigned char *out, + uint64_t now) { + size_t total = seal_packet(l->out, NULL, l->out_base, l->own_id, l->run, f->pn, body, size, out); + ++l->next_pn; + f->sent = now; + f->bytes = (uint32_t)total; + if (!total) { /* not sealed: what it carried is unsent again */ + for (uint16_t u = 0; u < f->count; ++u) { + Record *r = record_of(l, f->units[u].seq); + if (r && (*piece_of(r, f->units[u].piece)).pn == f->pn) set_piece(l, r, f->units[u].piece, UNSENT); + } + f->done = true; + f->eliciting = false; + f->count = 0; + push_flight(l, f); + return 0; + } + push_flight(l, f); + ++l->counts.packets; + if (f->eliciting) { + ++l->eliciting; + l->in_flight += f->bytes; + l->last_eliciting = now; + } + return total; +} + +size_t pure_link_sender_packet(PureLinkSender *l, unsigned char out[PURE_LINK_DATAGRAM], bool whole, uint64_t now, + uint64_t *number) { + bool probe = l->probes > 0; + if (!l->out || (!probe && l->in_flight + DATAGRAM > l->window) || !flight_room(l)) return 0; + unsigned char body[BODY]; + Flight f = {.pn = l->next_pn, .eliciting = true}; + size_t size = fill(l, body, &f, whole); + if (!size) { + if (!probe) return 0; + body[0] = F_PING; /* nothing to send again: a probe all the same */ + size = 1; + } + size_t total = transmit(l, &f, body, size, out, now); + if (!total) return 0; + if (probe) --l->probes; + *number = f.pn; + return total; +} + +static uint64_t pto_period(const PureLinkSender *l) { + uint64_t smoothed = l->sampled ? l->smoothed_rtt : FIRST_RTT, var = l->sampled ? l->rtt_var : FIRST_RTT / 2; + uint64_t period = smoothed + most(4 * var, GRANULARITY) + ACK_DELAY_MOST; + for (unsigned i = 0; i < l->pto_count && period < PTO_MOST; ++i) period *= 2; + return least(period, PTO_MOST); +} + +static bool waiting_for_save(const PureLinkSender *l) { return l->received > 0; } + +uint64_t pure_link_sender_due(const PureLinkSender *l) { + uint64_t due = l->loss_time; + if (l->eliciting) { + uint64_t pto = l->last_eliciting + pto_period(l); + if (!due || pto < due) due = pto; + } + if (waiting_for_save(l)) { + uint64_t stored = l->stored_changed + STORED_WAIT; + if (!due || stored < due) due = stored; + } + return due; +} + +void pure_link_sender_tick(PureLinkSender *l, uint64_t now) { + if (l->loss_time && l->loss_time <= now) detect_lost(l, now); + else if (l->eliciting && l->last_eliciting + pto_period(l) <= now) { + /* Nothing heard for a while: a probe, what flies longest sent again. */ + ++l->pto_count; + l->probes = 1; + for (size_t k = 0; k < l->count; ++k) { + Record *r = record_at(l, k); + uint32_t i = 0; + while (i < r->pieces && (*piece_of(r, i)).state != FLYING) ++i; + if (i < r->pieces) { + set_piece(l, r, i, UNSENT); + ++l->counts.resent; + break; + } + } + } + if (waiting_for_save(l) && l->stored_changed + STORED_WAIT <= now) { + /* Received long ago and still not stored: the receiver lost them (it + * restarted) or its word got lost. Sent again: it takes a number once. */ + for (size_t k = 0; k < l->count && l->received; ++k) { + Record *r = record_at(l, k); + for (uint32_t i = 0; i < r->pieces; ++i) + if ((*piece_of(r, i)).state == RECEIVED) set_piece(l, r, i, UNSENT), ++l->counts.resent; + } + l->stored_changed = now; + } +} + +PureLinkSender *pure_link_sender(const char *source, const char *input, uint32_t shape, const PureSealKey *own, + const unsigned char own_public[PURE_SEAL_KEY]) { + size_t a = strlen(source), b = strlen(input); + PureLinkSender *l = a && a <= PURE_LINK_NAME && b && b <= PURE_LINK_NAME ? calloc(1, sizeof(*l)) : NULL; + if (!l) return NULL; + memcpy(l->source, source, a); + memcpy(l->input, input, b); + l->shape = shape; + l->own = own; + memcpy(l->own_public, own_public, PURE_SEAL_KEY); + l->own_id = key_id(own_public); + l->run = random64(); + l->window = WINDOW_FIRST; + l->threshold = UINT64_MAX; + l->credit = CREDIT; + return l; +} + +bool pure_link_sender_peer(PureLinkSender *l, const unsigned char peer[PURE_SEAL_KEY]) { + memcpy(l->peer, peer, PURE_SEAL_KEY); + l->peer_id = key_id(peer); + pure_aead_free(l->out); + pure_aead_free(l->in); + l->out = l->in = NULL; + return l->own && pure_link_secret(l->own, l->peer, l->secret) && + (l->out = aead_of(l->secret, l->own_public, l->peer, l->run, 0, l->out_base)); +} + +uint64_t pure_link_sender_queued(const PureLinkSender *l) { return l->queued; } + +uint64_t pure_link_sender_stored(const PureLinkSender *l) { return l->stored; } + +void pure_link_sender_counts(const PureLinkSender *l, PureLinkCounts *out) { + *out = l->counts; + out->stored = l->stored; +} + +void pure_link_sender_free(PureLinkSender *l) { + if (!l) return; + for (size_t k = 0; k < l->count; ++k) free(record_at(l, k)->more); + free(l->records); + free(l->ring); + free(l->flights); + pure_aead_free(l->out); + pure_aead_free(l->in); + explicit_bzero(l->secret, sizeof(l->secret)); + free(l); +} + +/* ---- The receiving side + * + * A sender allowed is a peer: its key and the sources it may send as, a few + * bytes each, found by its key's id. What a sending peer needs besides (the + * secret its key shares with the node's, its runs' keys and addresses, its + * streams) is its active state: made at its first packet (one X25519), let + * go once the peer has been quiet for the idle period with nothing of it + * waiting here. So a link at rest costs the node its allowing, and the next + * packet goes on without a handshake. */ + +typedef struct { uint64_t first, last; } Range; +typedef struct Active Active; +typedef struct { + Active *active; + uint64_t run, seen; /* the sender's run; when it last sent */ + bool used, ack_due, due; + unsigned char in_key[PURE_AEAD_KEY], out_key[PURE_AEAD_KEY], in_base[PURE_AEAD_NONCE], out_base[PURE_AEAD_NONCE]; + uint64_t next_pn, largest_at; + _Alignas(8) unsigned char address[PURE_LINK_ADDRESS]; /* where it sends from: a socket's address */ + size_t address_size; + Range got[RANGES]; /* the numbers received, newest first */ + unsigned got_count; +} Run; +typedef struct { + uint64_t seq; + unsigned char *bytes; /* its size, then a byte a piece: had */ + uint32_t size, pieces, got; +} Held; +typedef struct Stream { + Active *active; + char source[PURE_LINK_NAME + 1], input_name[PURE_LINK_NAME + 1]; + uint32_t input, shape; /* the host's number for the input; the messages' shape */ + uint64_t heard; /* the last number heard (stored), once known */ + bool known, out, stored_due, gap, ready; + /* The line: once the last heard is known, whole records from the one + * after it on, in order, as a delivery carries them (u32 size, bytes), + * from line_head: handed over as they are, let go by moving the head. + * Records that come early or in pieces, or before the last heard is + * known, wait in held until the line reaches them. */ + unsigned char *line; + size_t line_head, line_size, line_capacity; + uint64_t line_first; + unsigned line_count; + Held *held; /* records not yet stored, by number: held_count from held_first */ + size_t held_first, held_count, held_capacity, held_bytes; + Run *run; /* the run that sent to it last: where STORED goes */ + uint64_t run_id; +} Stream; +struct Active { + uint32_t peer; /* its place among the peers */ + unsigned char secret[32]; + uint64_t seen; + size_t place; /* among the active */ + Run runs[RUNS]; + Stream **streams; + unsigned stream_count, stream_capacity; +}; +typedef struct { + unsigned char key[PURE_SEAL_KEY]; + uint32_t id, allowed; /* allowed: the place + 1 of its first source (0: none) */ + Active *active; +} Peer; +typedef struct { uint32_t name, next; } Allowed; /* name: where in names (u8 length, bytes); next: a place + 1 */ + +/* A peer quiet this long lets its active state go; the active are looked at + * that often; what they held goes back to the system once TRIM_AFTER have + * gone, or TRIM_WAIT after any went. */ +static const uint64_t IDLE_FIRST = 30000 * MS, SWEEP_EVERY = 250 * MS, TRIM_WAIT = 10000 * MS; +enum { TRIM_AFTER = 1024 }; + +struct PureLinkReceiver { + PureLinkHost host; + const PureSealKey *own; + unsigned char own_public[PURE_SEAL_KEY]; + uint32_t own_id; + uint64_t run, idle, swept, trimmed, untrimmed; + PureAead *aead; /* keyed by each packet's run */ + unsigned char open_key[PURE_AEAD_KEY], seal_key[PURE_AEAD_KEY]; /* the keys it has now */ + bool open_keyed, seal_keyed; + unsigned char *memo; /* the secrets remembered (pure_link_remember), mapped */ + size_t memo_size; + Peer *peers; + size_t peer_count, peer_capacity; + uint32_t *index; /* the peers by key id: open addressing, a place + 1 (0: empty) */ + size_t index_size; + Allowed *allowed; + size_t allowed_count, allowed_capacity; + unsigned char *names; + size_t names_size, names_capacity; + Active **actives; + size_t active_count, active_capacity; + Stream **ready; /* streams that may have whole records to hand over */ + size_t ready_count, ready_capacity; + Run **due; /* runs to answer */ + size_t due_count, due_capacity; + unsigned char *batch; + size_t batch_capacity; + PureLinkCounts counts; +}; + +PureLinkReceiver *pure_link_receiver(const PureSealKey *own, const unsigned char own_public[PURE_SEAL_KEY], + const PureLinkHost *host, char *error, size_t size) { + if (!own) { + snprintf(error, size, "a receiver needs the node's key"); + return NULL; + } + if (!pure_seal_ready(error, size)) return NULL; + PureLinkReceiver *s = calloc(1, sizeof(*s)); + unsigned char key[PURE_AEAD_KEY] = {0}; /* each packet brings its run's */ + if (!s || !(s->aead = pure_aead(key))) { + free(s); + snprintf(error, size, "out of memory"); + return NULL; + } + s->host = *host; + s->own = own; + memcpy(s->own_public, own_public, PURE_SEAL_KEY); + s->own_id = key_id(own_public); + s->run = random64(); + s->idle = IDLE_FIRST; + return s; +} + +void pure_link_idle(PureLinkReceiver *s, uint64_t ms) { s->idle = ms * MS; } + +/* items (count used of capacity) with room for one more: the same or grown, + * or NULL when memory runs out (items stay as they were). */ +static void *room(void *items, size_t *capacity, size_t count, size_t item_size, size_t first) { + if (count < *capacity) return items; + size_t more = *capacity ? 2 * *capacity : first; + void *grown = realloc(items, more * item_size); + if (grown) *capacity = more; + return grown; +} + +/* The peers by key id again, in an index twice as large. */ +static bool grow_index(PureLinkReceiver *s) { + size_t size = s->index_size ? 2 * s->index_size : 1024; + uint32_t *index = calloc(size, sizeof(*index)); + if (!index) return false; + for (size_t p = 0; p < s->peer_count; ++p) { + size_t slot = s->peers[p].id & (size - 1); + while (index[slot]) slot = (slot + 1) & (size - 1); + index[slot] = (uint32_t)p + 1; + } + free(s->index); + s->index = index; + s->index_size = size; + return true; +} + +static bool same_name(const PureLinkReceiver *s, uint32_t name, const char *source, size_t length) { + return s->names[name] == length && !memcmp(s->names + name + 1, source, length); +} + +/* Keys are found by their ids, which are 32 bits: two keys may share one. */ +bool pure_link_allow(PureLinkReceiver *s, const char *source, const unsigned char key[PURE_SEAL_KEY], char *error, + size_t size) { + size_t length = strlen(source), p = SIZE_MAX; + uint32_t id = key_id(key); + if (!length || length > PURE_LINK_NAME) { + snprintf(error, size, "a source's name is 1 to %d bytes", PURE_LINK_NAME); + return false; + } + for (size_t slot = s->index_size ? id & (s->index_size - 1) : 0; s->index_size && s->index[slot]; + slot = (slot + 1) & (s->index_size - 1)) { + const Peer *peer = &s->peers[s->index[slot] - 1]; + if (peer->id == id && !memcmp(peer->key, key, PURE_SEAL_KEY)) { + p = s->index[slot] - 1; + break; + } + } + if (p == SIZE_MAX) { + Peer *peers = s->peer_count < UINT32_MAX - 1 && ((s->peer_count + 1) * 2 <= s->index_size || grow_index(s)) ? + room(s->peers, &s->peer_capacity, s->peer_count, sizeof(Peer), 64) : NULL; + if (!peers) goto full; + s->peers = peers; + p = s->peer_count++; + s->peers[p] = (Peer){.id = id}; + memcpy(s->peers[p].key, key, PURE_SEAL_KEY); + size_t slot = id & (s->index_size - 1); + while (s->index[slot]) slot = (slot + 1) & (s->index_size - 1); + s->index[slot] = (uint32_t)p + 1; + } + for (uint32_t a = s->peers[p].allowed; a; a = s->allowed[a - 1].next) + if (same_name(s, s->allowed[a - 1].name, source, length)) return true; + Allowed *kept = s->allowed_count < UINT32_MAX - 1 && s->names_size + 1 + length <= UINT32_MAX ? + room(s->allowed, &s->allowed_capacity, s->allowed_count, sizeof(Allowed), 64) : NULL; + if (!kept) goto full; + s->allowed = kept; + if (s->names_size + 1 + length > s->names_capacity) { + size_t more = s->names_capacity ? 2 * s->names_capacity : 4096; + unsigned char *names = realloc(s->names, more); + if (!names) goto full; + s->names = names; + s->names_capacity = more; + } + s->names[s->names_size] = (unsigned char)length; + memcpy(s->names + s->names_size + 1, source, length); + s->allowed[s->allowed_count] = (Allowed){(uint32_t)s->names_size, s->peers[p].allowed}; + s->peers[p].allowed = (uint32_t)++s->allowed_count; + s->names_size += 1 + length; + return true; +full: + snprintf(error, size, "out of memory"); + return false; +} + +static Held *held_at(Stream *st, size_t k) { return &st->held[st->held_first + k]; } + +static uint64_t line_end(const Stream *st) { return st->line_first + st->line_count; } + +/* Room for more bytes at the line's end: its front let go given back + * first when that is at least half of it (so no byte moves twice in a + * doubling), else a longer line. */ +static bool line_room(Stream *st, size_t more) { + if (st->line_size + more <= st->line_capacity) return true; + if (st->line_head && st->line_head >= st->line_capacity / 2) { + memmove(st->line, st->line + st->line_head, st->line_size - st->line_head); + st->line_size -= st->line_head; + st->line_head = 0; + if (st->line_size + more <= st->line_capacity) return true; + } + size_t capacity = st->line_capacity ? st->line_capacity : 4096; + while (capacity < st->line_size + more) capacity *= 2; + unsigned char *grown = realloc(st->line, capacity); + if (!grown) return false; + st->line = grown; + st->line_capacity = capacity; + return true; +} + +static bool line_append(Stream *st, const unsigned char *bytes, uint32_t size) { + if (!line_room(st, 4 + (size_t)size)) return false; + put32(st->line + st->line_size, size); + if (size) memcpy(st->line + st->line_size + 4, bytes, size); + st->line_size += 4 + (size_t)size; + ++st->line_count; + return true; +} + +/* The line's records up to heard let go (all of them, when heard is past it). */ +static void line_drop(Stream *st, uint64_t heard) { + if (heard < st->line_first) return; + if (heard >= line_end(st)) { + st->line_head = st->line_size = 0; + st->line_count = 0; + st->line_first = heard + 1; + return; + } + for (uint64_t n = heard - st->line_first + 1; n; --n, --st->line_count, ++st->line_first) + st->line_head += 4 + (size_t)get32(st->line + st->line_head); +} + +/* Records held that the line reached, moved onto it; ones behind it let go. */ +static void drop_first(Stream *st); +static void line_pull(Stream *st) { + while (st->held_count) { + Held *h = held_at(st, 0); + if (h->seq < line_end(st)) drop_first(st); + else if (h->seq == line_end(st) && h->got == h->pieces && line_append(st, h->bytes, h->size)) drop_first(st); + else break; + } +} + +/* The first record held let go: the ones after it stay where they are. */ +static void drop_first(Stream *st) { + st->held_bytes -= held_at(st, 0)->size; + free(held_at(st, 0)->bytes); + ++st->held_first; + if (!--st->held_count) st->held_first = 0; +} + +/* The memo of secrets (pure_link_remember): a header, then a place a peer + * in the order allowed, each the first 28 bytes of its key, a check (FNV-1a + * of its whole key and the secret) and the secret. A place whose check + * fails (never made, another key, cut short by a crash) is made again. */ +enum { MEMO_HEAD = 64, MEMO_PLACE = 64 }; +static const unsigned char memo_magic[8] = {'P', 'L', 'M', '1', 0, 0, 0, 0}; + +static uint32_t memo_check(const unsigned char key[PURE_SEAL_KEY], const unsigned char secret[32]) { + return fnv(fnv(2166136261u, key, PURE_SEAL_KEY), secret, 32); +} + +bool pure_link_remember(PureLinkReceiver *s, const char *path, char *error, size_t size) { + int fd = open(path, O_RDWR | O_CREAT | O_NOFOLLOW | O_CLOEXEC, 0600); + struct stat info; + unsigned char head[MEMO_HEAD]; + size_t needed = MEMO_HEAD + (s->peer_count ? s->peer_count : 1) * MEMO_PLACE; + bool ok = fd >= 0 && !fstat(fd, &info) && S_ISREG(info.st_mode) && !(info.st_mode & 077); + /* Another node's key, or none yet: begun again. */ + bool ours = ok && pread(fd, head, MEMO_HEAD, 0) == MEMO_HEAD && !memcmp(head, memo_magic, 8) && + !memcmp(head + 8, s->own_public, PURE_SEAL_KEY); + if (ok && !ours) { + memset(head, 0, MEMO_HEAD); + memcpy(head, memo_magic, 8); + memcpy(head + 8, s->own_public, PURE_SEAL_KEY); + ok = !ftruncate(fd, 0) && pwrite(fd, head, MEMO_HEAD, 0) == MEMO_HEAD; + } + ok = ok && (ours && (size_t)info.st_size >= needed ? true : !ftruncate(fd, (off_t)needed)); + void *map = ok ? mmap(NULL, needed, PROT_READ | PROT_WRITE, MAP_SHARED, fd, 0) : MAP_FAILED; + if (fd >= 0) close(fd); + if (map == MAP_FAILED) { + snprintf(error, size, "%s: cannot keep secrets there (a file only its owner may read)", path); + return false; + } + s->memo = map; + s->memo_size = needed; + return true; +} + +/* A peer's active state, made at its first packet: the secret its key + * shares with ours, remembered or made (one X25519, then remembered: the + * secret first, the key that makes the place its own last). */ +static Active *activate(PureLinkReceiver *s, Peer *peer, uint64_t now) { + Active *a = calloc(1, sizeof(*a)), **actives = room(s->actives, &s->active_capacity, s->active_count, sizeof(*actives), 64); + if (actives) s->actives = actives; + if (!a || !actives) { + free(a); + return NULL; + } + size_t index = (size_t)(peer - s->peers); + unsigned char *place = s->memo && MEMO_HEAD + (index + 1) * MEMO_PLACE <= s->memo_size ? + s->memo + MEMO_HEAD + index * MEMO_PLACE : NULL; + if (place && !memcmp(place, peer->key, 28) && get32(place + 28) == memo_check(peer->key, place + 32)) { + memcpy(a->secret, place + 32, 32); + ++s->counts.remembered; + } else if (pure_link_secret(s->own, peer->key, a->secret)) { + if (place) { + memset(place, 0, 28); + __atomic_thread_fence(__ATOMIC_RELEASE); + memcpy(place + 32, a->secret, 32); + put32(place + 28, memo_check(peer->key, a->secret)); + __atomic_thread_fence(__ATOMIC_RELEASE); + memcpy(place, peer->key, 28); + } + } else { + explicit_bzero(a, sizeof(*a)); + free(a); + return NULL; + } + a->peer = (uint32_t)(peer - s->peers); + a->seen = now; + a->place = s->active_count; + s->actives[s->active_count++] = a; + peer->active = a; + return a; +} + +static void let_go(PureLinkReceiver *s, Active *a) { + for (unsigned k = 0; k < a->stream_count; ++k) { + while (a->streams[k]->held_count) drop_first(a->streams[k]); + free(a->streams[k]->held); + free(a->streams[k]->line); + free(a->streams[k]); + } + free(a->streams); + s->peers[a->peer].active = NULL; + s->actives[a->place] = s->actives[--s->active_count]; + s->actives[a->place]->place = a->place; + explicit_bzero(a, sizeof(*a)); + free(a); +} + +/* Whether any of a peer's records or answers waits here. */ +static bool busy(const Active *a) { + for (unsigned k = 0; k < a->stream_count; ++k) + if (a->streams[k]->out || a->streams[k]->held_count || a->streams[k]->line_count || a->streams[k]->ready) return true; + for (unsigned i = 0; i < RUNS; ++i) + if (a->runs[i].due) return true; + return false; +} + +/* Let go the active state of peers quiet for the idle period; what idle + * links held goes back to the system. */ +static void sweep(PureLinkReceiver *s, uint64_t now) { + if (now - s->swept < SWEEP_EVERY) return; + s->swept = now; + size_t freed = 0; + for (size_t k = 0; k < s->active_count;) { + Active *a = s->actives[k]; + if (now - a->seen < s->idle || busy(a)) ++k; + else let_go(s, a), ++freed; + } + s->counts.swept += freed; + s->untrimmed += freed; + if (s->untrimmed >= TRIM_AFTER || (s->untrimmed && now - s->trimmed >= TRIM_WAIT)) { + malloc_trim(0); + s->untrimmed = 0; + s->trimmed = now; + } +} + +static void mark_ready(PureLinkReceiver *s, Stream *st) { + Stream **ready = st->ready ? NULL : room(s->ready, &s->ready_capacity, s->ready_count, sizeof(*ready), 64); + if (!ready) return; + s->ready = ready; + s->ready[s->ready_count++] = st; + st->ready = true; +} + +static void mark_due(PureLinkReceiver *s, Run *r) { + Run **due = r->due ? NULL : room(s->due, &s->due_capacity, s->due_count, sizeof(*due), 64); + if (!due) return; + s->due = due; + s->due[s->due_count++] = r; + r->due = true; +} + +static Run *run_found(Active *a, uint64_t run) { + for (unsigned i = 0; i < RUNS; ++i) + if (a->runs[i].used && a->runs[i].run == run) return &a->runs[i]; + return NULL; +} + +/* A new run's keys, in the slot unused longest: only once its first packet + * opened with them, so a forged one displaces nothing. */ +static Run *install_run(Active *a, const Run *made) { + Run *slot = &a->runs[0]; + for (unsigned i = 0; i < RUNS; ++i) { + Run *r = &a->runs[i]; + if (!r->used || (slot->used && r->seen < slot->seen)) slot = r; + } + bool due = slot->due; /* the answer list may hold the slot */ + *slot = *made; + slot->due = due; + return slot; +} + +static bool had(const Run *r, uint64_t pn) { + for (unsigned i = 0; i < r->got_count; ++i) + if (pn >= r->got[i].first && pn <= r->got[i].last) return true; + return false; +} + +static void note(Run *r, uint64_t pn, uint64_t now) { + unsigned i = 0, n = r->got_count; + while (i < n && r->got[i].first > pn + 1) ++i; + if (i < n && r->got[i].first == pn + 1) { /* joins the range above it */ + r->got[i].first = pn; + if (i + 1 < n && r->got[i + 1].last + 1 == pn) { + r->got[i].first = r->got[i + 1].first; + memmove(&r->got[i + 1], &r->got[i + 2], (n - i - 2) * sizeof(Range)); + --r->got_count; + } + } else if (i < n && r->got[i].last + 1 == pn) { + r->got[i].last = pn; + } else if (i < RANGES) { + if (n == RANGES) --n; /* the oldest range is forgotten */ + memmove(&r->got[i + 1], &r->got[i], (n - i) * sizeof(Range)); + r->got[i] = (Range){pn, pn}; + r->got_count = n + 1; + } + if (r->got_count && r->got[0].last == pn) r->largest_at = now; +} + +static bool allowed(const PureLinkReceiver *s, const Peer *peer, const char *source) { + for (uint32_t a = peer->allowed; a; a = s->allowed[a - 1].next) + if (same_name(s, s->allowed[a - 1].name, source, strlen(source))) return true; + return false; +} + +/* The stream of a peer's source to input, made the first time: NULL when the + * source is not allowed from the peer, or the host has no input that takes + * such messages. */ +static Stream *stream_of(PureLinkReceiver *s, Active *a, const char *source, const char *input, uint32_t shape) { + for (unsigned k = 0; k < a->stream_count; ++k) { + Stream *st = a->streams[k]; + if (!strcmp(st->input_name, input) && !strcmp(st->source, source)) return st->shape == shape ? st : NULL; + } + int index = allowed(s, &s->peers[a->peer], source) ? s->host.input(s->host.context, input, shape) : -1; + if (index < 0) return NULL; + size_t capacity = a->stream_capacity; + Stream **streams = room(a->streams, &capacity, a->stream_count, sizeof(*streams), 2), *st; + if (!streams) return NULL; + a->streams = streams; + a->stream_capacity = (unsigned)capacity; + if (!(st = calloc(1, sizeof(*st)))) return NULL; + st->active = a; + st->input = (uint32_t)index; + st->shape = shape; + snprintf(st->source, sizeof(st->source), "%s", source); + snprintf(st->input_name, sizeof(st->input_name), "%s", input); + a->streams[a->stream_count++] = st; + return st; +} + +/* A record's piece, kept until the record is whole and handed over. */ +static void hold(Stream *st, uint64_t seq, uint32_t size, uint32_t offset, const unsigned char *bytes, uint32_t length) { + if (st->known && seq <= st->heard) { + st->stored_due = true; /* sent again: say again how far it is stored */ + return; + } + if ((st->known && seq > st->heard + CREDIT) || size > RECORD_MOST) return; + if (st->known && seq < line_end(st)) return; /* on the line already */ + if (st->known && seq == line_end(st) && !offset && length == size) { /* the next, whole */ + if (st->held_bytes + (st->line_size - st->line_head) + size <= HELD_MOST && line_append(st, bytes, size)) + line_pull(st); + return; + } + /* In order it goes last; out of order, its place is found by halving. */ + size_t k = st->held_count; + if (k && held_at(st, k - 1)->seq >= seq) { + size_t low = 0; + while (low < k) { + size_t middle = low + (k - low) / 2; + if (held_at(st, middle)->seq < seq) low = middle + 1; + else k = middle; + } + } + if (k == st->held_count || held_at(st, k)->seq != seq) { + if (st->held_bytes + size > HELD_MOST) return; + if (st->held_first && st->held_first + st->held_count == st->held_capacity) { /* the front is free: used again */ + memmove(st->held, held_at(st, 0), st->held_count * sizeof(Held)); + st->held_first = 0; + } + Held *kept = room(st->held, &st->held_capacity, st->held_first + st->held_count, sizeof(Held), 16); + if (!kept) return; + st->held = kept; + Held h = {.seq = seq, .size = size, .pieces = size ? (size + PIECE - 1) / PIECE : 1}; + if (!(h.bytes = malloc(size + h.pieces))) return; + memset(h.bytes + size, 0, h.pieces); + memmove(held_at(st, k + 1), held_at(st, k), (st->held_count - k) * sizeof(Held)); + *held_at(st, k) = h; + ++st->held_count; + st->held_bytes += size; + } + Held *h = held_at(st, k); + if (h->size != size) return; + if (!offset && length == size) { /* whole: every piece had */ + if (size) memcpy(h->bytes, bytes, size); + memset(h->bytes + size, 1, h->pieces); + h->got = h->pieces; + return; + } + uint32_t piece = offset / PIECE; + if (offset % PIECE || piece >= h->pieces || length != least(PIECE, size - offset) || h->bytes[size + piece]) return; + if (length) memcpy(h->bytes + offset, bytes, length); + h->bytes[size + piece] = 1; + ++h->got; +} + +/* Whole records from the one after the last heard, in order, to the host: + * one delivery a stream at a time. Before the last heard is known (after a + * restart) they go from the first held: the host says how far it heard. + * False when the host could not take them now. */ +static bool hand_over(PureLinkReceiver *s, Stream *st) { + if (st->out) return true; + if (st->known) { /* the line as it is, up to BATCH records */ + line_pull(st); + if (!st->line_count) return true; + unsigned n = 0; + size_t bytes = 0; + for (; n < st->line_count && n < BATCH; ++n) bytes += 4 + (size_t)get32(st->line + st->line_head + bytes); + PureLinkDelivery d = {st->input, st->source, st->line_first, n, st->line + st->line_head, bytes, st}; + if (!s->host.deliver(s->host.context, &d)) return false; + st->out = true; + ++s->counts.deliveries; + return true; + } + if (!st->held_count || (st->known && held_at(st, 0)->seq != st->heard + 1)) return true; + size_t n = 0, bytes = 0; + while (n < st->held_count && n < BATCH && held_at(st, n)->got == held_at(st, n)->pieces && + held_at(st, n)->seq == held_at(st, 0)->seq + n) + bytes += 4 + held_at(st, n++)->size; + if (!n) return true; + if (bytes > s->batch_capacity) { + unsigned char *grown = realloc(s->batch, bytes); + if (!grown) return false; + s->batch = grown; + s->batch_capacity = bytes; + } + size_t at = 0; + for (size_t k = 0; k < n; ++k) { + put32(s->batch + at, held_at(st, k)->size); + memcpy(s->batch + at + 4, held_at(st, k)->bytes, held_at(st, k)->size); + at += 4 + held_at(st, k)->size; + } + PureLinkDelivery d = {st->input, st->source, held_at(st, 0)->seq, (unsigned)n, s->batch, bytes, st}; + if (!s->host.deliver(s->host.context, &d)) return false; + st->out = true; + ++s->counts.deliveries; + return true; +} + +/* A packet from peer, if it opens with the keys of peer's run (made for a + * run not seen yet, and for a peer at rest its active state): its frames. */ +static bool take_from(PureLinkReceiver *s, Peer *peer, const unsigned char *packet, size_t n, const void *from, + size_t from_size, uint64_t now) { + uint64_t pn = get64(packet + 13), run = get64(packet + 5); + Active *a = peer->active; + Run *r = a ? run_found(a, run) : NULL; + if (r && had(r, pn)) { + r->ack_due = true; /* its acknowledgment was lost, perhaps */ + mark_due(s, r); + return true; + } + bool woke = !a; + if (woke && !(a = activate(s, peer, now))) return false; + Run made = {.active = a, .run = run, .seen = now, .used = true}; + bool keyed = r || (derive(a->secret, peer->key, s->own_public, run, 0, made.in_key, made.in_base) && + derive(a->secret, s->own_public, peer->key, s->run, run, made.out_key, made.out_base)); + unsigned char plain[DATAGRAM], nonce[PURE_AEAD_NONCE]; + nonce_of(nonce, r ? r->in_base : made.in_base, pn); + size_t size = n - HEADER - PURE_AEAD_TAG; + const unsigned char *key = r ? r->in_key : made.in_key, *give = key; + if (s->open_keyed && !memcmp(s->open_key, key, PURE_AEAD_KEY)) give = NULL; /* the key it has */ + else if (keyed) memcpy(s->open_key, key, PURE_AEAD_KEY), s->open_keyed = true; + if (!keyed || !pure_aead_open(s->aead, give, nonce, packet, HEADER, packet + HEADER, n - HEADER, plain)) { + explicit_bzero(&made, sizeof(made)); + if (woke) let_go(s, a); + return false; + } + if (woke) ++s->counts.woken; + if (!r) r = install_run(a, &made); + a->seen = r->seen = now; + note(r, pn, now); + r->address_size = from_size <= sizeof(r->address) ? from_size : sizeof(r->address); + memcpy(r->address, from, r->address_size); /* where it is now: answers go there */ + ++s->counts.received; + Stream *st = NULL; + bool eliciting = false; + for (size_t at = 0; at < size;) { + unsigned type = plain[at++]; + if (type == F_STREAM) { + char source[PURE_LINK_NAME + 1], input[PURE_LINK_NAME + 1]; + size_t names = get_names(plain + at, size - at, source, input); + if (!names || at + names + 4 > size) break; + st = stream_of(s, a, source, input, get32(plain + at + names)); + if (st) { + st->run = r; + st->run_id = r->run; + } + at += names + 4; + eliciting = true; + } else if (type == F_RECORD) { + if (at + RECORD_HEAD - 1 > size) break; + uint32_t length = get16(plain + at + 16); + if (at + RECORD_HEAD - 1 + length > size) break; + if (st) { + hold(st, get64(plain + at), get32(plain + at + 8), get32(plain + at + 12), plain + at + 18, length); + mark_ready(s, st); + } + at += RECORD_HEAD - 1 + length; + eliciting = true; + } else if (type == F_RECORDS) { + if (at + RECORDS_HEAD - 1 > size) break; + uint64_t first = get64(plain + at), length = 0; + unsigned count = plain[at + 8]; + at += RECORDS_HEAD - 1; + unsigned i = 0; + for (size_t used; i < count && (used = get_varint(plain + at, size - at, &length)) && + length <= size - at - used; ++i) { + if (st) hold(st, first + i, (uint32_t)length, 0, plain + at + used, (uint32_t)length); + at += used + length; + } + if (st) mark_ready(s, st); + eliciting = true; + if (i < count) break; /* cut short: the rest is not read */ + } else if (type == F_PING) { + eliciting = true; + } else + break; /* padding, or what this side does not know: the rest */ + } + if (eliciting) { + r->ack_due = true; + mark_due(s, r); + } + return true; +} + +void pure_link_receiver_take(PureLinkReceiver *s, const unsigned char *packet, size_t n, const void *from, + size_t from_size, uint64_t now) { + if (n < HEADER + PURE_AEAD_TAG || n > DATAGRAM || packet[0] != VERSION || !s->index_size) return; + uint32_t id = get32(packet + 1); + size_t mask = s->index_size - 1; + uint64_t run = get64(packet + 5); + /* The peers with the packet's key id: first one whose run is known (no + * key to make), then any. */ + for (size_t slot = id & mask; s->index[slot]; slot = (slot + 1) & mask) { + Peer *peer = &s->peers[s->index[slot] - 1]; + if (peer->id == id && peer->active && run_found(peer->active, run) && + take_from(s, peer, packet, n, from, from_size, now)) + return; + } + for (size_t slot = id & mask; s->index[slot]; slot = (slot + 1) & mask) { + Peer *peer = &s->peers[s->index[slot] - 1]; + if (peer->id == id && !(peer->active && run_found(peer->active, run)) && + take_from(s, peer, packet, n, from, from_size, now)) + return; + } +} + +/* Acknowledgments and how far each stream is stored, a packet to each run + * that has any. How far a stream is stored goes with every acknowledgment + * to its run, so a lost word is said again at the next packet. */ +static void answer(PureLinkReceiver *s, uint64_t now, PureLinkSend *send, void *context) { + unsigned char body[BODY], packet[DATAGRAM]; + for (size_t k = 0; k < s->due_count; ++k) { + Run *r = s->due[k]; + Active *a = r->active; + r->due = false; + if (!r->used) continue; + size_t at = 0; + if (r->ack_due && r->got_count) { + body[at++] = F_ACK; + put64(body + at, r->got[0].last); + put32(body + at + 8, (uint32_t)least((now - r->largest_at) / 1000, UINT32_MAX)); + body[at + 12] = (unsigned char)r->got_count; + at += 13; + for (unsigned i = 0; i < r->got_count; ++i, at += 16) { + put64(body + at, r->got[i].first); + put64(body + at + 8, r->got[i].last); + } + } + for (unsigned j = 0; j < a->stream_count; ++j) { + Stream *st = a->streams[j]; + if (!(st->stored_due || r->ack_due) || st->run != r || st->run_id != r->run || !st->known) continue; + if (at + strlen(st->source) + strlen(st->input_name) + 16 > BODY) break; /* type, names, 13 */ + body[at++] = F_STORED; + at += put_names(body + at, st->source, st->input_name); + put64(body + at, st->heard); + body[at + 8] = st->gap ? STORED_GAP : 0; + size_t room = st->held_bytes + (st->line_size - st->line_head) < HELD_MOST ? CREDIT : 1; + put32(body + at + 9, (uint32_t)room); + at += 13; + st->stored_due = st->gap = false; + } + r->ack_due = false; + if (!at) continue; + const unsigned char *give = r->out_key; + if (s->seal_keyed && !memcmp(s->seal_key, r->out_key, PURE_AEAD_KEY)) give = NULL; /* the key it has */ + else memcpy(s->seal_key, r->out_key, PURE_AEAD_KEY), s->seal_keyed = true; + size_t total = seal_packet(s->aead, give, r->out_base, s->own_id, s->run, r->next_pn++, body, at, packet); + if (total && send(context, packet, total, r->address, r->address_size)) ++s->counts.packets; + } + s->due_count = 0; +} + +/* What the host said of its deliveries: the saved records let go, and how + * far they are stored said. */ +static void apply_done(PureLinkReceiver *s, const PureLinkDone *done, size_t count) { + for (size_t k = 0; k < count; ++k) { + Stream *st = done[k].token; + st->out = false; + mark_ready(s, st); + if (!done[k].stored) continue; /* not saved: handed over again */ + if (!st->known || done[k].heard > st->heard) st->heard = done[k].heard; + if (!st->known) st->line_first = st->heard + 1; /* the line begins */ + line_drop(st, st->heard); + while (st->held_count && held_at(st, 0)->seq <= st->heard) drop_first(st); + st->known = true; + st->gap = done[k].gap; + st->stored_due = true; + if (st->run && st->run->used && st->run_id == st->run->run) mark_due(s, st->run); + s->counts.stored = st->heard; + } +} + +/* The streams ready, each handed over; one the host could not take now is + * tried again at the next turn. */ +static void hand_over_ready(PureLinkReceiver *s) { + size_t count = s->ready_count; + for (size_t k = 0; k < count; ++k) { + Stream *st = s->ready[k]; + st->ready = false; + if (!hand_over(s, st)) mark_ready(s, st); + } + memmove(s->ready, s->ready + count, (s->ready_count - count) * sizeof(*s->ready)); + s->ready_count -= count; +} + +void pure_link_receiver_turn(PureLinkReceiver *s, const PureLinkDone *done, size_t count, uint64_t now, + PureLinkSend *send, void *context) { + if (!s->swept) s->swept = s->trimmed = now; + apply_done(s, done, count); + hand_over_ready(s); + answer(s, now, send, context); + sweep(s, now); +} + +uint64_t pure_link_receiver_due(const PureLinkReceiver *s) { + return s->active_count || s->ready_count ? s->swept + SWEEP_EVERY : 0; +} + +void pure_link_receiver_counts(const PureLinkReceiver *s, PureLinkCounts *counts) { + *counts = s->counts; + counts->active = s->active_count; +} + +void pure_link_receiver_free(PureLinkReceiver *s) { + if (!s) return; + while (s->active_count) let_go(s, s->actives[0]); + pure_aead_free(s->aead); + if (s->memo) munmap(s->memo, s->memo_size); + free(s->actives); + free(s->peers); + free(s->index); + free(s->allowed); + free(s->names); + free(s->ready); + free(s->due); + free(s->batch); + free(s); +} diff --git a/link/link-protocol.h b/link/link-protocol.h new file mode 100644 index 0000000..b0f8ef3 --- /dev/null +++ b/link/link-protocol.h @@ -0,0 +1,160 @@ +#ifndef PURE_LINK_PROTOCOL_H +#define PURE_LINK_PROTOCOL_H +#include "seal.h" + +/* Links without connections (docs/design/011-pure-transport.md), as state + * alone: no socket, thread or clock of their own. The owner gives each side + * the packets that came and the time (ns, monotonic), and sends the packets + * each side makes; link/link.h is such an owner over UDP. So one thread may + * carry many links, and a test may carry both sides through a network of its + * own making. + * + * A link's lasting state is the apps' own: the sender's outbox (what to send + * and send again) and the receiver's heard numbers (what it applied, saved + * with the events). What either side keeps in memory is a cache, so either + * may restart and the other goes on with the next packet. + * + * A packet is a header in the clear, then frames sealed with AES-128-GCM + * (the header as associated data): + * u8 0x70 (version 0), u32le the key id of the node that sent it (FNV-1a + * of its public key; two keys may share an id, and a receiver tries the + * keys it allows with it), u64le that node's run (random, new each time it + * starts, so no nonce repeats under a key across restarts), u64le the + * packet's number in its run. + * Keys come from the secret the two nodes' keys share (X25519) by + * HKDF-SHA256, with both public keys and the sending node's run: a sender's + * packets use its own run; a receiver's answers add the run they answer, + * so an answer opens only in that run. The nonce is the key's base with the + * packet's number in its last eight bytes (XOR, big-endian, as QUIC does). + * + * Frames (numbers little-endian): + * 0x01 STREAM source (u8 length, bytes), input (u8 length, bytes), u32 the + * shape of the messages (the sender's output and the + * receiver's input must have the same): the stream the + * records after it are of + * 0x02 RECORD u64 number, u32 size, u32 offset, u16 length, bytes: a piece + * of a saved send (pieces are 1000 bytes, the last shorter) + * 0x03 ACK u64 largest number, u32 delay (µs), u8 ranges, each as u64 + * first and u64 last, newest first + * 0x04 STORED source, input (as STREAM), u64 the last number heard + * (applied and saved), u8 flags (1: a record after a gap was + * not taken: send what follows it again), u32 credit (records + * it takes past that) + * 0x05 PING to be acknowledged + * 0x06 RECORDS u64 the first one's number, u8 count, then whole records one + * after another (numbers following on), each its size (a + * varint: LEB128) and its bytes: a send that fits the packet + * whole, with 1 to 3 bytes of framing, not the 19 of RECORD + * 0x00 PADDING to the end of the packet: a full packet sent in a burst is + * a whole datagram + * Losses are detected as RFC 9002 says; congestion control is its NewReno + * (RFC 9002 §7). */ + +/* A packet is at most PURE_LINK_DATAGRAM bytes; a source's or an input's + * name is 1 to PURE_LINK_NAME bytes; a receiver keeps where a packet came + * from (the socket's address) in PURE_LINK_ADDRESS bytes. */ +enum { PURE_LINK_DATAGRAM = 1200, PURE_LINK_NAME = 64, PURE_LINK_ADDRESS = 28 }; + +/* Counts for a host's report: packets and records sent, records sent again, + * packets lost and acknowledged, packets received, deliveries handed over, + * the last number stored; a receiver's senders woken (their active state + * made), of them with a secret remembered (no X25519), let go when idle, + * and active now. */ +typedef struct { + uint64_t packets, records, resent, lost, acknowledged, received, deliveries, stored, woken, remembered, swept, active; +} PureLinkCounts; + +/* ---- The sending side: one output's saved sends to a receiver's input, as + * source (the receiver's name for it), their messages of shape (the + * receiver's input must have the same: core/module.h's pure_module_shape), + * from own (this node's key, whose public half is own_public). NULL when + * the names are not 1 to PURE_LINK_NAME bytes, or memory runs out. */ +typedef struct PureLinkSender PureLinkSender; +PureLinkSender *pure_link_sender(const char *source, const char *input, uint32_t shape, const PureSealKey *own, + const unsigned char own_public[PURE_SEAL_KEY]); +/* The receiver's public key: the keys made from it (one X25519). A sender + * makes no packet before it has one. */ +bool pure_link_sender_peer(PureLinkSender *, const unsigned char peer[PURE_SEAL_KEY]); +/* A saved send, numbered one past the last queued (any number when none + * waits); false when it cannot be kept. */ +bool pure_link_sender_queue(PureLinkSender *, uint64_t seq, const unsigned char *bytes, size_t size); +/* A packet that came (the receiver's answers; others are let be). */ +void pure_link_sender_take(PureLinkSender *, const unsigned char *packet, size_t size, uint64_t now); +/* The timers due by now: losses found, probes, records sent again. */ +void pure_link_sender_tick(PureLinkSender *, uint64_t now); +/* The next packet to send into out, and its number: its size, or 0 when + * there is none (nothing to send, or the window is full). whole: a full + * packet is padded to PURE_LINK_DATAGRAM (so packets may go in a burst of + * whole datagrams, as UDP GSO sends them). */ +size_t pure_link_sender_packet(PureLinkSender *, unsigned char out[PURE_LINK_DATAGRAM], bool whole, uint64_t now, + uint64_t *number); +/* A packet made but not sent (the socket refused it): what it carried is + * sent again, and it counts as never sent. */ +void pure_link_sender_refused(PureLinkSender *, uint64_t number); +/* When the next timer is due (ns), or 0 for none. */ +uint64_t pure_link_sender_due(const PureLinkSender *); +/* The last number queued; the last the receiver stored. */ +uint64_t pure_link_sender_queued(const PureLinkSender *); +uint64_t pure_link_sender_stored(const PureLinkSender *); +void pure_link_sender_counts(const PureLinkSender *, PureLinkCounts *); +void pure_link_sender_free(PureLinkSender *); + +/* ---- The receiving side: packets from the senders allowed (a source name + * and the key it seals with, a few bytes each however many there are: see + * pure_link_idle). A stream (a source to an input) goes to the host's input + * its name and shape find (its number, or -1: the stream is refused). Whole + * records of a stream, in order, go to deliver as one delivery; the host + * applies them from the one after the source's last heard number, stops at + * a gap, saves, and then says so with a done (pure_link_receiver_turn). + * Records wait here until then, and only the saved ones are acknowledged as + * stored. */ +typedef struct { + uint32_t input; + const char *source; + uint64_t first; + unsigned count; + const unsigned char *records; /* count of: u32le size, bytes */ + size_t size; + void *token; /* for its done */ +} PureLinkDelivery; +typedef struct { + int (*input)(void *context, const char *name, uint32_t shape); + bool (*deliver)(void *context, const PureLinkDelivery *); + void *context; +} PureLinkHost; +/* A delivery's end: heard, the last number heard; stored, its save + * succeeded (else it is handed over again); gap, it stopped at a gap. */ +typedef struct { void *token; uint64_t heard; bool stored, gap; } PureLinkDone; +/* A packet to send to an address a packet came from: whether it went. */ +typedef bool PureLinkSend(void *context, const unsigned char *packet, size_t size, const unsigned char *to, + size_t to_size); + +typedef struct PureLinkReceiver PureLinkReceiver; +PureLinkReceiver *pure_link_receiver(const PureSealKey *own, const unsigned char own_public[PURE_SEAL_KEY], + const PureLinkHost *host, char *error, size_t size); +bool pure_link_allow(PureLinkReceiver *, const char *source, const unsigned char key[PURE_SEAL_KEY], char *error, + size_t size); +/* A sender quiet for ms (30 s at first) with nothing of it waiting lets its + * keys, addresses and streams go: an idle link costs the receiver only its + * allowing, and the sender's next packet makes them again (one X25519). */ +void pure_link_idle(PureLinkReceiver *, uint64_t ms); +/* Remember the secret each sender's key shares with the node's in a file + * (path, 0600; begun again for another node's key): a place a sender in + * the order allowed, mapped, so a sender at rest takes no memory for it. + * A node that starts again then makes no X25519 for a sender it heard from + * before. After the senders are allowed, before the first packet. */ +bool pure_link_remember(PureLinkReceiver *, const char *path, char *error, size_t size); +/* A packet that came from an address (from_size bytes, at most + * PURE_LINK_ADDRESS kept): answers to its sender go there. */ +void pure_link_receiver_take(PureLinkReceiver *, const unsigned char *packet, size_t size, const void *from, + size_t from_size, uint64_t now); +/* After the packets that came: the dones (count of them) applied, whole + * records handed over, answers sent through send, and senders at rest let + * go. */ +void pure_link_receiver_turn(PureLinkReceiver *, const PureLinkDone *done, size_t count, uint64_t now, + PureLinkSend *send, void *context); +/* When a turn is due with no packet (ns), or 0 for none. */ +uint64_t pure_link_receiver_due(const PureLinkReceiver *); +void pure_link_receiver_counts(const PureLinkReceiver *, PureLinkCounts *); +void pure_link_receiver_free(PureLinkReceiver *); +#endif diff --git a/link/link.c b/link/link.c index 103dcf5..0588633 100644 --- a/link/link.c +++ b/link/link.c @@ -1,10 +1,8 @@ #define _GNU_SOURCE -/* Links without connections (link/link.h, docs/design/011-pure-transport.md). */ +/* Links over UDP (link/link.h): link/link-protocol.h's two sides with + * sockets and threads. */ #include "link.h" -#include <arpa/inet.h> #include <errno.h> -#include <fcntl.h> -#include <malloc.h> #include <netdb.h> #include <netinet/in.h> #include <netinet/udp.h> @@ -14,16 +12,11 @@ #include <stdlib.h> #include <string.h> #include <sys/eventfd.h> -#include <sys/mman.h> -#include <sys/random.h> #include <sys/socket.h> -#include <sys/stat.h> #include <time.h> #include <unistd.h> -enum { VERSION = 0x70, HEADER = 21, DATAGRAM = 1200, BODY = DATAGRAM - HEADER - PURE_AEAD_TAG, PIECE = 1000 }; -enum { F_PADDING = 0, F_STREAM = 1, F_RECORD = 2, F_ACK = 3, F_STORED = 4, F_PING = 5, F_RECORDS = 6, STORED_GAP = 1 }; -enum { RECORDS_HEAD = 10 }; /* type, the first number, the count */ +enum { DATAGRAM = PURE_LINK_DATAGRAM }; /* Packets sent at once with UDP GSO: each but the last is a whole datagram * (a full one padded), and 54 of them fit a UDP datagram's 65,507 bytes. */ enum { BURST = 54 }; @@ -33,47 +26,7 @@ enum { BURST = 54 }; #ifndef UDP_GRO #define UDP_GRO 104 #endif -enum { RANGES = 32, RUNS = 2, UNITS = 64, CREDIT = 1024, BATCH = 256, RECORD_HEAD = 19 }; -enum { WINDOW_FIRST = 10 * DATAGRAM, WINDOW_LEAST = 2 * DATAGRAM }; #define MS UINT64_C(1000000) -/* RFC 9002's values, but a first RTT of 100 ms (a LAN's, not the 333 ms of - * the Internet at large); a probe at least every second while a receiver is - * away; a record received but not stored after a second is sent again. */ -static const uint64_t FIRST_RTT = 100 * MS, ACK_DELAY_MOST = 25 * MS, GRANULARITY = 1 * MS, PTO_MOST = 1000 * MS, - STORED_WAIT = 1000 * MS, RECORD_MOST = UINT64_C(64) << 20, HELD_MOST = UINT64_C(16) << 20; - -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 get16(const unsigned char *p) { return (uint32_t)p[0] | (uint32_t)p[1] << 8; } -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; } -/* LEB128: seven bits a byte, low first. */ -static size_t put_varint(unsigned char *p, uint64_t v) { - size_t n = 0; - for (; v >= 0x80; v >>= 7) p[n++] = (unsigned char)(v | 0x80); - p[n++] = (unsigned char)v; - return n; -} -static size_t varint_size(uint64_t v) { - size_t n = 1; - for (; v >= 0x80; v >>= 7) ++n; - return n; -} -/* A varint from p (size bytes): how many bytes it took, or 0. */ -static size_t get_varint(const unsigned char *p, size_t size, uint64_t *v) { - *v = 0; - for (size_t n = 0; n < size && n < 10; ++n) { - *v |= (uint64_t)(p[n] & 0x7f) << (7 * n); - if (!(p[n] & 0x80)) return n + 1; - } - return 0; -} - -static uint64_t least(uint64_t a, uint64_t b) { return a < b ? a : b; } -static uint64_t most(uint64_t a, uint64_t b) { return a > b ? a : b; } static uint64_t now_ns(void) { struct timespec t; @@ -81,88 +34,6 @@ static uint64_t now_ns(void) { 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 uint32_t key_id(const unsigned char key[PURE_SEAL_KEY]) { return fnv(2166136261u, key, PURE_SEAL_KEY); } - -static uint64_t random64(void) { - uint64_t value = 0; - while (getrandom(&value, sizeof(value), 0) != sizeof(value) || !value) {} - return value; -} - -/* The key and nonce base of packets from `from` (in its run from_run) to - * `to`, answering to_run (0 for none). */ -static bool derive(const unsigned char secret[32], const unsigned char from[PURE_SEAL_KEY], - const unsigned char to[PURE_SEAL_KEY], uint64_t from_run, uint64_t to_run, - unsigned char key[PURE_AEAD_KEY], unsigned char base[PURE_AEAD_NONCE]) { - unsigned char info[16 + 2 * PURE_SEAL_KEY + 16], out[PURE_AEAD_KEY + PURE_AEAD_NONCE]; - memcpy(info, "pure link 0 keys", 16); - memcpy(info + 16, from, PURE_SEAL_KEY); - memcpy(info + 16 + PURE_SEAL_KEY, to, PURE_SEAL_KEY); - put64(info + 16 + 2 * PURE_SEAL_KEY, from_run); - put64(info + 24 + 2 * PURE_SEAL_KEY, to_run); - if (!pure_link_derive(secret, info, sizeof(info), out, sizeof(out))) return false; - memcpy(key, out, PURE_AEAD_KEY); - memcpy(base, out + PURE_AEAD_KEY, PURE_AEAD_NONCE); - explicit_bzero(out, sizeof(out)); - return true; -} - -/* The same as a context of its own, for a sender's one link. */ -static PureAead *aead_of(const unsigned char secret[32], const unsigned char from[PURE_SEAL_KEY], - const unsigned char to[PURE_SEAL_KEY], uint64_t from_run, uint64_t to_run, - unsigned char base[PURE_AEAD_NONCE]) { - unsigned char key[PURE_AEAD_KEY]; - PureAead *aead = derive(secret, from, to, from_run, to_run, key, base) ? pure_aead(key) : NULL; - explicit_bzero(key, sizeof(key)); - return aead; -} - -static void nonce_of(unsigned char out[PURE_AEAD_NONCE], const unsigned char base[PURE_AEAD_NONCE], uint64_t pn) { - memcpy(out, base, PURE_AEAD_NONCE); - for (int i = 0; i < 8; ++i) out[4 + i] ^= (unsigned char)(pn >> (56 - 8 * i)); -} - -/* A packet: the header, then body sealed (with key, or the context's own - * when NULL); its size, or 0. */ -static size_t seal_packet(PureAead *aead, const unsigned char *key, const unsigned char base[PURE_AEAD_NONCE], uint32_t id, - uint64_t run, uint64_t pn, const unsigned char *body, size_t size, unsigned char *out) { - unsigned char nonce[PURE_AEAD_NONCE]; - out[0] = VERSION; - put32(out + 1, id); - put64(out + 5, run); - put64(out + 13, pn); - nonce_of(nonce, base, pn); - return pure_aead_seal(aead, key, nonce, out, HEADER, body, size, out + HEADER) ? HEADER + size + PURE_AEAD_TAG : 0; -} - -static size_t put_names(unsigned char *p, const char *source, const char *input) { - size_t a = strlen(source), b = strlen(input); - p[0] = (unsigned char)a; - memcpy(p + 1, source, a); - p[1 + a] = (unsigned char)b; - memcpy(p + 2 + a, input, b); - return 2 + a + b; -} - -/* A source and an input's names from p (size bytes); how many it takes, or 0. */ -static size_t get_names(const unsigned char *p, size_t size, char source[PURE_LINK_NAME + 1], - char input[PURE_LINK_NAME + 1]) { - if (size < 1 || !p[0] || p[0] > PURE_LINK_NAME || size < 2u + p[0]) return 0; - size_t a = p[0], b = p[1 + a]; - if (!b || b > PURE_LINK_NAME || size < 2 + a + b) return 0; - memcpy(source, p + 1, a); - source[a] = 0; - memcpy(input, p + 2 + a, b); - input[b] = 0; - return 2 + a + b; -} - /* Datagrams taken from a socket at once (recvmmsg). */ enum { TAKE = 64, SLOT = DATAGRAM + 64 }; typedef struct { @@ -221,63 +92,27 @@ static int take_datagrams(int fd, Taking *t) { /* ---- The sending side */ -enum { UNSENT, FLYING, RECEIVED }; -typedef struct { uint64_t pn; uint8_t state; } Piece; -/* A record queued: its bytes at `at` in the link's byte ring, its pieces' - * states (one kept here, more beside). */ -typedef struct { uint64_t seq; size_t at; uint32_t size, pieces; Piece *more, one; } Record; -static Piece *piece_of(Record *r, uint32_t i) { return r->pieces == 1 ? &r->one : &r->more[i]; } -typedef struct { uint64_t seq; uint32_t piece; } Unit; -typedef struct { - uint64_t pn, sent; /* its number; when it went (ns) */ - uint32_t bytes; - bool eliciting, done; /* acknowledged or lost */ - uint16_t count; - Unit units[UNITS]; -} Flight; - struct PureLink { - uint32_t shape, own_id, peer_id; - char host[256], port[8], input[PURE_LINK_NAME + 1], source[PURE_LINK_NAME + 1]; - const PureSealKey *own; - unsigned char own_public[PURE_SEAL_KEY], peer[PURE_SEAL_KEY], secret[32]; - bool peered; + PureLinkSender *sender; /* under the lock */ + char host[256], port[8], input[PURE_LINK_NAME + 1]; + bool peered, keyed; /* given the receiver's key; the keys made from it */ pthread_mutex_t lock; - Record *records; /* the sends not yet stored, oldest first (a ring of consecutive numbers) */ - size_t head, count, capacity; - /* Their bytes, one after another as queued (a record's never cut by the - * ring's end): taken back in order as they are stored. */ - unsigned char *ring; - size_t ring_capacity; - uint64_t queued, stored, told; /* the last queued; saved by the receiver; told the owner */ bool stopping, started; /* The thread sleeps (in poll), and has been woken since; records were * queued since it last sent. A queue wakes it only when it sleeps, once. */ bool sleeping, signaled, queued_new; + uint64_t told; /* the last stored the owner was told of */ int fd, wake; pthread_t thread; void (*notify)(void *); void *context; - /* The thread's own, under the lock. */ - uint64_t run, next_pn, in_run; - PureAead *out, *in; - unsigned char out_base[PURE_AEAD_NONCE], in_base[PURE_AEAD_NONCE]; - Flight *flights; /* one for each number sent, oldest first (a ring) */ - size_t flight_head, flight_count, flight_capacity; - uint64_t in_flight, window, threshold, recovery_start, eliciting; - uint64_t largest_acked, latest_rtt, smoothed_rtt, rtt_var, min_rtt; - bool acked_any, sampled; - uint64_t loss_time, last_eliciting, credit, stored_changed; - uint64_t received; /* pieces received and not yet stored */ - uint64_t unsent_from; /* no record before this number has a piece unsent */ + /* The thread's own. */ Taking *taking; /* answers, received at once */ unsigned char *burst; /* packets sealed and not yet sent (BURST datagrams) */ size_t burst_size; unsigned burst_count; uint64_t burst_pn[BURST]; bool gso; - unsigned pto_count, probes; - PureLinkCounts counts; }; PureLink *pure_link_open(const char *url, const char *source, uint32_t shape, const PureSealKey *own, @@ -300,41 +135,22 @@ PureLink *pure_link_open(const char *url, const char *source, uint32_t shape, co snprintf(error, size, "out of memory"); return NULL; } - l->shape = shape; memcpy(l->host, host, (size_t)(colon - host)); memcpy(l->port, colon + 1, (size_t)(slash - colon - 1)); strcpy(l->input, slash + 1); - strcpy(l->source, source); - l->own = own; - memcpy(l->own_public, own_public, PURE_SEAL_KEY); - l->own_id = key_id(own_public); l->fd = l->wake = -1; + if (!(l->sender = pure_link_sender(source, l->input, shape, own, own_public))) { + pthread_mutex_destroy(&l->lock); + free(l); + snprintf(error, size, "out of memory"); + return NULL; + } return l; } void pure_link_peer(PureLink *l, const unsigned char peer[PURE_SEAL_KEY]) { - memcpy(l->peer, peer, PURE_SEAL_KEY); - l->peer_id = key_id(peer); l->peered = true; -} - -static Record *record_at(PureLink *l, size_t k) { return &l->records[(l->head + k) % l->capacity]; } - -/* The queued record numbered seq, or NULL (stored already, or not queued). */ -static Record *record_of(PureLink *l, uint64_t seq) { - if (!l->count || seq < record_at(l, 0)->seq || seq - record_at(l, 0)->seq >= l->count) return NULL; - return record_at(l, (size_t)(seq - record_at(l, 0)->seq)); -} - -/* Every change of a piece's state goes through here: the link knows at once - * how many pieces wait to be stored, and where the unsent ones begin. */ -static void set_piece(PureLink *l, Record *r, uint32_t i, uint8_t state) { - uint8_t was = (*piece_of(r, i)).state; - if (was == state) return; - if (was == RECEIVED) --l->received; - if (state == RECEIVED) ++l->received; - if (state == UNSENT && r->seq < l->unsent_from) l->unsent_from = r->seq; - (*piece_of(r, i)).state = state; + l->keyed = pure_link_sender_peer(l->sender, peer); } static void wake(int fd) { @@ -342,372 +158,39 @@ static void wake(int fd) { if (fd >= 0 && write(fd, &one, sizeof(one)) != sizeof(one)) { /* full: it is readable already */ } } -/* Where size bytes go at the end of the byte ring (after the newest record, - * or from its start when that is free and the end is not), or SIZE_MAX. */ -static size_t ring_place(PureLink *l, size_t size) { - if (!l->count) return size <= l->ring_capacity ? 0 : SIZE_MAX; - Record *oldest = record_at(l, 0), *newest = record_at(l, l->count - 1); - size_t head = oldest->at, tail = newest->at + newest->size; - if (newest->at >= head) /* the bytes are [head, tail): after them, or before them from the start */ - return l->ring_capacity - tail >= size ? tail : size <= head ? 0 : SIZE_MAX; - return head - tail >= size ? tail : SIZE_MAX; /* around the end: between the newest and the oldest */ -} - -/* A ring room for size bytes more: twice as large (or more), the records - * moved into it one after another. */ -static bool ring_grow(PureLink *l, size_t size) { - size_t used = 0; - for (size_t k = 0; k < l->count; ++k) used += record_at(l, k)->size; - size_t capacity = l->ring_capacity ? 2 * l->ring_capacity : 64 << 10; - while (capacity < used + size + 1) capacity *= 2; - unsigned char *ring = malloc(capacity); - if (!ring) return false; - for (size_t k = 0, at = 0; k < l->count; ++k) { - Record *r = record_at(l, k); - if (r->size) memcpy(ring + at, l->ring + r->at, r->size); - r->at = at; - at += r->size; - } - free(l->ring); - l->ring = ring; - l->ring_capacity = capacity; - return true; -} - bool pure_link_queue(PureLink *l, uint64_t seq, const unsigned char *bytes, size_t size) { - Record r = {.seq = seq, .size = (uint32_t)size}; - r.pieces = size ? (uint32_t)((size + PIECE - 1) / PIECE) : 1; - if (size > RECORD_MOST || (r.pieces > 1 && !(r.more = calloc(r.pieces, sizeof(Piece))))) return false; pthread_mutex_lock(&l->lock); - bool ok = seq > l->queued && (!l->count || seq == l->queued + 1); - size_t at = ok ? ring_place(l, size) : SIZE_MAX; - if (ok && at == SIZE_MAX) ok = ring_grow(l, size) && (at = ring_place(l, size)) != SIZE_MAX; - if (ok) { - r.at = at; - if (size) memcpy(l->ring + at, bytes, size); - } - if (ok && l->count == l->capacity) { - size_t capacity = l->capacity ? 2 * l->capacity : 64; - Record *grown = malloc(capacity * sizeof(*grown)); - if (grown) { - for (size_t k = 0; k < l->count; ++k) grown[k] = *record_at(l, k); - free(l->records); - l->records = grown; - l->capacity = capacity; - l->head = 0; - } else - ok = false; - } - bool signal = false; + bool ok = pure_link_sender_queue(l->sender, seq, bytes, size), signal = false; if (ok) { - *record_at(l, l->count++) = r; - l->queued = seq; l->queued_new = true; signal = l->sleeping && !l->signaled; l->signaled = l->signaled || signal; } pthread_mutex_unlock(&l->lock); - if (!ok) { - free(r.more); - return false; - } if (signal) wake(l->wake); - return true; + return ok; } uint64_t pure_link_queued(PureLink *l) { pthread_mutex_lock(&l->lock); - uint64_t queued = l->queued; + uint64_t queued = pure_link_sender_queued(l->sender); pthread_mutex_unlock(&l->lock); return queued; } uint64_t pure_link_delivered(PureLink *l) { pthread_mutex_lock(&l->lock); - uint64_t stored = l->stored; + uint64_t stored = pure_link_sender_stored(l->sender); pthread_mutex_unlock(&l->lock); return stored; } -static Flight *flight_of(PureLink *l, uint64_t pn) { - if (!l->flight_count) return NULL; - uint64_t first = l->flights[l->flight_head].pn; - if (pn < first || pn - first >= l->flight_count) return NULL; - return &l->flights[(l->flight_head + (size_t)(pn - first)) % l->flight_capacity]; -} - -static bool push_flight(PureLink *l, const Flight *f) { - if (l->flight_count == l->flight_capacity) { - size_t capacity = l->flight_capacity ? 2 * l->flight_capacity : 256; - Flight *grown = malloc(capacity * sizeof(*grown)); - if (!grown) return false; - for (size_t k = 0; k < l->flight_count; ++k) grown[k] = l->flights[(l->flight_head + k) % l->flight_capacity]; - free(l->flights); - l->flights = grown; - l->flight_capacity = capacity; - l->flight_head = 0; - } - l->flights[(l->flight_head + l->flight_count++) % l->flight_capacity] = *f; - return true; -} - -static void settle(PureLink *l, Flight *f) { - f->done = true; - if (f->eliciting) { - --l->eliciting; - l->in_flight -= f->bytes; - } -} - -/* A packet lost: what it carried and was not received otherwise goes again. */ -static void lose(PureLink *l, Flight *f, uint64_t now) { - settle(l, f); - ++l->counts.lost; - for (uint16_t u = 0; u < f->count; ++u) { - Record *r = record_of(l, f->units[u].seq); - Piece *p = r ? &(*piece_of(r, f->units[u].piece)) : NULL; - if (p && p->state == FLYING && p->pn == f->pn) { - set_piece(l, r, f->units[u].piece, UNSENT); - ++l->counts.resent; - } - } - if (f->sent > l->recovery_start) { /* one congestion event per round trip */ - l->recovery_start = now; - l->window = most(l->window / 2, WINDOW_LEAST); - l->threshold = l->window; - } -} - -static void detect_lost(PureLink *l, uint64_t now) { - l->loss_time = 0; - if (!l->acked_any) return; - uint64_t rtt = most(l->latest_rtt, l->sampled ? l->smoothed_rtt : FIRST_RTT); - uint64_t delay = most(rtt * 9 / 8, GRANULARITY); - for (size_t k = 0; k < l->flight_count; ++k) { - Flight *f = &l->flights[(l->flight_head + k) % l->flight_capacity]; - if (f->pn >= l->largest_acked) break; - if (f->done) continue; - if (f->sent + delay <= now || l->largest_acked >= f->pn + 3) lose(l, f, now); - else if (!l->loss_time || f->sent + delay < l->loss_time) l->loss_time = f->sent + delay; - } -} - -static void sample_rtt(PureLink *l, uint64_t latest, uint64_t ack_delay) { - l->latest_rtt = latest; - if (!l->sampled) { - l->min_rtt = l->smoothed_rtt = latest; - l->rtt_var = latest / 2; - l->sampled = true; - return; - } - l->min_rtt = least(l->min_rtt, latest); - ack_delay = least(ack_delay, ACK_DELAY_MOST); - uint64_t adjusted = latest >= l->min_rtt + ack_delay ? latest - ack_delay : latest; - uint64_t difference = l->smoothed_rtt > adjusted ? l->smoothed_rtt - adjusted : adjusted - l->smoothed_rtt; - l->rtt_var = (3 * l->rtt_var + difference) / 4; - l->smoothed_rtt = (7 * l->smoothed_rtt + adjusted) / 8; -} - -static void on_ack(PureLink *l, const unsigned char *p, size_t size, uint64_t now) { - if (size < 13 || size < 13 + 16u * p[12]) return; - uint64_t largest = get64(p), delay = (uint64_t)get32(p + 8) * 1000; - Flight *top = NULL; - for (unsigned k = 0; k < p[12]; ++k) { - uint64_t first = get64(p + 13 + 16 * k), last = get64(p + 21 + 16 * k); - if (first > last || !l->flight_count) continue; - uint64_t oldest = l->flights[l->flight_head].pn, newest = oldest + l->flight_count - 1; - for (uint64_t pn = most(first, oldest); pn <= least(last, newest); ++pn) { - Flight *f = flight_of(l, pn); - if (!f || f->done) continue; - settle(l, f); - ++l->counts.acknowledged; - for (uint16_t u = 0; u < f->count; ++u) { - Record *r = record_of(l, f->units[u].seq); - if (r && (*piece_of(r, f->units[u].piece)).state != RECEIVED) { - set_piece(l, r, f->units[u].piece, RECEIVED); - l->stored_changed = now; /* waits for its save from now */ - } - } - if (f->eliciting && f->sent > l->recovery_start) - l->window += l->window < l->threshold ? f->bytes : (uint64_t)DATAGRAM * f->bytes / l->window; - if (pn == largest) top = f; - } - } - if (!l->acked_any || largest > l->largest_acked) l->largest_acked = largest; - l->acked_any = true; - if (top && top->eliciting) sample_rtt(l, now - top->sent, delay); - l->pto_count = 0; - detect_lost(l, now); - while (l->flight_count && l->flights[l->flight_head].done) { - l->flight_head = (l->flight_head + 1) % l->flight_capacity; - --l->flight_count; - } -} - -static void on_stored(PureLink *l, const unsigned char *p, size_t size, uint64_t now) { - char source[PURE_LINK_NAME + 1], input[PURE_LINK_NAME + 1]; - size_t names = get_names(p, size, source, input); - if (!names || size < names + 13 || strcmp(source, l->source) || strcmp(input, l->input)) return; - uint64_t heard = get64(p + names); - unsigned flags = p[names + 8]; - uint32_t credit = get32(p + names + 9); - if (heard > l->stored) { - while (l->count && record_at(l, 0)->seq <= heard) { - Record *r = record_at(l, 0); - for (uint32_t i = 0; i < r->pieces; ++i) l->received -= (*piece_of(r, i)).state == RECEIVED; - free(r->more); /* its bytes go back with the ring's head */ - l->head = (l->head + 1) % l->capacity; - --l->count; - } - l->stored = heard; - l->stored_changed = now; - } - if (flags & STORED_GAP) /* it did not keep what came after a gap: send it again */ - for (size_t k = 0; k < l->count && l->received; ++k) { - Record *r = record_at(l, k); - for (uint32_t i = 0; i < r->pieces; ++i) - if ((*piece_of(r, i)).state == RECEIVED) set_piece(l, r, i, UNSENT); - } - l->credit = credit ? least(credit, 65536) : 1; -} - -static void take_answer(PureLink *l, unsigned char *packet, ssize_t n, uint64_t now); - static void take_answers(PureLink *l, uint64_t now) { for (unsigned rounds = 0, n = TAKE; rounds < 4 && n == TAKE; ++rounds) { n = (unsigned)take_datagrams(l->fd, l->taking); - for (unsigned k = 0; k < n; ++k) take_answer(l, l->taking->bytes[k], l->taking->messages[k].msg_len, now); - } -} - -static void take_answer(PureLink *l, unsigned char *packet, ssize_t n, uint64_t now) { - unsigned char plain[DATAGRAM], nonce[PURE_AEAD_NONCE]; - if (n < HEADER + PURE_AEAD_TAG || (size_t)n > DATAGRAM || packet[0] != VERSION || get32(packet + 1) != l->peer_id) - return; /* not ours, or an error from an earlier send (nobody listening) */ - uint64_t run = get64(packet + 5), pn = get64(packet + 13); - if (!l->in || l->in_run != run) { - pure_aead_free(l->in); - l->in = aead_of(l->secret, l->peer, l->own_public, run, l->run, l->in_base); - l->in_run = run; - if (!l->in) return; - } - nonce_of(nonce, l->in_base, pn); - size_t size = (size_t)n - HEADER - PURE_AEAD_TAG; - if (!pure_aead_open(l->in, NULL, nonce, packet, HEADER, packet + HEADER, (size_t)n - HEADER, plain)) return; - ++l->counts.received; - for (size_t at = 0; at < size;) { - unsigned type = plain[at++]; - if (type == F_ACK && at + 13 <= size && at + 13 + 16u * plain[at + 12] <= size) { - on_ack(l, plain + at, size - at, now); - at += 13 + 16u * plain[at + 12]; - } else if (type == F_STORED) { - char source[PURE_LINK_NAME + 1], input[PURE_LINK_NAME + 1]; - size_t names = get_names(plain + at, size - at, source, input); - if (!names || at + names + 13 > size) break; - on_stored(l, plain + at, size - at, now); - at += names + 13; - } else if (type != F_PING) - break; /* padding, or what this side does not know: the rest */ - } -} - -/* A packet of pieces not sent yet, in order, within the receiver's credit: - * its body's size, or 0 when there is none. */ -static size_t fill(PureLink *l, unsigned char *body, Flight *f) { - size_t at = 0; - body[at++] = F_STREAM; - at += put_names(body + at, l->source, l->input); - put32(body + at, l->shape); - at += 4; - /* The credit counts from the oldest record not stored: after a restart - * the sender does not know yet how far the receiver stored. */ - uint64_t base = l->count ? record_at(l, 0)->seq - 1 : l->stored; - /* From the first record that may have an unsent piece: the ones before - * it are sent, and every one found sent moves that mark on. */ - size_t k = l->count && l->unsent_from > base + 1 ? (size_t)(l->unsent_from - base - 1) : 0; - bool before = true; - /* A record none of whose pieces went yet goes whole when it fits what is - * left of the packet: whole records whose numbers follow on share a - * RECORDS frame (its count at run). Else its unsent pieces go, each in a - * RECORD frame. */ - size_t run = 0; - uint64_t next = 0; - for (; k < l->count && f->count < UNITS; ++k) { - Record *r = record_at(l, k); - if (r->seq > base + l->credit) break; - bool unsent = false, fresh = true, joins = run && r->seq == next; - for (uint32_t i = 0; i < r->pieces && fresh; ++i) fresh = piece_of(r, i)->state == UNSENT; - size_t whole = (joins ? 0 : RECORDS_HEAD) + varint_size(r->size) + r->size; - if (fresh && f->count + r->pieces <= UNITS && at + whole <= BODY) { - if (!joins) { - body[at] = F_RECORDS; - put64(body + at + 1, r->seq); - body[at + 9] = 0; - run = at + 9; - at += RECORDS_HEAD; - } - at += put_varint(body + at, r->size); - if (r->size) memcpy(body + at, l->ring + r->at, r->size); - at += r->size; - ++body[run]; - next = r->seq + 1; - for (uint32_t i = 0; i < r->pieces; ++i) { - set_piece(l, r, i, FLYING); - piece_of(r, i)->pn = f->pn; - f->units[f->count++] = (Unit){r->seq, i}; - } - ++l->counts.records; - } else if (fresh && r->pieces == 1) - unsent = true; /* too large for what is left: the next packet */ - else - for (uint32_t i = 0; i < r->pieces; ++i) { - if (piece_of(r, i)->state != UNSENT) continue; - uint32_t offset = i * PIECE, length = (uint32_t)least(PIECE, r->size - offset); - if (f->count == UNITS || at + RECORD_HEAD + length > BODY) { - unsent = true; - break; - } - body[at] = F_RECORD; - put64(body + at + 1, r->seq); - put32(body + at + 9, r->size); - put32(body + at + 13, offset); - put16(body + at + 17, length); - if (length) memcpy(body + at + RECORD_HEAD, l->ring + r->at + offset, length); - at += RECORD_HEAD + length; - run = 0; - set_piece(l, r, i, FLYING); - piece_of(r, i)->pn = f->pn; - f->units[f->count++] = (Unit){r->seq, i}; - ++l->counts.records; - } - if (before && !unsent) l->unsent_from = r->seq + 1; - before = before && !unsent; - if (unsent) { /* the packet is full: whole, so it can go in a burst */ - if (l->gso && f->count) { - memset(body + at, F_PADDING, BODY - at); - at = BODY; - } - break; - } - } - return f->count ? at : 0; -} - -/* Seal and send f with body; its number is spent even when the socket - * refuses it (what it carried is then marked unsent again). */ -/* A packet the socket refused: what it carried is unsent again, and it - * counts as never sent. Its number is spent all the same. */ -static void refused(PureLink *l, uint64_t pn) { - Flight *f = flight_of(l, pn); - if (!f || f->done) return; - for (uint16_t u = 0; u < f->count; ++u) { - Record *r = record_of(l, f->units[u].seq); - if (r && (*piece_of(r, f->units[u].piece)).pn == pn) set_piece(l, r, f->units[u].piece, UNSENT); + for (unsigned k = 0; k < n; ++k) + pure_link_sender_take(l->sender, l->taking->bytes[k], l->taking->messages[k].msg_len, now); } - settle(l, f); - f->eliciting = false; - f->count = 0; - --l->counts.packets; } static bool transient(void) { return errno == EAGAIN || errno == EWOULDBLOCK || errno == ENOBUFS || errno == EINTR; } @@ -724,119 +207,32 @@ static void flush(PureLink *l) { if (send(l->fd, l->burst + at, size, 0) < 0 && transient()) break; at += size; } - for (unsigned k = sent; k < n; ++k) refused(l, l->burst_pn[k]); + for (unsigned k = sent; k < n; ++k) pure_link_sender_refused(l->sender, l->burst_pn[k]); l->burst_count = 0; l->burst_size = 0; } -/* Seal f with body into the burst (a whole datagram each but the last), and - * count it sent; its number is spent even when the socket then refuses it. */ -static bool transmit(PureLink *l, Flight *f, const unsigned char *body, size_t size, uint64_t now) { - if (l->burst_count && l->burst_size != (size_t)l->burst_count * DATAGRAM) flush(l); /* the last was short */ - size_t total = seal_packet(l->out, NULL, l->out_base, l->own_id, l->run, f->pn, body, size, l->burst + l->burst_size); - ++l->next_pn; - f->sent = now; - f->bytes = (uint32_t)total; - if (!total) { /* not sealed: what it carried is unsent again */ - for (uint16_t u = 0; u < f->count; ++u) { - Record *r = record_of(l, f->units[u].seq); - if (r && (*piece_of(r, f->units[u].piece)).pn == f->pn) set_piece(l, r, f->units[u].piece, UNSENT); - } - f->done = true; - f->eliciting = false; - f->count = 0; - push_flight(l, f); - return false; - } - if (!push_flight(l, f)) return false; - l->burst_pn[l->burst_count++] = f->pn; - l->burst_size += total; - ++l->counts.packets; - if (f->eliciting) { - ++l->eliciting; - l->in_flight += f->bytes; - l->last_eliciting = now; - } - if (l->burst_count == BURST) flush(l); - return true; -} - +/* Packets while the sender has them (up to 128 a turn), into bursts: a + * short packet ends one. */ static void send_packets(PureLink *l, uint64_t now) { - unsigned char body[BODY]; l->queued_new = false; /* all queued is looked at now */ for (unsigned n = 0; n < 128; ++n) { - bool probe = l->probes > 0; - if (!probe && l->in_flight + DATAGRAM > l->window) break; - Flight f = {.pn = l->next_pn, .eliciting = true}; - size_t size = fill(l, body, &f); - if (!size) { - if (!probe) break; - body[0] = F_PING; /* nothing to send again: a probe all the same */ - size = 1; - } - if (!transmit(l, &f, body, size, now)) break; - if (probe) --l->probes; + if (l->burst_count && l->burst_size != (size_t)l->burst_count * DATAGRAM) flush(l); /* the last was short */ + uint64_t pn; + size_t size = pure_link_sender_packet(l->sender, l->burst + l->burst_size, l->gso, now, &pn); + if (!size) break; + l->burst_pn[l->burst_count++] = pn; + l->burst_size += size; + if (l->burst_count == BURST) flush(l); } flush(l); } -static uint64_t pto_period(const PureLink *l) { - uint64_t smoothed = l->sampled ? l->smoothed_rtt : FIRST_RTT, var = l->sampled ? l->rtt_var : FIRST_RTT / 2; - uint64_t period = smoothed + most(4 * var, GRANULARITY) + ACK_DELAY_MOST; - for (unsigned i = 0; i < l->pto_count && period < PTO_MOST; ++i) period *= 2; - return least(period, PTO_MOST); -} - -static bool waiting_for_save(PureLink *l) { return l->received > 0; } - -/* The next time a timer is due (ns), or 0. */ -static uint64_t next_timer(PureLink *l) { - uint64_t due = l->loss_time; - if (l->eliciting) { - uint64_t pto = l->last_eliciting + pto_period(l); - if (!due || pto < due) due = pto; - } - if (waiting_for_save(l)) { - uint64_t stored = l->stored_changed + STORED_WAIT; - if (!due || stored < due) due = stored; - } - return due; -} - -static void run_timers(PureLink *l, uint64_t now) { - if (l->loss_time && l->loss_time <= now) detect_lost(l, now); - else if (l->eliciting && l->last_eliciting + pto_period(l) <= now) { - /* Nothing heard for a while: a probe, what flies longest sent again. */ - ++l->pto_count; - l->probes = 1; - for (size_t k = 0; k < l->count; ++k) { - Record *r = record_at(l, k); - uint32_t i = 0; - while (i < r->pieces && (*piece_of(r, i)).state != FLYING) ++i; - if (i < r->pieces) { - set_piece(l, r, i, UNSENT); - ++l->counts.resent; - break; - } - } - } - if (waiting_for_save(l) && l->stored_changed + STORED_WAIT <= now) { - /* Received long ago and still not stored: the receiver lost them (it - * restarted) or its word got lost. Sent again: it takes a number once. */ - for (size_t k = 0; k < l->count && l->received; ++k) { - Record *r = record_at(l, k); - for (uint32_t i = 0; i < r->pieces; ++i) - if ((*piece_of(r, i)).state == RECEIVED) set_piece(l, r, i, UNSENT), ++l->counts.resent; - } - l->stored_changed = now; - } -} - static void *run_link(void *context) { PureLink *l = context; for (;;) { pthread_mutex_lock(&l->lock); - uint64_t now = now_ns(), due = next_timer(l); + uint64_t now = now_ns(), due = pure_link_sender_due(l->sender); bool stopping = l->stopping, more = l->queued_new; /* queued while it sent: no sleep */ l->sleeping = !more; pthread_mutex_unlock(&l->lock); @@ -852,10 +248,11 @@ static void *run_link(void *context) { l->sleeping = l->signaled = false; now = now_ns(); if (waits[0].revents) take_answers(l, now); - run_timers(l, now); + pure_link_sender_tick(l->sender, now); send_packets(l, now); - bool told = l->stored != l->told; - l->told = l->stored; + uint64_t stored = pure_link_sender_stored(l->sender); + bool told = stored != l->told; + l->told = stored; pthread_mutex_unlock(&l->lock); if (told && l->notify) l->notify(l->context); } @@ -883,13 +280,7 @@ bool pure_link_start(PureLink *l, void (*notify)(void *), void *context, char *e } int segment = DATAGRAM; l->gso = ok && !setsockopt(l->fd, SOL_UDP, UDP_SEGMENT, &segment, sizeof(segment)); - ok = ok && (l->burst = malloc((size_t)BURST * DATAGRAM)) && (l->taking = malloc(sizeof(*l->taking))); - l->run = random64(); - l->window = WINDOW_FIRST; - l->threshold = UINT64_MAX; - l->credit = CREDIT; - ok = ok && pure_link_secret(l->own, l->peer, l->secret) && - (l->out = aead_of(l->secret, l->own_public, l->peer, l->run, 0, l->out_base)) && + ok = ok && l->keyed && (l->burst = malloc((size_t)BURST * DATAGRAM)) && (l->taking = malloc(sizeof(*l->taking))) && (l->wake = eventfd(0, EFD_NONBLOCK | EFD_CLOEXEC)) >= 0; l->notify = notify; l->context = context; @@ -904,8 +295,7 @@ bool pure_link_start(PureLink *l, void (*notify)(void *), void *context, char *e void pure_link_counts(PureLink *l, PureLinkCounts *out) { pthread_mutex_lock(&l->lock); - *out = l->counts; - out->stored = l->stored; + pure_link_sender_counts(l->sender, out); pthread_mutex_unlock(&l->lock); } @@ -918,803 +308,46 @@ void pure_link_close(PureLink *l) { wake(l->wake); pthread_join(l->thread, NULL); } - for (size_t k = 0; k < l->count; ++k) { - free(record_at(l, k)->more); - } - free(l->records); - free(l->ring); - free(l->flights); + pure_link_sender_free(l->sender); free(l->burst); free(l->taking); - pure_aead_free(l->out); - pure_aead_free(l->in); - explicit_bzero(l->secret, sizeof(l->secret)); if (l->fd >= 0) close(l->fd); if (l->wake >= 0) close(l->wake); pthread_mutex_destroy(&l->lock); free(l); } -/* ---- The receiving side - * - * A sender allowed is a peer: its key and the sources it may send as, a few - * bytes each, found by its key's id. What a sending peer needs besides (the - * secret its key shares with the node's, its runs' keys and addresses, its - * streams) is its active state: made at its first packet (one X25519), let - * go once the peer has been quiet for the idle period with nothing of it - * waiting here. So a link at rest costs the node its allowing, and the next - * packet goes on without a handshake. */ - -typedef struct { uint64_t first, last; } Range; -typedef struct Active Active; -typedef struct { - Active *active; - uint64_t run, seen; /* the sender's run; when it last sent */ - bool used, ack_due, due; - unsigned char in_key[PURE_AEAD_KEY], out_key[PURE_AEAD_KEY], in_base[PURE_AEAD_NONCE], out_base[PURE_AEAD_NONCE]; - uint64_t next_pn, largest_at; - struct sockaddr_in6 address; /* or an IPv4 one */ - socklen_t address_size; - Range got[RANGES]; /* the numbers received, newest first */ - unsigned got_count; -} Run; -typedef struct { - uint64_t seq; - unsigned char *bytes; /* its size, then a byte a piece: had */ - uint32_t size, pieces, got; -} Held; -typedef struct Stream { - Active *active; - char source[PURE_LINK_NAME + 1], input_name[PURE_LINK_NAME + 1]; - uint32_t input, shape; /* the host's number for the input; the messages' shape */ - uint64_t heard; /* the last number heard (stored), once known */ - bool known, out, stored_due, gap, ready; - /* The line: once the last heard is known, whole records from the one - * after it on, in order, as a delivery carries them (u32 size, bytes), - * from line_head: handed over as they are, let go by moving the head. - * Records that come early or in pieces, or before the last heard is - * known, wait in held until the line reaches them. */ - unsigned char *line; - size_t line_head, line_size, line_capacity; - uint64_t line_first; - unsigned line_count; - Held *held; /* records not yet stored, by number: held_count from held_first */ - size_t held_first, held_count, held_capacity, held_bytes; - Run *run; /* the run that sent to it last: where STORED goes */ - uint64_t run_id; -} Stream; -struct Active { - uint32_t peer; /* its place among the peers */ - unsigned char secret[32]; - uint64_t seen; - size_t place; /* among the active */ - Run runs[RUNS]; - Stream **streams; - unsigned stream_count, stream_capacity; -}; -typedef struct { - unsigned char key[PURE_SEAL_KEY]; - uint32_t id, allowed; /* allowed: the place + 1 of its first source (0: none) */ - Active *active; -} Peer; -typedef struct { uint32_t name, next; } Allowed; /* name: where in names (u8 length, bytes); next: a place + 1 */ -typedef struct { Stream *stream; uint64_t heard; bool stored, gap; } Done; - -/* A peer quiet this long lets its active state go; the active are looked at - * that often; what they held goes back to the system once TRIM_AFTER have - * gone, or TRIM_WAIT after any went. */ -static const uint64_t IDLE_FIRST = 30000 * MS, SWEEP_EVERY = 250 * MS, TRIM_WAIT = 10000 * MS; -enum { TRIM_AFTER = 1024 }; +/* ---- The receiving side */ struct PureLinkListener { - PureLinkHost host; - const PureSealKey *own; - unsigned char own_public[PURE_SEAL_KEY]; - uint32_t own_id; + PureLinkReceiver *receiver; /* the thread's */ unsigned port; - uint64_t run, idle, swept, trimmed, untrimmed; - PureAead *aead; /* keyed by each packet's run */ - unsigned char open_key[PURE_AEAD_KEY], seal_key[PURE_AEAD_KEY]; /* the keys it has now */ - bool open_keyed, seal_keyed; - unsigned char *memo; /* the secrets remembered (pure_link_remember), mapped */ - size_t memo_size; int fd, wake; pthread_t thread; bool started; pthread_mutex_t lock; /* the done list and stopping */ - Done *done; + PureLinkDone *done; size_t done_count, done_capacity; bool stopping; - Peer *peers; - size_t peer_count, peer_capacity; - uint32_t *index; /* the peers by key id: open addressing, a place + 1 (0: empty) */ - size_t index_size; - Allowed *allowed; - size_t allowed_count, allowed_capacity; - unsigned char *names; - size_t names_size, names_capacity; - Active **actives; - size_t active_count, active_capacity; - Stream **ready; /* streams that may have whole records to hand over */ - size_t ready_count, ready_capacity; - Run **due; /* runs to answer */ - size_t due_count, due_capacity; - unsigned char *batch; - size_t batch_capacity; - PureLinkCounts counts; }; -PureLinkListener *pure_link_listen(unsigned port, const PureSealKey *own, const unsigned char own_public[PURE_SEAL_KEY], - const PureLinkHost *host, char *error, size_t size) { - PureLinkListener *s = own ? calloc(1, sizeof(*s)) : NULL; +PureLinkListener *pure_link_listen(PureLinkReceiver *r, unsigned port, char *error, size_t size) { + PureLinkListener *s = calloc(1, sizeof(*s)); if (!s || pthread_mutex_init(&s->lock, NULL)) { free(s); - snprintf(error, size, own ? "out of memory" : "--link-port needs the node's --key"); + snprintf(error, size, "out of memory"); return NULL; } - s->host = *host; - s->own = own; - memcpy(s->own_public, own_public, PURE_SEAL_KEY); - s->own_id = key_id(own_public); + s->receiver = r; s->port = port; - s->idle = IDLE_FIRST; s->fd = s->wake = -1; return s; } -void pure_link_idle(PureLinkListener *s, uint64_t ms) { s->idle = ms * MS; } - -/* items (count used of capacity) with room for one more: the same or grown, - * or NULL when memory runs out (items stay as they were). */ -static void *room(void *items, size_t *capacity, size_t count, size_t item_size, size_t first) { - if (count < *capacity) return items; - size_t more = *capacity ? 2 * *capacity : first; - void *grown = realloc(items, more * item_size); - if (grown) *capacity = more; - return grown; -} - -/* The peers by key id again, in an index twice as large. */ -static bool grow_index(PureLinkListener *s) { - size_t size = s->index_size ? 2 * s->index_size : 1024; - uint32_t *index = calloc(size, sizeof(*index)); - if (!index) return false; - for (size_t p = 0; p < s->peer_count; ++p) { - size_t slot = s->peers[p].id & (size - 1); - while (index[slot]) slot = (slot + 1) & (size - 1); - index[slot] = (uint32_t)p + 1; - } - free(s->index); - s->index = index; - s->index_size = size; - return true; -} - -static bool same_name(const PureLinkListener *s, uint32_t name, const char *source, size_t length) { - return s->names[name] == length && !memcmp(s->names + name + 1, source, length); -} - -/* Keys are found by their ids, which are 32 bits: two keys may share one. */ -bool pure_link_allow(PureLinkListener *s, const char *source, const unsigned char key[PURE_SEAL_KEY], char *error, - size_t size) { - size_t length = strlen(source), p = SIZE_MAX; - uint32_t id = key_id(key); - if (!length || length > PURE_LINK_NAME) { - snprintf(error, size, "a source's name is 1 to %d bytes", PURE_LINK_NAME); - return false; - } - for (size_t slot = s->index_size ? id & (s->index_size - 1) : 0; s->index_size && s->index[slot]; - slot = (slot + 1) & (s->index_size - 1)) { - const Peer *peer = &s->peers[s->index[slot] - 1]; - if (peer->id == id && !memcmp(peer->key, key, PURE_SEAL_KEY)) { - p = s->index[slot] - 1; - break; - } - } - if (p == SIZE_MAX) { - Peer *peers = s->peer_count < UINT32_MAX - 1 && ((s->peer_count + 1) * 2 <= s->index_size || grow_index(s)) ? - room(s->peers, &s->peer_capacity, s->peer_count, sizeof(Peer), 64) : NULL; - if (!peers) goto full; - s->peers = peers; - p = s->peer_count++; - s->peers[p] = (Peer){.id = id}; - memcpy(s->peers[p].key, key, PURE_SEAL_KEY); - size_t slot = id & (s->index_size - 1); - while (s->index[slot]) slot = (slot + 1) & (s->index_size - 1); - s->index[slot] = (uint32_t)p + 1; - } - for (uint32_t a = s->peers[p].allowed; a; a = s->allowed[a - 1].next) - if (same_name(s, s->allowed[a - 1].name, source, length)) return true; - Allowed *kept = s->allowed_count < UINT32_MAX - 1 && s->names_size + 1 + length <= UINT32_MAX ? - room(s->allowed, &s->allowed_capacity, s->allowed_count, sizeof(Allowed), 64) : NULL; - if (!kept) goto full; - s->allowed = kept; - if (s->names_size + 1 + length > s->names_capacity) { - size_t more = s->names_capacity ? 2 * s->names_capacity : 4096; - unsigned char *names = realloc(s->names, more); - if (!names) goto full; - s->names = names; - s->names_capacity = more; - } - s->names[s->names_size] = (unsigned char)length; - memcpy(s->names + s->names_size + 1, source, length); - s->allowed[s->allowed_count] = (Allowed){(uint32_t)s->names_size, s->peers[p].allowed}; - s->peers[p].allowed = (uint32_t)++s->allowed_count; - s->names_size += 1 + length; - return true; -full: - snprintf(error, size, "out of memory"); - return false; -} - -static Held *held_at(Stream *st, size_t k) { return &st->held[st->held_first + k]; } - -static uint64_t line_end(const Stream *st) { return st->line_first + st->line_count; } - -/* Room for more bytes at the line's end: its front let go given back - * first when that is at least half of it (so no byte moves twice in a - * doubling), else a longer line. */ -static bool line_room(Stream *st, size_t more) { - if (st->line_size + more <= st->line_capacity) return true; - if (st->line_head && st->line_head >= st->line_capacity / 2) { - memmove(st->line, st->line + st->line_head, st->line_size - st->line_head); - st->line_size -= st->line_head; - st->line_head = 0; - if (st->line_size + more <= st->line_capacity) return true; - } - size_t capacity = st->line_capacity ? st->line_capacity : 4096; - while (capacity < st->line_size + more) capacity *= 2; - unsigned char *grown = realloc(st->line, capacity); - if (!grown) return false; - st->line = grown; - st->line_capacity = capacity; - return true; -} - -static bool line_append(Stream *st, const unsigned char *bytes, uint32_t size) { - if (!line_room(st, 4 + (size_t)size)) return false; - put32(st->line + st->line_size, size); - if (size) memcpy(st->line + st->line_size + 4, bytes, size); - st->line_size += 4 + (size_t)size; - ++st->line_count; - return true; -} - -/* The line's records up to heard let go (all of them, when heard is past it). */ -static void line_drop(Stream *st, uint64_t heard) { - if (heard < st->line_first) return; - if (heard >= line_end(st)) { - st->line_head = st->line_size = 0; - st->line_count = 0; - st->line_first = heard + 1; - return; - } - for (uint64_t n = heard - st->line_first + 1; n; --n, --st->line_count, ++st->line_first) - st->line_head += 4 + (size_t)get32(st->line + st->line_head); -} - -/* Records held that the line reached, moved onto it; ones behind it let go. */ -static void drop_first(Stream *st); -static void line_pull(Stream *st) { - while (st->held_count) { - Held *h = held_at(st, 0); - if (h->seq < line_end(st)) drop_first(st); - else if (h->seq == line_end(st) && h->got == h->pieces && line_append(st, h->bytes, h->size)) drop_first(st); - else break; - } -} - -/* The first record held let go: the ones after it stay where they are. */ -static void drop_first(Stream *st) { - st->held_bytes -= held_at(st, 0)->size; - free(held_at(st, 0)->bytes); - ++st->held_first; - if (!--st->held_count) st->held_first = 0; -} - -/* The memo of secrets (pure_link_remember): a header, then a place a peer - * in the order allowed, each the first 28 bytes of its key, a check (FNV-1a - * of its whole key and the secret) and the secret. A place whose check - * fails (never made, another key, cut short by a crash) is made again. */ -enum { MEMO_HEAD = 64, MEMO_PLACE = 64 }; -static const unsigned char memo_magic[8] = {'P', 'L', 'M', '1', 0, 0, 0, 0}; - -static uint32_t memo_check(const unsigned char key[PURE_SEAL_KEY], const unsigned char secret[32]) { - return fnv(fnv(2166136261u, key, PURE_SEAL_KEY), secret, 32); -} - -bool pure_link_remember(PureLinkListener *s, const char *path, char *error, size_t size) { - int fd = open(path, O_RDWR | O_CREAT | O_NOFOLLOW | O_CLOEXEC, 0600); - struct stat info; - unsigned char head[MEMO_HEAD]; - size_t needed = MEMO_HEAD + (s->peer_count ? s->peer_count : 1) * MEMO_PLACE; - bool ok = fd >= 0 && !fstat(fd, &info) && S_ISREG(info.st_mode) && !(info.st_mode & 077); - /* Another node's key, or none yet: begun again. */ - bool ours = ok && pread(fd, head, MEMO_HEAD, 0) == MEMO_HEAD && !memcmp(head, memo_magic, 8) && - !memcmp(head + 8, s->own_public, PURE_SEAL_KEY); - if (ok && !ours) { - memset(head, 0, MEMO_HEAD); - memcpy(head, memo_magic, 8); - memcpy(head + 8, s->own_public, PURE_SEAL_KEY); - ok = !ftruncate(fd, 0) && pwrite(fd, head, MEMO_HEAD, 0) == MEMO_HEAD; - } - ok = ok && (ours && (size_t)info.st_size >= needed ? true : !ftruncate(fd, (off_t)needed)); - void *map = ok ? mmap(NULL, needed, PROT_READ | PROT_WRITE, MAP_SHARED, fd, 0) : MAP_FAILED; - if (fd >= 0) close(fd); - if (map == MAP_FAILED) { - snprintf(error, size, "%s: cannot keep secrets there (a file only its owner may read)", path); - return false; - } - s->memo = map; - s->memo_size = needed; - return true; -} - -/* A peer's active state, made at its first packet: the secret its key - * shares with ours, remembered or made (one X25519, then remembered: the - * secret first, the key that makes the place its own last). */ -static Active *activate(PureLinkListener *s, Peer *peer, uint64_t now) { - Active *a = calloc(1, sizeof(*a)), **actives = room(s->actives, &s->active_capacity, s->active_count, sizeof(*actives), 64); - if (actives) s->actives = actives; - if (!a || !actives) { - free(a); - return NULL; - } - size_t index = (size_t)(peer - s->peers); - unsigned char *place = s->memo && MEMO_HEAD + (index + 1) * MEMO_PLACE <= s->memo_size ? - s->memo + MEMO_HEAD + index * MEMO_PLACE : NULL; - if (place && !memcmp(place, peer->key, 28) && get32(place + 28) == memo_check(peer->key, place + 32)) { - memcpy(a->secret, place + 32, 32); - ++s->counts.remembered; - } else if (pure_link_secret(s->own, peer->key, a->secret)) { - if (place) { - memset(place, 0, 28); - __atomic_thread_fence(__ATOMIC_RELEASE); - memcpy(place + 32, a->secret, 32); - put32(place + 28, memo_check(peer->key, a->secret)); - __atomic_thread_fence(__ATOMIC_RELEASE); - memcpy(place, peer->key, 28); - } - } else { - explicit_bzero(a, sizeof(*a)); - free(a); - return NULL; - } - a->peer = (uint32_t)(peer - s->peers); - a->seen = now; - a->place = s->active_count; - s->actives[s->active_count++] = a; - peer->active = a; - return a; -} - -static void let_go(PureLinkListener *s, Active *a) { - for (unsigned k = 0; k < a->stream_count; ++k) { - while (a->streams[k]->held_count) drop_first(a->streams[k]); - free(a->streams[k]->held); - free(a->streams[k]->line); - free(a->streams[k]); - } - free(a->streams); - s->peers[a->peer].active = NULL; - s->actives[a->place] = s->actives[--s->active_count]; - s->actives[a->place]->place = a->place; - explicit_bzero(a, sizeof(*a)); - free(a); -} - -/* Whether any of a peer's records or answers waits here. */ -static bool busy(const Active *a) { - for (unsigned k = 0; k < a->stream_count; ++k) - if (a->streams[k]->out || a->streams[k]->held_count || a->streams[k]->line_count || a->streams[k]->ready) return true; - for (unsigned i = 0; i < RUNS; ++i) - if (a->runs[i].due) return true; - return false; -} - -/* Let go the active state of peers quiet for the idle period; what idle - * links held goes back to the system. */ -static void sweep(PureLinkListener *s, uint64_t now) { - if (now - s->swept < SWEEP_EVERY) return; - s->swept = now; - size_t freed = 0; - for (size_t k = 0; k < s->active_count;) { - Active *a = s->actives[k]; - if (now - a->seen < s->idle || busy(a)) ++k; - else let_go(s, a), ++freed; - } - s->counts.swept += freed; - s->untrimmed += freed; - if (s->untrimmed >= TRIM_AFTER || (s->untrimmed && now - s->trimmed >= TRIM_WAIT)) { - malloc_trim(0); - s->untrimmed = 0; - s->trimmed = now; - } -} - -static void mark_ready(PureLinkListener *s, Stream *st) { - Stream **ready = st->ready ? NULL : room(s->ready, &s->ready_capacity, s->ready_count, sizeof(*ready), 64); - if (!ready) return; - s->ready = ready; - s->ready[s->ready_count++] = st; - st->ready = true; -} - -static void mark_due(PureLinkListener *s, Run *r) { - Run **due = r->due ? NULL : room(s->due, &s->due_capacity, s->due_count, sizeof(*due), 64); - if (!due) return; - s->due = due; - s->due[s->due_count++] = r; - r->due = true; -} - -static Run *run_found(Active *a, uint64_t run) { - for (unsigned i = 0; i < RUNS; ++i) - if (a->runs[i].used && a->runs[i].run == run) return &a->runs[i]; - return NULL; -} - -/* A new run's keys, in the slot unused longest: only once its first packet - * opened with them, so a forged one displaces nothing. */ -static Run *install_run(Active *a, const Run *made) { - Run *slot = &a->runs[0]; - for (unsigned i = 0; i < RUNS; ++i) { - Run *r = &a->runs[i]; - if (!r->used || (slot->used && r->seen < slot->seen)) slot = r; - } - bool due = slot->due; /* the answer list may hold the slot */ - *slot = *made; - slot->due = due; - return slot; -} - -static bool had(const Run *r, uint64_t pn) { - for (unsigned i = 0; i < r->got_count; ++i) - if (pn >= r->got[i].first && pn <= r->got[i].last) return true; - return false; -} - -static void note(Run *r, uint64_t pn, uint64_t now) { - unsigned i = 0, n = r->got_count; - while (i < n && r->got[i].first > pn + 1) ++i; - if (i < n && r->got[i].first == pn + 1) { /* joins the range above it */ - r->got[i].first = pn; - if (i + 1 < n && r->got[i + 1].last + 1 == pn) { - r->got[i].first = r->got[i + 1].first; - memmove(&r->got[i + 1], &r->got[i + 2], (n - i - 2) * sizeof(Range)); - --r->got_count; - } - } else if (i < n && r->got[i].last + 1 == pn) { - r->got[i].last = pn; - } else if (i < RANGES) { - if (n == RANGES) --n; /* the oldest range is forgotten */ - memmove(&r->got[i + 1], &r->got[i], (n - i) * sizeof(Range)); - r->got[i] = (Range){pn, pn}; - r->got_count = n + 1; - } - if (r->got_count && r->got[0].last == pn) r->largest_at = now; -} - -static bool allowed(const PureLinkListener *s, const Peer *peer, const char *source) { - for (uint32_t a = peer->allowed; a; a = s->allowed[a - 1].next) - if (same_name(s, s->allowed[a - 1].name, source, strlen(source))) return true; - return false; -} - -/* The stream of a peer's source to input, made the first time: NULL when the - * source is not allowed from the peer, or the host has no input that takes - * such messages. */ -static Stream *stream_of(PureLinkListener *s, Active *a, const char *source, const char *input, uint32_t shape) { - for (unsigned k = 0; k < a->stream_count; ++k) { - Stream *st = a->streams[k]; - if (!strcmp(st->input_name, input) && !strcmp(st->source, source)) return st->shape == shape ? st : NULL; - } - int index = allowed(s, &s->peers[a->peer], source) ? s->host.input(s->host.context, input, shape) : -1; - if (index < 0) return NULL; - size_t capacity = a->stream_capacity; - Stream **streams = room(a->streams, &capacity, a->stream_count, sizeof(*streams), 2), *st; - if (!streams) return NULL; - a->streams = streams; - a->stream_capacity = (unsigned)capacity; - if (!(st = calloc(1, sizeof(*st)))) return NULL; - st->active = a; - st->input = (uint32_t)index; - st->shape = shape; - snprintf(st->source, sizeof(st->source), "%s", source); - snprintf(st->input_name, sizeof(st->input_name), "%s", input); - a->streams[a->stream_count++] = st; - return st; -} - -/* A record's piece, kept until the record is whole and handed over. */ -static void hold(Stream *st, uint64_t seq, uint32_t size, uint32_t offset, const unsigned char *bytes, uint32_t length) { - if (st->known && seq <= st->heard) { - st->stored_due = true; /* sent again: say again how far it is stored */ - return; - } - if ((st->known && seq > st->heard + CREDIT) || size > RECORD_MOST) return; - if (st->known && seq < line_end(st)) return; /* on the line already */ - if (st->known && seq == line_end(st) && !offset && length == size) { /* the next, whole */ - if (st->held_bytes + (st->line_size - st->line_head) + size <= HELD_MOST && line_append(st, bytes, size)) - line_pull(st); - return; - } - /* In order it goes last; out of order, its place is found by halving. */ - size_t k = st->held_count; - if (k && held_at(st, k - 1)->seq >= seq) { - size_t low = 0; - while (low < k) { - size_t middle = low + (k - low) / 2; - if (held_at(st, middle)->seq < seq) low = middle + 1; - else k = middle; - } - } - if (k == st->held_count || held_at(st, k)->seq != seq) { - if (st->held_bytes + size > HELD_MOST) return; - if (st->held_first && st->held_first + st->held_count == st->held_capacity) { /* the front is free: used again */ - memmove(st->held, held_at(st, 0), st->held_count * sizeof(Held)); - st->held_first = 0; - } - Held *kept = room(st->held, &st->held_capacity, st->held_first + st->held_count, sizeof(Held), 16); - if (!kept) return; - st->held = kept; - Held h = {.seq = seq, .size = size, .pieces = size ? (size + PIECE - 1) / PIECE : 1}; - if (!(h.bytes = malloc(size + h.pieces))) return; - memset(h.bytes + size, 0, h.pieces); - memmove(held_at(st, k + 1), held_at(st, k), (st->held_count - k) * sizeof(Held)); - *held_at(st, k) = h; - ++st->held_count; - st->held_bytes += size; - } - Held *h = held_at(st, k); - if (h->size != size) return; - if (!offset && length == size) { /* whole: every piece had */ - if (size) memcpy(h->bytes, bytes, size); - memset(h->bytes + size, 1, h->pieces); - h->got = h->pieces; - return; - } - uint32_t piece = offset / PIECE; - if (offset % PIECE || piece >= h->pieces || length != least(PIECE, size - offset) || h->bytes[size + piece]) return; - if (length) memcpy(h->bytes + offset, bytes, length); - h->bytes[size + piece] = 1; - ++h->got; -} - -/* Whole records from the one after the last heard, in order, to the host: - * one delivery a stream at a time. Before the last heard is known (after a - * restart) they go from the first held: the host says how far it heard. - * False when the host could not take them now. */ -static bool hand_over(PureLinkListener *s, Stream *st) { - if (st->out) return true; - if (st->known) { /* the line as it is, up to BATCH records */ - line_pull(st); - if (!st->line_count) return true; - unsigned n = 0; - size_t bytes = 0; - for (; n < st->line_count && n < BATCH; ++n) bytes += 4 + (size_t)get32(st->line + st->line_head + bytes); - PureLinkDelivery d = {st->input, st->source, st->line_first, n, st->line + st->line_head, bytes, st}; - if (!s->host.deliver(s->host.context, &d)) return false; - st->out = true; - ++s->counts.deliveries; - return true; - } - if (!st->held_count || (st->known && held_at(st, 0)->seq != st->heard + 1)) return true; - size_t n = 0, bytes = 0; - while (n < st->held_count && n < BATCH && held_at(st, n)->got == held_at(st, n)->pieces && - held_at(st, n)->seq == held_at(st, 0)->seq + n) - bytes += 4 + held_at(st, n++)->size; - if (!n) return true; - if (bytes > s->batch_capacity) { - unsigned char *grown = realloc(s->batch, bytes); - if (!grown) return false; - s->batch = grown; - s->batch_capacity = bytes; - } - size_t at = 0; - for (size_t k = 0; k < n; ++k) { - put32(s->batch + at, held_at(st, k)->size); - memcpy(s->batch + at + 4, held_at(st, k)->bytes, held_at(st, k)->size); - at += 4 + held_at(st, k)->size; - } - PureLinkDelivery d = {st->input, st->source, held_at(st, 0)->seq, (unsigned)n, s->batch, bytes, st}; - if (!s->host.deliver(s->host.context, &d)) return false; - st->out = true; - ++s->counts.deliveries; - return true; -} - -/* A packet from peer, if it opens with the keys of peer's run (made for a - * run not seen yet, and for a peer at rest its active state): its frames. */ -static bool take_from(PureLinkListener *s, Peer *peer, unsigned char *packet, size_t n, - const struct sockaddr_storage *from, socklen_t from_size, uint64_t now) { - uint64_t pn = get64(packet + 13), run = get64(packet + 5); - Active *a = peer->active; - Run *r = a ? run_found(a, run) : NULL; - if (r && had(r, pn)) { - r->ack_due = true; /* its acknowledgment was lost, perhaps */ - mark_due(s, r); - return true; - } - bool woke = !a; - if (woke && !(a = activate(s, peer, now))) return false; - Run made = {.active = a, .run = run, .seen = now, .used = true}; - bool keyed = r || (derive(a->secret, peer->key, s->own_public, run, 0, made.in_key, made.in_base) && - derive(a->secret, s->own_public, peer->key, s->run, run, made.out_key, made.out_base)); - unsigned char plain[DATAGRAM], nonce[PURE_AEAD_NONCE]; - nonce_of(nonce, r ? r->in_base : made.in_base, pn); - size_t size = n - HEADER - PURE_AEAD_TAG; - const unsigned char *key = r ? r->in_key : made.in_key, *give = key; - if (s->open_keyed && !memcmp(s->open_key, key, PURE_AEAD_KEY)) give = NULL; /* the key it has */ - else if (keyed) memcpy(s->open_key, key, PURE_AEAD_KEY), s->open_keyed = true; - if (!keyed || !pure_aead_open(s->aead, give, nonce, packet, HEADER, packet + HEADER, n - HEADER, plain)) { - explicit_bzero(&made, sizeof(made)); - if (woke) let_go(s, a); - return false; - } - if (woke) ++s->counts.woken; - if (!r) r = install_run(a, &made); - a->seen = r->seen = now; - note(r, pn, now); - r->address_size = from_size <= sizeof(r->address) ? from_size : sizeof(r->address); - memcpy(&r->address, from, r->address_size); /* where it is now: answers go there */ - ++s->counts.received; - Stream *st = NULL; - bool eliciting = false; - for (size_t at = 0; at < size;) { - unsigned type = plain[at++]; - if (type == F_STREAM) { - char source[PURE_LINK_NAME + 1], input[PURE_LINK_NAME + 1]; - size_t names = get_names(plain + at, size - at, source, input); - if (!names || at + names + 4 > size) break; - st = stream_of(s, a, source, input, get32(plain + at + names)); - if (st) { - st->run = r; - st->run_id = r->run; - } - at += names + 4; - eliciting = true; - } else if (type == F_RECORD) { - if (at + RECORD_HEAD - 1 > size) break; - uint32_t length = get16(plain + at + 16); - if (at + RECORD_HEAD - 1 + length > size) break; - if (st) { - hold(st, get64(plain + at), get32(plain + at + 8), get32(plain + at + 12), plain + at + 18, length); - mark_ready(s, st); - } - at += RECORD_HEAD - 1 + length; - eliciting = true; - } else if (type == F_RECORDS) { - if (at + RECORDS_HEAD - 1 > size) break; - uint64_t first = get64(plain + at), length = 0; - unsigned count = plain[at + 8]; - at += RECORDS_HEAD - 1; - unsigned i = 0; - for (size_t used; i < count && (used = get_varint(plain + at, size - at, &length)) && - length <= size - at - used; ++i) { - if (st) hold(st, first + i, (uint32_t)length, 0, plain + at + used, (uint32_t)length); - at += used + length; - } - if (st) mark_ready(s, st); - eliciting = true; - if (i < count) break; /* cut short: the rest is not read */ - } else if (type == F_PING) { - eliciting = true; - } else - break; /* padding, or what this side does not know: the rest */ - } - if (eliciting) { - r->ack_due = true; - mark_due(s, r); - } - return true; -} - -static void take_packet(PureLinkListener *s, unsigned char *packet, size_t n, const struct sockaddr_storage *from, - socklen_t from_size, uint64_t now) { - if (n < HEADER + PURE_AEAD_TAG || n > DATAGRAM || packet[0] != VERSION || !s->index_size) return; - uint32_t id = get32(packet + 1); - size_t mask = s->index_size - 1; - uint64_t run = get64(packet + 5); - /* The peers with the packet's key id: first one whose run is known (no - * key to make), then any. */ - for (size_t slot = id & mask; s->index[slot]; slot = (slot + 1) & mask) { - Peer *peer = &s->peers[s->index[slot] - 1]; - if (peer->id == id && peer->active && run_found(peer->active, run) && - take_from(s, peer, packet, n, from, from_size, now)) - return; - } - for (size_t slot = id & mask; s->index[slot]; slot = (slot + 1) & mask) { - Peer *peer = &s->peers[s->index[slot] - 1]; - if (peer->id == id && !(peer->active && run_found(peer->active, run)) && - take_from(s, peer, packet, n, from, from_size, now)) - return; - } -} - -/* Acknowledgments and how far each stream is stored, a packet to each run - * that has any. How far a stream is stored goes with every acknowledgment - * to its run, so a lost word is said again at the next packet. */ -static void answer(PureLinkListener *s, uint64_t now) { - unsigned char body[BODY], packet[DATAGRAM]; - for (size_t k = 0; k < s->due_count; ++k) { - Run *r = s->due[k]; - Active *a = r->active; - r->due = false; - if (!r->used) continue; - size_t at = 0; - if (r->ack_due && r->got_count) { - body[at++] = F_ACK; - put64(body + at, r->got[0].last); - put32(body + at + 8, (uint32_t)least((now - r->largest_at) / 1000, UINT32_MAX)); - body[at + 12] = (unsigned char)r->got_count; - at += 13; - for (unsigned i = 0; i < r->got_count; ++i, at += 16) { - put64(body + at, r->got[i].first); - put64(body + at + 8, r->got[i].last); - } - } - for (unsigned j = 0; j < a->stream_count; ++j) { - Stream *st = a->streams[j]; - if (!(st->stored_due || r->ack_due) || st->run != r || st->run_id != r->run || !st->known) continue; - if (at + strlen(st->source) + strlen(st->input_name) + 16 > BODY) break; /* type, names, 13 */ - body[at++] = F_STORED; - at += put_names(body + at, st->source, st->input_name); - put64(body + at, st->heard); - body[at + 8] = st->gap ? STORED_GAP : 0; - size_t room = st->held_bytes + (st->line_size - st->line_head) < HELD_MOST ? CREDIT : 1; - put32(body + at + 9, (uint32_t)room); - at += 13; - st->stored_due = st->gap = false; - } - r->ack_due = false; - if (!at) continue; - const unsigned char *give = r->out_key; - if (s->seal_keyed && !memcmp(s->seal_key, r->out_key, PURE_AEAD_KEY)) give = NULL; /* the key it has */ - else memcpy(s->seal_key, r->out_key, PURE_AEAD_KEY), s->seal_keyed = true; - size_t total = seal_packet(s->aead, give, r->out_base, s->own_id, s->run, r->next_pn++, body, at, packet); - if (total && sendto(s->fd, packet, total, 0, (const struct sockaddr *)&r->address, r->address_size) >= 0) - ++s->counts.packets; - } - s->due_count = 0; -} - -static void take_done(PureLinkListener *s) { - pthread_mutex_lock(&s->lock); - Done *done = s->done; - size_t count = s->done_count; - s->done = NULL; - s->done_count = s->done_capacity = 0; - pthread_mutex_unlock(&s->lock); - for (size_t k = 0; k < count; ++k) { - Stream *st = done[k].stream; - st->out = false; - mark_ready(s, st); - if (!done[k].stored) continue; /* not saved: handed over again */ - if (!st->known || done[k].heard > st->heard) st->heard = done[k].heard; - if (!st->known) st->line_first = st->heard + 1; /* the line begins */ - line_drop(st, st->heard); - while (st->held_count && held_at(st, 0)->seq <= st->heard) drop_first(st); - st->known = true; - st->gap = done[k].gap; - st->stored_due = true; - if (st->run && st->run->used && st->run_id == st->run->run) mark_due(s, st->run); - s->counts.stored = st->heard; - } - free(done); -} - -/* The streams ready, each handed over; one the host could not take now is - * tried again at the next turn. */ -static void hand_over_ready(PureLinkListener *s) { - size_t count = s->ready_count; - for (size_t k = 0; k < count; ++k) { - Stream *st = s->ready[k]; - st->ready = false; - if (!hand_over(s, st)) mark_ready(s, st); - } - memmove(s->ready, s->ready + count, (s->ready_count - count) * sizeof(*s->ready)); - s->ready_count -= count; +static bool send_answer(void *context, const unsigned char *packet, size_t size, const unsigned char *to, + size_t to_size) { + const PureLinkListener *s = context; + return sendto(s->fd, packet, size, 0, (const struct sockaddr *)(const void *)to, (socklen_t)to_size) >= 0; } static void *run_listener(void *context) { @@ -1722,8 +355,9 @@ static void *run_listener(void *context) { Gathering *gathering = malloc(sizeof(*gathering)); if (!gathering) return NULL; for (;;) { + uint64_t now = now_ns(), due = pure_link_receiver_due(s->receiver); + int timeout = due ? (int)((due > now ? due - now : 0) / MS) + 1 : -1; struct pollfd waits[2] = {{s->fd, POLLIN, 0}, {s->wake, POLLIN, 0}}; - int timeout = s->active_count || s->ready_count ? (int)(SWEEP_EVERY / MS) : -1; if (poll(waits, 2, timeout) < 0 && errno != EINTR) break; pthread_mutex_lock(&s->lock); bool stopping = s->stopping; @@ -1733,21 +367,25 @@ static void *run_listener(void *context) { uint64_t signals; if (read(s->wake, &signals, sizeof(signals)) < 0) { /* cleared already */ } } - uint64_t now = now_ns(); + now = now_ns(); for (unsigned rounds = 0, n = GATHER; rounds < 8 && n == GATHER; ++rounds) { n = (unsigned)gather(s->fd, gathering); for (unsigned k = 0; k < n; ++k) { struct mmsghdr *m = &gathering->messages[k]; size_t length = m->msg_len, step = segment_of(&m->msg_hdr, length); for (size_t at = 0; at < length; at += step) - take_packet(s, gathering->bytes[k] + at, least(step, length - at), &gathering->from[k], - m->msg_hdr.msg_namelen, now); + pure_link_receiver_take(s->receiver, gathering->bytes[k] + at, step < length - at ? step : length - at, + &gathering->from[k], m->msg_hdr.msg_namelen, now); } } - take_done(s); - hand_over_ready(s); - answer(s, now); - sweep(s, now); + pthread_mutex_lock(&s->lock); + PureLinkDone *done = s->done; + size_t count = s->done_count; + s->done = NULL; + s->done_count = s->done_capacity = 0; + pthread_mutex_unlock(&s->lock); + pure_link_receiver_turn(s->receiver, done, count, now, send_answer, s); + free(done); } free(gathering); return NULL; @@ -1773,11 +411,8 @@ bool pure_link_serve(PureLinkListener *s, char *error, size_t size) { setsockopt(s->fd, SOL_SOCKET, SO_SNDBUF, &buffer, sizeof(buffer)); int one = 1; /* without GRO, a buffer is a datagram */ setsockopt(s->fd, SOL_UDP, UDP_GRO, &one, sizeof(one)); - s->run = random64(); - s->swept = s->trimmed = now_ns(); s->wake = eventfd(0, EFD_NONBLOCK | EFD_CLOEXEC); - unsigned char key[PURE_AEAD_KEY] = {0}; /* each packet brings its run's */ - if (s->wake < 0 || !(s->aead = pure_aead(key)) || pthread_create(&s->thread, NULL, run_listener, s)) { + if (s->wake < 0 || pthread_create(&s->thread, NULL, run_listener, s)) { snprintf(error, size, "--link-port %u: cannot start", s->port); return false; } @@ -1790,12 +425,12 @@ void pure_link_done(PureLinkListener *s, void *token, uint64_t heard, bool store pthread_mutex_lock(&s->lock); if (s->done_count == s->done_capacity) { size_t capacity = s->done_capacity ? 2 * s->done_capacity : 16; - Done *grown = realloc(s->done, capacity * sizeof(*grown)); + PureLinkDone *grown = realloc(s->done, capacity * sizeof(*grown)); if (!grown) abort(); /* a delivery's word cannot be lost */ s->done = grown; s->done_capacity = capacity; } - s->done[s->done_count++] = (Done){token, heard, stored, gap}; + s->done[s->done_count++] = (PureLinkDone){token, heard, stored, gap}; pthread_mutex_unlock(&s->lock); wake(s->wake); } @@ -1809,21 +444,8 @@ void pure_link_listener_close(PureLinkListener *s, PureLinkCounts *counts) { wake(s->wake); pthread_join(s->thread, NULL); } - if (counts) { - *counts = s->counts; - counts->active = s->active_count; - } - while (s->active_count) let_go(s, s->actives[0]); - pure_aead_free(s->aead); - if (s->memo) munmap(s->memo, s->memo_size); - free(s->actives); - free(s->peers); - free(s->index); - free(s->allowed); - free(s->names); - free(s->ready); - free(s->due); - free(s->batch); + if (counts) pure_link_receiver_counts(s->receiver, counts); + pure_link_receiver_free(s->receiver); free(s->done); if (s->fd >= 0) close(s->fd); if (s->wake >= 0) close(s->wake); diff --git a/link/link.h b/link/link.h index 1ee9473..5237262 100644 --- a/link/link.h +++ b/link/link.h @@ -1,53 +1,11 @@ #ifndef PURE_LINK_H #define PURE_LINK_H -#include "seal.h" +#include "link-protocol.h" -/* Links without connections (docs/design/011-pure-transport.md): one node's - * saved sends to another node's input over UDP, sealed with a key the two - * nodes' keys share, with no handshake: the first packet carries records. - * A link's lasting state is the apps' own: the sender's outbox (what to send - * and send again) and the receiver's heard numbers (what it applied, saved - * with the events). What either side keeps in memory is a cache, so either - * may restart and the other goes on with the next packet. - * - * A packet is a header in the clear, then frames sealed with AES-128-GCM - * (the header as associated data): - * u8 0x70 (version 0), u32le the key id of the node that sent it (FNV-1a - * of its public key; two keys may share an id, and a receiver tries the - * keys it allows with it), u64le that node's run (random, new each time it - * starts, so no nonce repeats under a key across restarts), u64le the - * packet's number in its run. - * Keys come from the secret the two nodes' keys share (X25519) by - * HKDF-SHA256, with both public keys and the sending node's run: a sender's - * packets use its own run; a receiver's answers add the run they answer, - * so an answer opens only in that run. The nonce is the key's base with the - * packet's number in its last eight bytes (XOR, big-endian, as QUIC does). - * - * Frames (numbers little-endian): - * 0x01 STREAM source (u8 length, bytes), input (u8 length, bytes), u32 the - * shape of the messages (the sender's output and the - * receiver's input must have the same): the stream the - * records after it are of - * 0x02 RECORD u64 number, u32 size, u32 offset, u16 length, bytes: a piece - * of a saved send (pieces are 1000 bytes, the last shorter) - * 0x03 ACK u64 largest number, u32 delay (µs), u8 ranges, each as u64 - * first and u64 last, newest first - * 0x04 STORED source, input (as STREAM), u64 the last number heard - * (applied and saved), u8 flags (1: a record after a gap was - * not taken: send what follows it again), u32 credit (records - * it takes past that) - * 0x05 PING to be acknowledged - * 0x06 RECORDS u64 the first one's number, u8 count, then whole records one - * after another (numbers following on), each its size (a - * varint: LEB128) and its bytes: a send that fits the packet - * whole, with 1 to 3 bytes of framing, not the 19 of RECORD - * 0x00 PADDING to the end of the packet: a full packet sent in a burst is - * a whole datagram - * Losses are detected as RFC 9002 says; congestion control is its NewReno - * (RFC 9002 §7). */ - -/* A source's or an input's name is 1 to PURE_LINK_NAME bytes. */ -enum { PURE_LINK_NAME = 64 }; +/* Links over UDP: the two sides of link/link-protocol.h (what goes on the + * wire, and what each side keeps) with sockets and threads. A sender sends + * on a thread of its own, packets in bursts with UDP GSO; a listener takes + * a receiver's packets on a port, with UDP GRO. */ /* The sending side: one output's saved sends to url, pure://host:port/INPUT, * as source (a Pure receiver's name for it), their messages of shape (the @@ -66,59 +24,17 @@ uint64_t pure_link_queued(PureLink *); uint64_t pure_link_delivered(PureLink *); /* Send on a thread of the link's own, calling notify when more was saved. */ bool pure_link_start(PureLink *, void (*notify)(void *), void *context, char *error, size_t size); +void pure_link_counts(PureLink *, PureLinkCounts *); void pure_link_close(PureLink *); -/* The receiving side: packets on a UDP port, from the senders allowed (a - * source name and the key it seals with, a few bytes each however many - * there are: see pure_link_idle). A stream (a source to an input) goes to - * the host's input its name and shape find (its number, or -1: the stream - * is refused). Whole records of a stream, in order, go to deliver (both on - * the listener's thread) as one delivery; the host applies them from the - * one after the source's last heard number, stops at a gap, saves, and then - * calls pure_link_done (any thread) with the last number heard. Records - * wait here until then, and only the saved ones are acknowledged as - * stored. */ -typedef struct { - uint32_t input; - const char *source; - uint64_t first; - unsigned count; - const unsigned char *records; /* count of: u32le size, bytes */ - size_t size; - void *token; /* for pure_link_done */ -} PureLinkDelivery; -typedef struct { - int (*input)(void *context, const char *name, uint32_t shape); - bool (*deliver)(void *context, const PureLinkDelivery *); - void *context; -} PureLinkHost; +/* The receiving side: a receiver's packets on a UDP port, taken on a thread + * of the listener's own, where the host's input and deliver are called. The + * listener takes the receiver: closing it frees the receiver, and gives its + * counts. pure_link_done (any thread) says how a delivery's save went. */ typedef struct PureLinkListener PureLinkListener; -PureLinkListener *pure_link_listen(unsigned port, const PureSealKey *own, const unsigned char own_public[PURE_SEAL_KEY], - const PureLinkHost *host, char *error, size_t size); -bool pure_link_allow(PureLinkListener *, const char *source, const unsigned char key[PURE_SEAL_KEY], char *error, - size_t size); -/* A sender quiet for ms (30 s at first) with nothing of it waiting lets its - * keys, addresses and streams go: an idle link costs the listener only its - * allowing, and the sender's next packet makes them again (one X25519). */ -void pure_link_idle(PureLinkListener *, uint64_t ms); -/* Remember the secret each sender's key shares with the node's in a file - * (path, 0600; begun again for another node's key): a place a sender in - * the order allowed, mapped, so a sender at rest takes no memory for it. - * A node that starts again then makes no X25519 for a sender it heard from - * before. After the senders are allowed, before serving. */ -bool pure_link_remember(PureLinkListener *, const char *path, char *error, size_t size); +PureLinkListener *pure_link_listen(PureLinkReceiver *, unsigned port, char *error, size_t size); bool pure_link_serve(PureLinkListener *, char *error, size_t size); /* stored: the delivery's save succeeded; gap: it stopped at a gap. */ void pure_link_done(PureLinkListener *, void *token, uint64_t heard, bool stored, bool gap); - -/* Counts for a host's report: packets and records sent, records sent again, - * packets lost and acknowledged, packets received, deliveries handed over, - * the last number stored; a listener's senders woken (their active state - * made), of them with a secret remembered (no X25519), let go when idle, - * and active when it closed. A listener's are given when it closes. */ -typedef struct { - uint64_t packets, records, resent, lost, acknowledged, received, deliveries, stored, woken, remembered, swept, active; -} PureLinkCounts; -void pure_link_counts(PureLink *, PureLinkCounts *); void pure_link_listener_close(PureLinkListener *, PureLinkCounts *counts); #endif diff --git a/link/test-link.c b/link/test-link.c new file mode 100644 index 0000000..8be32a9 --- /dev/null +++ b/link/test-link.c @@ -0,0 +1,300 @@ +#define _GNU_SOURCE +/* A link's two sides (link/link-protocol.h) through a network of the test's + * own: time is the test's, and packets are lost, held back (so they come out + * of order) and doubled at random, from a seed. The receiving host applies a + * record once, from the one after its last heard, and says so a turn later, + * now and then failing a save; the receiver and the sender start again + * midway, each from what its app kept. Every record must be applied once, + * in order, as it was sent. */ +#include "link-protocol.h" +#include <stdio.h> +#include <stdlib.h> +#include <string.h> +#include <unistd.h> + +#define MS UINT64_C(1000000) +enum { PACKETS = 1 << 16 }; + +static uint64_t seed; +static uint64_t next_random(void) { + seed ^= seed << 13, seed ^= seed >> 7, seed ^= seed << 17; + return seed; +} +static bool chance(unsigned percent) { return next_random() % 100 < percent; } + +static int failures; +#define CHECK(condition, ...) \ + do { \ + if (!(condition)) { \ + fprintf(stderr, "test-link: " __VA_ARGS__); \ + fputc('\n', stderr); \ + ++failures; \ + } \ + } while (0) + +/* A record's size and bytes, from its number: small ones, ones of a packet or + * so, and some of many pieces. */ +static size_t record_size(uint64_t seq) { + uint64_t h = seq * 0x9e3779b97f4a7c15u; + unsigned kind = (unsigned)(h >> 60) % 10; + return kind < 3 ? (size_t)(h >> 8) % 61 : kind < 8 ? 61 + (size_t)(h >> 8) % 1040 : 1101 + (size_t)(h >> 8) % 19000; +} +static unsigned char record_byte(uint64_t seq, size_t i) { return (unsigned char)(seq * 31 + i * 7 + (i >> 8)); } + +typedef struct { + uint64_t at; + bool to_receiver; + size_t size; + unsigned char bytes[PURE_LINK_DATAGRAM]; +} Packet; + +typedef struct { + /* The network: packets on their way, each arriving at its time. */ + Packet *packets; + size_t count; + unsigned loss, doubled, spread_ms; /* percent, percent, the most a packet is held back */ + uint64_t now, sent; + /* The receiving app: its last heard (kept across the receiver's + * restarts), what it said of deliveries (given at the next turn). */ + uint64_t heard, applied, deliveries; + PureLinkDone done[64], told[64]; + size_t done_count; + unsigned fail_every; /* a delivery's save fails every so many (0: never) */ +} World; + +static void put_packet(World *w, bool to_receiver, const unsigned char *bytes, size_t size) { + ++w->sent; + for (int copies = chance(w->doubled) ? 2 : 1; copies--;) { + if (chance(w->loss) || w->count == PACKETS) continue; + Packet *p = &w->packets[w->count++]; + p->at = w->now + 5 * MS + (w->spread_ms ? next_random() % (w->spread_ms * MS) : 0); + p->to_receiver = to_receiver; + p->size = size; + memcpy(p->bytes, bytes, size); + } +} + +static bool answer(void *context, const unsigned char *packet, size_t size, const unsigned char *to, size_t to_size) { + CHECK(to_size == 6 && !memcmp(to, "sender", 6), "an answer to an address no packet came from"); + put_packet(context, false, packet, size); + return true; +} + +static int input_of(void *context, const char *name, uint32_t shape) { + (void)context; + return !strcmp(name, "order") && shape == 0x5eed ? 3 : -1; +} + +/* Applied from the one after the last heard, in order; a gap stops it. */ +static bool deliver(void *context, const PureLinkDelivery *d) { + World *w = context; + CHECK(d->input == 3 && !strcmp(d->source, "test.ordered"), "a delivery to input %u from %s", d->input, d->source); + if (w->done_count == sizeof(w->done) / sizeof(w->done[0])) return false; /* busy: tried again */ + ++w->deliveries; + bool fail = w->fail_every && w->deliveries % w->fail_every == 0, gap = false; + uint64_t heard = w->heard; + const unsigned char *at = d->records; + for (unsigned k = 0; k < d->count; ++k) { + uint32_t size = (uint32_t)at[0] | (uint32_t)at[1] << 8 | (uint32_t)at[2] << 16 | (uint32_t)at[3] << 24; + uint64_t seq = d->first + k; + if (seq == heard + 1) { + bool same = size == record_size(seq); + for (size_t i = 0; same && i < size; ++i) same = at[4 + i] == record_byte(seq, i); + CHECK(same, "record %llu is not as it was sent", (unsigned long long)seq); + heard = seq; + } else if (seq > heard + 1) { + gap = true; + break; + } + at += 4 + size; + } + if (!fail) { + w->applied += heard - w->heard; + w->heard = heard; + } + w->done[w->done_count++] = (PureLinkDone){d->token, w->heard, !fail, gap}; + return true; +} + +static PureSealKey *key(const char *dir, const char *name, unsigned char public_key[PURE_SEAL_KEY]) { + char path[512], hex[2 * PURE_SEAL_KEY + 1], error[256]; + snprintf(path, sizeof(path), "%s/%s", dir, name); + PureSealKey *k = pure_seal_key(path, hex, error, sizeof(error)); + if (!k || !pure_seal_peer(hex, public_key)) { + fprintf(stderr, "test-link: %s\n", error); + exit(1); + } + return k; +} + +typedef struct { + const char *name; + uint64_t records, seed; + unsigned loss, doubled, spread_ms, fail_every; + uint64_t pause_after, receiver_restart_after, sender_restart_after, idle_ms; +} Case; + +typedef struct { PureLinkCounts sent, heard; uint64_t packets, seconds_ms; } Outcome; + +static PureLinkReceiver *receiver_of(World *w, const PureSealKey *b, const unsigned char *public_b, + const unsigned char *public_a, const char *memo, uint64_t idle_ms) { + char error[256]; + PureLinkReceiver *r = pure_link_receiver(b, public_b, &(PureLinkHost){input_of, deliver, w}, error, sizeof(error)); + if (!r || !pure_link_allow(r, "test.ordered", public_a, error, sizeof(error)) || + !pure_link_remember(r, memo, error, sizeof(error))) { + fprintf(stderr, "test-link: %s\n", error); + exit(1); + } + if (idle_ms) pure_link_idle(r, idle_ms); + return r; +} + +/* The sender as its app makes it: the sends its outbox keeps, from the one + * after the last stored, up to limit. */ +static PureLinkSender *sender_of(const PureSealKey *a, const unsigned char *public_a, const unsigned char *public_b, + uint64_t from, uint64_t limit) { + PureLinkSender *s = pure_link_sender("test.ordered", "order", 0x5eed, a, public_a); + if (!s || !pure_link_sender_peer(s, public_b)) { + fprintf(stderr, "test-link: cannot make a sender\n"); + exit(1); + } + static unsigned char bytes[20200]; + for (uint64_t seq = from; seq <= limit; ++seq) { + size_t size = record_size(seq); + for (size_t i = 0; i < size; ++i) bytes[i] = record_byte(seq, i); + if (!pure_link_sender_queue(s, seq, bytes, size)) { + fprintf(stderr, "test-link: cannot queue %llu\n", (unsigned long long)seq); + exit(1); + } + } + return s; +} + +static Outcome run(const Case *c, const char *dir) { + unsigned char public_a[PURE_SEAL_KEY], public_b[PURE_SEAL_KEY]; + PureSealKey *a = key(dir, "a.key", public_a), *b = key(dir, "b.key", public_b); + char memo[512]; + snprintf(memo, sizeof(memo), "%s/b.key.links", dir); + unlink(memo); + World w = {.packets = malloc(PACKETS * sizeof(Packet)), .loss = c->loss, .doubled = c->doubled, + .spread_ms = c->spread_ms, .fail_every = c->fail_every}; + if (!w.packets) exit(1); + seed = c->seed; + uint64_t queued = c->pause_after ? c->pause_after : c->records, resumed = 0; + PureLinkSender *s = sender_of(a, public_a, public_b, 1, queued); + PureLinkReceiver *r = receiver_of(&w, b, public_b, public_a, memo, c->idle_ms); + PureLinkCounts sent_before = {0}, heard_before = {0}; + bool receiver_restarted = !c->receiver_restart_after, sender_restarted = !c->sender_restart_after; + unsigned char out[PURE_LINK_DATAGRAM]; + while (w.heard < c->records || pure_link_sender_stored(s) < c->records) { + CHECK(w.now < 600000 * MS, "%s: not done after ten minutes (heard %llu of %llu)", c->name, + (unsigned long long)w.heard, (unsigned long long)c->records); + if (w.now >= 600000 * MS) break; + uint64_t number; + for (size_t size; (size = pure_link_sender_packet(s, out, false, w.now, &number));) + put_packet(&w, true, out, size); + /* Then the next thing to happen. */ + uint64_t next = UINT64_MAX, due = pure_link_sender_due(s), turn = pure_link_receiver_due(r); + for (size_t k = 0; k < w.count; ++k) + if (w.packets[k].at < next) next = w.packets[k].at; + if (due && due < next) next = due; + if (turn && turn < next) next = turn; + if (w.done_count && w.now + MS < next) next = w.now + MS; /* the host's word, a turn later */ + if (queued < c->records && pure_link_sender_stored(s) >= queued && !resumed) resumed = w.now + 2000 * MS; + if (resumed && queued < c->records && resumed < next) next = resumed; + if (next == UINT64_MAX) { + CHECK(false, "%s: nothing more happens (heard %llu of %llu)", c->name, (unsigned long long)w.heard, + (unsigned long long)c->records); + break; + } + w.now = next > w.now ? next : w.now; + if (resumed && resumed <= w.now && queued < c->records) { /* after a pause: the rest */ + static unsigned char bytes[20200]; + for (uint64_t seq = queued + 1; seq <= c->records; ++seq) { + size_t size = record_size(seq); + for (size_t i = 0; i < size; ++i) bytes[i] = record_byte(seq, i); + CHECK(pure_link_sender_queue(s, seq, bytes, size), "%s: cannot queue %llu", c->name, (unsigned long long)seq); + } + queued = c->records; + } + for (size_t k = 0; k < w.count;) { + Packet *p = &w.packets[k]; + if (p->at > w.now) { + ++k; + continue; + } + if (p->to_receiver) pure_link_receiver_take(r, p->bytes, p->size, "sender", 6, w.now); + else pure_link_sender_take(s, p->bytes, p->size, w.now); + *p = w.packets[--w.count]; + } + pure_link_sender_tick(s, w.now); + size_t told = w.done_count; + memcpy(w.told, w.done, told * sizeof(*w.done)); + w.done_count = 0; + pure_link_receiver_turn(r, w.told, told, w.now, answer, &w); + if (!receiver_restarted && w.heard >= c->receiver_restart_after) { /* kill -9: what it held is gone */ + receiver_restarted = true; + pure_link_receiver_counts(r, &heard_before); + pure_link_receiver_free(r); + w.done_count = 0; + r = receiver_of(&w, b, public_b, public_a, memo, c->idle_ms); + } + if (!sender_restarted && pure_link_sender_stored(s) >= c->sender_restart_after) { + sender_restarted = true; + pure_link_sender_counts(s, &sent_before); + uint64_t stored = pure_link_sender_stored(s); + pure_link_sender_free(s); + s = sender_of(a, public_a, public_b, stored + 1, queued); + } + } + Outcome o = {.packets = w.sent, .seconds_ms = w.now / MS}; + pure_link_sender_counts(s, &o.sent); + pure_link_receiver_counts(r, &o.heard); + o.sent.packets += sent_before.packets, o.sent.resent += sent_before.resent; + o.heard.woken += heard_before.woken, o.heard.swept += heard_before.swept; + CHECK(w.heard == c->records && w.applied == c->records, "%s: applied %llu, heard %llu of %llu", c->name, + (unsigned long long)w.applied, (unsigned long long)w.heard, (unsigned long long)c->records); + CHECK(pure_link_sender_stored(s) == c->records, "%s: the sender knows %llu stored of %llu", c->name, + (unsigned long long)pure_link_sender_stored(s), (unsigned long long)c->records); + printf("{\"case\":\"%s\",\"records\":%llu,\"virtual_ms\":%llu,\"packets\":%llu,\"resent\":%llu,\"lost\":%llu," + "\"deliveries\":%llu,\"woken\":%llu,\"remembered\":%llu,\"swept\":%llu}\n", + c->name, (unsigned long long)c->records, (unsigned long long)o.seconds_ms, (unsigned long long)o.sent.packets, + (unsigned long long)o.sent.resent, (unsigned long long)o.sent.lost, (unsigned long long)w.deliveries, + (unsigned long long)o.heard.woken, (unsigned long long)o.heard.remembered, (unsigned long long)o.heard.swept); + pure_link_sender_free(s); + pure_link_receiver_free(r); + pure_seal_key_free(a); + pure_seal_key_free(b); + free(w.packets); + return o; +} + +int main(void) { + char dir[] = "/tmp/pure-test-link-XXXXXX", error[256]; + if (!mkdtemp(dir) || !pure_seal_ready(error, sizeof(error))) { + fprintf(stderr, "test-link: %s\n", error); + return 1; + } + /* A clean network: nothing is sent twice. */ + Case clean = {"clean", 3000, 1, 0, 0, 0, 0, 0, 0, 0, 0}; + Outcome o = run(&clean, dir); + CHECK(o.sent.resent == 0 && o.sent.lost == 0, "clean: %llu sent again, %llu lost", (unsigned long long)o.sent.resent, + (unsigned long long)o.sent.lost); + /* Loss, packets out of order and doubled, failed saves, a pause long + * enough for the receiver to let the sender's state go, and both sides + * starting again: the receiver remembers its secret, so it wakes the + * sender without an X25519. */ + for (uint64_t s = 1; s <= 4; ++s) { + Case rough = {"rough", 3000, s * 7919, 10, 3, 20, 37, 1500, 800, 2200, 300}; + o = run(&rough, dir); + CHECK(o.sent.resent > 0 && o.heard.swept >= 1 && o.heard.woken >= 1 && o.heard.remembered >= 1, + "rough: sent again %llu, swept %llu, woken %llu, remembered %llu", (unsigned long long)o.sent.resent, + (unsigned long long)o.heard.swept, (unsigned long long)o.heard.woken, + (unsigned long long)o.heard.remembered); + } + char command[600]; + snprintf(command, sizeof(command), "rm -rf %s", dir); + if (system(command)) { /* left behind in /tmp */ } + if (failures) fprintf(stderr, "test-link: %d failed\n", failures); + return failures ? 1 : 0; +} diff --git a/research/experiments/019-idle-links/transport.c b/research/experiments/019-idle-links/transport.c index c4c9de1..5827311 100644 --- a/research/experiments/019-idle-links/transport.c +++ b/research/experiments/019-idle-links/transport.c @@ -138,9 +138,9 @@ int main(int argc, char **argv) { return 1; unsigned port = ntohs(any.sin_port); close(probe); - if (!(listener = pure_link_listen(port, b, public_b, &(PureLinkHost){input_of, deliver, NULL}, error, sizeof(error))) || - !pure_link_allow(listener, "bench.ordered", public_a, error, sizeof(error)) || - !pure_link_serve(listener, error, sizeof(error))) + PureLinkReceiver *r = pure_link_receiver(b, public_b, &(PureLinkHost){input_of, deliver, NULL}, error, sizeof(error)); + if (!r || !pure_link_allow(r, "bench.ordered", public_a, error, sizeof(error)) || + !(listener = pure_link_listen(r, port, error, sizeof(error))) || !pure_link_serve(listener, error, sizeof(error))) return fprintf(stderr, "transport: %s\n", error), 1; char url[128]; snprintf(url, sizeof(url), "pure://127.0.0.1:%u/order", port); diff --git a/serve/serve.c b/serve/serve.c index 5688d79..f466ee6 100644 --- a/serve/serve.c +++ b/serve/serve.c @@ -2493,7 +2493,7 @@ static bool name_senders(const char *const *named, unsigned count) { /* --senders FILE: senders a --link-port takes besides those named, a line * each (NAME KEY, the key as 64 hex digits), as many as memory holds. */ -static bool allow_senders(PureLinkListener *listener, const char *path, char *error, size_t size) { +static bool allow_senders(PureLinkReceiver *receiver, const char *path, char *error, size_t size) { if (!path) return true; FILE *file = fopen(path, "r"); if (!file) { @@ -2516,7 +2516,7 @@ static bool allow_senders(PureLinkListener *listener, const char *path, char *er *space = 0; for (const char *c = line; *c && ok; ++c) ok = *c > ' ' && *c <= '~'; if (!ok) snprintf(error, size, "--senders %s: line %zu names no source", path, count + 1); - else ok = pure_link_allow(listener, line, key, error, size); + else ok = pure_link_allow(receiver, line, key, error, size); } free(line); fclose(file); @@ -2675,15 +2675,16 @@ int main(int argc, char **argv) { fputs("--link-port needs an app with states and the node's --key\n", stderr); return 2; } - link_listener = pure_link_listen(link_port, seal_key, own_public, &(PureLinkHost){link_input, link_deliver, &owner}, - error, sizeof(error)); - bool allowed = link_listener != NULL && allow_senders(link_listener, senders_path, error, sizeof(error)); + PureLinkReceiver *receiver = + pure_link_receiver(seal_key, own_public, &(PureLinkHost){link_input, link_deliver, &owner}, error, sizeof(error)); + bool allowed = receiver != NULL && allow_senders(receiver, senders_path, error, sizeof(error)); for (unsigned i = 0; allowed && i < sender_count; ++i) - allowed = pure_link_allow(link_listener, senders[i].name, senders[i].key, error, sizeof(error)); - if (allowed && link_idle_ms) pure_link_idle(link_listener, link_idle_ms); + allowed = pure_link_allow(receiver, senders[i].name, senders[i].key, error, sizeof(error)); + if (allowed && link_idle_ms) pure_link_idle(receiver, link_idle_ms); char memo[4096]; snprintf(memo, sizeof(memo), "%s.links", key_path); /* beside the key the secrets come from */ - allowed = allowed && pure_link_remember(link_listener, memo, error, sizeof(error)); + allowed = allowed && pure_link_remember(receiver, memo, error, sizeof(error)) && + (link_listener = pure_link_listen(receiver, link_port, error, sizeof(error))); if (!allowed || !pure_link_serve(link_listener, error, sizeof(error))) { fprintf(stderr, "--link-port: %s\n", error); return 2; |