diff options
Diffstat (limited to 'link')
| -rw-r--r-- | link/link-protocol.c | 136 | ||||
| -rw-r--r-- | link/link-protocol.h | 26 | ||||
| -rw-r--r-- | link/link.c | 10 | ||||
| -rw-r--r-- | link/link.h | 4 | ||||
| -rw-r--r-- | link/test-link.c | 84 |
5 files changed, 228 insertions, 32 deletions
diff --git a/link/link-protocol.c b/link/link-protocol.c index 4fa5aab..169705f 100644 --- a/link/link-protocol.c +++ b/link/link-protocol.c @@ -19,7 +19,7 @@ enum { DATAGRAM_MOST = PURE_LINK_DATAGRAM_MOST, BODY_MOST = DATAGRAM_MOST - HEAD * at least this much of it fits. */ enum { PART_LEAST = 64 }; enum { F_PADDING = 0, F_STREAM = 1, F_RECORD = 2, F_ACK = 3, F_STORED = 4, F_PING = 5, F_RECORDS = 6, F_CHALLENGE = 7, - F_RESPONSE = 8, F_OFFER = 9, F_ACCEPT = 10, F_LOST = 11, F_REFUSED = 12 }; + F_RESPONSE = 8, F_OFFER = 9, F_ACCEPT = 10, F_LOST = 11, F_REFUSED = 12, F_FIELDS = 13 }; /* STORED's flags: what came after a gap was not kept; the app is full (no * room for more now), and what came is kept for when it has room. */ enum { STORED_GAP = 1, STORED_FULL = 2 }; @@ -307,6 +307,11 @@ struct PureLinkSender { * answer with no REFUSED. */ unsigned refusal, tried; uint64_t refused_until, refused_wait, tried_pn; + /* The fields its records are laid out with, named (FIELDS) in each + * packet from a refusal for another shape until the stream is taken. */ + unsigned char *fields; + size_t fields_size; + bool naming; PureLinkCounts counts; /* Epochs (forward secrecy). Keys come from the secret of the epoch it * sends in (epoch 0's is the two nodes' static secret), each made from @@ -696,6 +701,7 @@ static void on_stored(PureLinkSender *l, const unsigned char *p, size_t size, ui unsigned flags = p[names + 8]; uint32_t credit = get32(p + names + 9); l->refusal = l->tried = 0, l->refused_until = l->refused_wait = 0; /* the stream is taken, and stored */ + l->naming = false; if (heard > l->stored) { while (l->count && record_at(l, 0)->seq <= heard) { Record *r = record_at(l, 0); @@ -725,8 +731,12 @@ static void on_stored(PureLinkSender *l, const unsigned char *p, size_t size, ui static void refused(PureLinkSender *l, unsigned why, uint64_t now) { if (l->refusal && l->tried != TRIED) return; l->refusal = why ? why : PURE_LINK_NO_INPUT; - l->refused_wait = l->refused_wait ? least(2 * l->refused_wait, REFUSED_MOST) : REFUSED_FIRST; - l->refused_until = now + l->refused_wait; + /* Another shape, and its fields not named yet: named in a packet at once. */ + bool naming = (why == PURE_LINK_OTHER_SHAPE || why == PURE_LINK_FIELDS_LACKED) && l->fields; + bool at_once = naming && !l->naming; + l->naming = l->naming || naming; + if (!at_once) l->refused_wait = l->refused_wait ? least(2 * l->refused_wait, REFUSED_MOST) : REFUSED_FIRST; + l->refused_until = at_once ? 0 : now + l->refused_wait; l->tried = 0; ++l->counts.refused; for (size_t k = 0; k < l->count; ++k) { @@ -804,7 +814,8 @@ void pure_link_sender_take(PureLinkSender *l, const unsigned char *packet, size_ } else if (type != F_PING) break; /* padding, or what this side does not know: the rest */ } - if (l->refusal && l->tried == TRIED && !refusing) l->refusal = l->tried = 0, l->refused_until = 0; /* taken now */ + if (l->refusal && l->tried == TRIED && !refusing) /* taken now */ + l->refusal = l->tried = 0, l->refused_until = 0, l->naming = false; } /* A packet of what is not sent yet, in order, within the receiver's credit @@ -831,6 +842,17 @@ static size_t fill(PureLinkSender *l, unsigned char *body, Flight *f, bool pad, memcpy(body + at + 5, l->offer_public, PURE_SEAL_KEY); at += EPOCH_FRAME; } + /* The fields, for a receiver whose input takes another shape: after the + * frames one that knows no FIELDS reads, before the records. */ + size_t names = 2 + strlen(l->source) + strlen(l->input); + if (l->naming && at + 1 + names + 4 + l->fields_size <= most) { + body[at++] = F_FIELDS; + at += put_names(body + at, l->source, l->input); + put32(body + at, l->shape); + at += 4; + memcpy(body + at, l->fields, l->fields_size); + at += l->fields_size; + } if (l->renegotiating) { /* no record under the static keys */ l->responding = 0; return at; @@ -1196,6 +1218,15 @@ uint64_t pure_link_sender_stored(const PureLinkSender *l) { return l->stored; } size_t pure_link_sender_held(const PureLinkSender *l) { return l->held; } +bool pure_link_sender_fields(PureLinkSender *l, const unsigned char *fields, size_t size) { + unsigned char *kept = size ? malloc(size) : NULL; + if (size && !kept) return false; + if (size) memcpy(kept, fields, size); + free(l->fields); + l->fields = kept, l->fields_size = size; + return true; +} + void pure_link_sender_counts(const PureLinkSender *l, PureLinkCounts *out) { *out = l->counts; out->stored = l->stored; @@ -1213,6 +1244,7 @@ void pure_link_sender_free(PureLinkSender *l) { explicit_bzero(l->epoch_secret, sizeof(l->epoch_secret)); explicit_bzero(l->before_secret, sizeof(l->before_secret)); explicit_bzero(l->offer_private, sizeof(l->offer_private)); + free(l->fields); free(l); } @@ -1298,6 +1330,8 @@ typedef struct Stream { 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; + unsigned char *fields; /* the fields its records are laid out with, when its input took them by name */ + size_t fields_size; } Stream; struct Active { uint32_t peer; /* its place among the peers */ @@ -1720,6 +1754,7 @@ static void let_go(PureLinkReceiver *s, Active *a) { 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]->fields); free(a->streams[k]); } free(a->streams); @@ -1845,30 +1880,56 @@ static bool allowed(const PureLinkReceiver *s, const Peer *peer, const char *sou return false; } +/* Whose fields a stream's records are laid out with, kept (NULL, 0: its + * input's). False when memory runs out. */ +static bool keep_fields(Stream *st, const unsigned char *fields, size_t size) { + unsigned char *kept = fields && size ? malloc(size) : NULL; + if (fields && size && !kept) return false; + if (kept) memcpy(kept, fields, size); + free(st->fields); + st->fields = kept, st->fields_size = kept ? size : 0; + return true; +} + /* The stream of a peer's source to input, made the first time: NULL when it - * is refused (why: the source is not allowed from the peer, or the host has - * no input that takes such messages), or memory runs out (why 0). */ + * is refused (why: the source is not allowed from the peer, the host has no + * input that takes such messages, one of another shape unless it takes the + * fields named, which come with FIELDS), or memory runs out (why 0). A + * stream the sender now sends in another shape (it started again with + * another module) starts over with what it holds let go: the sender sends + * those again, laid out anew. */ static Stream *stream_of(PureLinkReceiver *s, Active *a, const char *source, const char *input, uint32_t shape, - unsigned *why) { + const unsigned char *fields, size_t fields_size, unsigned *why) { *why = 0; - for (unsigned k = 0; k < a->stream_count; ++k) { - Stream *st = a->streams[k]; - if (strcmp(st->input_name, input) || strcmp(st->source, source)) continue; - if (st->shape != shape) *why = PURE_LINK_OTHER_SHAPE; - return *why ? NULL : st; - } - int index = allowed(s, &s->peers[a->peer], source) ? s->host.input(s->host.context, input, shape) - : -PURE_LINK_NOT_ALLOWED; + Stream *st = NULL; + for (unsigned k = 0; k < a->stream_count && !st; ++k) + if (!strcmp(a->streams[k]->input_name, input) && !strcmp(a->streams[k]->source, source)) st = a->streams[k]; + if (st && st->shape == shape) return st; + int index = !allowed(s, &s->peers[a->peer], source) ? -PURE_LINK_NOT_ALLOWED : + s->host.input(s->host.context, input, shape, fields, fields ? fields_size : 0); if (index < 0) { - *why = index == -PURE_LINK_OTHER_SHAPE || index == -PURE_LINK_NOT_ALLOWED ? (unsigned)-index : PURE_LINK_NO_INPUT; + *why = index == -PURE_LINK_OTHER_SHAPE || index == -PURE_LINK_NOT_ALLOWED || index == -PURE_LINK_FIELDS_LACKED + ? (unsigned)-index : PURE_LINK_NO_INPUT; return NULL; } + if (st) { /* another shape now */ + if (st->out) { *why = PURE_LINK_OTHER_SHAPE; return NULL; } /* its last delivery first */ + if (!keep_fields(st, fields, fields_size)) return NULL; + while (st->held_count) drop_first(st); + st->line_head = st->line_size = 0; + st->line_count = 0; + st->tail = st->known = false; + st->input = (uint32_t)index; + st->shape = shape; + return st; + } size_t capacity = a->stream_capacity; - Stream **streams = room(a->streams, &capacity, a->stream_count, sizeof(*streams), 2), *st; + Stream **streams = room(a->streams, &capacity, a->stream_count, sizeof(*streams), 2); if (!streams) return NULL; a->streams = streams; a->stream_capacity = (unsigned)capacity; if (!(st = calloc(1, sizeof(*st)))) return NULL; + if (!keep_fields(st, fields, fields_size)) { free(st); return NULL; } st->active = a; st->input = (uint32_t)index; st->shape = shape; @@ -2023,7 +2084,8 @@ static bool hand_over(PureLinkReceiver *s, Stream *st) { size_t bytes = st->line_size - st->line_head; /* the whole line, when it is a batch or less */ if (n > BATCH) for (n = 0, bytes = 0; 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}; + PureLinkDelivery d = {st->input, st->source, st->line_first, n, st->line + st->line_head, bytes, st, + st->fields, st->fields_size}; if (!s->host.deliver(s->host.context, &d)) return false; st->out = true; st->out_count = n, st->out_bytes = bytes; @@ -2052,7 +2114,8 @@ static bool hand_over(PureLinkReceiver *s, Stream *st) { 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, first)->seq, (unsigned)n, s->batch, bytes, st}; + PureLinkDelivery d = {st->input, st->source, held_at(st, first)->seq, (unsigned)n, s->batch, bytes, st, + st->fields, st->fields_size}; if (!s->host.deliver(s->host.context, &d)) return false; st->out = true; ++s->counts.deliveries; @@ -2215,8 +2278,10 @@ static void lost(PureLinkReceiver *s, const Peer *peer, const Active *a, uint64_ * to its run: once a turn. */ static void refuse(PureLinkReceiver *s, Run *r, const unsigned char *names, size_t size, unsigned why) { for (size_t k = 0; k < s->refusal_count; ++k) - if (s->refusals[k].run == r && s->refusals[k].size == size + 2 && !memcmp(s->refusals[k].frame + 1, names, size)) + if (s->refusals[k].run == r && s->refusals[k].size == size + 2 && !memcmp(s->refusals[k].frame + 1, names, size)) { + s->refusals[k].frame[1 + size] = (unsigned char)why; /* said once, as last judged (named fields, say) */ return; + } void *more = room(s->refusals, &s->refusal_capacity, s->refusal_count, sizeof(*s->refusals), 4); if (!more) return; s->refusals = more; @@ -2318,6 +2383,10 @@ static bool take_from(PureLinkReceiver *s, Peer *peer, const unsigned char *pack Stream *st = NULL; bool eliciting = false, records = false; const unsigned char *offer = NULL; /* taken after the frames: whether records came with it */ + /* A stream of another shape: refused at the end unless this packet's + * FIELDS names fields its input takes. */ + const unsigned char *other = NULL; + size_t other_names = 0; for (size_t at = 0; at < size;) { unsigned type = plain[at++]; if (type == F_STREAM) { @@ -2325,14 +2394,36 @@ static bool take_from(PureLinkReceiver *s, Peer *peer, const unsigned char *pack size_t names = get_names(plain + at, size - at, source, input); if (!names || at + names + 4 > size) break; unsigned why; - st = stream_of(s, a, source, input, get32(plain + at + names), &why); + st = stream_of(s, a, source, input, get32(plain + at + names), NULL, 0, &why); if (st) { st->run = r; st->run_id = r->run; - } else if (why) + } else if (why == PURE_LINK_OTHER_SHAPE) + other = plain + at, other_names = names; + else if (why) refuse(s, r, plain + at, names, why); at += names + 4; eliciting = true; + } else if (type == F_FIELDS) { + char source[PURE_LINK_NAME + 1], input[PURE_LINK_NAME + 1]; + size_t names = get_names(plain + at, size - at, source, input), length = 1; + if (!names || at + names + 5 > size) break; + const unsigned char *fields = plain + at + names + 4; + for (unsigned f = 0; f < fields[0] && length != SIZE_MAX; ++f) /* each {type, name length, name} */ + length = at + names + 4 + length + 2 <= size ? length + 2 + fields[length + 1] : SIZE_MAX; + if (length == SIZE_MAX || at + names + 4 + length > size) break; + if (other && names == other_names && !memcmp(plain + at, other, names + 4)) { + unsigned why; + st = stream_of(s, a, source, input, get32(plain + at + names), fields, length, &why); + if (st) { + st->run = r; + st->run_id = r->run; + } else if (why) + refuse(s, r, other, other_names, why); + other = NULL; + } + at += names + 4 + length; + eliciting = true; } else if (type == F_RECORD) { if (at + RECORD_HEAD - 1 > size) break; uint32_t length = get16(plain + at + 16); @@ -2384,6 +2475,7 @@ static bool take_from(PureLinkReceiver *s, Peer *peer, const unsigned char *pack } else break; /* padding, or what this side does not know: the rest */ } + if (other) refuse(s, r, other, other_names, PURE_LINK_OTHER_SHAPE); /* its fields not named */ if (offer) offer_taken(s, a, r, get32(offer), offer + 4, e, !records, now); if (eliciting) { r->ack_due = true; diff --git a/link/link-protocol.h b/link/link-protocol.h index e4d3080..feff543 100644 --- a/link/link-protocol.h +++ b/link/link-protocol.h @@ -111,6 +111,12 @@ * sender sends one packet of them a wait from then (a * second, twice as long each time, 64 s at most) until one * is stored + * 0x0d FIELDS source, input, u32 shape (as STREAM), then the fields the + * stream's records are laid out with (core/module.h's + * pure_module_fields): after a REFUSED for another shape, + * in every packet (after OFFER, before records) until the + * stream is taken. A receiver whose input has every field + * takes it by name (docs/design/014-fields-by-name.md) * 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, with RACK's patience for packets @@ -129,9 +135,10 @@ enum { PURE_LINK_DATAGRAM = 1200, PURE_LINK_DATAGRAM_MOST = 1452, PURE_LINK_NAME = 64, PURE_LINK_ADDRESS = 48, PURE_LINK_RECORD_MOST = 16 << 20 }; /* Why a receiver refuses a stream (REFUSED): no input of that name takes - * messages, the input takes messages of another shape, or the sender's key - * may not send as that source. */ -enum { PURE_LINK_NO_INPUT = 1, PURE_LINK_OTHER_SHAPE = 2, PURE_LINK_NOT_ALLOWED = 3 }; + * messages, the input takes messages of another shape (send FIELDS), the + * sender's key may not send as that source, or the input lacks fields the + * stream sends (it would drop them). */ +enum { PURE_LINK_NO_INPUT = 1, PURE_LINK_OTHER_SHAPE = 2, PURE_LINK_NOT_ALLOWED = 3, PURE_LINK_FIELDS_LACKED = 4 }; /* Counts for a host's report: packets and records sent, records sent again, * packets lost and acknowledged, packets received, deliveries handed over, @@ -193,6 +200,9 @@ unsigned pure_link_sender_refusal(const PureLinkSender *); /* How often it moves on to a new epoch (ns), and after how many bytes sent * (an hour and a GiB at first). */ void pure_link_sender_epochs(PureLinkSender *, uint64_t every_ns, uint64_t bytes); +/* The fields its records are laid out with (pure_module_fields), for a + * receiver whose input takes another shape (FIELDS); none at first. */ +bool pure_link_sender_fields(PureLinkSender *, const unsigned char *fields, size_t size); /* 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 *); @@ -206,7 +216,9 @@ void pure_link_sender_free(PureLinkSender *); * pure_link_idle). A stream (a source to an input) goes to the host's input * its name and shape find: its number, or -PURE_LINK_OTHER_SHAPE when it * takes another shape, or -1 (PURE_LINK_NO_INPUT) when none takes messages - * by that name; a stream refused is said to be (REFUSED). Whole + * by that name; asked again with the fields the sender names (FIELDS), + * its number when it takes them by name, else -PURE_LINK_FIELDS_LACKED. A + * stream refused is said to be (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). @@ -220,9 +232,13 @@ typedef struct { const unsigned char *records; /* count of: u32le size, bytes (the receiver's: the host's during the call) */ size_t size; void *token; /* for its done */ + /* The fields the records are laid out with when the input took them by + * name (the host lays each out anew for it), else NULL. */ + const unsigned char *fields; + size_t fields_size; } PureLinkDelivery; typedef struct { - int (*input)(void *context, const char *name, uint32_t shape); + int (*input)(void *context, const char *name, uint32_t shape, const unsigned char *fields, size_t fields_size); bool (*deliver)(void *context, const PureLinkDelivery *); void *context; } PureLinkHost; diff --git a/link/link.c b/link/link.c index ad9376c..8f36e21 100644 --- a/link/link.c +++ b/link/link.c @@ -210,6 +210,13 @@ uint64_t pure_link_delivered(PureLink *l) { return stored; } +bool pure_link_fields(PureLink *l, const unsigned char *fields, size_t size) { + pthread_mutex_lock(&l->lock); + bool kept = pure_link_sender_fields(l->sender, fields, size); + pthread_mutex_unlock(&l->lock); + return kept; +} + size_t pure_link_held(PureLink *l) { pthread_mutex_lock(&l->lock); size_t held = pure_link_sender_held(l->sender); @@ -380,7 +387,8 @@ static void *run_link(void *context) { "trying again\"}\n", l->host, l->port, l->input, l->source, refusal == PURE_LINK_NOT_ALLOWED ? "this node's key may not send as that source" : refusal == PURE_LINK_OTHER_SHAPE ? "its input takes messages of another shape" - : "it has no input of that name taking messages"); + : refusal == PURE_LINK_FIELDS_LACKED ? "its input lacks fields this output sends (it would drop them)" + : "it has no input of that name taking messages"); } return NULL; } diff --git a/link/link.h b/link/link.h index 64846ce..c8d8635 100644 --- a/link/link.h +++ b/link/link.h @@ -18,6 +18,10 @@ typedef struct PureLink PureLink; PureLink *pure_link_open(const char *url, const char *source, uint32_t shape, const PureSealKey *own, const unsigned char own_public[PURE_SEAL_KEY], char *error, size_t size); void pure_link_peer(PureLink *, const unsigned char peer[PURE_SEAL_KEY]); +/* The fields its sends are laid out with (core/module.h's + * pure_module_fields): a receiver whose input takes another shape may take + * them by name (docs/design/014-fields-by-name.md). */ +bool pure_link_fields(PureLink *, const unsigned char *fields, size_t size); bool pure_link_queue(PureLink *, uint64_t seq, const unsigned char *bytes, size_t size); uint64_t pure_link_queued(PureLink *); /* The last number the receiver saved; any thread. */ diff --git a/link/test-link.c b/link/test-link.c index 66899ff..2e7509a 100644 --- a/link/test-link.c +++ b/link/test-link.c @@ -102,6 +102,8 @@ typedef struct { * handed records meanwhile, and how long after it had room again. */ uint64_t full_at, full_ms, full_until, full_tries, full_late; bool later_open; /* the app has an input "later" (as "order") */ + bool takes; /* its input takes the fields a sender of another shape names (NAMED) */ + uint64_t named_deliveries; /* deliveries that came with those fields */ } World; static void put_packet(World *w, bool to_receiver, const unsigned char *bytes, size_t size) { @@ -146,10 +148,15 @@ static bool answer(void *context, const unsigned char *packet, size_t size, cons return true; } -static int input_of(void *context, const char *name, uint32_t shape) { +/* The fields a sender of another shape (0x5eee) names; opaque to the link. */ +static const unsigned char NAMED[] = {2, 4, 4, 'i', 't', 'e', 'm', 1, 5, 'c', 'o', 'u', 'n', 't'}; + +static int input_of(void *context, const char *name, uint32_t shape, const unsigned char *fields, size_t size) { const World *w = context; if (strcmp(name, "order") && !(w && w->later_open && !strcmp(name, "later"))) return -PURE_LINK_NO_INPUT; - return shape == 0x5eed ? 3 : -PURE_LINK_OTHER_SHAPE; + if (shape == 0x5eed) return 3; + if (!fields) return -PURE_LINK_OTHER_SHAPE; + return w && w->takes && size == sizeof(NAMED) && !memcmp(fields, NAMED, size) ? 3 : -PURE_LINK_FIELDS_LACKED; } /* Applied from the one after the last heard, in order; a gap stops it. */ @@ -158,6 +165,7 @@ static bool deliver(void *context, const PureLinkDelivery *d) { 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; + w->named_deliveries += d->fields && d->fields_size == sizeof(NAMED) && !memcmp(d->fields, NAMED, sizeof(NAMED)); if (w->full_ms && !w->full_until && w->heard >= w->full_at) w->full_until = w->now + w->full_ms * MS; if (w->now < w->full_until) { ++w->full_tries; @@ -682,8 +690,8 @@ typedef struct { PureLinkDone done[64]; size_t done_count; } Outputs; -static int output_input(void *context, const char *name, uint32_t shape) { - (void)context; +static int output_input(void *context, const char *name, uint32_t shape, const unsigned char *fields, size_t size) { + (void)context, (void)fields, (void)size; return shape != 0x5eed ? -1 : !strcmp(name, "one") ? 0 : !strcmp(name, "two") ? 1 : !strcmp(name, "three") ? 2 : -1; } static bool output_deliver(void *context, const PureLinkDelivery *d) { @@ -958,6 +966,73 @@ static void refused_restarted(const char *dir) { free(w.packets); } +/* A stream of another shape whose sender names its fields (FIELDS): the + * receiver's input takes them by name and every record is stored, each + * delivery naming them for the host to lay the records out anew; an input + * that does not take them refuses the stream (it would drop fields), and the + * sender tries a packet a wait as for any refusal. One that names none stays + * refused for another shape (refusals, above). */ +static void named(const char *dir) { + for (int takes = 1; takes >= 0; --takes) { + unsigned char public_a[PURE_SEAL_KEY], public_b[PURE_SEAL_KEY], out[PURE_LINK_DATAGRAM_MOST]; + 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)), .takes = takes}; + if (!w.packets) exit(1); + PureLinkReceiver *r = receiver_of(&w, b, public_b, public_a, memo, 120000); + PureLinkSender *s = pure_link_sender("test.ordered", "order", 0x5eee, a, public_a); + if (!s || !pure_link_sender_peer(s, public_b) || !pure_link_sender_fields(s, NAMED, sizeof(NAMED))) exit(1); + static unsigned char bytes[20200]; + for (uint64_t seq = 1; seq <= 200; ++seq) { + size_t size = record_size(seq); + for (size_t k = 0; k < size; ++k) bytes[k] = record_byte(seq, k); + if (!pure_link_sender_queue(s, seq, bytes, size)) exit(1); + } + uint64_t number, packets = 0, stored_at = 0; + for (; w.now < 30000 * MS; w.now += MS) { + for (size_t size; (size = pure_link_sender_packet(s, out, false, w.now, &number)); ++packets) + pure_link_receiver_take(r, out, size, "sender0", 7, 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); + for (size_t k = 0; k < w.count;) { + if (w.packets[k].at > w.now) { + ++k; + continue; + } + pure_link_sender_take(s, w.packets[k].bytes, w.packets[k].size, w.now); + w.packets[k] = w.packets[--w.count]; + } + pure_link_sender_tick(s, w.now); + if (!stored_at && pure_link_sender_stored(s) == 200) stored_at = w.now; + } + PureLinkCounts c; + pure_link_sender_counts(s, &c); + unsigned why = pure_link_sender_refusal(s); + printf("{\"case\":\"named\",\"takes\":%d,\"why\":%u,\"refused\":%llu,\"packets\":%llu,\"stored\":%llu," + "\"stored_ms\":%llu,\"named_deliveries\":%llu}\n", takes, why, (unsigned long long)c.refused, + (unsigned long long)packets, (unsigned long long)pure_link_sender_stored(s), + (unsigned long long)(stored_at / MS), (unsigned long long)w.named_deliveries); + if (takes) + CHECK(!why && c.refused == 1 && stored_at && stored_at < 1000 * MS && w.heard == 200 && + w.named_deliveries == w.deliveries, + "named and taken: why %u, refused %llu, stored at %llu ms, %llu of %llu deliveries named", why, + (unsigned long long)c.refused, (unsigned long long)(stored_at / MS), + (unsigned long long)w.named_deliveries, (unsigned long long)w.deliveries); + else /* refused at once, named at once, then at 1, 3, 7 and 15 s */ + CHECK(why == PURE_LINK_FIELDS_LACKED && c.refused == 6 && !pure_link_sender_stored(s) && packets <= 20, + "named and not taken: why %u, refused %llu, %llu packets", why, (unsigned long long)c.refused, + (unsigned long long)packets); + pure_link_sender_free(s); + pure_link_receiver_free(r); + pure_seal_key_free(a), pure_seal_key_free(b); + free(w.packets); + } +} + /* The largest send (PURE_LINK_RECORD_MOST) goes whole, over a receiver that * never heard of the stream before; one byte more is not taken to send. */ typedef struct { World *w; PureLinkSender *s; } Largest; @@ -1089,6 +1164,7 @@ int main(void) { outputs(dir); refusals(dir); refused_restarted(dir); + named(dir); largest(dir); char command[600]; snprintf(command, sizeof(command), "rm -rf %s", dir); |