aboutsummaryrefslogtreecommitdiffstats
path: root/serve
diff options
context:
space:
mode:
Diffstat (limited to 'serve')
-rw-r--r--serve/link.c106
1 files changed, 93 insertions, 13 deletions
diff --git a/serve/link.c b/serve/link.c
index 8b0ea0d..d69080a 100644
--- a/serve/link.c
+++ b/serve/link.c
@@ -361,11 +361,10 @@ static void wake(int fd) {
bool pure_link_queue(PureLink *l, uint64_t seq, PureSpan payload) {
Record r = {.seq = seq, .size = (uint32_t)payload.size};
r.pieces = payload.size ? (uint32_t)((payload.size + PIECE - 1) / PIECE) : 1;
- if (payload.size > RECORD_MOST || !(r.bytes = malloc(payload.size ? payload.size : 1)) ||
- !(r.piece = calloc(r.pieces, sizeof(*r.piece)))) {
- free(r.bytes);
- return false;
- }
+ /* One allocation: its pieces' states, then its bytes. */
+ if (payload.size > RECORD_MOST || !(r.piece = malloc(r.pieces * sizeof(Piece) + payload.size))) return false;
+ memset(r.piece, 0, r.pieces * sizeof(Piece));
+ r.bytes = (unsigned char *)(r.piece + r.pieces);
if (payload.size) memcpy(r.bytes, payload.data, payload.size);
pthread_mutex_lock(&l->lock);
bool ok = seq > l->queued && (!l->count || seq == l->queued + 1);
@@ -391,7 +390,6 @@ bool pure_link_queue(PureLink *l, uint64_t seq, PureSpan payload) {
}
pthread_mutex_unlock(&l->lock);
if (!ok) {
- free(r.bytes);
free(r.piece);
return false;
}
@@ -539,8 +537,7 @@ static void on_stored(PureLink *l, const unsigned char *p, size_t size, uint64_t
while (l->count && record_at(l, 0)->seq <= heard) {
Record *r = record_at(l, 0);
for (uint32_t i = 0; i < r->pieces; ++i) l->received -= r->piece[i].state == RECEIVED;
- free(r->bytes);
- free(r->piece);
+ free(r->piece); /* and its bytes */
l->head = (l->head + 1) % l->capacity;
--l->count;
}
@@ -895,8 +892,7 @@ void pure_link_close(PureLink *l) {
pthread_join(l->thread, NULL);
}
for (size_t k = 0; k < l->count; ++k) {
- free(record_at(l, k)->bytes);
- free(record_at(l, k)->piece);
+ free(record_at(l, k)->piece); /* and its bytes */
}
free(l->records);
free(l->flights);
@@ -945,6 +941,15 @@ typedef struct Stream {
uint32_t input;
uint64_t heard; /* the last number heard (stored), once known */
bool known, out, stored_due, gap, ready;
+ /* The line: once the last heard is known, whole records from the one
+ * after it on, in order, as a delivery carries them (u32 size, bytes),
+ * from line_head: handed over as they are, let go by moving the head.
+ * Records that come early or in pieces, or before the last heard is
+ * known, wait in held until the line reaches them. */
+ unsigned char *line;
+ size_t line_head, line_size, line_capacity;
+ uint64_t line_first;
+ unsigned line_count;
Held *held; /* records not yet stored, by number: held_count from held_first */
size_t held_first, held_count, held_capacity, held_bytes;
Run *run; /* the run that sent to it last: where STORED goes */
@@ -1122,6 +1127,60 @@ full:
static Held *held_at(Stream *st, size_t k) { return &st->held[st->held_first + k]; }
+static uint64_t line_end(const Stream *st) { return st->line_first + st->line_count; }
+
+/* Room for more bytes at the line's end: its front let go given back
+ * first, else a longer line. */
+static bool line_room(Stream *st, size_t more) {
+ if (st->line_size + more <= st->line_capacity) return true;
+ if (st->line_head) {
+ memmove(st->line, st->line + st->line_head, st->line_size - st->line_head);
+ st->line_size -= st->line_head;
+ st->line_head = 0;
+ if (st->line_size + more <= st->line_capacity) return true;
+ }
+ size_t capacity = st->line_capacity ? st->line_capacity : 4096;
+ while (capacity < st->line_size + more) capacity *= 2;
+ unsigned char *grown = realloc(st->line, capacity);
+ if (!grown) return false;
+ st->line = grown;
+ st->line_capacity = capacity;
+ return true;
+}
+
+static bool line_append(Stream *st, const unsigned char *bytes, uint32_t size) {
+ if (!line_room(st, 4 + (size_t)size)) return false;
+ put32(st->line + st->line_size, size);
+ if (size) memcpy(st->line + st->line_size + 4, bytes, size);
+ st->line_size += 4 + (size_t)size;
+ ++st->line_count;
+ return true;
+}
+
+/* The line's records up to heard let go (all of them, when heard is past it). */
+static void line_drop(Stream *st, uint64_t heard) {
+ if (heard < st->line_first) return;
+ if (heard >= line_end(st)) {
+ st->line_head = st->line_size = 0;
+ st->line_count = 0;
+ st->line_first = heard + 1;
+ return;
+ }
+ for (uint64_t n = heard - st->line_first + 1; n; --n, --st->line_count, ++st->line_first)
+ st->line_head += 4 + (size_t)get32(st->line + st->line_head);
+}
+
+/* Records held that the line reached, moved onto it; ones behind it let go. */
+static void drop_first(Stream *st);
+static void line_pull(Stream *st) {
+ while (st->held_count) {
+ Held *h = held_at(st, 0);
+ if (h->seq < line_end(st)) drop_first(st);
+ else if (h->seq == line_end(st) && h->got == h->pieces && line_append(st, h->bytes, h->size)) drop_first(st);
+ else break;
+ }
+}
+
/* The first record held let go: the ones after it stay where they are. */
static void drop_first(Stream *st) {
st->held_bytes -= held_at(st, 0)->size;
@@ -1210,6 +1269,7 @@ static void let_go(PureLinkListener *s, Active *a) {
for (unsigned k = 0; k < a->stream_count; ++k) {
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]);
}
free(a->streams);
@@ -1223,7 +1283,7 @@ static void let_go(PureLinkListener *s, Active *a) {
/* Whether any of a peer's records or answers waits here. */
static bool busy(const Active *a) {
for (unsigned k = 0; k < a->stream_count; ++k)
- if (a->streams[k]->out || a->streams[k]->held_count || a->streams[k]->ready) return true;
+ if (a->streams[k]->out || a->streams[k]->held_count || a->streams[k]->line_count || a->streams[k]->ready) return true;
for (unsigned i = 0; i < RUNS; ++i)
if (a->runs[i].due) return true;
return false;
@@ -1351,6 +1411,12 @@ static void hold(Stream *st, uint64_t seq, uint32_t size, uint32_t offset, const
return;
}
if ((st->known && seq > st->heard + CREDIT) || size > RECORD_MOST) return;
+ if (st->known && seq < line_end(st)) return; /* on the line already */
+ if (st->known && seq == line_end(st) && !offset && length == size && size <= PIECE) { /* the next, whole */
+ if (st->held_bytes + (st->line_size - st->line_head) + size <= HELD_MOST && line_append(st, bytes, size))
+ line_pull(st);
+ return;
+ }
/* In order it goes last; out of order, its place is found by halving. */
size_t k = st->held_count;
if (k && held_at(st, k - 1)->seq >= seq) {
@@ -1394,7 +1460,18 @@ 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(PureLinkListener *s, Stream *st) {
if (st->out) return true;
- while (st->known && st->held_count && held_at(st, 0)->seq <= st->heard) drop_first(st);
+ if (st->known) { /* the line as it is, up to BATCH records */
+ line_pull(st);
+ if (!st->line_count) return true;
+ unsigned n = 0;
+ size_t bytes = 0;
+ for (; n < st->line_count && 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};
+ if (!s->deliver(s->context, &d)) return false;
+ st->out = true;
+ ++s->counts.deliveries;
+ return true;
+ }
if (!st->held_count || (st->known && held_at(st, 0)->seq != st->heard + 1)) return true;
size_t n = 0, bytes = 0;
while (n < st->held_count && n < BATCH && held_at(st, n)->got == held_at(st, n)->pieces &&
@@ -1558,7 +1635,7 @@ static void answer(PureLinkListener *s, uint64_t now) {
at += put_names(body + at, st->source, s->module->input_names[st->input]);
put64(body + at, st->heard);
body[at + 8] = st->gap ? STORED_GAP : 0;
- size_t room = st->held_bytes < HELD_MOST ? CREDIT : 1;
+ size_t room = st->held_bytes + (st->line_size - st->line_head) < HELD_MOST ? CREDIT : 1;
put32(body + at + 9, (uint32_t)room);
at += 13;
st->stored_due = st->gap = false;
@@ -1588,6 +1665,9 @@ static void take_done(PureLinkListener *s) {
mark_ready(s, st);
if (!done[k].stored) continue; /* not saved: handed over again */
if (!st->known || done[k].heard > st->heard) st->heard = done[k].heard;
+ if (!st->known) st->line_first = st->heard + 1; /* the line begins */
+ line_drop(st, st->heard);
+ while (st->held_count && held_at(st, 0)->seq <= st->heard) drop_first(st);
st->known = true;
st->gap = done[k].gap;
st->stored_due = true;
generated by cgit v1.2.3 (git 2.46.0) at 2026-10-20 13:58:23 +0000