From b15d39421234e0562b68e043a5e8c55d3b22bfc8 Mon Sep 17 00:00:00 2001 From: Robert Schmidt Date: Wed, 15 Apr 2026 12:57:44 +0200 Subject: [PATCH] Improve spsc_q through atomics Because it is SPSC, atomic variables are enough to synchronize the two threads. Concretely put() is modified to ensure that read_idx is "acquired" so that the read index has been written, including memory, by the other thread. "Release" ensures that the write operation (including the memory to the buffer) is written before it is visible to the other thread in get() (which in turn "acquires" it). For more information, see also [1]. The rest of the library has been simplified to work only with put() and get(), reducing total code and the surface for possible bugs. This (and the previous) version has been tested with the threadSanitizer: TSAN_OPTIONS=halt_on_error=1 ./common/utils/ds/tests/test_spsc_q_perf On my machine, using Google Benchmark, I measure a considerable 5x speed improvement: $ /tmp/benchmark/tools/compare.py benchmarks pthread.json atomic.json Comparing pthread.json to atomic.json Benchmark Time CPU Time Old Time New CPU Old CPU New -------------------------------------------------------------------------------------------------------------- BM_spsc_q/10 -0.8201 -0.0740 266779912 47989020 52387 48512 BM_spsc_q/16 -0.8301 -0.0520 249656592 42428540 51462 48784 BM_spsc_q/32 -0.8003 -0.0841 230248798 45972155 53841 49311 BM_spsc_q/64 -0.7995 -0.0506 210429791 42199674 50690 48124 BM_spsc_q/128 -0.7930 -0.1101 205212273 42483155 52745 46936 BM_spsc_q/160 -0.7880 -0.1663 216644738 45938247 53400 44518 OVERALL_GEOMEAN -0.8057 -0.0904 0 0 0 0 [1] https://en.cppreference.com/w/c/atomic/memory_order.html Assisted-By: Claude:claude-sonnet-4-6 Signed-off-by: Robert Schmidt --- common/utils/ds/spsc_q.c | 48 +++++++++++----------------------------- common/utils/ds/spsc_q.h | 13 +++++++---- 2 files changed, 22 insertions(+), 39 deletions(-) diff --git a/common/utils/ds/spsc_q.c b/common/utils/ds/spsc_q.c index fdc97f00b8..6f05a19865 100644 --- a/common/utils/ds/spsc_q.c +++ b/common/utils/ds/spsc_q.c @@ -8,9 +8,6 @@ #include "spsc_q.h" -#define rb_lock(LOCK) if (pthread_mutex_lock(LOCK) != 0) abort() -#define rb_unlock(LOCK) if (pthread_mutex_unlock(LOCK) != 0) abort() - spsc_q_t spsc_q_alloc(size_t cnt, size_t elsiz) { /* internally, use one element more: the ringbuffer is full if @@ -20,51 +17,47 @@ spsc_q_t spsc_q_alloc(size_t cnt, size_t elsiz) cnt += 1; spsc_q_t rb = {.cnt = cnt, .elsiz = elsiz}; rb.buf = calloc(cnt, elsiz); - pthread_mutex_init(&rb.mtx, NULL); return rb; } void spsc_q_free(spsc_q_t *rb) { free(rb->buf); - pthread_mutex_destroy(&rb->mtx); memset(rb, 0, sizeof(*rb)); } bool spsc_q_put(spsc_q_t *rb, const void *src, size_t elsiz) { assert(elsiz == rb->elsiz); - rb_lock(&rb->mtx); - if ((rb->write_idx + 1) % rb->cnt == rb->read_idx) { - rb_unlock(&rb->mtx); + size_t w = atomic_load_explicit(&rb->write_idx, memory_order_relaxed); + size_t r = atomic_load_explicit(&rb->read_idx, memory_order_acquire); + if ((w + 1) % rb->cnt == r) { return false; } - uint8_t *bufpos = &rb->buf[rb->write_idx * rb->elsiz]; + uint8_t *bufpos = &rb->buf[w * rb->elsiz]; memcpy(bufpos, src, elsiz); - rb->write_idx = (rb->write_idx + 1) % rb->cnt; - rb_unlock(&rb->mtx); + atomic_store_explicit(&rb->write_idx, (w + 1) % rb->cnt, memory_order_release); return true; } bool spsc_q_get_if(spsc_q_t *rb, pred p, void *user, void *dest, size_t elsiz) { assert(elsiz == rb->elsiz); - rb_lock(&rb->mtx); - if (rb->write_idx == rb->read_idx) { - rb_unlock(&rb->mtx); + size_t w = atomic_load_explicit(&rb->write_idx, memory_order_acquire); + size_t r = atomic_load_explicit(&rb->read_idx, memory_order_relaxed); + if (w == r) { return false; } uint8_t *bufpos = &rb->buf[rb->read_idx * rb->elsiz]; if (!p(bufpos, user)) { - rb_unlock(&rb->mtx); return false; } - memcpy(dest, bufpos, elsiz); - rb->read_idx = (rb->read_idx + 1) % rb->cnt; - rb_unlock(&rb->mtx); + if (dest) + memcpy(dest, bufpos, elsiz); + atomic_store_explicit(&rb->read_idx, (r + 1) % rb->cnt, memory_order_release); return true; } @@ -82,31 +75,16 @@ bool spsc_q_get(spsc_q_t *rb, void *dest, size_t elsiz) int spsc_q_get_while(spsc_q_t *rb, pred p, void *user, void *dest, size_t elsiz, size_t max_len) { assert(elsiz == rb->elsiz); - rb_lock(&rb->mtx); size_t count = 0; - while (rb->write_idx != rb->read_idx && count < max_len) { - uint8_t *bufpos = &rb->buf[rb->read_idx * rb->elsiz]; - if (!p(bufpos, user)) - break; - memcpy(dest + count * elsiz, bufpos, elsiz); + while (count < max_len && spsc_q_get_if(rb, p, user, dest + elsiz * count, elsiz)) count++; - rb->read_idx = (rb->read_idx + 1) % rb->cnt; - } - rb_unlock(&rb->mtx); return count; } int spsc_q_drop_while(spsc_q_t *rb, pred p, void *user) { - rb_lock(&rb->mtx); size_t dropped = 0; - while (rb->write_idx != rb->read_idx) { - uint8_t *bufpos = &rb->buf[rb->read_idx * rb->elsiz]; - if (!p(bufpos, user)) - break; + while (spsc_q_get_if(rb, p, user, NULL, rb->elsiz)) dropped++; - rb->read_idx = (rb->read_idx + 1) % rb->cnt; - } - rb_unlock(&rb->mtx); return dropped; } diff --git a/common/utils/ds/spsc_q.h b/common/utils/ds/spsc_q.h index d00cb4b77d..7fd968b2f0 100644 --- a/common/utils/ds/spsc_q.h +++ b/common/utils/ds/spsc_q.h @@ -7,15 +7,20 @@ #include #include -#include +#include + +#if defined(__cplusplus) +// use atomic_size_t to help interworking with C++ +using std::atomic_size_t; +#endif typedef struct spsc_q { uint8_t *buf; size_t elsiz; size_t cnt; - size_t write_idx; - size_t read_idx; - pthread_mutex_t mtx; + + atomic_size_t write_idx; + atomic_size_t read_idx; } spsc_q_t; spsc_q_t spsc_q_alloc(size_t cnt, size_t elsiz);