diff options
Diffstat (limited to 'prototypes')
| -rw-r--r-- | prototypes/connections/module.c | 7 | ||||
| -rw-r--r-- | prototypes/connections/module.h | 2 | ||||
| -rw-r--r-- | prototypes/connections/serve.c | 43 |
3 files changed, 42 insertions, 10 deletions
diff --git a/prototypes/connections/module.c b/prototypes/connections/module.c index 5cb7f11..697e918 100644 --- a/prototypes/connections/module.c +++ b/prototypes/connections/module.c @@ -1668,6 +1668,13 @@ void pure_app_outbox(const PureApp *app, PureSentVisit visit, void *context) { } } +void pure_app_outbox_newest(const PureApp *app, PureSentVisit visit, void *context) { + for (size_t i = app->sent_durable; i-- > 0;) { + const struct PureSent *sent = &app->sent[i]; + if (!sent->trimmed && !visit(context, sent->output, sent->seq, (PureSpan){sent->payload, sent->size})) return; + } +} + void pure_app_trim(PureApp *app, uint32_t output, uint64_t through) { for (size_t i = 0; i < app->sent_durable; ++i) if (app->sent[i].output == output && app->sent[i].seq <= through && !app->sent[i].trimmed) { diff --git a/prototypes/connections/module.h b/prototypes/connections/module.h index 3441c03..f62326b 100644 --- a/prototypes/connections/module.h +++ b/prototypes/connections/module.h @@ -309,6 +309,8 @@ PureResult pure_app_replace(PureApp *, const PureModule *); * and trims what its receiver has saved; a receiver applies each once. */ typedef bool (*PureSentVisit)(void *context, uint32_t output, uint64_t seq, PureSpan payload); void pure_app_outbox(const PureApp *, PureSentVisit visit, void *context); +/* The same, newest first (a host looking only for what is new stops early). */ +void pure_app_outbox_newest(const PureApp *, PureSentVisit visit, void *context); /* Forget output's saved sends numbered up to through (saved with the next save). */ void pure_app_trim(PureApp *, uint32_t output, uint64_t through); /* Deliver a message event from source (a sender's name, 1 to 64 printable diff --git a/prototypes/connections/serve.c b/prototypes/connections/serve.c index 60771e1..2ea7adb 100644 --- a/prototypes/connections/serve.c +++ b/prototypes/connections/serve.c @@ -1033,23 +1033,46 @@ static void publish(Executor *ex) { ++ex->publications; } -/* The saved sends of wired outputs not queued yet go to their wires, in order; - * a wire that cannot take one takes the rest at a later save. */ -typedef struct { Executor *executor; bool blocked[PURE_MODULE_INPUTS]; } Queueing; - -static bool queue_send(void *context, uint32_t output, uint64_t seq, PureSpan payload) { +/* The saved sends of wired outputs not queued yet go to their wires, in + * order; a wire that cannot take one takes the rest at a later save. The + * outbox is looked at newest first, only as far back as the new sends go, + * however many older ones wait for their receivers. */ +typedef struct { uint32_t wire; uint64_t seq; PureSpan payload; } Fresh; +typedef struct { Executor *executor; Fresh *fresh; size_t count, capacity; unsigned done; bool failed; } Queueing; + +static bool find_fresh(void *context, uint32_t output, uint64_t seq, PureSpan payload) { Queueing *q = context; Executor *ex = q->executor; - for (unsigned i = 0; i < ex->wire_count; ++i) - if (pure_wire_output(ex->wires[i]) == output && !q->blocked[i] && seq > pure_wire_queued(ex->wires[i])) - q->blocked[i] = !pure_wire_queue(ex->wires[i], seq, payload); - return true; + for (unsigned i = 0; i < ex->wire_count; ++i) { + if (pure_wire_output(ex->wires[i]) != output || (q->done >> i & 1)) continue; + if (seq <= pure_wire_queued(ex->wires[i])) { + q->done |= 1u << i; /* older ones were queued before */ + break; + } + if (q->count == q->capacity) { + size_t capacity = q->capacity ? 2 * q->capacity : 64; + Fresh *grown = realloc(q->fresh, capacity * sizeof(*grown)); + if (!grown) { q->failed = true; return false; } + q->fresh = grown; + q->capacity = capacity; + } + q->fresh[q->count++] = (Fresh){i, seq, payload}; + break; + } + return q->done != (ex->wire_count >= 32 ? UINT32_MAX : (1u << ex->wire_count) - 1); } static void after_save(Executor *ex) { publish(ex); + if (!ex->wire_count) return; Queueing q = {.executor = ex}; - if (ex->wire_count) pure_app_outbox(&ex->app, queue_send, &q); + pure_app_outbox_newest(&ex->app, find_fresh, &q); + bool blocked[PURE_MODULE_INPUTS] = {false}; + for (size_t k = q.failed ? 0 : q.count; k-- > 0;) { + const Fresh *f = &q.fresh[k]; + if (!blocked[f->wire]) blocked[f->wire] = !pure_wire_queue(ex->wires[f->wire], f->seq, f->payload); + } + free(q.fresh); } /* The answers the wires made become events, in order; then what the wires |