aboutsummaryrefslogtreecommitdiffstats
path: root/prototypes
diff options
context:
space:
mode:
authoreunhhu <[email protected]>2026-09-27 08:24:08 +0100
committereunhhu <[email protected]>2026-09-27 08:24:08 +0100
commit5372d03b5968796515966e2cfc89628d6448f0c1 (patch)
tree4ddefd226b2089aafdca06fd981ad3a17a37dbbf /prototypes
parentddff7e15d1d0e90a5f5a52fcc93828c8ef0bbd5c (diff)
downloadpure-5372d03b5968796515966e2cfc89628d6448f0c1.tar.gz
pure-5372d03b5968796515966e2cfc89628d6448f0c1.zip
Look for new sends newest first, only as far back as they go
After each save the owner walked the whole outbox to find the sends not yet queued, so a receiver long down made every save walk the backlog. pure_app_outbox_newest visits saved sends newest first, and the owner stops once every wired output reaches the number it queued last, then queues what it found in order.
Diffstat (limited to 'prototypes')
-rw-r--r--prototypes/connections/module.c7
-rw-r--r--prototypes/connections/module.h2
-rw-r--r--prototypes/connections/serve.c43
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