From 94e7bb87bce6821e91734f1fcbb6cfd61a008049 Mon Sep 17 00:00:00 2001 From: eunhhu Date: Thu, 1 Oct 2026 08:52:43 +0100 Subject: Push a turn's sends to a link at once, and round only the links that want one (017, 2b-1) The sending loop was woken by every record queued while it slept: the owner queues one by one, the loop caught up and slept between them, and every link took a round on every wake. Four links sent 64k packets of 12 records where threads sent 40k of 20, and ran at 257k sends/s against 581k. The owner now queues a turn's sends and pushes each link once; the loop rounds a link only when its socket has answers, it was pushed, or its time came. --- link/link.c | 37 ++++++++++++++----------- link/link.h | 13 ++++++--- research/experiments/019-idle-links/run.py | 4 +-- research/experiments/019-idle-links/transport.c | 4 ++- serve/serve.c | 2 ++ 5 files changed, 37 insertions(+), 23 deletions(-) diff --git a/link/link.c b/link/link.c index 3ec138b..8b57ba6 100644 --- a/link/link.c +++ b/link/link.c @@ -118,15 +118,16 @@ struct PureLink { bool peered, keyed; /* given the receiver's key; the keys made from it */ pthread_mutex_t lock; bool started; - bool queued_new; /* records were queued since it last sent */ - uint64_t told; /* the last stored the owner was told of */ + uint64_t told; /* the last stored the owner was told of */ int fd; void (*notify)(void *); void *context; /* The sending loop's own (below). */ struct PureLink *next; /* in the loop */ uint64_t id; /* its epoll events' */ + _Atomic bool pushed; /* records were queued and pushed since its last round */ bool readable; /* its socket said so this round */ + uint64_t due; /* when it wants its next round (0: when told) */ unsigned char *burst; /* packets sealed and not yet sent (BURST datagrams) */ size_t burst_size; unsigned burst_count; @@ -187,9 +188,11 @@ static void wake(int fd) { /* The sending loop: every link of the process sends on one thread * (docs/design/017-apps-on-one-host.md, 2b), an epoll over their sockets * and one wake, sleeping until the soonest link's next time. Made with the - * first link started, stopped with the last closed. A queue wakes it only - * when it sleeps, once: queued tells it records came (it then does not - * sleep), sleeping that it is about to, signaled that it was woken. */ + * first link started, stopped with the last closed. A round goes to the + * links that want one: answers on its socket, records pushed, or its time + * come. A push wakes it only when it sleeps, once: queued tells it records + * came (it then does not sleep), sleeping that it is about to, signaled + * that it was woken. */ static struct { pthread_mutex_t lock; /* the links, and its making and stopping */ PureLink *first; @@ -204,15 +207,16 @@ static struct { bool pure_link_queue(PureLink *l, uint64_t seq, const unsigned char *bytes, size_t size) { pthread_mutex_lock(&l->lock); bool ok = pure_link_sender_queue(l->sender, seq, bytes, size); - if (ok) l->queued_new = true; pthread_mutex_unlock(&l->lock); - if (ok) { - atomic_store(&loop.queued, true); - if (atomic_load(&loop.sleeping) && !atomic_exchange(&loop.signaled, true)) wake(loop.wake); - } return ok; } +void pure_link_push(PureLink *l) { + atomic_store(&l->pushed, true); + atomic_store(&loop.queued, true); + if (atomic_load(&loop.sleeping) && !atomic_exchange(&loop.signaled, true)) wake(loop.wake); +} + uint64_t pure_link_queued(PureLink *l) { pthread_mutex_lock(&l->lock); uint64_t queued = pure_link_sender_queued(l->sender); @@ -297,7 +301,6 @@ static void flush(PureLink *l) { /* Packets while the sender has them (up to 128 a turn), into bursts: a * short packet ends one, and a probe of a larger datagram goes alone. */ static void send_packets(PureLink *l, uint64_t now) { - l->queued_new = false; /* all queued is looked at now */ l->segment = pure_link_sender_datagram(l->sender); /* larger once the path is found to carry it */ unsigned most = BURST_BYTES / l->segment < BURST ? (unsigned)(BURST_BYTES / l->segment) : BURST; for (unsigned n = 0; n < 128; ++n) { @@ -365,7 +368,7 @@ enum { LOOK_PROBES = 3 }; /* A link's round in the loop (its lock not held): answers taken when its * socket had them, its times kept, the receiver looked for again when * probes go unanswered, packets sent, and the owner told of what was - * stored. When it wants its next round (ns on now's clock; 0: none). */ + * stored. When it wants its next round (ns on now's clock; 0: when told). */ static uint64_t link_round(PureLink *l) { pthread_mutex_lock(&l->lock); uint64_t now = now_ns(); @@ -386,7 +389,7 @@ static uint64_t link_round(PureLink *l) { } l->looked = probes; send_packets(l, now); - uint64_t stored = pure_link_sender_stored(l->sender), due = l->queued_new ? now : pure_link_sender_due(l->sender); + uint64_t stored = pure_link_sender_stored(l->sender), due = pure_link_sender_due(l->sender); bool told = stored != l->told; l->told = stored; unsigned refusal = pure_link_sender_refusal(l->sender), before = l->refusal; @@ -430,8 +433,9 @@ static void *run_sending(void *context) { } uint64_t soonest = 0, now = now_ns(); for (PureLink *l = loop.first; l; l = l->next) { - uint64_t due = link_round(l); - if (due && (!soonest || due < soonest)) soonest = due; + bool pushed = atomic_exchange(&l->pushed, false); /* pushed after: the next round's */ + if (l->readable || pushed || (l->due && l->due <= now)) l->due = link_round(l); + if (l->due && (!soonest || l->due < soonest)) soonest = l->due; } pthread_mutex_unlock(&loop.lock); timeout = !soonest ? -1 : soonest <= now ? 0 : (int)((soonest - now) / MS) + 1; @@ -471,7 +475,8 @@ bool pure_link_start(PureLink *l, void (*notify)(void *), void *context, char *e l->next = loop.first; loop.first = l; l->started = true; - atomic_store(&loop.queued, true); /* a first round for it */ + atomic_store(&l->pushed, true); /* a first round for it */ + atomic_store(&loop.queued, true); } pthread_mutex_unlock(&loop.lock); if (!ok) { diff --git a/link/link.h b/link/link.h index c8d8635..9846af4 100644 --- a/link/link.h +++ b/link/link.h @@ -3,9 +3,9 @@ #include "link-protocol.h" /* Links over UDP: the two sides of link/link-protocol.h (what goes on the - * wire, and what each side keeps) with sockets and threads. A sender sends - * on a thread of its own, packets in bursts with UDP GSO; a listener takes - * a receiver's packets on a port, with UDP GRO. */ + * wire, and what each side keeps) with sockets and threads. The process's + * senders send on one thread, packets in bursts with UDP GSO; a listener + * takes a receiver's packets on a port, with UDP GRO. */ /* The sending side: one output's saved sends to url, pure://host:port/INPUT, * as source (a Pure receiver's name for it), their messages of shape (the @@ -22,7 +22,10 @@ void pure_link_peer(PureLink *, const unsigned char peer[PURE_SEAL_KEY]); * pure_module_fields): a receiver whose input takes another shape may take * them by name (docs/design/014-fields-by-name.md). */ bool pure_link_fields(PureLink *, const unsigned char *fields, size_t size); +/* Sends are queued, then pushed: what was queued goes out once pushed, so + * the owner's sends of a turn go together, in full packets. */ bool pure_link_queue(PureLink *, uint64_t seq, const unsigned char *bytes, size_t size); +void pure_link_push(PureLink *); uint64_t pure_link_queued(PureLink *); /* The last number the receiver saved; any thread. */ uint64_t pure_link_delivered(PureLink *); @@ -32,7 +35,9 @@ size_t pure_link_held(PureLink *); * off, each in a call of its own: for a path whose queues take a burst as * one packet (netem does). Before pure_link_start. */ void pure_link_gso(PureLink *, bool on); -/* Send on a thread of the link's own, calling notify when more was saved. */ +/* Send on the process's sending loop (its thread made with the first link + * started, stopped with the last closed), calling notify (on that thread) + * when more was saved. */ bool pure_link_start(PureLink *, void (*notify)(void *), void *context, char *error, size_t size); void pure_link_counts(PureLink *, PureLinkCounts *); void pure_link_close(PureLink *); diff --git a/research/experiments/019-idle-links/run.py b/research/experiments/019-idle-links/run.py index 0d18c1d..bfe5b92 100644 --- a/research/experiments/019-idle-links/run.py +++ b/research/experiments/019-idle-links/run.py @@ -490,7 +490,7 @@ def drain_pure(work, n, size): b, _ = start([ROOT / 'build/pure-serve', '--module', receiver_module, '--store', work / 'drain-b', '--port', port_b, '--threads', '2', '--key', work / 'b.key', '--link-port', link_port, '--senders', work / 'drain-senders']) cpu = {name: (thread_cpu(pid, None), thread_cpu(pid, thread)) for name, pid, thread in - (('sender', a.pid, 'pure-link'), ('receiver', b.pid, 'pure-listen'))} + (('sender', a.pid, 'pure-send'), ('receiver', b.pid, 'pure-listen'))} import http.client connection = http.client.HTTPConnection('127.0.0.1', port_b, timeout=30) first = None @@ -507,7 +507,7 @@ def drain_pure(work, n, size): seconds = now - first[0] row = {'messages': n, 'item_bytes': size, 'first_seen': first[1], 'seconds': round(seconds, 4), 'per_second': round((n - first[1]) / seconds) if seconds else None} - for name, pid, thread in (('sender', a.pid, 'pure-link'), ('receiver', b.pid, 'pure-listen')): + for name, pid, thread in (('sender', a.pid, 'pure-send'), ('receiver', b.pid, 'pure-listen')): total, link = thread_cpu(pid, None) - cpu[name][0], thread_cpu(pid, thread) - cpu[name][1] row[f'{name}_cpu_us_per_message'] = round(total * 1000 / n, 2) row[f'{name}_link_cpu_us_per_message'] = round(link * 1000 / n, 2) diff --git a/research/experiments/019-idle-links/transport.c b/research/experiments/019-idle-links/transport.c index 3a531ef..4560850 100644 --- a/research/experiments/019-idle-links/transport.c +++ b/research/experiments/019-idle-links/transport.c @@ -191,6 +191,7 @@ int main(int argc, char **argv) { if (!window) /* a backlog: all queued before the link starts */ for (; first <= count; ++first) if (!pure_link_queue(link, first, payload, payload_size)) return fputs("transport: cannot queue\n", stderr), 1; + if (!window) pure_link_push(link); began = now_ns(); uint64_t cpu_link = 0, cpu_listen = thread_cpu("pure-listen"), cpu_all = thread_cpu(NULL); uint64_t end = limit > 0 ? began + (uint64_t)(limit * 1e9) : UINT64_MAX; @@ -204,13 +205,14 @@ int main(int argc, char **argv) { if (now_ns() >= end) break; if (queued_at) queued_at[seq] = now_ns(); if (!pure_link_queue(link, seq, payload, payload_size)) return fputs("transport: cannot queue\n", stderr), 1; + pure_link_push(link); } while (pure_link_delivered(link) < count && now_ns() < end) { struct timespec pause = {0, 50000}; nanosleep(&pause, NULL); } double seconds = (double)((end < now_ns() ? end : now_ns()) - began) / 1e9; - uint64_t link_ns = thread_cpu("pure-link") - cpu_link, listen_ns = thread_cpu("pure-listen") - cpu_listen, + uint64_t link_ns = thread_cpu("pure-send") - cpu_link, listen_ns = thread_cpu("pure-listen") - cpu_listen, all_ns = thread_cpu(NULL) - cpu_all; PureLinkCounts sent, got; pure_link_counts(link, &sent); diff --git a/serve/serve.c b/serve/serve.c index b5112d2..86e38f8 100644 --- a/serve/serve.c +++ b/serve/serve.c @@ -1829,6 +1829,7 @@ static void feed_outlets(Executor *ex) { delivered = wire ? pure_wire_delivered(wire) : link ? pure_link_delivered(link) : local_delivered(local); size_t held = wire ? pure_wire_held(wire) : link ? pure_link_held(link) : local_held(local), at = ex->outlet_from[i]; at = at > gone ? at - (size_t)gone : 0; + uint64_t before = queued; for (; at < ex->app.sent_saved; ++at) { uint32_t of; uint64_t seq; @@ -1844,6 +1845,7 @@ static void feed_outlets(Executor *ex) { queued = seq; held += payload.size; } + if (link && queued != before) pure_link_push(link); /* the turn's together */ ex->outlet_from[i] = at; } uint64_t now = ex->local_count ? now_ms() : 0; -- cgit v1.2.3