From 9b5514ccc7043439c3ddff432c09d2dda53d5aad Mon Sep 17 00:00:00 2001 From: Larry Ruane Date: Tue, 6 Oct 2026 11:44:31 -0600 Subject: [PATCH 1/8] Add message queues 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 ; 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 Claude-Session: https://claude.ai/code/session_01SR7v9BVsFrHiEhQBSqZmAs --- .github/workflows/ci.yml | 28 ++++- CHANGELOG.md | 8 ++ CMakeLists.txt | 1 + README.md | 88 +++++++++++++- protothread.h | 3 +- protothread_queue.h | 225 +++++++++++++++++++++++++++++++++++ protothread_test.c | 247 +++++++++++++++++++++++++++++++++++++++ 7 files changed, 590 insertions(+), 10 deletions(-) create mode 100644 protothread_queue.h diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 1fe7413..7e9d0bb 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -156,10 +156,15 @@ jobs: #include "protothread_sem.h" #include "protothread_lock.h" #include "protothread_timer.h" + #include "protothread_queue.h" + PT_QUEUE_DEFINE(uq, unsigned) typedef struct { pt_thread_t pt_thread ; pt_func_t pt_func ; pt_sem_env_t se ; pt_lock_env_t le ; pt_timer_env_t te ; + pt_queue_env_t qe ; unsigned v ; } ctx_t ; + static unsigned qslots[4] ; + static uq_t q ; static struct protothread_s state ; static pt_timers_t timers ; static pt_lock_t lock ; @@ -173,10 +178,13 @@ jobs: pt_lock_acquire_write(x, &x->le, &lock) ; pt_lock_release_write(&x->le, &lock) ; pt_sleep(x, &x->te, &timers, 10) ; + pt_call(x, uq_send, &x->qe, &q, &x->v) ; + pt_call(x, uq_receive, &x->qe, &q, &x->v) ; return PT_DONE ; } void app_init(void) { protothread_init(&state) ; pt_lock_init(&lock) ; + uq_init(&q, qslots, 4) ; pt_timers_init(&timers, 0) ; pt_create(&state, &c.pt_thread, thr, &c) ; } @@ -198,7 +206,7 @@ jobs: # The headers are C, but a C++ project should be able to include them. # This is a narrow claim and easy to break with one implicit void* # conversion, so compile a real protothread -- yields, nesting through the - # semaphore, lock and timer helpers, and the computed goto -- as C++. + # semaphore, lock, timer and queue helpers, and the computed goto -- as C++. # Three internal functions require the caller to already be in a critical # section. PT_CRITICAL_* are no-ops by default, so a future call from thread # context would break that silently and nothing would complain. Define them @@ -396,13 +404,21 @@ jobs: #include "protothread_sem.h" #include "protothread_lock.h" #include "protothread_timer.h" + #include "protothread_queue.h" + + PT_QUEUE_DEFINE(intq, int) typedef struct { pt_thread_t pt_thread ; pt_func_t pt_func ; pt_sem_env_t se ; pt_lock_env_t le ; pt_timer_env_t te ; + pt_queue_env_t qe ; int i ; + int v ; } ctx_t ; + static int qslots[2] ; + static intq_t q ; + static struct protothread_s state ; static pt_timers_t timers ; static pt_lock_t lock ; @@ -423,6 +439,10 @@ jobs: pt_lock_acquire_write(x, &x->le, &lock) ; pt_lock_release_write(&x->le, &lock) ; pt_sleep(x, &x->te, &timers, 10) ; + x->v = 5 ; + pt_call(x, intq_send, &x->qe, &q, &x->v) ; + x->v = 0 ; + pt_call(x, intq_receive, &x->qe, &q, &x->v) ; return PT_DONE ; } @@ -432,14 +452,15 @@ jobs: pt_time_t t ; protothread_init(&state) ; pt_lock_init(&lock) ; + intq_init(&q, qslots, 2) ; pt_timers_init(&timers, 0) ; pt_create(&state, &c.pt_thread, thr, &c) ; for (t = 0; t <= 20; t++) { pt_timer_run(&state, &timers, t) ; while (protothread_run(&state)) { } } - if (c.i != 3) { - std::printf("FAIL: i = %d\n", c.i) ; + if (c.i != 3 || c.v != 5) { + std::printf("FAIL: i = %d, v = %d\n", c.i, c.v) ; return 1 ; } protothread_deinit(&state) ; @@ -469,6 +490,7 @@ jobs: set -euo pipefail cmake --install build test -f inst/include/protothread_timer.h + test -f inst/include/protothread_queue.h export PKG_CONFIG_PATH="$PWD/inst/lib/pkgconfig" pkg-config --exists protothread echo "version: $(pkg-config --modversion protothread)" diff --git a/CHANGELOG.md b/CHANGELOG.md index 30c2433..69d313e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -23,6 +23,14 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 `pt_timer_cancel()` is for one that might be sleeping. Without it such a kill left the lock granted to a dead thread and never released. +- `protothread_queue.h`: message queues, what Go calls channels. A queue is a + fixed-capacity ring buffer in storage the caller supplies, so it needs no + allocator, and `PT_QUEUE_DEFINE(name, type)` makes it type-safe; `pt_queue_t` + is a predefined queue of `void *`. Sending and receiving block when full and + empty, `try_send()` is safe to call from an interrupt handler, and `count()`, + `at()`, `remove()` and `wait_send()` let a receiver take one kind of message + while others wait, which with a tagged union stands in for Go's `select`. + ### Changed - The blocking macros name their goto labels with `__COUNTER__` instead of diff --git a/CMakeLists.txt b/CMakeLists.txt index cf7a130..2bf8c7a 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -79,6 +79,7 @@ configure_file( install (TARGETS pttest DESTINATION bin) install (FILES protothread.h protothread_lock.h protothread_sem.h protothread_timer.h + protothread_queue.h DESTINATION include) # Install pkgconfig files to libdata on BSD, otherwise lib diff --git a/README.md b/README.md index 9c292e0..817e5c2 100644 --- a/README.md +++ b/README.md @@ -19,7 +19,7 @@ I wrote this from scratch, and it is not compatible with other versions of proto * **Blocking functions nest with an ordinary call.** A protothread function can `pt_call()` another protothread function, which can block, to any depth, passing it arguments and carrying on when it returns, so you can factor blocking code into subroutines the way you would anywhere else. Dunkels' library nests too, with `PT_SPAWN()`; what differs is how a blocked thread gets resumed, the third point below. * **You can block anywhere**, including inside a `switch` statement. Implementations built on Duff's device cannot, because they have already spent the `switch`. - * **A real scheduler, with wait channels.** `pt_wait()`/`pt_signal()`/`pt_broadcast()` on an arbitrary address deliberately mirror condition variables (`pthread_cond_wait()`, except a `pthread_cond_t` variable isn't needed) and the classic Unix kernel `sleep()`/`wakeup()`. Threads live on a run list or a wait list, and a blocked thread does not run at all until its channel is signalled. Dunkels' library leaves scheduling to the application: a blocked protothread is simply called again later, and re-tests its condition each time, re-running every level of any nesting to get there. Semaphores, reader-writer locks and timers are built on top of it, in optional headers of their own. + * **A real scheduler, with wait channels.** `pt_wait()`/`pt_signal()`/`pt_broadcast()` on an arbitrary address deliberately mirror condition variables (`pthread_cond_wait()`, except a `pthread_cond_t` variable isn't needed) and the classic Unix kernel `sleep()`/`wakeup()`. Threads live on a run list or a wait list, and a blocked thread does not run at all until its channel is signalled. Dunkels' library leaves scheduling to the application: a blocked protothread is simply called again later, and re-tests its condition each time, re-running every level of any nesting to get there. Semaphores, reader-writer locks, timers and message queues are built on top of it, in optional headers of their own. The cost of nesting and arbitrary blocking is that this implementation uses [gcc label variables](http://gcc.gnu.org/onlinedocs/gcc/Labels-as-Values.html), which is the one part of it that is not standard C, so it requires **gcc or clang** (see [Compiler requirements](#compiler-requirements)). Dunkels' `switch`-based version is portable to any C compiler; this one trades that for a better interface. @@ -39,7 +39,7 @@ The cost of nesting and arbitrary blocking is that this implementation uses [gcc | memory per thread | 56 bytes | 16,384 bytes | **293x** | Protothreads are faster here because they do less: no kernel transition, no scheduler, no per-thread stack, no mutex locking. POSIX threads provide preemption and real parallelism, which protothreads do not. See [Memory overhead and performance](#memory-overhead-and-performance-benchmarks) for a detailed comparison and [Protothreads on a multi-core system](#protothreads-on-a-multi-core-system) for using both together. - * optional semaphores, reader-writer locks and timers, each in its own header and built entirely on the core + * optional semaphores, reader-writer locks, timers and message queues, each in its own header and built entirely on the core * about 1000 lines of test code * gdb (debugger) macros to print the stack traces of a given protothread or all protothreads. * a CMake find module, `FindPROTOTHREAD.cmake`: copy it into another CMake project, and `find_package(PROTOTHREAD)` will locate the installed headers @@ -101,13 +101,14 @@ For contrast, POSIX [`pthread.h`](https://pubs.opengroup.org/onlinepubs/96999197 Nearly all of that exists because a pthread can be interrupted between any two instructions. A protothread can't. It runs until it blocks, at a line you can point at, so there is no priority to invert, no cancellation point to reason about, and no per-thread stack to size. Those functions aren't missing here so much as they have nothing to do. -These three headers built on it are entirely optional, but provide commonly-used abstractions: +These four headers built on it are entirely optional, but provide commonly-used abstractions: * `protothread_sem.h` -- counting semaphores * `protothread_timer.h` -- sleeping, driven by a clock you supply * `protothread_lock.h` -- reader-writer locks + * `protothread_queue.h` -- message queues, what Go calls channels -Each is ordinary protothread code over `pt_wait()` and `pt_broadcast()`, as much a demonstration of what the core can express as a facility to use. They are described under [Built on top](#built-on-top-semaphores-timers-and-locks). +Each is ordinary protothread code over `pt_wait()` and `pt_broadcast()`, as much a demonstration of what the core can express as a facility to use. They are described under [Built on top](#built-on-top-semaphores-timers-locks-and-queues). ## Threads that share one stack ## @@ -784,7 +785,7 @@ These are macros (designed to look and act like function calls) whose first argu > `bool_t pt_call_waited(struct context_t *c)` > -> Returns TRUE if the most recent `pt_call()` blocked (either directly in the called function, or in a function that it called, recursively). If function **A** calls **B** and **B** blocks, then when it finally returns to **A**, it's sometimes helpful for **A** to know that other threads might have run, so it should reevaluate the state of the world. But if **B** didn't block, then **A** knows that only a limited change of state (namely, whatever **B** might do) could have occurred. This works after acquiring a semaphore or a lock too, since those are `pt_call()`s; see [Built on top](#built-on-top-semaphores-timers-and-locks) for an example. +> Returns TRUE if the most recent `pt_call()` blocked (either directly in the called function, or in a function that it called, recursively). If function **A** calls **B** and **B** blocks, then when it finally returns to **A**, it's sometimes helpful for **A** to know that other threads might have run, so it should reevaluate the state of the world. But if **B** didn't block, then **A** knows that only a limited change of state (namely, whatever **B** might do) could have occurred. This works after acquiring a semaphore or a lock too, since those are `pt_call()`s; see [Built on top](#built-on-top-semaphores-timers-locks-and-queues) for an example. > `void pt_reset(struct context_t *c)` > @@ -862,7 +863,7 @@ None of these schedule or wake a protothread, so none has the lost-wakeup hazard > > To prevent a sequence of protothread executions from holding onto the CPU for too long, the function can limit the number of times it calls `protothread_run()`; for example it may run no more than 20 threads before returning to the main scheduler to let other things (outside of protothreads) run. But if it does so (if the last call to `protothread_run()` returns TRUE), it should reschedule itself because there is still work to do. -## Built on top: semaphores, timers and locks ## +## Built on top: semaphores, timers, locks and queues ## None of this is needed to use protothreads, and none of it is part of the core. Each header is ordinary protothread code over `pt_wait()` and `pt_broadcast()`, with no privileged access to the scheduler -- closer in spirit to the programs in [`demo/`](demo) than to the API above, and worth reading as examples of how to build primitives of your own. They are tested and maintained like the rest of the library. @@ -981,6 +982,81 @@ if (pt_call_waited(c)) { Releasing never blocks, so if acquiring didn't either, nothing else ran in between and the upgrade was atomic. That's the case when you were the only reader and nobody was queued. Otherwise you still end up holding the write lock, but anything you read under the read lock may be stale. The answer can only be wrong in the safe direction: if another reader was holding the lock, it reports a wait even though nothing changed. +### Message queues ### + +`#include "protothread_queue.h"`. A queue is a first-in, first-out buffer of fixed capacity in storage you supply, so it needs no allocator; Go calls the same thing a channel. Queues are type-safe: `PT_QUEUE_DEFINE(name, type)` generates a queue type `name_t` and its functions, much as `` generates trees, and `pt_queue_t` is a predefined queue of `void *`. + +```c +typedef struct { int kind; int value; } msg_t; +PT_QUEUE_DEFINE(msgq, msg_t) /* msgq_t, msgq_send() and the rest */ + +msg_t slots[8]; +msgq_t q; +msgq_init(&q, slots, 8); +``` + +Sending and receiving can block, so like any blocking function they are protothread functions, called through `pt_call()`, with a `pt_queue_env_t` in your context structure. Items go in and come out through pointers, so each is copied once each way: + +```c +pt_call(c, msgq_send, &c->queue_env, &q, &c->msg); /* blocks while full */ +pt_call(c, msgq_receive, &c->queue_env, &q, &c->msg); /* blocks while empty */ +``` + +> `name_send(pt_queue_env_t *queue_env, name_t *q, type const *item)` +> +> Block while the queue is full, then append a copy of `*item`. Call through `pt_call()`. + +> `name_receive(pt_queue_env_t *queue_env, name_t *q, type *item)` +> +> Block while the queue is empty, then remove the oldest item into `*item`. Call through `pt_call()`. + +> `name_wait_send(pt_queue_env_t *queue_env, name_t *q)` +> +> Block until another item is sent, without taking anything. Call through `pt_call()`. + +> `bool_t name_try_send(protothread_t, name_t *q, type const *item)` +> +> `bool_t name_try_receive(protothread_t, name_t *q, type *item)` +> +> The same as send and receive, but return FALSE instead of blocking. `try_send()` is safe to call from an interrupt handler once the `PT_CRITICAL_*` macros are defined, and a protothread blocked in `receive()` or `wait_send()` cannot miss what it sends. + +> `unsigned int name_count(name_t const *q)` +> +> `type *name_at(name_t *q, unsigned int i)` +> +> `bool_t name_remove(protothread_t, name_t *q, unsigned int i, type *item)` +> +> Look inside the queue without blocking: the number of items, a pointer to the `i`-th oldest (0 is the next to be received), and removal of the `i`-th oldest, which keeps the rest in order. `item` may be NULL to discard. A pointer from `at()` stays valid until something is removed. + +Like the semaphore, a queue isn't strictly fair: a protothread that finds it non-empty takes an item even if another was woken for it first, for the same reason, to avoid convoys. Every wakeup is a broadcast, and senders and receivers wait on different channels, so sending wakes only receivers and receiving wakes only senders. Items are copied by assignment, and for an item larger than a few words the compiler may emit a call to `memcpy` (clang targeting Cortex-M does so from 16 bytes at `-Os`). Every embedded toolchain provides it, but if you are avoiding it, queue pointers instead. + +**There is no `select`.** Go's `select` waits on several channels at once; a protothread waits on one channel at a time, and that stays true here. Pike's [Go Concurrency Patterns](https://www.youtube.com/watch?v=f6kdp27TYZs) talk shows the usual alternative just before it introduces `select`: *fan-in*, where everything a receiver might want arrives on one queue. With different kinds of message, the item is a tagged union: + +```c +typedef struct { + enum { REPLY, STATUS, TIMEOUT } kind; + union { struct reply reply; struct status status; } u; +} msg_t; +``` + +A timeout is just another sender: a small protothread that calls `pt_sleep()` and then sends a `TIMEOUT` message, which is all Go's `time.After()` does. + +When a receiver wants one kind of message while others wait their turn, it can look before it takes. `wait_send()` blocks only until something new arrives, so messages it has already passed over don't wake it again: + +```c +for (;;) { + for (c->i = 0; c->i < msgq_count(&q); c->i++) { + if (msgq_at(&q, c->i)->kind == REPLY) { + msgq_remove(pt_get_pt(c), &q, c->i, &c->msg); + goto got_reply; + } + } + pt_call(c, msgq_wait_send, &c->queue_env, &q); +} +``` + +That is linear in the length of the queue, which is fine for the queues a microcontroller has. + ## License ## MIT. See [LICENSE](LICENSE). diff --git a/protothread.h b/protothread.h index 3dcbb2c..5d1524d 100644 --- a/protothread.h +++ b/protothread.h @@ -54,7 +54,8 @@ * PT_SIGNAL_WAKES_ALL, pt_assert * Version PT_VERSION_{MAJOR,MINOR,PATCH,NUMBER,STRING}, * PT_VERSION_AT_LEAST - * Companions protothread_sem.h, protothread_lock.h, protothread_timer.h + * Companions protothread_sem.h, protothread_lock.h, protothread_timer.h, + * protothread_queue.h */ /* Version, semantic versioning (https://semver.org). This is the single diff --git a/protothread_queue.h b/protothread_queue.h new file mode 100644 index 0000000..5fb8fca --- /dev/null +++ b/protothread_queue.h @@ -0,0 +1,225 @@ +/**************************************************************/ +/* PROTOTHREAD_QUEUE.H */ +/* https://github.com/LarryRuane/protothread */ +/* Copyright (c) 2008-present Larry Ruane */ +/* Distributed under the MIT software license, see the accompanying */ +/* file LICENSE or https://opensource.org/licenses/MIT. */ +/* SPDX-License-Identifier: MIT */ +/* Message queues (what Go calls channels) */ +/**************************************************************/ +#ifndef PROTOTHREAD_QUEUE_H +#define PROTOTHREAD_QUEUE_H + +#include "protothread.h" + +/* Message queue. The public API; pt_i_ names are internal. + * + * PT_QUEUE_DEFINE(name, type) generates a queue of named name_t. + * The three blocking functions are protothread functions, so like any other + * they are called through pt_call(), as pt_call(c, name_send, env, q, item): + * name_send(queue_env, q, item_p) block while full, then append + * name_receive(queue_env, q, item_p) block while empty, then take oldest + * name_wait_send(queue_env, q) block until another item is sent + * The rest never block, and are called directly: + * name_init(q, slots, capacity) use slots[capacity] as storage + * name_try_send(s, q, item_p) append if room; true if sent + * name_try_receive(s, q, item_p) take oldest if any; true if taken + * name_count(q) number of items queued + * name_at(q, i) pointer to the i-th oldest item + * name_remove(s, q, i, item_p) take the i-th oldest, keep order + * pt_queue_env_t one per protothread that blocks + * + * pt_queue_t and pt_queue_*() are a predefined queue of void *. + * + * Items go in and come out through pointers, so each is copied exactly once + * each way, by assignment, inside a critical section; for a large structure, + * consider queueing a pointer to it instead. try_send() may be called from an + * interrupt handler when the PT_CRITICAL_* macros are defined; everything + * else is for thread context. + */ + +/* One per protothread that blocks on a queue, in its context structure */ +typedef struct pt_queue_env_s { + pt_func_t pt_func ; + unsigned int seen ; /* the send count wait_send() started at */ +} pt_queue_env_t ; + +/* The untyped part of every queue: indices and counts, never the items. + * Receivers wait on &sent, senders wait on &head, so that a send wakes only + * receivers and a receive wakes only senders. + */ +typedef struct pt_i_queue_s { + unsigned int head ; /* slot of the oldest item */ + unsigned int count ; /* items queued */ + unsigned int capacity ; /* slots */ + unsigned int sent ; /* sends so far, wrapping */ +} pt_i_queue_t ; + +static inline void +pt_i_queue_init(pt_i_queue_t * const q, unsigned int const capacity) +{ + pt_assert(capacity > 0) ; + q->head = 0 ; + q->count = 0 ; + q->capacity = capacity ; + q->sent = 0 ; +} + +/* the slot holding the i-th oldest item (or, at i == count, the next free) */ +static inline unsigned int +pt_i_queue_slot(pt_i_queue_t const * const q, unsigned int const i) +{ + unsigned int const j = q->head + i ; + return j < q->capacity ? j : j - q->capacity ; +} + +/* the caller has just filled the next free slot */ +static inline void +pt_i_queue_appended(pt_i_queue_t * const q) +{ + PT_CRITICAL_ASSERT() ; + q->count++ ; + q->sent++ ; +} + +/* the caller has just copied out the oldest item */ +static inline void +pt_i_queue_took_oldest(pt_i_queue_t * const q) +{ + PT_CRITICAL_ASSERT() ; + q->head = pt_i_queue_slot(q, 1) ; + q->count-- ; +} + +/* An instance needn't use every function it generates, and clang warns about + * unused static functions expanded outside a header. + */ +#define PT_I_MAYBE_UNUSED __attribute__((__unused__)) + +/* Generate a queue of named name_t, and its functions. The blocking + * ones are a wait on the untyped part followed by the non-blocking one, and + * retry if another protothread got there first. + */ +#define PT_QUEUE_DEFINE(name, type) \ +\ +typedef struct name ## _s { \ + pt_i_queue_t q ; \ + type * slots ; \ +} name ## _t ; \ +\ +static inline PT_I_MAYBE_UNUSED void \ +name ## _init(name ## _t * const q, type * const slots, \ + unsigned int const capacity) \ +{ \ + pt_i_queue_init(&q->q, capacity) ; \ + q->slots = slots ; \ +} \ +\ +static inline PT_I_MAYBE_UNUSED unsigned int \ +name ## _count(name ## _t const * const q) \ +{ \ + return q->q.count ; \ +} \ +\ +static inline PT_I_MAYBE_UNUSED type * \ +name ## _at(name ## _t * const q, unsigned int const i) \ +{ \ + pt_assert(i < q->q.count) ; \ + return &q->slots[pt_i_queue_slot(&q->q, i)] ; \ +} \ +\ +static inline PT_I_MAYBE_UNUSED bool_t \ +name ## _try_send(protothread_t const s, name ## _t * const q, \ + type const * const item) \ +{ \ + bool_t sent = false ; \ + pt_i_critical_t const saved = PT_CRITICAL_ENTER() ; \ + if (q->q.count < q->q.capacity) { \ + q->slots[pt_i_queue_slot(&q->q, q->q.count)] = *item ; \ + pt_i_queue_appended(&q->q) ; \ + sent = true ; \ + } \ + PT_CRITICAL_EXIT(saved) ; \ + if (sent) { \ + pt_broadcast(s, &q->q.sent) ; \ + } \ + return sent ; \ +} \ +\ +static inline PT_I_MAYBE_UNUSED bool_t \ +name ## _remove(protothread_t const s, name ## _t * const q, \ + unsigned int const i, type * const item) \ +{ \ + bool_t taken = false ; \ + pt_i_critical_t const saved = PT_CRITICAL_ENTER() ; \ + if (i < q->q.count) { \ + unsigned int k ; \ + if (item) { \ + *item = q->slots[pt_i_queue_slot(&q->q, i)] ; \ + } \ + /* close the gap by moving the older items up by one */ \ + for (k = i; k > 0; k--) { \ + q->slots[pt_i_queue_slot(&q->q, k)] = \ + q->slots[pt_i_queue_slot(&q->q, k - 1)] ; \ + } \ + pt_i_queue_took_oldest(&q->q) ; \ + taken = true ; \ + } \ + PT_CRITICAL_EXIT(saved) ; \ + if (taken) { \ + pt_broadcast(s, &q->q.head) ; \ + } \ + return taken ; \ +} \ +\ +static inline PT_I_MAYBE_UNUSED bool_t \ +name ## _try_receive(protothread_t const s, name ## _t * const q, \ + type * const item) \ +{ \ + return name ## _remove(s, q, 0, item) ; \ +} \ +\ +static inline PT_I_MAYBE_UNUSED pt_t \ +name ## _send(pt_queue_env_t * const c, name ## _t * const q, \ + type const * const item) \ +{ \ + pt_resume(c) ; \ + for (;;) { \ + pt_wait_until(c, &q->q.head, q->q.count < q->q.capacity) ; \ + if (name ## _try_send(pt_get_pt(c), q, item)) { \ + return PT_DONE ; \ + } \ + } \ +} \ +\ +static inline PT_I_MAYBE_UNUSED pt_t \ +name ## _receive(pt_queue_env_t * const c, name ## _t * const q, \ + type * const item) \ +{ \ + pt_resume(c) ; \ + for (;;) { \ + pt_wait_until(c, &q->q.sent, q->q.count > 0) ; \ + if (name ## _try_receive(pt_get_pt(c), q, item)) { \ + return PT_DONE ; \ + } \ + } \ +} \ +\ +/* the send count is read with interrupts masked, since on a small target \ + * an unsigned int can take two loads */ \ +static inline PT_I_MAYBE_UNUSED pt_t \ +name ## _wait_send(pt_queue_env_t * const c, name ## _t * const q) \ +{ \ + pt_resume(c) ; \ + { \ + pt_i_critical_t const saved = PT_CRITICAL_ENTER() ; \ + c->seen = q->q.sent ; \ + PT_CRITICAL_EXIT(saved) ; \ + } \ + pt_wait_until(c, &q->q.sent, q->q.sent != c->seen) ; \ + return PT_DONE ; \ +} + +PT_QUEUE_DEFINE(pt_queue, void *) + +#endif /* PROTOTHREAD_QUEUE_H */ diff --git a/protothread_test.c b/protothread_test.c index 52500cb..373e436 100644 --- a/protothread_test.c +++ b/protothread_test.c @@ -25,6 +25,7 @@ #include "protothread_sem.h" #include "protothread_lock.h" #include "protothread_timer.h" +#include "protothread_queue.h" /******************************************************************************/ @@ -1451,6 +1452,203 @@ test_lock_cancel(void) /******************************************************************************/ +PT_QUEUE_DEFINE(intq, int) + +typedef struct { + pt_thread_t pt_thread ; + pt_func_t pt_func ; + pt_queue_env_t queue_env ; + intq_t * q ; + int i ; + int got ; + int waited ; /* times a send or receive blocked */ +} intq_context_t ; + +static pt_t +intq_producer_thr(env_t const env) +{ + intq_context_t * const c = env ; + pt_resume(c) ; + for (c->i = 1; c->i <= 100; c->i++) { + pt_call(c, intq_send, &c->queue_env, c->q, &c->i) ; + c->waited += pt_call_waited(c) ; + } + return PT_DONE ; +} + +static pt_t +intq_consumer_thr(env_t const env) +{ + intq_context_t * const c = env ; + pt_resume(c) ; + for (c->i = 1; c->i <= 100; c->i++) { + pt_call(c, intq_receive, &c->queue_env, c->q, &c->got) ; + c->waited += pt_call_waited(c) ; + check(c->got == c->i) ; + } + return PT_DONE ; +} + +/* the producer/consumer example, through a queue of four */ +static void +test_queue_fifo(void) +{ + protothread_t const pt = protothread_create() ; + int slots[4] ; + intq_t q ; + intq_context_t p, c ; + + intq_init(&q, slots, 4) ; + memset(&p, 0, sizeof(p)) ; + memset(&c, 0, sizeof(c)) ; + p.q = c.q = &q ; + pt_create(pt, &c.pt_thread, intq_consumer_thr, &c) ; + pt_create(pt, &p.pt_thread, intq_producer_thr, &p) ; + while (protothread_run(pt)) ; + check(p.i == 101 && c.i == 101) ; + check(intq_count(&q) == 0) ; + /* both sides had to wait for the other at some point */ + check(p.waited > 0 && c.waited > 0) ; + protothread_free(pt) ; +} + +/* the non-blocking calls, including across the wrap of the ring */ +static void +test_queue_nonblocking(void) +{ + protothread_t const pt = protothread_create() ; + int slots[4] ; + intq_t q ; + int x ; + int i ; + + intq_init(&q, slots, 4) ; + check(!intq_try_receive(pt, &q, &x)) ; + for (i = 1; i <= 4; i++) { + check(intq_try_send(pt, &q, &i)) ; + } + check(!intq_try_send(pt, &q, &i)) ; + check(intq_count(&q) == 4) ; + + /* take two, add two: the ring now wraps */ + check(intq_try_receive(pt, &q, &x) && x == 1) ; + check(intq_try_receive(pt, &q, &x) && x == 2) ; + i = 5 ; + check(intq_try_send(pt, &q, &i)) ; + i = 6 ; + check(intq_try_send(pt, &q, &i)) ; + for (i = 0; i < 4; i++) { + check(*intq_at(&q, (unsigned int)i) == 3 + i) ; + } + + /* remove from the middle, then from the end, keeping order */ + check(intq_remove(pt, &q, 2, &x) && x == 5) ; + check(intq_remove(pt, &q, 2, NULL)) ; + check(!intq_remove(pt, &q, 2, &x)) ; + check(intq_count(&q) == 2) ; + check(*intq_at(&q, 0) == 3 && *intq_at(&q, 1) == 4) ; + check(intq_try_receive(pt, &q, &x) && x == 3) ; + check(intq_try_receive(pt, &q, &x) && x == 4) ; + check(intq_count(&q) == 0) ; + protothread_free(pt) ; +} + +/* Selective receive, standing in for Go's select: several kinds of message + * share one queue, and the receiver takes the one it wants, leaving the + * others in order. + */ +enum { MSG_STATUS, MSG_REPLY } ; +typedef struct { + int kind ; + int value ; +} msg_t ; +PT_QUEUE_DEFINE(msgq, msg_t) + +typedef struct { + pt_thread_t pt_thread ; + pt_func_t pt_func ; + pt_queue_env_t queue_env ; + msgq_t * q ; + unsigned int i ; + msg_t got ; +} msgq_context_t ; + +static pt_t +msgq_sender_thr(env_t const env) +{ + msgq_context_t * const c = env ; + msg_t m ; + pt_resume(c) ; + m.kind = MSG_STATUS ; m.value = 1 ; + check(msgq_try_send(pt_get_pt(c), c->q, &m)) ; + m.value = 2 ; + check(msgq_try_send(pt_get_pt(c), c->q, &m)) ; + pt_yield(c) ; + m.kind = MSG_REPLY ; m.value = 42 ; + check(msgq_try_send(pt_get_pt(c), c->q, &m)) ; + return PT_DONE ; +} + +static pt_t +msgq_selective_thr(env_t const env) +{ + msgq_context_t * const c = env ; + pt_resume(c) ; + for (;;) { + for (c->i = 0; c->i < msgq_count(c->q); c->i++) { + if (msgq_at(c->q, c->i)->kind == MSG_REPLY) { + check(msgq_remove(pt_get_pt(c), c->q, c->i, &c->got)) ; + return PT_DONE ; + } + } + pt_call(c, msgq_wait_send, &c->queue_env, c->q) ; + } +} + +static void +test_queue_select(void) +{ + protothread_t const pt = protothread_create() ; + msg_t slots[4] ; + msgq_t q ; + msgq_context_t r, w ; + + msgq_init(&q, slots, 4) ; + memset(&r, 0, sizeof(r)) ; + memset(&w, 0, sizeof(w)) ; + r.q = w.q = &q ; + pt_create(pt, &r.pt_thread, msgq_selective_thr, &r) ; + pt_create(pt, &w.pt_thread, msgq_sender_thr, &w) ; + while (protothread_run(pt)) ; + check(r.got.kind == MSG_REPLY && r.got.value == 42) ; + /* the status messages are still there, in order */ + check(msgq_count(&q) == 2) ; + check(msgq_at(&q, 0)->value == 1 && msgq_at(&q, 1)->value == 2) ; + protothread_free(pt) ; +} + +/* the predefined queue of void * */ +static void +test_queue_void(void) +{ + protothread_t const pt = protothread_create() ; + void * slots[2] ; + pt_queue_t q ; + int a, b ; + void * x ; + + pt_queue_init(&q, slots, 2) ; + x = &a ; + check(pt_queue_try_send(pt, &q, &x)) ; + x = &b ; + check(pt_queue_try_send(pt, &q, &x)) ; + check(pt_queue_try_receive(pt, &q, &x) && x == &a) ; + check(pt_queue_try_receive(pt, &q, &x) && x == &b) ; + protothread_free(pt) ; +} + +/******************************************************************************/ + /* Interrupt races. These need the critical-section harness from CI, which * counts nesting depth and can fire a fake interrupt handler at the moment * interrupts are about to be masked -- the only moment a real one can land. @@ -1480,6 +1678,8 @@ typedef struct { pt_thread_t pt_thread ; pt_func_t pt_func ; pt_timer_env_t timer_env ; + pt_queue_env_t queue_env ; + int value ; int done ; } irq_context_t ; @@ -1552,6 +1752,25 @@ irq_signal(void) pt_signal(irq_pt, &irq_until_channel) ; } +static intq_t irq_q ; + +static pt_t +irq_receiver_thr(env_t const env) +{ + irq_context_t * const c = env ; + pt_resume(c) ; + pt_call(c, intq_receive, &c->queue_env, &irq_q, &c->value) ; + c->done = 1 ; + return PT_DONE ; +} + +static void +irq_queue_send(void) +{ + int const seven = 7 ; + check(intq_try_send(irq_pt, &irq_q, &seven)) ; +} + static void test_interrupts(void) { @@ -1639,6 +1858,30 @@ test_interrupts(void) check(c.done) ; protothread_free(irq_pt) ; } + + /* an interrupt that sends to a queue, at any point while a protothread + * is blocking to receive from it, is never lost + */ + for (k = 1; k <= 8; k++) { + irq_context_t c ; + int slots[2] ; + + memset(&c, 0, sizeof(c)) ; + intq_init(&irq_q, slots, 2) ; + irq_pt = protothread_create() ; + pt_create(irq_pt, &c.pt_thread, irq_receiver_thr, &c) ; + pt_cs_irq = irq_queue_send ; + pt_cs_irq_at = k ; + while (protothread_run(irq_pt)) ; + if (pt_cs_irq) { + /* the interrupt never fired; send from thread context instead */ + pt_cs_irq = NULL ; + irq_queue_send() ; + while (protothread_run(irq_pt)) ; + } + check(c.done && c.value == 7) ; + protothread_free(irq_pt) ; + } } #endif /* PT_TEST_IRQ */ @@ -1667,6 +1910,10 @@ main() test_join() ; test_same_line() ; test_lock_cancel() ; + test_queue_fifo() ; + test_queue_nonblocking() ; + test_queue_select() ; + test_queue_void() ; test_version() ; #ifdef PT_TEST_IRQ test_interrupts() ; From eb0ca1dda74c8b69a8fdde659c0dba383efd03a8 Mon Sep 17 00:00:00 2001 From: Larry Ruane Date: Tue, 6 Oct 2026 11:48:07 -0600 Subject: [PATCH 2/8] README: producer/consumer again, with a queue 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 Claude-Session: https://claude.ai/code/session_01SR7v9BVsFrHiEhQBSqZmAs --- README.md | 44 ++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 44 insertions(+) diff --git a/README.md b/README.md index 817e5c2..675967b 100644 --- a/README.md +++ b/README.md @@ -269,6 +269,50 @@ The event-driven version of the consumer is a two-state machine, a `switch` on " The difference grows with the sequence. Every `pt_wait()` in a protothread is a state that the event-driven version has to name, store and dispatch on by hand, and naming is the hard part: finding a good name for every intermediate state of a long sequence is often harder than writing the sequence. A protothread's state is its position in the function, and a position needs no name. `pt_resume()` is the `switch` you no longer have to write. +### The same, with a queue ### + +The mailbox above is a queue of one, written by hand. With a [message queue](#message-queues) from `protothread_queue.h`, the two protothreads shrink to their loops: +``` + PT_QUEUE_DEFINE(intq, int) + + typedef struct { + pt_thread_t pt_thread; + pt_func_t pt_func; + pt_queue_env_t queue_env; + intq_t * q; + int i; + int value; + } qpc_context_t; + + static pt_t + producer_thr(void * const env) + { + qpc_context_t * const c = env; + pt_resume(c); + + for (c->i = 1; c->i <= 100; c->i++) { + pt_call(c, intq_send, &c->queue_env, c->q, &c->i); + } + return PT_DONE; + } + + static pt_t + consumer_thr(void * const env) + { + qpc_context_t * const c = env; + pt_resume(c); + + for (c->i = 1; c->i <= 100; c->i++) { + pt_call(c, intq_receive, &c->queue_env, c->q, &c->value); + assert(c->value == c->i); + } + return PT_DONE; + } +``` +The queue needs storage, `int slots[4]; intq_t q; intq_init(&q, slots, 4);`, and the rest of the test is as before. + +The two versions differ in what moves. In the mailbox version the data stays put and the protothreads come to it: each one checks the mailbox, waits on it, changes it, and wakes the other. In the queue version the protothreads stay put, each in its own loop, and the data moves, one value at a time, from the producer to the consumer, as if along a wire between them. The waiting and waking haven't gone away, they're inside `send` and `receive`, but neither protothread mentions the other any more. And with room for four values, the producer can run up to four ahead before it has to wait, so the two take turns less often: passing 100 values takes 51 protothread runs here, against 201 through the mailbox. + ## Bare-metal and embedded use ## Protothreads were invented for memory-constrained embedded systems, and this implementation is usable on a bare microcontroller: no operating system, no heap, and no C library. From 0f8026a774c93a89b8b711063598d320fe0bdbbf Mon Sep 17 00:00:00 2001 From: Larry Ruane Date: Tue, 6 Oct 2026 11:53:20 -0600 Subject: [PATCH 3/8] Queues of queues: request and reply 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 Claude-Session: https://claude.ai/code/session_01SR7v9BVsFrHiEhQBSqZmAs --- README.md | 23 ++++++++++++++ protothread_test.c | 77 ++++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 100 insertions(+) diff --git a/README.md b/README.md index 675967b..ff4ea48 100644 --- a/README.md +++ b/README.md @@ -1101,6 +1101,29 @@ for (;;) { That is linear in the length of the queue, which is fine for the queues a microcontroller has. +**Queues of queues.** In Go a channel can be sent over a channel, and it's what makes request and reply easy: the client sends its question together with a channel for the answer, and the server replies on whatever channel it was given. A Go channel is a reference, so the equivalent here is a pointer to a queue: + +```c +typedef struct { + int x; + intq_t * reply; /* where the answer goes */ +} request_t; +PT_QUEUE_DEFINE(requestq, request_t) + +/* server */ +pt_call(c, requestq_receive, &c->queue_env, &requests, &c->req); +c->answer = c->req.x * c->req.x; +pt_call(c, intq_send, &c->queue_env, c->req.reply, &c->answer); + +/* client, with its own reply queue in its context */ +c->req.x = 7; +c->req.reply = &c->replies; +pt_call(c, requestq_send, &c->queue_env, &requests, &c->req); +pt_call(c, intq_receive, &c->queue_env, &c->replies, &c->answer); +``` + +Send the pointer, never the queue itself: a copy of a queue would be a second queue sharing the first one's storage. The reply queue also has to outlive the request, which it does here because the client waits for its answer. + ## License ## MIT. See [LICENSE](LICENSE). diff --git a/protothread_test.c b/protothread_test.c index 373e436..fb46d54 100644 --- a/protothread_test.c +++ b/protothread_test.c @@ -1627,6 +1627,82 @@ test_queue_select(void) protothread_free(pt) ; } +/* A queue of requests, each carrying the queue its reply goes to: Go's + * channel of channels, with a pointer standing in for Go's reference. + */ +typedef struct { + int x ; + intq_t * reply ; +} request_t ; +PT_QUEUE_DEFINE(requestq, request_t) + +typedef struct { + pt_thread_t pt_thread ; + pt_func_t pt_func ; + pt_queue_env_t queue_env ; + requestq_t * requests ; + request_t req ; + int answer ; + int slots[1] ; + intq_t replies ; /* each client's own */ + int x ; +} rr_context_t ; + +static pt_t +rr_server_thr(env_t const env) +{ + rr_context_t * const c = env ; + pt_resume(c) ; + for (;;) { + pt_call(c, requestq_receive, &c->queue_env, c->requests, &c->req) ; + c->answer = c->req.x * c->req.x ; + pt_call(c, intq_send, &c->queue_env, c->req.reply, &c->answer) ; + } +} + +static pt_t +rr_client_thr(env_t const env) +{ + rr_context_t * const c = env ; + pt_resume(c) ; + c->req.x = c->x ; + c->req.reply = &c->replies ; + pt_call(c, requestq_send, &c->queue_env, c->requests, &c->req) ; + pt_call(c, intq_receive, &c->queue_env, &c->replies, &c->answer) ; + return PT_DONE ; +} + +static void +test_queue_of_queues(void) +{ + protothread_t const pt = protothread_create() ; + request_t slots[2] ; + requestq_t requests ; + rr_context_t server, client[3] ; + int i ; + + requestq_init(&requests, slots, 2) ; + memset(&server, 0, sizeof(server)) ; + server.requests = &requests ; + pt_create(pt, &server.pt_thread, rr_server_thr, &server) ; + for (i = 0; i < 3; i++) { + memset(&client[i], 0, sizeof(client[i])) ; + client[i].requests = &requests ; + client[i].x = i + 2 ; + intq_init(&client[i].replies, client[i].slots, 1) ; + pt_create(pt, &client[i].pt_thread, rr_client_thr, &client[i]) ; + } + while (protothread_run(pt)) ; + /* each client got the answer to its own question */ + for (i = 0; i < 3; i++) { + check(client[i].answer == (i + 2) * (i + 2)) ; + check(!pt_is_alive(&client[i].pt_thread)) ; + } + /* the server is still waiting for the next request */ + check(pt_kill(&server.pt_thread)) ; + protothread_free(pt) ; +} + /* the predefined queue of void * */ static void test_queue_void(void) @@ -1914,6 +1990,7 @@ main() test_queue_nonblocking() ; test_queue_select() ; test_queue_void() ; + test_queue_of_queues() ; test_version() ; #ifdef PT_TEST_IRQ test_interrupts() ; From 9a93a55a5b6e9b7452f88313f3de2f13bbacab6c Mon Sep 17 00:00:00 2001 From: Larry Ruane Date: Tue, 6 Oct 2026 12:04:20 -0600 Subject: [PATCH 4/8] Show and test a receive with a timeout 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 Claude-Session: https://claude.ai/code/session_01SR7v9BVsFrHiEhQBSqZmAs --- README.md | 31 +++++++++- protothread_test.c | 142 +++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 172 insertions(+), 1 deletion(-) diff --git a/README.md b/README.md index ff4ea48..e80d979 100644 --- a/README.md +++ b/README.md @@ -1079,11 +1079,40 @@ Like the semaphore, a queue isn't strictly fair: a protothread that finds it non ```c typedef struct { enum { REPLY, STATUS, TIMEOUT } kind; + unsigned int gen; /* for timeouts; see below */ union { struct reply reply; struct status status; } u; } msg_t; ``` -A timeout is just another sender: a small protothread that calls `pt_sleep()` and then sends a `TIMEOUT` message, which is all Go's `time.After()` does. +**Timeouts.** A timeout is just another sender: a small protothread that sleeps, using `pt_sleep()` from [timers](#timers), and then sends a `TIMEOUT` message into the same queue. That is all Go's `time.After()` does: + +```c +static pt_t +after_thr(void * const env) +{ + after_context_t * const c = env; + pt_resume(c); + pt_sleep(c, &c->timer_env, c->timers, 10); + pt_call(c, msgq_send, &c->queue_env, c->q, &c->msg); + return PT_DONE; +} +``` + +Go gets one thing for free here. Its timeout has a channel of its own, so when the reply wins, that channel is simply forgotten. Here the timeout shares the reply's queue, so two things are needed. When the reply wins, stop the timer, which also frees its context for the next wait. And since the timer may already have sent by then, number each wait, and skip a `TIMEOUT` left over from an earlier one: + +```c +c->gen++; +c->after.msg.kind = TIMEOUT; +c->after.msg.gen = c->gen; +pt_create(pt_get_pt(c), &c->after.pt_thread, after_thr, &c->after); +do { + pt_call(c, msgq_receive, &c->queue_env, &q, &c->msg); +} while (c->msg.kind == TIMEOUT && c->msg.gen != c->gen); +if (c->msg.kind != TIMEOUT) { + pt_timer_cancel(&timers, &c->after.timer_env); + pt_kill(&c->after.pt_thread); +} +``` When a receiver wants one kind of message while others wait their turn, it can look before it takes. `wait_send()` blocks only until something new arrives, so messages it has already passed over don't wake it again: diff --git a/protothread_test.c b/protothread_test.c index fb46d54..9f0e325 100644 --- a/protothread_test.c +++ b/protothread_test.c @@ -1703,6 +1703,147 @@ test_queue_of_queues(void) protothread_free(pt) ; } +/* A receive with a timeout, Go's time.After(): a helper protothread sleeps, + * then sends TIMEOUT into the same queue the reply will arrive on. Whichever + * comes first wins. If the reply does, the helper is stopped; but it may + * already have sent, so each wait has a generation and a TIMEOUT from an + * earlier one is discarded. + */ +enum { TO_REPLY, TO_TIMEOUT } ; +typedef struct { + int kind ; + unsigned int gen ; +} tmsg_t ; +PT_QUEUE_DEFINE(tmsgq, tmsg_t) + +typedef struct { + pt_thread_t pt_thread ; + pt_func_t pt_func ; + pt_timer_env_t timer_env ; + pt_queue_env_t queue_env ; + pt_timers_t * timers ; + tmsgq_t * q ; + tmsg_t msg ; +} after_context_t ; + +static pt_t +after_thr(env_t const env) +{ + after_context_t * const c = env ; + pt_resume(c) ; + pt_sleep(c, &c->timer_env, c->timers, 10) ; + pt_call(c, tmsgq_send, &c->queue_env, c->q, &c->msg) ; + return PT_DONE ; +} + +typedef struct { + pt_thread_t pt_thread ; + pt_func_t pt_func ; + pt_queue_env_t queue_env ; + after_context_t after ; + pt_timers_t * timers ; + tmsgq_t * q ; + tmsg_t msg ; + unsigned int gen ; + int round ; + int result[2] ; + int stale ; +} timeout_context_t ; + +static pt_t +timeout_waiter_thr(env_t const env) +{ + timeout_context_t * const c = env ; + pt_resume(c) ; + for (c->round = 0; c->round < 2; c->round++) { + /* start the timer */ + c->gen++ ; + c->after.timers = c->timers ; + c->after.q = c->q ; + c->after.msg.kind = TO_TIMEOUT ; + c->after.msg.gen = c->gen ; + pt_create(pt_get_pt(c), &c->after.pt_thread, after_thr, &c->after) ; + + /* take the first reply or current timeout, skipping stale ones */ + for (;;) { + pt_call(c, tmsgq_receive, &c->queue_env, c->q, &c->msg) ; + if (c->msg.kind != TO_TIMEOUT || c->msg.gen == c->gen) { + break ; + } + c->stale++ ; + } + + /* if the reply won, stop the timer, if it hasn't already fired */ + if (c->msg.kind == TO_REPLY) { + (void)pt_timer_cancel(c->timers, &c->after.timer_env) ; + (void)pt_kill(&c->after.pt_thread) ; + } + c->result[c->round] = c->msg.kind ; + } + return PT_DONE ; +} + +static void +test_queue_timeout(void) +{ + protothread_t const pt = protothread_create() ; + pt_timers_t timers ; + tmsg_t slots[4] ; + tmsgq_t q ; + timeout_context_t w ; + tmsg_t reply ; + + reply.kind = TO_REPLY ; + reply.gen = 0 ; + + /* first the reply wins, then the timeout does */ + pt_timers_init(&timers, 0) ; + tmsgq_init(&q, slots, 4) ; + memset(&w, 0, sizeof(w)) ; + w.timers = &timers ; + w.q = &q ; + pt_create(pt, &w.pt_thread, timeout_waiter_thr, &w) ; + while (protothread_run(pt)) ; + pt_timer_run(pt, &timers, 2) ; + check(tmsgq_try_send(pt, &q, &reply)) ; + while (protothread_run(pt)) ; + check(w.result[0] == TO_REPLY) ; + pt_timer_run(pt, &timers, 12) ; /* the second wait's deadline */ + while (protothread_run(pt)) ; + check(w.result[1] == TO_TIMEOUT) ; + check(!pt_is_alive(&w.pt_thread)) ; + /* the first timer was stopped, so nothing is left over */ + pt_timer_run(pt, &timers, 100) ; + while (protothread_run(pt)) ; + check(tmsgq_count(&q) == 0) ; + check(w.stale == 0) ; + + /* Now the timer fires just before the reply arrives, so its TIMEOUT is + * queued behind the reply. The reply still wins, and the next wait finds + * the leftover TIMEOUT and discards it. + */ + pt_timers_init(&timers, 0) ; + tmsgq_init(&q, slots, 4) ; + memset(&w, 0, sizeof(w)) ; + w.timers = &timers ; + w.q = &q ; + pt_create(pt, &w.pt_thread, timeout_waiter_thr, &w) ; + while (protothread_run(pt)) ; + pt_timer_run(pt, &timers, 10) ; /* the helper is ready first */ + check(tmsgq_try_send(pt, &q, &reply)) ; /* then the waiter */ + while (protothread_run(pt)) ; + check(w.result[0] == TO_REPLY) ; + check(w.stale == 1) ; + check(tmsgq_try_send(pt, &q, &reply)) ; + while (protothread_run(pt)) ; + check(w.result[1] == TO_REPLY) ; + check(!pt_is_alive(&w.pt_thread)) ; + pt_timer_run(pt, &timers, 100) ; + while (protothread_run(pt)) ; + check(tmsgq_count(&q) == 0) ; + protothread_free(pt) ; +} + /* the predefined queue of void * */ static void test_queue_void(void) @@ -1991,6 +2132,7 @@ main() test_queue_select() ; test_queue_void() ; test_queue_of_queues() ; + test_queue_timeout() ; test_version() ; #ifdef PT_TEST_IRQ test_interrupts() ; From 7a7d2e2f105a139dcf10711251df19276b80cd87 Mon Sep 17 00:00:00 2001 From: Larry Ruane Date: Thu, 8 Oct 2026 08:55:09 -0600 Subject: [PATCH 5/8] Close a removal gap from whichever side is shorter 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 Claude-Session: https://claude.ai/code/session_01SR7v9BVsFrHiEhQBSqZmAs --- README.md | 2 +- protothread_queue.h | 28 +++++++++++++++---- protothread_test.c | 67 +++++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 90 insertions(+), 7 deletions(-) diff --git a/README.md b/README.md index e80d979..7c58c31 100644 --- a/README.md +++ b/README.md @@ -1070,7 +1070,7 @@ pt_call(c, msgq_receive, &c->queue_env, &q, &c->msg); /* blocks while empty */ > > `bool_t name_remove(protothread_t, name_t *q, unsigned int i, type *item)` > -> Look inside the queue without blocking: the number of items, a pointer to the `i`-th oldest (0 is the next to be received), and removal of the `i`-th oldest, which keeps the rest in order. `item` may be NULL to discard. A pointer from `at()` stays valid until something is removed. +> Look inside the queue without blocking: the number of items, a pointer to the `i`-th oldest (0 is the next to be received), and removal of the `i`-th oldest, which keeps the rest in order. Removal moves whichever side of the gap is shorter, so taking the oldest or the newest moves nothing. `item` may be NULL to discard. A pointer from `at()` stays valid until something is removed. Like the semaphore, a queue isn't strictly fair: a protothread that finds it non-empty takes an item even if another was woken for it first, for the same reason, to avoid convoys. Every wakeup is a broadcast, and senders and receivers wait on different channels, so sending wakes only receivers and receiving wakes only senders. Items are copied by assignment, and for an item larger than a few words the compiler may emit a call to `memcpy` (clang targeting Cortex-M does so from 16 bytes at `-Os`). Every embedded toolchain provides it, but if you are avoiding it, queue pointers instead. diff --git a/protothread_queue.h b/protothread_queue.h index 5fb8fca..0accb72 100644 --- a/protothread_queue.h +++ b/protothread_queue.h @@ -82,7 +82,7 @@ pt_i_queue_appended(pt_i_queue_t * const q) q->sent++ ; } -/* the caller has just copied out the oldest item */ +/* the caller has just closed a gap by moving the older items up by one */ static inline void pt_i_queue_took_oldest(pt_i_queue_t * const q) { @@ -91,6 +91,14 @@ pt_i_queue_took_oldest(pt_i_queue_t * const q) q->count-- ; } +/* the caller has just closed a gap by moving the newer items down by one */ +static inline void +pt_i_queue_took_newest(pt_i_queue_t * const q) +{ + PT_CRITICAL_ASSERT() ; + q->count-- ; +} + /* An instance needn't use every function it generates, and clang warns about * unused static functions expanded outside a header. */ @@ -157,12 +165,20 @@ name ## _remove(protothread_t const s, name ## _t * const q, \ if (item) { \ *item = q->slots[pt_i_queue_slot(&q->q, i)] ; \ } \ - /* close the gap by moving the older items up by one */ \ - for (k = i; k > 0; k--) { \ - q->slots[pt_i_queue_slot(&q->q, k)] = \ - q->slots[pt_i_queue_slot(&q->q, k - 1)] ; \ + /* close the gap from whichever side has fewer items to move */ \ + if (i < q->q.count - 1 - i) { \ + for (k = i; k > 0; k--) { \ + q->slots[pt_i_queue_slot(&q->q, k)] = \ + q->slots[pt_i_queue_slot(&q->q, k - 1)] ; \ + } \ + pt_i_queue_took_oldest(&q->q) ; \ + } else { \ + for (k = i + 1; k < q->q.count; k++) { \ + q->slots[pt_i_queue_slot(&q->q, k - 1)] = \ + q->slots[pt_i_queue_slot(&q->q, k)] ; \ + } \ + pt_i_queue_took_newest(&q->q) ; \ } \ - pt_i_queue_took_oldest(&q->q) ; \ taken = true ; \ } \ PT_CRITICAL_EXIT(saved) ; \ diff --git a/protothread_test.c b/protothread_test.c index 9f0e325..3c7401d 100644 --- a/protothread_test.c +++ b/protothread_test.c @@ -1553,6 +1553,72 @@ test_queue_nonblocking(void) protothread_free(pt) ; } +/* Removal closes the gap from whichever side is shorter. Comparing the + * addresses at() returns shows which items moved: an item that didn't move + * stays in the same slot. + */ +static void +test_queue_remove(void) +{ + protothread_t const pt = protothread_create() ; + int slots[5] ; + intq_t q ; + int const * oldest ; + int x ; + int i ; + + /* fill, then take two and add two, so that the items wrap the ring */ + intq_init(&q, slots, 5) ; + for (i = 1; i <= 5; i++) { + check(intq_try_send(pt, &q, &i)) ; + } + check(intq_try_receive(pt, &q, &x) && x == 1) ; + check(intq_try_receive(pt, &q, &x) && x == 2) ; + for (i = 6; i <= 7; i++) { + check(intq_try_send(pt, &q, &i)) ; + } + + /* 3 4 5 6 7: removing the newest moves nothing */ + oldest = intq_at(&q, 0) ; + check(intq_remove(pt, &q, 4, &x) && x == 7) ; + check(intq_at(&q, 0) == oldest) ; + + /* 3 4 5 6: the gap at 5 is nearer the new end, so only 6 moves */ + check(intq_remove(pt, &q, 2, &x) && x == 5) ; + check(intq_at(&q, 0) == oldest && *intq_at(&q, 2) == 6) ; + + /* 3 4 6: removing the oldest moves nothing either */ + { + int const * const second = intq_at(&q, 1) ; + check(intq_remove(pt, &q, 0, &x) && x == 3) ; + check(intq_at(&q, 0) == second) ; + } + + /* 4 6 8 9 10: the gap at 6 is nearer the old end, so only 4 moves */ + for (i = 8; i <= 10; i++) { + check(intq_try_send(pt, &q, &i)) ; + } + { + int const * const third = intq_at(&q, 2) ; + check(intq_remove(pt, &q, 1, &x) && x == 6) ; + check(intq_at(&q, 1) == third) ; + } + + /* 4 8 9 10, in order */ + check(intq_count(&q) == 4) ; + check(*intq_at(&q, 0) == 4 && *intq_at(&q, 1) == 8) ; + check(*intq_at(&q, 2) == 9 && *intq_at(&q, 3) == 10) ; + + /* empty it from the new end, then reuse it */ + while (intq_count(&q) > 0) { + check(intq_remove(pt, &q, intq_count(&q) - 1, NULL)) ; + } + i = 11 ; + check(intq_try_send(pt, &q, &i)) ; + check(intq_try_receive(pt, &q, &x) && x == 11) ; + protothread_free(pt) ; +} + /* Selective receive, standing in for Go's select: several kinds of message * share one queue, and the receiver takes the one it wants, leaving the * others in order. @@ -2129,6 +2195,7 @@ main() test_lock_cancel() ; test_queue_fifo() ; test_queue_nonblocking() ; + test_queue_remove() ; test_queue_select() ; test_queue_void() ; test_queue_of_queues() ; From c0e4cc7b1cbeca4b0ec4792b3666d9845a1c40fb Mon Sep 17 00:00:00 2001 From: Larry Ruane Date: Thu, 8 Oct 2026 09:35:07 -0600 Subject: [PATCH 6/8] Make the blocking queue calls macros that wait in the caller's context 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 Claude-Session: https://claude.ai/code/session_01SR7v9BVsFrHiEhQBSqZmAs --- .github/workflows/ci.yml | 11 ++- CHANGELOG.md | 10 ++- README.md | 48 ++++++----- protothread_queue.h | 173 ++++++++++++++++++++++----------------- protothread_test.c | 75 +++++++++++++---- 5 files changed, 195 insertions(+), 122 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 7e9d0bb..075c20d 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -161,7 +161,7 @@ jobs: typedef struct { pt_thread_t pt_thread ; pt_func_t pt_func ; pt_sem_env_t se ; pt_lock_env_t le ; pt_timer_env_t te ; - pt_queue_env_t qe ; unsigned v ; + unsigned v ; } ctx_t ; static unsigned qslots[4] ; static uq_t q ; @@ -178,8 +178,8 @@ jobs: pt_lock_acquire_write(x, &x->le, &lock) ; pt_lock_release_write(&x->le, &lock) ; pt_sleep(x, &x->te, &timers, 10) ; - pt_call(x, uq_send, &x->qe, &q, &x->v) ; - pt_call(x, uq_receive, &x->qe, &q, &x->v) ; + pt_queue_send(x, &q, &x->v) ; + pt_queue_receive(x, &q, &x->v) ; return PT_DONE ; } void app_init(void) { @@ -411,7 +411,6 @@ jobs: typedef struct { pt_thread_t pt_thread ; pt_func_t pt_func ; pt_sem_env_t se ; pt_lock_env_t le ; pt_timer_env_t te ; - pt_queue_env_t qe ; int i ; int v ; } ctx_t ; @@ -440,9 +439,9 @@ jobs: pt_lock_release_write(&x->le, &lock) ; pt_sleep(x, &x->te, &timers, 10) ; x->v = 5 ; - pt_call(x, intq_send, &x->qe, &q, &x->v) ; + pt_queue_send(x, &q, &x->v) ; x->v = 0 ; - pt_call(x, intq_receive, &x->qe, &q, &x->v) ; + pt_queue_receive(x, &q, &x->v) ; return PT_DONE ; } diff --git a/CHANGELOG.md b/CHANGELOG.md index 69d313e..e57ba5e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -26,10 +26,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - `protothread_queue.h`: message queues, what Go calls channels. A queue is a fixed-capacity ring buffer in storage the caller supplies, so it needs no allocator, and `PT_QUEUE_DEFINE(name, type)` makes it type-safe; `pt_queue_t` - is a predefined queue of `void *`. Sending and receiving block when full and - empty, `try_send()` is safe to call from an interrupt handler, and `count()`, - `at()`, `remove()` and `wait_send()` let a receiver take one kind of message - while others wait, which with a tagged union stands in for Go's `select`. + is a predefined queue of `void *`. `pt_queue_send()` and `pt_queue_receive()` + block when full and empty, in the caller's own context like `pt_wait()`; + `try_send()` is safe to call from an interrupt handler; and `count()`, + `at()`, `remove()` and `pt_queue_wait_send()` let a receiver take one kind + of message while others wait, which with a tagged union stands in for Go's + `select`. ### Changed diff --git a/README.md b/README.md index 7c58c31..8cea0b3 100644 --- a/README.md +++ b/README.md @@ -278,7 +278,6 @@ The mailbox above is a queue of one, written by hand. With a [message queue](#me typedef struct { pt_thread_t pt_thread; pt_func_t pt_func; - pt_queue_env_t queue_env; intq_t * q; int i; int value; @@ -291,7 +290,7 @@ The mailbox above is a queue of one, written by hand. With a [message queue](#me pt_resume(c); for (c->i = 1; c->i <= 100; c->i++) { - pt_call(c, intq_send, &c->queue_env, c->q, &c->i); + pt_queue_send(c, c->q, &c->i); } return PT_DONE; } @@ -303,7 +302,7 @@ The mailbox above is a queue of one, written by hand. With a [message queue](#me pt_resume(c); for (c->i = 1; c->i <= 100; c->i++) { - pt_call(c, intq_receive, &c->queue_env, c->q, &c->value); + pt_queue_receive(c, c->q, &c->value); assert(c->value == c->i); } return PT_DONE; @@ -1032,37 +1031,41 @@ Releasing never blocks, so if acquiring didn't either, nothing else ran in betwe ```c typedef struct { int kind; int value; } msg_t; -PT_QUEUE_DEFINE(msgq, msg_t) /* msgq_t, msgq_send() and the rest */ +PT_QUEUE_DEFINE(msgq, msg_t) /* msgq_t, msgq_try_send() and the rest */ msg_t slots[8]; msgq_t q; msgq_init(&q, slots, 8); ``` -Sending and receiving can block, so like any blocking function they are protothread functions, called through `pt_call()`, with a `pt_queue_env_t` in your context structure. Items go in and come out through pointers, so each is copied once each way: +Sending and receiving block like `pt_wait()` does, in your own protothread, so they are macros, and they work on any queue. Items go in and come out through pointers, so each is copied once each way: ```c -pt_call(c, msgq_send, &c->queue_env, &q, &c->msg); /* blocks while full */ -pt_call(c, msgq_receive, &c->queue_env, &q, &c->msg); /* blocks while empty */ +pt_queue_send(c, &q, &c->msg); /* blocks while full */ +pt_queue_receive(c, &q, &c->msg); /* blocks while empty */ ``` -> `name_send(pt_queue_env_t *queue_env, name_t *q, type const *item)` +> `void pt_queue_send(struct context_t *c, name_t *q, type const *item)` > -> Block while the queue is full, then append a copy of `*item`. Call through `pt_call()`. +> Block while the queue is full, then append a copy of `*item`. -> `name_receive(pt_queue_env_t *queue_env, name_t *q, type *item)` +> `void pt_queue_receive(struct context_t *c, name_t *q, type *item)` > -> Block while the queue is empty, then remove the oldest item into `*item`. Call through `pt_call()`. +> Block while the queue is empty, then remove the oldest item into `*item`. -> `name_wait_send(pt_queue_env_t *queue_env, name_t *q)` +> `void pt_queue_wait_send(struct context_t *c, name_t *q, unsigned int seen)` > -> Block until another item is sent, without taking anything. Call through `pt_call()`. +> `unsigned int name_sent(name_t const *q)` +> +> Block, without taking anything, until something has been sent since `name_sent()` returned `seen`. See the example below. + +Like the other blocking macros, these evaluate `q` and `item` more than once, so neither may have side effects. They aren't `pt_call()`s, so `pt_call_waited()` doesn't apply to them; with queues it is seldom needed, since what a protothread needs to know arrives in the message rather than in memory that others may have changed. > `bool_t name_try_send(protothread_t, name_t *q, type const *item)` > > `bool_t name_try_receive(protothread_t, name_t *q, type *item)` > -> The same as send and receive, but return FALSE instead of blocking. `try_send()` is safe to call from an interrupt handler once the `PT_CRITICAL_*` macros are defined, and a protothread blocked in `receive()` or `wait_send()` cannot miss what it sends. +> The same as send and receive, but return FALSE instead of blocking. `try_send()` is safe to call from an interrupt handler once the `PT_CRITICAL_*` macros are defined, and a protothread blocked in `pt_queue_receive()` or `pt_queue_wait_send()` cannot miss what it sends. > `unsigned int name_count(name_t const *q)` > @@ -1093,7 +1096,7 @@ after_thr(void * const env) after_context_t * const c = env; pt_resume(c); pt_sleep(c, &c->timer_env, c->timers, 10); - pt_call(c, msgq_send, &c->queue_env, c->q, &c->msg); + pt_queue_send(c, c->q, &c->msg); return PT_DONE; } ``` @@ -1106,7 +1109,7 @@ c->after.msg.kind = TIMEOUT; c->after.msg.gen = c->gen; pt_create(pt_get_pt(c), &c->after.pt_thread, after_thr, &c->after); do { - pt_call(c, msgq_receive, &c->queue_env, &q, &c->msg); + pt_queue_receive(c, &q, &c->msg); } while (c->msg.kind == TIMEOUT && c->msg.gen != c->gen); if (c->msg.kind != TIMEOUT) { pt_timer_cancel(&timers, &c->after.timer_env); @@ -1114,17 +1117,18 @@ if (c->msg.kind != TIMEOUT) { } ``` -When a receiver wants one kind of message while others wait their turn, it can look before it takes. `wait_send()` blocks only until something new arrives, so messages it has already passed over don't wake it again: +When a receiver wants one kind of message while others wait their turn, it can look before it takes. `pt_queue_wait_send()` blocks only until something new arrives, so messages it has already passed over don't wake it again. Note where `seen` is read: *before* looking, so that a message sent while the receiver looks, even by an interrupt, ends the wait at once instead of being slept through: ```c for (;;) { + c->seen = msgq_sent(&q); for (c->i = 0; c->i < msgq_count(&q); c->i++) { if (msgq_at(&q, c->i)->kind == REPLY) { msgq_remove(pt_get_pt(c), &q, c->i, &c->msg); goto got_reply; } } - pt_call(c, msgq_wait_send, &c->queue_env, &q); + pt_queue_wait_send(c, &q, c->seen); } ``` @@ -1140,15 +1144,15 @@ typedef struct { PT_QUEUE_DEFINE(requestq, request_t) /* server */ -pt_call(c, requestq_receive, &c->queue_env, &requests, &c->req); +pt_queue_receive(c, &requests, &c->req); c->answer = c->req.x * c->req.x; -pt_call(c, intq_send, &c->queue_env, c->req.reply, &c->answer); +pt_queue_send(c, c->req.reply, &c->answer); /* client, with its own reply queue in its context */ c->req.x = 7; c->req.reply = &c->replies; -pt_call(c, requestq_send, &c->queue_env, &requests, &c->req); -pt_call(c, intq_receive, &c->queue_env, &c->replies, &c->answer); +pt_queue_send(c, &requests, &c->req); +pt_queue_receive(c, &c->replies, &c->answer); ``` Send the pointer, never the queue itself: a copy of a queue would be a second queue sharing the first one's storage. The reply queue also has to outlive the request, which it does here because the client waits for its answer. diff --git a/protothread_queue.h b/protothread_queue.h index 0accb72..ffe2472 100644 --- a/protothread_queue.h +++ b/protothread_queue.h @@ -14,20 +14,20 @@ /* Message queue. The public API; pt_i_ names are internal. * - * PT_QUEUE_DEFINE(name, type) generates a queue of named name_t. - * The three blocking functions are protothread functions, so like any other - * they are called through pt_call(), as pt_call(c, name_send, env, q, item): - * name_send(queue_env, q, item_p) block while full, then append - * name_receive(queue_env, q, item_p) block while empty, then take oldest - * name_wait_send(queue_env, q) block until another item is sent - * The rest never block, and are called directly: + * PT_QUEUE_DEFINE(name, type) generates a queue of named name_t, and + * these functions, which never block: * name_init(q, slots, capacity) use slots[capacity] as storage * name_try_send(s, q, item_p) append if room; true if sent * name_try_receive(s, q, item_p) take oldest if any; true if taken * name_count(q) number of items queued * name_at(q, i) pointer to the i-th oldest item * name_remove(s, q, i, item_p) take the i-th oldest, keep order - * pt_queue_env_t one per protothread that blocks + * name_sent(q) count of items ever sent; wraps + * + * The blocking calls are macros, like pt_wait(), and work on any queue: + * pt_queue_send(c, q, item_p) block while full, then append + * pt_queue_receive(c, q, item_p) block while empty, then take oldest + * pt_queue_wait_send(c, q, seen) block until name_sent(q) != seen * * pt_queue_t and pt_queue_*() are a predefined queue of void *. * @@ -38,12 +38,6 @@ * else is for thread context. */ -/* One per protothread that blocks on a queue, in its context structure */ -typedef struct pt_queue_env_s { - pt_func_t pt_func ; - unsigned int seen ; /* the send count wait_send() started at */ -} pt_queue_env_t ; - /* The untyped part of every queue: indices and counts, never the items. * Receivers wait on &sent, senders wait on &head, so that a send wakes only * receivers and a receive wakes only senders. @@ -99,15 +93,51 @@ pt_i_queue_took_newest(pt_i_queue_t * const q) q->count-- ; } +/* read with interrupts masked, since on a small target this can be two loads */ +static inline unsigned int +pt_i_queue_sent(pt_i_queue_t const * const q) +{ + pt_i_critical_t const saved = PT_CRITICAL_ENTER() ; + unsigned int const sent = q->sent ; + PT_CRITICAL_EXIT(saved) ; + return sent ; +} + +/* Append *item_p if there is room, and set ok to whether it was. The copy is + * by assignment from the queue's own element type, so it is type-checked. + */ +#define pt_i_queue_put(queue, item_p, ok) \ + do { \ + pt_i_critical_t const pt_i_qsaved = PT_CRITICAL_ENTER() ; \ + (ok) = (queue)->q.count < (queue)->q.capacity ; \ + if (ok) { \ + (queue)->slots[pt_i_queue_slot(&(queue)->q, (queue)->q.count)] = \ + *(item_p) ; \ + pt_i_queue_appended(&(queue)->q) ; \ + } \ + PT_CRITICAL_EXIT(pt_i_qsaved) ; \ + } while (0) + +/* Take the oldest item into *item_p, if there is one, and set ok to whether + * there was. + */ +#define pt_i_queue_take(queue, item_p, ok) \ + do { \ + pt_i_critical_t const pt_i_qsaved = PT_CRITICAL_ENTER() ; \ + (ok) = (queue)->q.count > 0 ; \ + if (ok) { \ + *(item_p) = (queue)->slots[(queue)->q.head] ; \ + pt_i_queue_took_oldest(&(queue)->q) ; \ + } \ + PT_CRITICAL_EXIT(pt_i_qsaved) ; \ + } while (0) + /* An instance needn't use every function it generates, and clang warns about * unused static functions expanded outside a header. */ #define PT_I_MAYBE_UNUSED __attribute__((__unused__)) -/* Generate a queue of named name_t, and its functions. The blocking - * ones are a wait on the untyped part followed by the non-blocking one, and - * retry if another protothread got there first. - */ +/* Generate a queue of named name_t, and its non-blocking functions */ #define PT_QUEUE_DEFINE(name, type) \ \ typedef struct name ## _s { \ @@ -129,6 +159,12 @@ name ## _count(name ## _t const * const q) \ return q->q.count ; \ } \ \ +static inline PT_I_MAYBE_UNUSED unsigned int \ +name ## _sent(name ## _t const * const q) \ +{ \ + return pt_i_queue_sent(&q->q) ; \ +} \ +\ static inline PT_I_MAYBE_UNUSED type * \ name ## _at(name ## _t * const q, unsigned int const i) \ { \ @@ -140,14 +176,8 @@ static inline PT_I_MAYBE_UNUSED bool_t \ name ## _try_send(protothread_t const s, name ## _t * const q, \ type const * const item) \ { \ - bool_t sent = false ; \ - pt_i_critical_t const saved = PT_CRITICAL_ENTER() ; \ - if (q->q.count < q->q.capacity) { \ - q->slots[pt_i_queue_slot(&q->q, q->q.count)] = *item ; \ - pt_i_queue_appended(&q->q) ; \ - sent = true ; \ - } \ - PT_CRITICAL_EXIT(saved) ; \ + bool_t sent ; \ + pt_i_queue_put(q, item, sent) ; \ if (sent) { \ pt_broadcast(s, &q->q.sent) ; \ } \ @@ -155,6 +185,18 @@ name ## _try_send(protothread_t const s, name ## _t * const q, \ } \ \ static inline PT_I_MAYBE_UNUSED bool_t \ +name ## _try_receive(protothread_t const s, name ## _t * const q, \ + type * const item) \ +{ \ + bool_t taken ; \ + pt_i_queue_take(q, item, taken) ; \ + if (taken) { \ + pt_broadcast(s, &q->q.head) ; \ + } \ + return taken ; \ +} \ +\ +static inline PT_I_MAYBE_UNUSED bool_t \ name ## _remove(protothread_t const s, name ## _t * const q, \ unsigned int const i, type * const item) \ { \ @@ -186,56 +228,39 @@ name ## _remove(protothread_t const s, name ## _t * const q, \ pt_broadcast(s, &q->q.head) ; \ } \ return taken ; \ -} \ -\ -static inline PT_I_MAYBE_UNUSED bool_t \ -name ## _try_receive(protothread_t const s, name ## _t * const q, \ - type * const item) \ -{ \ - return name ## _remove(s, q, 0, item) ; \ -} \ -\ -static inline PT_I_MAYBE_UNUSED pt_t \ -name ## _send(pt_queue_env_t * const c, name ## _t * const q, \ - type const * const item) \ -{ \ - pt_resume(c) ; \ - for (;;) { \ - pt_wait_until(c, &q->q.head, q->q.count < q->q.capacity) ; \ - if (name ## _try_send(pt_get_pt(c), q, item)) { \ - return PT_DONE ; \ - } \ - } \ -} \ -\ -static inline PT_I_MAYBE_UNUSED pt_t \ -name ## _receive(pt_queue_env_t * const c, name ## _t * const q, \ - type * const item) \ -{ \ - pt_resume(c) ; \ - for (;;) { \ - pt_wait_until(c, &q->q.sent, q->q.count > 0) ; \ - if (name ## _try_receive(pt_get_pt(c), q, item)) { \ - return PT_DONE ; \ - } \ - } \ -} \ -\ -/* the send count is read with interrupts masked, since on a small target \ - * an unsigned int can take two loads */ \ -static inline PT_I_MAYBE_UNUSED pt_t \ -name ## _wait_send(pt_queue_env_t * const c, name ## _t * const q) \ -{ \ - pt_resume(c) ; \ - { \ - pt_i_critical_t const saved = PT_CRITICAL_ENTER() ; \ - c->seen = q->q.sent ; \ - PT_CRITICAL_EXIT(saved) ; \ - } \ - pt_wait_until(c, &q->q.sent, q->q.sent != c->seen) ; \ - return PT_DONE ; \ } +/* The blocking calls wait in the caller's context, as pt_wait() does, and + * retry if another protothread or an interrupt got there first. q and item_p + * are evaluated more than once, so they must not have side effects. + */ +#define pt_queue_send(c, queue, item_p) \ + do { \ + bool_t pt_i_ok ; \ + do { \ + pt_wait_until(c, &(queue)->q.head, \ + (queue)->q.count < (queue)->q.capacity) ; \ + pt_i_queue_put(queue, item_p, pt_i_ok) ; \ + } while (!pt_i_ok) ; \ + pt_broadcast(pt_get_pt(c), &(queue)->q.sent) ; \ + } while (0) + +#define pt_queue_receive(c, queue, item_p) \ + do { \ + bool_t pt_i_ok ; \ + do { \ + pt_wait_until(c, &(queue)->q.sent, (queue)->q.count > 0) ; \ + pt_i_queue_take(queue, item_p, pt_i_ok) ; \ + } while (!pt_i_ok) ; \ + pt_broadcast(pt_get_pt(c), &(queue)->q.head) ; \ + } while (0) + +/* Read seen from name_sent() before looking at the queue, so that anything + * sent after that, even by an interrupt, ends the wait. + */ +#define pt_queue_wait_send(c, queue, seen) \ + pt_wait_until(c, &(queue)->q.sent, (queue)->q.sent != (seen)) + PT_QUEUE_DEFINE(pt_queue, void *) #endif /* PROTOTHREAD_QUEUE_H */ diff --git a/protothread_test.c b/protothread_test.c index 3c7401d..ede8ddf 100644 --- a/protothread_test.c +++ b/protothread_test.c @@ -1457,7 +1457,6 @@ PT_QUEUE_DEFINE(intq, int) typedef struct { pt_thread_t pt_thread ; pt_func_t pt_func ; - pt_queue_env_t queue_env ; intq_t * q ; int i ; int got ; @@ -1470,7 +1469,7 @@ intq_producer_thr(env_t const env) intq_context_t * const c = env ; pt_resume(c) ; for (c->i = 1; c->i <= 100; c->i++) { - pt_call(c, intq_send, &c->queue_env, c->q, &c->i) ; + pt_queue_send(c, c->q, &c->i) ; c->waited += pt_call_waited(c) ; } return PT_DONE ; @@ -1482,7 +1481,7 @@ intq_consumer_thr(env_t const env) intq_context_t * const c = env ; pt_resume(c) ; for (c->i = 1; c->i <= 100; c->i++) { - pt_call(c, intq_receive, &c->queue_env, c->q, &c->got) ; + pt_queue_receive(c, c->q, &c->got) ; c->waited += pt_call_waited(c) ; check(c->got == c->i) ; } @@ -1633,9 +1632,9 @@ PT_QUEUE_DEFINE(msgq, msg_t) typedef struct { pt_thread_t pt_thread ; pt_func_t pt_func ; - pt_queue_env_t queue_env ; msgq_t * q ; unsigned int i ; + unsigned int seen ; msg_t got ; } msgq_context_t ; @@ -1661,13 +1660,14 @@ msgq_selective_thr(env_t const env) msgq_context_t * const c = env ; pt_resume(c) ; for (;;) { + c->seen = msgq_sent(c->q) ; for (c->i = 0; c->i < msgq_count(c->q); c->i++) { if (msgq_at(c->q, c->i)->kind == MSG_REPLY) { check(msgq_remove(pt_get_pt(c), c->q, c->i, &c->got)) ; return PT_DONE ; } } - pt_call(c, msgq_wait_send, &c->queue_env, c->q) ; + pt_queue_wait_send(c, c->q, c->seen) ; } } @@ -1705,7 +1705,6 @@ PT_QUEUE_DEFINE(requestq, request_t) typedef struct { pt_thread_t pt_thread ; pt_func_t pt_func ; - pt_queue_env_t queue_env ; requestq_t * requests ; request_t req ; int answer ; @@ -1720,9 +1719,9 @@ rr_server_thr(env_t const env) rr_context_t * const c = env ; pt_resume(c) ; for (;;) { - pt_call(c, requestq_receive, &c->queue_env, c->requests, &c->req) ; + pt_queue_receive(c, c->requests, &c->req) ; c->answer = c->req.x * c->req.x ; - pt_call(c, intq_send, &c->queue_env, c->req.reply, &c->answer) ; + pt_queue_send(c, c->req.reply, &c->answer) ; } } @@ -1733,8 +1732,8 @@ rr_client_thr(env_t const env) pt_resume(c) ; c->req.x = c->x ; c->req.reply = &c->replies ; - pt_call(c, requestq_send, &c->queue_env, c->requests, &c->req) ; - pt_call(c, intq_receive, &c->queue_env, &c->replies, &c->answer) ; + pt_queue_send(c, c->requests, &c->req) ; + pt_queue_receive(c, &c->replies, &c->answer) ; return PT_DONE ; } @@ -1786,7 +1785,6 @@ typedef struct { pt_thread_t pt_thread ; pt_func_t pt_func ; pt_timer_env_t timer_env ; - pt_queue_env_t queue_env ; pt_timers_t * timers ; tmsgq_t * q ; tmsg_t msg ; @@ -1798,14 +1796,13 @@ after_thr(env_t const env) after_context_t * const c = env ; pt_resume(c) ; pt_sleep(c, &c->timer_env, c->timers, 10) ; - pt_call(c, tmsgq_send, &c->queue_env, c->q, &c->msg) ; + pt_queue_send(c, c->q, &c->msg) ; return PT_DONE ; } typedef struct { pt_thread_t pt_thread ; pt_func_t pt_func ; - pt_queue_env_t queue_env ; after_context_t after ; pt_timers_t * timers ; tmsgq_t * q ; @@ -1832,7 +1829,7 @@ timeout_waiter_thr(env_t const env) /* take the first reply or current timeout, skipping stale ones */ for (;;) { - pt_call(c, tmsgq_receive, &c->queue_env, c->q, &c->msg) ; + pt_queue_receive(c, c->q, &c->msg) ; if (c->msg.kind != TO_TIMEOUT || c->msg.gen == c->gen) { break ; } @@ -1961,8 +1958,9 @@ typedef struct { pt_thread_t pt_thread ; pt_func_t pt_func ; pt_timer_env_t timer_env ; - pt_queue_env_t queue_env ; int value ; + unsigned int seen ; + unsigned int i ; int done ; } irq_context_t ; @@ -2042,11 +2040,30 @@ irq_receiver_thr(env_t const env) { irq_context_t * const c = env ; pt_resume(c) ; - pt_call(c, intq_receive, &c->queue_env, &irq_q, &c->value) ; + pt_queue_receive(c, &irq_q, &c->value) ; c->done = 1 ; return PT_DONE ; } +/* waits for a 7, leaving anything else in the queue */ +static pt_t +irq_select_thr(env_t const env) +{ + irq_context_t * const c = env ; + pt_resume(c) ; + for (;;) { + c->seen = intq_sent(&irq_q) ; + for (c->i = 0; c->i < intq_count(&irq_q); c->i++) { + if (*intq_at(&irq_q, c->i) == 7) { + check(intq_remove(irq_pt, &irq_q, c->i, &c->value)) ; + c->done = 1 ; + return PT_DONE ; + } + } + pt_queue_wait_send(c, &irq_q, c->seen) ; + } +} + static void irq_queue_send(void) { @@ -2165,6 +2182,32 @@ test_interrupts(void) check(c.done && c.value == 7) ; protothread_free(irq_pt) ; } + + /* a selective receiver passing over an unwanted item never misses the + * one it wants, whenever the interrupt sends it + */ + for (k = 1; k <= 8; k++) { + irq_context_t c ; + int slots[4] ; + int const one = 1 ; + + memset(&c, 0, sizeof(c)) ; + intq_init(&irq_q, slots, 4) ; + irq_pt = protothread_create() ; + check(intq_try_send(irq_pt, &irq_q, &one)) ; + pt_create(irq_pt, &c.pt_thread, irq_select_thr, &c) ; + pt_cs_irq = irq_queue_send ; + pt_cs_irq_at = k ; + while (protothread_run(irq_pt)) ; + if (pt_cs_irq) { + pt_cs_irq = NULL ; + irq_queue_send() ; + while (protothread_run(irq_pt)) ; + } + check(c.done && c.value == 7) ; + check(intq_count(&irq_q) == 1 && *intq_at(&irq_q, 0) == 1) ; + protothread_free(irq_pt) ; + } } #endif /* PT_TEST_IRQ */ From 5bd7158d68b8c0755c432f3a1b23bf1c7bb9cc4d Mon Sep 17 00:00:00 2001 From: Larry Ruane Date: Sun, 11 Oct 2026 22:51:10 -0600 Subject: [PATCH 7/8] Rename pt_queue_wait_send() to pt_queue_wait_new() wait_send read as "wait, then send", next to pt_queue_send(). Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01SR7v9BVsFrHiEhQBSqZmAs --- CHANGELOG.md | 2 +- README.md | 8 ++++---- protothread_queue.h | 4 ++-- protothread_test.c | 4 ++-- 4 files changed, 9 insertions(+), 9 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index e57ba5e..4dc8a22 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -29,7 +29,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 is a predefined queue of `void *`. `pt_queue_send()` and `pt_queue_receive()` block when full and empty, in the caller's own context like `pt_wait()`; `try_send()` is safe to call from an interrupt handler; and `count()`, - `at()`, `remove()` and `pt_queue_wait_send()` let a receiver take one kind + `at()`, `remove()` and `pt_queue_wait_new()` let a receiver take one kind of message while others wait, which with a tagged union stands in for Go's `select`. diff --git a/README.md b/README.md index 8cea0b3..3084143 100644 --- a/README.md +++ b/README.md @@ -1053,7 +1053,7 @@ pt_queue_receive(c, &q, &c->msg); /* blocks while empty */ > > Block while the queue is empty, then remove the oldest item into `*item`. -> `void pt_queue_wait_send(struct context_t *c, name_t *q, unsigned int seen)` +> `void pt_queue_wait_new(struct context_t *c, name_t *q, unsigned int seen)` > > `unsigned int name_sent(name_t const *q)` > @@ -1065,7 +1065,7 @@ Like the other blocking macros, these evaluate `q` and `item` more than once, so > > `bool_t name_try_receive(protothread_t, name_t *q, type *item)` > -> The same as send and receive, but return FALSE instead of blocking. `try_send()` is safe to call from an interrupt handler once the `PT_CRITICAL_*` macros are defined, and a protothread blocked in `pt_queue_receive()` or `pt_queue_wait_send()` cannot miss what it sends. +> The same as send and receive, but return FALSE instead of blocking. `try_send()` is safe to call from an interrupt handler once the `PT_CRITICAL_*` macros are defined, and a protothread blocked in `pt_queue_receive()` or `pt_queue_wait_new()` cannot miss what it sends. > `unsigned int name_count(name_t const *q)` > @@ -1117,7 +1117,7 @@ if (c->msg.kind != TIMEOUT) { } ``` -When a receiver wants one kind of message while others wait their turn, it can look before it takes. `pt_queue_wait_send()` blocks only until something new arrives, so messages it has already passed over don't wake it again. Note where `seen` is read: *before* looking, so that a message sent while the receiver looks, even by an interrupt, ends the wait at once instead of being slept through: +When a receiver wants one kind of message while others wait their turn, it can look before it takes. `pt_queue_wait_new()` blocks only until something new arrives, so messages it has already passed over don't wake it again. Note where `seen` is read: *before* looking, so that a message sent while the receiver looks, even by an interrupt, ends the wait at once instead of being slept through: ```c for (;;) { @@ -1128,7 +1128,7 @@ for (;;) { goto got_reply; } } - pt_queue_wait_send(c, &q, c->seen); + pt_queue_wait_new(c, &q, c->seen); } ``` diff --git a/protothread_queue.h b/protothread_queue.h index ffe2472..f9210d6 100644 --- a/protothread_queue.h +++ b/protothread_queue.h @@ -27,7 +27,7 @@ * The blocking calls are macros, like pt_wait(), and work on any queue: * pt_queue_send(c, q, item_p) block while full, then append * pt_queue_receive(c, q, item_p) block while empty, then take oldest - * pt_queue_wait_send(c, q, seen) block until name_sent(q) != seen + * pt_queue_wait_new(c, q, seen) block until name_sent(q) != seen * * pt_queue_t and pt_queue_*() are a predefined queue of void *. * @@ -258,7 +258,7 @@ name ## _remove(protothread_t const s, name ## _t * const q, \ /* Read seen from name_sent() before looking at the queue, so that anything * sent after that, even by an interrupt, ends the wait. */ -#define pt_queue_wait_send(c, queue, seen) \ +#define pt_queue_wait_new(c, queue, seen) \ pt_wait_until(c, &(queue)->q.sent, (queue)->q.sent != (seen)) PT_QUEUE_DEFINE(pt_queue, void *) diff --git a/protothread_test.c b/protothread_test.c index ede8ddf..b3d513b 100644 --- a/protothread_test.c +++ b/protothread_test.c @@ -1667,7 +1667,7 @@ msgq_selective_thr(env_t const env) return PT_DONE ; } } - pt_queue_wait_send(c, c->q, c->seen) ; + pt_queue_wait_new(c, c->q, c->seen) ; } } @@ -2060,7 +2060,7 @@ irq_select_thr(env_t const env) return PT_DONE ; } } - pt_queue_wait_send(c, &irq_q, c->seen) ; + pt_queue_wait_new(c, &irq_q, c->seen) ; } } From 0fac56a7b19d890fae6fcb02aa5421c1c638e2e6 Mon Sep 17 00:00:00 2001 From: Larry Ruane Date: Sun, 11 Oct 2026 23:09:10 -0600 Subject: [PATCH 8/8] Name the internal queue macros' parameter qp Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01SR7v9BVsFrHiEhQBSqZmAs --- protothread_queue.h | 16 ++++++++-------- 1 file changed, 8 insertions(+), 8 deletions(-) diff --git a/protothread_queue.h b/protothread_queue.h index f9210d6..ffa4035 100644 --- a/protothread_queue.h +++ b/protothread_queue.h @@ -106,14 +106,14 @@ pt_i_queue_sent(pt_i_queue_t const * const q) /* Append *item_p if there is room, and set ok to whether it was. The copy is * by assignment from the queue's own element type, so it is type-checked. */ -#define pt_i_queue_put(queue, item_p, ok) \ +#define pt_i_queue_put(qp, item_p, ok) \ do { \ pt_i_critical_t const pt_i_qsaved = PT_CRITICAL_ENTER() ; \ - (ok) = (queue)->q.count < (queue)->q.capacity ; \ + (ok) = (qp)->q.count < (qp)->q.capacity ; \ if (ok) { \ - (queue)->slots[pt_i_queue_slot(&(queue)->q, (queue)->q.count)] = \ + (qp)->slots[pt_i_queue_slot(&(qp)->q, (qp)->q.count)] = \ *(item_p) ; \ - pt_i_queue_appended(&(queue)->q) ; \ + pt_i_queue_appended(&(qp)->q) ; \ } \ PT_CRITICAL_EXIT(pt_i_qsaved) ; \ } while (0) @@ -121,13 +121,13 @@ pt_i_queue_sent(pt_i_queue_t const * const q) /* Take the oldest item into *item_p, if there is one, and set ok to whether * there was. */ -#define pt_i_queue_take(queue, item_p, ok) \ +#define pt_i_queue_take(qp, item_p, ok) \ do { \ pt_i_critical_t const pt_i_qsaved = PT_CRITICAL_ENTER() ; \ - (ok) = (queue)->q.count > 0 ; \ + (ok) = (qp)->q.count > 0 ; \ if (ok) { \ - *(item_p) = (queue)->slots[(queue)->q.head] ; \ - pt_i_queue_took_oldest(&(queue)->q) ; \ + *(item_p) = (qp)->slots[(qp)->q.head] ; \ + pt_i_queue_took_oldest(&(qp)->q) ; \ } \ PT_CRITICAL_EXIT(pt_i_qsaved) ; \ } while (0)