diff options
| -rw-r--r-- | link/link-protocol.c | 36 | ||||
| -rw-r--r-- | link/link-protocol.h | 6 | ||||
| -rw-r--r-- | link/link.c | 30 | ||||
| -rw-r--r-- | link/link.h | 4 | ||||
| -rw-r--r-- | link/test-link.c | 102 |
5 files changed, 173 insertions, 5 deletions
diff --git a/link/link-protocol.c b/link/link-protocol.c index 169705f..ea38fab 100644 --- a/link/link-protocol.c +++ b/link/link-protocol.c @@ -1332,6 +1332,7 @@ typedef struct Stream { 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; + bool stale; /* its input to be found again (pure_link_receiver_rebind) */ } Stream; struct Active { uint32_t peer; /* its place among the peers */ @@ -1897,21 +1898,32 @@ static bool keep_fields(Stream *st, const unsigned char *fields, size_t size) { * 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. */ + * those again, laid out anew. A stale one (its input may be another now) + * finds its input again, with the fields it was laid out with unless these + * come, and keeps what it holds. */ static Stream *stream_of(PureLinkReceiver *s, Active *a, const char *source, const char *input, uint32_t shape, const unsigned char *fields, size_t fields_size, unsigned *why) { *why = 0; 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; + if (st && st->shape == shape && !st->stale) return st; + bool again = st && st->shape == shape; /* stale */ + const unsigned char *named = fields ? fields : again ? st->fields : NULL; + size_t named_size = fields ? fields_size : again ? st->fields_size : 0; 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); + s->host.input(s->host.context, input, shape, named, named_size); if (index < 0) { *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 (again) { /* its input found again; what it holds stays */ + if (fields && !keep_fields(st, fields, fields_size)) return NULL; + st->input = (uint32_t)index; + st->stale = false; + return st; + } 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; @@ -2077,6 +2089,12 @@ static void hold(Stream *st, uint64_t seq, uint32_t size, uint32_t offset, const * False when the host could not take them now. */ static bool hand_over(PureLinkReceiver *s, Stream *st) { if (st->out) return true; + if (st->stale) { /* its input found again first: none takes them now, and they wait */ + int index = s->host.input(s->host.context, st->input_name, st->shape, st->fields, st->fields_size); + if (index < 0) return false; + st->input = (uint32_t)index; + st->stale = false; + } if (st->known) { /* the line as it is, up to BATCH records */ line_pull(st); if (!st->line_count) return true; @@ -2641,6 +2659,18 @@ void pure_link_receiver_turn(PureLinkReceiver *s, const PureLinkDone *done, size sweep(s, now); } +void pure_link_receiver_rebind(PureLinkReceiver *s, const char *prefix) { + size_t length = strlen(prefix); + for (size_t k = 0; k < s->active_count; ++k) { + Active *a = s->actives[k]; + for (unsigned i = 0; i < a->stream_count; ++i) + if (!strncmp(a->streams[i]->input_name, prefix, length)) { + a->streams[i]->stale = true; + mark_ready(s, a->streams[i]); /* what it holds handed over, its input found again */ + } + } +} + uint64_t pure_link_receiver_due(const PureLinkReceiver *s) { uint64_t due = s->active_count || s->ready_count ? s->swept + SWEEP_EVERY : 0; return s->waiting && s->waiting < due ? s->waiting : due; diff --git a/link/link-protocol.h b/link/link-protocol.h index feff543..4f263b2 100644 --- a/link/link-protocol.h +++ b/link/link-protocol.h @@ -278,6 +278,12 @@ void pure_link_receiver_take(PureLinkReceiver *, const unsigned char *packet, si * go. */ void pure_link_receiver_turn(PureLinkReceiver *, const PureLinkDone *done, size_t count, uint64_t now, PureLinkSend *send, void *context); +/* The host's inputs whose names begin with prefix may be others now (an app + * started again with another module): their streams find theirs again, by + * name and with the fields they were laid out with, before their next + * delivery, a delivery out first done; what they hold stays. One no input + * takes now is refused at its sender's next packet, as a new one would be. */ +void pure_link_receiver_rebind(PureLinkReceiver *, const char *prefix); /* 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 *); diff --git a/link/link.c b/link/link.c index a9270fe..2e1f359 100644 --- a/link/link.c +++ b/link/link.c @@ -426,9 +426,11 @@ struct PureLinkListener { int fd, wake, family; pthread_t thread; bool started; - pthread_mutex_t lock; /* the done list and stopping */ + pthread_mutex_t lock; /* the done and rebind lists, and stopping */ PureLinkDone *done; size_t done_count, done_capacity; + char **rebinds; /* the prefixes of the inputs to find again */ + size_t rebind_count, rebind_capacity; bool stopping; }; @@ -503,10 +505,18 @@ static void *run_listener(void *context) { } pthread_mutex_lock(&s->lock); PureLinkDone *done = s->done; - size_t count = s->done_count; + size_t count = s->done_count, rebinds = s->rebind_count; + char **prefixes = s->rebinds; s->done = NULL; s->done_count = s->done_capacity = 0; + s->rebinds = NULL; + s->rebind_count = s->rebind_capacity = 0; pthread_mutex_unlock(&s->lock); + for (size_t k = 0; k < rebinds; ++k) { + pure_link_receiver_rebind(s->receiver, prefixes[k]); + free(prefixes[k]); + } + free(prefixes); pure_link_receiver_turn(s->receiver, done, count, now, send_answer, s); free(done); } @@ -564,6 +574,20 @@ void pure_link_done(PureLinkListener *s, PureLinkDone done) { wake(s->wake); } +void pure_link_listener_rebind(PureLinkListener *s, const char *prefix) { + char *kept = strdup(prefix); + pthread_mutex_lock(&s->lock); + if (kept && s->rebind_count == s->rebind_capacity) { + size_t capacity = s->rebind_capacity ? 2 * s->rebind_capacity : 4; + char **grown = realloc(s->rebinds, capacity * sizeof(*grown)); + if (grown) s->rebinds = grown, s->rebind_capacity = capacity; + } + if (!kept || s->rebind_count == s->rebind_capacity) abort(); /* an input to find anew cannot be lost */ + s->rebinds[s->rebind_count++] = kept; + pthread_mutex_unlock(&s->lock); + wake(s->wake); +} + void pure_link_listener_close(PureLinkListener *s, PureLinkCounts *counts) { if (!s) return; if (s->started) { @@ -576,6 +600,8 @@ void pure_link_listener_close(PureLinkListener *s, PureLinkCounts *counts) { if (counts) pure_link_receiver_counts(s->receiver, counts); pure_link_receiver_free(s->receiver); free(s->done); + for (size_t k = 0; k < s->rebind_count; ++k) free(s->rebinds[k]); + free(s->rebinds); if (s->fd >= 0) close(s->fd); if (s->wake >= 0) close(s->wake); pthread_mutex_destroy(&s->lock); diff --git a/link/link.h b/link/link.h index 9846af4..40fc9f9 100644 --- a/link/link.h +++ b/link/link.h @@ -51,5 +51,9 @@ PureLinkListener *pure_link_listen(PureLinkReceiver *, unsigned port, char *erro bool pure_link_serve(PureLinkListener *, char *error, size_t size); /* How a delivery went: link-protocol.h's PureLinkDone, with its token. */ void pure_link_done(PureLinkListener *, PureLinkDone); +/* The host's inputs whose names begin with prefix may be others now: the + * receiver's streams to them find theirs again (pure_link_receiver_rebind), + * on the listener's thread; any thread. */ +void pure_link_listener_rebind(PureLinkListener *, const char *prefix); void pure_link_listener_close(PureLinkListener *, PureLinkCounts *counts); #endif diff --git a/link/test-link.c b/link/test-link.c index 2e7509a..fb64d52 100644 --- a/link/test-link.c +++ b/link/test-link.c @@ -774,6 +774,107 @@ static void outputs_until(Outputs *o, PureLinkReceiver *r, bool epochs) { } } } +/* The receiving app starts again while records come (docs/design/017's + * stage 3): first with no input of that name (the receiver holds what + * comes), then with the input at another number, then with one of another + * shape that takes the sender's fields by name. Told each time + * (pure_link_receiver_rebind), the stream finds its input again and keeps + * what it holds: every record is applied once, in order, and a delivery + * for the input as it was is never taken. */ +typedef struct { + PureLinkSender *sender; + uint64_t now, heard, stale, named, to[3]; + unsigned phase; /* 0: input 3; 1: none; 2: input 4; 3: input 5, by name */ + PureLinkDone done[64]; + size_t done_count; +} Rebound; +static int rebound_input(void *context, const char *name, uint32_t shape, const unsigned char *fields, size_t size) { + const Rebound *r = context; + if (strcmp(name, "order") || r->phase == 1) return -PURE_LINK_NO_INPUT; + if (r->phase < 3) return shape == 0x5eed ? (int)r->phase + 2 + (r->phase == 0) : -PURE_LINK_OTHER_SHAPE; + if (!fields) return -PURE_LINK_OTHER_SHAPE; + return size == sizeof(NAMED) && !memcmp(fields, NAMED, size) ? 5 : -PURE_LINK_FIELDS_LACKED; +} +static bool rebound_deliver(void *context, const PureLinkDelivery *d) { + Rebound *r = context; + unsigned now = r->phase == 0 ? 3 : r->phase == 2 ? 4 : r->phase == 3 ? 5 : 0; + if (d->input != now) { /* found for the app as it was: not taken, kept */ + ++r->stale; + return false; + } + if (r->done_count == sizeof(r->done) / sizeof(r->done[0])) return false; + CHECK(now != 5 || (d->fields && d->fields_size == sizeof(NAMED)), "rebound: records by name came without fields"); + r->named += now == 5; + ++r->to[now - 3]; + uint64_t heard = r->heard; + bool gap = false; + 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) { + gap = true; + break; + } + 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, "rebound: record %llu is not as it was sent", (unsigned long long)seq); + heard = seq; + } + at += 4 + size; + } + r->heard = heard; + r->done[r->done_count++] = (PureLinkDone){d->token, heard, true, gap, false}; + return true; +} +static bool rebound_answer(void *context, const unsigned char *packet, size_t size, const unsigned char *to, + size_t to_size) { + (void)to, (void)to_size; + Rebound *r = context; + pure_link_sender_take(r->sender, packet, size, r->now); + return true; +} +static void rebound(const char *dir) { + unsigned char public_a[PURE_SEAL_KEY], public_b[PURE_SEAL_KEY], out[PURE_LINK_DATAGRAM_MOST]; + char memo[512], error[256]; + PureSealKey *a = key(dir, "a.key", public_a), *b = key(dir, "b.key", public_b); + snprintf(memo, sizeof(memo), "%s/b.key.links", dir); + unlink(memo); + Rebound r = {0}; + PureLinkReceiver *receiver = pure_link_receiver(b, public_b, &(PureLinkHost){rebound_input, rebound_deliver, &r}, + error, sizeof(error)); + if (!receiver || !pure_link_allow(receiver, "test.ordered", public_a, error, sizeof(error))) exit(1); + r.sender = sender_of(a, public_a, public_b, 1, 3000); + if (!pure_link_sender_fields(r.sender, NAMED, sizeof(NAMED))) exit(1); + uint64_t number, back = 0, end = 60000 * MS; + for (; r.now < end && pure_link_sender_stored(r.sender) < 3000; r.now += MS) { + unsigned was = r.phase; + if (r.phase == 0 && r.heard >= 1000) r.phase = 1, back = r.now + 300 * MS; /* stopped */ + else if (r.phase == 1 && r.now >= back) r.phase = 2; /* started again, the input another */ + else if (r.phase == 2 && r.heard >= 2000) r.phase = 3; /* and again, of another shape */ + if (r.phase != was) pure_link_receiver_rebind(receiver, "ord"); + for (size_t size; (size = pure_link_sender_packet(r.sender, out, false, r.now, &number));) + pure_link_receiver_take(receiver, out, size, "sender0", 7, r.now); + PureLinkDone told[64]; + size_t count = r.done_count; + memcpy(told, r.done, count * sizeof(*told)); + r.done_count = 0; + pure_link_receiver_turn(receiver, told, count, r.now, rebound_answer, &r); + pure_link_sender_tick(r.sender, r.now); + } + CHECK(r.heard == 3000 && pure_link_sender_stored(r.sender) == 3000, "rebound: heard %llu, stored %llu of 3000", + (unsigned long long)r.heard, (unsigned long long)pure_link_sender_stored(r.sender)); + CHECK(r.phase == 3 && r.to[0] && r.to[1] && r.to[2], "rebound: deliveries %llu, %llu, %llu (phase %u)", + (unsigned long long)r.to[0], (unsigned long long)r.to[1], (unsigned long long)r.to[2], r.phase); + printf("{\"case\":\"rebound\",\"records\":3000,\"virtual_ms\":%llu,\"to_3\":%llu,\"to_4\":%llu,\"to_5_by_name\":%llu," + "\"stale_refused\":%llu}\n", (unsigned long long)(r.now / MS), (unsigned long long)r.to[0], + (unsigned long long)r.to[1], (unsigned long long)r.to[2], (unsigned long long)r.stale); + pure_link_sender_free(r.sender); + pure_link_receiver_free(receiver); + pure_seal_key_free(a), pure_seal_key_free(b); +} + static void outputs(const char *dir) { unsigned char public_a[PURE_SEAL_KEY], public_b[PURE_SEAL_KEY]; char memo[512]; @@ -1162,6 +1263,7 @@ int main(void) { impostors(dir); flood(dir); outputs(dir); + rebound(dir); refusals(dir); refused_restarted(dir); named(dir); |