diff options
Diffstat (limited to 'link')
| -rw-r--r-- | link/link.c | 37 | ||||
| -rw-r--r-- | link/link.h | 13 |
2 files changed, 30 insertions, 20 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 *); |