Repository navigation
Add message queues - #23
Merged
Merged
Conversation
protothread_queue.h provides what Go calls channels: a first-in, first-out buffer of fixed capacity in storage the caller supplies, so no allocator. PT_QUEUE_DEFINE(name, type) generates a type-safe queue name_t and its functions, in the manner of <sys/tree.h>; pt_queue_t is a predefined queue of void *. The blocking operations, send, receive and wait_send, are protothread functions called through pt_call(), like any other blocking function. A generator macro can't define the convenience macros that semaphores and locks use to hide pt_call(), and calling it explicitly is uniform anyway. Items go in and out through pointers, so each is copied exactly once each way. Passing by value cost nothing for the inlined try_send(), but the blocking send can't be inlined (it contains a computed goto), so a 64-byte item was copied onto the stack at every call and again on every resume. Only the element copies are generated per type. Indices, counts and the send counter are shared, untyped code. Senders and receivers wait on different addresses in the queue, so a send wakes only receivers and a receive only senders, and every wakeup is a broadcast: a send must wake both receive() and wait_send() waiters, and pt_signal() could pick a wait_send() waiter and strand the receiver. There is no select. Instead, as in Pike's fan-in pattern, several senders share one queue of tagged messages, and count(), at(), remove() and wait_send() let a receiver take the kind it wants while the others keep their order. wait_send() blocks until the send counter moves, so messages already passed over don't wake it again. try_send() is safe from an interrupt handler with PT_CRITICAL_* defined, and the blocking calls wait with pt_wait_until(), so a send from an interrupt can't be lost. The interrupt test fires try_send() at each point where a blocked receiver masks interrupts; with receive() written as a plain wait loop it fails in all 24 critical-section configurations. Generated functions are marked __unused__, since an instance needn't use them all and clang warns about unused static functions expanded in a .c file. Item copies of 16 bytes or more can become a memcpy call under clang for Cortex-M; the README says so, and the freestanding CI check uses a queue of unsigned, which never does. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01SR7v9BVsFrHiEhQBSqZmAs
The same example rewritten with protothread_queue.h, right after the mailbox version. Each protothread shrinks to its loop, and the text points out what changed: in the mailbox version the data stays put and the protothreads come to it, while with a queue the protothreads stay put and the data moves from one to the other. With four slots the producer runs ahead, so passing 100 values takes 51 protothread runs instead of 201; both counts measured from the README code as written. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01SR7v9BVsFrHiEhQBSqZmAs
Pike makes a point of Go channels being sendable over channels, chiefly for request and reply: the client sends its question with a channel for the answer. A Go channel is a reference, so the equivalent here is a queue whose items hold a pointer to another queue. test_queue_of_queues() has one server answering three clients, each through its own reply queue; the README shows the pattern, compiled as written, and warns against sending a queue by value, which would make a second queue sharing the first one's storage. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01SR7v9BVsFrHiEhQBSqZmAs
The README described a timeout in one sentence: a protothread that sleeps and then sends TIMEOUT into the same queue, which is Go's time.After(). That hid a catch. Go's timeout has its own channel, so when the reply wins it is simply forgotten; here it shares the reply's queue. So when the reply wins, the timer has to be stopped, which also frees its context for the next wait, and because it may already have sent, each wait is numbered and a TIMEOUT from an earlier one is skipped. The README now shows the helper and the waiting side, both compiled as written. test_queue_timeout() covers the reply winning, the timeout winning, and the timer firing just before the reply so its TIMEOUT is left queued behind it. Removing the generation check fails the test (the leftover TIMEOUT isn't skipped), and so does removing the cancel and kill (a stray TIMEOUT is left in the queue). Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01SR7v9BVsFrHiEhQBSqZmAs
remove() always closed the gap by moving the older items up one slot and advancing head, so removing near the front was cheap and removing near the back moved nearly the whole queue. That back end is the common case for selective receive: the message a receiver waits for, a reply say, is often the newest, behind several it is skipping. Now the newer items move down instead when they are fewer, so taking either end moves nothing and the worst case is half the queue. test_queue_remove() checks which items moved by comparing the addresses at() returns, across the wrap of the ring, removing the newest, the oldest, and gaps nearer each end. Against the previous remove() it fails on the first of those, removing the newest, which moved all four older items. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01SR7v9BVsFrHiEhQBSqZmAs
Larry's suggestion: send, receive and wait_send were generated protothread functions, called through pt_call() with a pt_queue_env_t in the caller's context. They are now macros, pt_queue_send(), pt_queue_receive() and pt_queue_wait_send(), that wait in the caller's own context as pt_wait() does. No pt_call(), no queue env in the context structure, and no function call per resume. The element type comes from the queue's slots, so the macros type-check the copy without knowing the queue's name and work on any queue; the non-blocking calls remain generated functions, which an interrupt handler needs. wait_send now takes the send count the caller read before looking at the queue, from the new name_sent(). That fixes a race in the previous version: it read the count on entry, after the receiver had scanned, so an interrupt sending the wanted message in between was already counted and the receiver slept through it. A new interrupt test runs a selective receiver past an unwanted item while an interrupt sends the wanted one at each masking point; with the count read after scanning it fails in all 24 critical-section configurations. pt_call_waited() doesn't apply to the macros. With queues it's seldom needed, since what a protothread needs arrives in the message rather than in memory others may have changed, and the README says so. The pt_call() version is kept on the queue-pt-call branch. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01SR7v9BVsFrHiEhQBSqZmAs
wait_send read as "wait, then send", next to pt_queue_send(). Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01SR7v9BVsFrHiEhQBSqZmAs
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01SR7v9BVsFrHiEhQBSqZmAs
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Adds
protothread_queue.h: message queues, what Go calls channels, built on the core the same way the semaphore, lock and timers are.A queue is a fixed-capacity ring buffer in storage you supply, so there's no allocator.
PT_QUEUE_DEFINE(name, type)generates a type-safe queue and its non-blocking functions, along the lines of<sys/tree.h>, andpt_queue_tis a predefined queue ofvoid *.Design choices worth a look:
pt_wait()does, so there's nopt_call()and nothing extra in the context structure. They take the element type from the queue's slots, so the copy is type-checked and one set of macros works on any queue. (An earlier version made them generated protothread functions called throughpt_call(); it's kept on thequeue-pt-callbranch.) The non-blocking calls stay generated functions, since an interrupt handler needs a real function that returns a result.wait_newwaiters, andpt_signal()could pick thewait_newone and strand the receiver. Senders and receivers do wait on separate addresses, so a send only wakes receivers.pt_queue_wait_new()takes a send count the caller read before looking at the queue. Reading it inside the wait, as the earlier version did, let an interrupt's send slip in between the receiver's scan and its wait, and be slept through.pt_call_waited()doesn't apply to queue calls. With queues it's seldom needed, since what a protothread needs to know arrives in the message rather than in memory others may have changed.select. The README shows the alternatives: fan-in to one queue of tagged messages (the pattern Pike shows in "Go Concurrency Patterns" just beforeselect); a timeout as a protothread that sleeps and then sends, with how to discard a stale one; selective receive withcount(),at(),remove()andwait_new; and queues of queues for request and reply.remove()closes the gap from whichever side is shorter, so taking the oldest or the newest moves nothing.The README also gets a queue version of producer/consumer right after the mailbox one, framed around what moves: with the mailbox the data stays put and the protothreads come to it, with a queue the protothreads stay put and the data moves. With four slots, passing 100 values takes 51 protothread runs instead of 201.
Testing:
void *instance.try_send()at every point where a protothread masks interrupts, for a blocked receiver and for a selective receiver. Each fails in all 24 critical-section configurations against the broken variant: a plain wait loop for the receiver, and a send count read after scanning for the selective receiver.One caveat, documented: clang targeting Cortex-M turns struct copies of 16 bytes or more into a
__aeabi_memcpycall at-Os. That's the compiler, and user code hits it too, but this is the first library code that copies a user's struct. Queues of integers or pointers never need it.🤖 Generated with Claude Code
https://claude.ai/code/session_01SR7v9BVsFrHiEhQBSqZmAs