aboutsummaryrefslogtreecommitdiffstats
path: root/serve/deliver.c
diff options
context:
space:
mode:
authoreunhhu <[email protected]>2026-09-30 03:32:07 +0100
committereunhhu <[email protected]>2026-09-30 03:32:07 +0100
commitd9e0b6870201d9200b3c225903b43b850475600b (patch)
treeea4eff8cff3b84b33995e2faa3955f50a0868a89 /serve/deliver.c
parent41cb1cca55c67dd2aac2afa2f96bdc09a5d50b78 (diff)
downloadpure-d9e0b6870201d9200b3c225903b43b850475600b.tar.gz
pure-d9e0b6870201d9200b3c225903b43b850475600b.zip
Hold a window of each output's sends in its transport, not the outbox
A link's ring and a wire's queue took every saved send of their output and kept it until delivered: with the receiver down they held the outbox a second time, 178 B and 313 B more a send (104). Each outlet now holds a window, --window bytes (4 MiB) and 65,536 sends past what its receiver has, at least one; the outbox keeps the rest. The owner remembers each outlet's place in the outbox and looks on from there after a save and after its transports deliver. Places move back by the sends that left the outbox since (PureApp.sends_gone), so a trim before a place skips nothing, and sends queued before are not looked at again however many wait. pure_app_sent reads a send by its place (the newest-first visit it replaces is gone); pure_link_held and pure_wire_held say what a transport holds. With the receiver down a node now grows by the outbox alone once the window is full: 219 and 218 B a send over the link and HTTP (399 and 534 B before). check.py server_windows keeps the receivers of a sender's two outputs, one linked and one over HTTP, down through 300 buys with a 2,000-byte window, then starts them: each applies all 300 once. Without the refill after deliveries the link stops at 117.
Diffstat (limited to 'serve/deliver.c')
-rw-r--r--serve/deliver.c15
1 files changed, 13 insertions, 2 deletions
diff --git a/serve/deliver.c b/serve/deliver.c
index 511a2cf..b00984b 100644
--- a/serve/deliver.c
+++ b/serve/deliver.c
@@ -35,6 +35,7 @@ struct PureWire {
pthread_mutex_t lock;
pthread_cond_t changed;
Pending *queue; /* the sends to deliver, in order, from head */
+ size_t held; /* their bytes */
size_t head, count, capacity;
uint64_t queued, delivered, attempts, deliveries;
unsigned batch; /* how many a request may carry, as the receiver said (X-Pure-Batch); 0: one */
@@ -269,6 +270,13 @@ uint64_t pure_wire_delivered(PureWire *w) {
return delivered;
}
+size_t pure_wire_held(PureWire *w) {
+ pthread_mutex_lock(&w->lock);
+ size_t held = w->held;
+ pthread_mutex_unlock(&w->lock);
+ return held;
+}
+
void pure_wire_counts(PureWire *w, uint64_t *delivered, uint64_t *attempts) {
pthread_mutex_lock(&w->lock);
*delivered = w->deliveries;
@@ -312,6 +320,7 @@ bool pure_wire_queue(PureWire *w, uint64_t seq, PureSpan payload) {
if (sent && payload.size) memcpy(sent, payload.data, payload.size);
w->queue[w->head + w->count++] = (Pending){t.bytes, t.size, seq, sent, payload.size};
w->queued = seq;
+ w->held += t.size + (sent ? payload.size : 0);
pthread_cond_signal(&w->changed);
pthread_mutex_unlock(&w->lock);
return true;
@@ -717,8 +726,10 @@ static void *deliver(void *context) {
}
if (outcome == DELIVERED) {
for (unsigned i = 0; i < count; ++i) {
- free(w->queue[w->head + i].body);
- free(w->queue[w->head + i].sent);
+ const Pending *p = &w->queue[w->head + i];
+ w->held -= p->size + (p->sent ? p->sent_size : 0);
+ free(p->body);
+ free(p->sent);
}
w->head += count;
w->count -= count;