From 93a5287fba3cf9eccffdaab6f016c0964d8132e9 Mon Sep 17 00:00:00 2001 From: zwe Date: Sun, 18 Oct 2026 12:05:37 +0800 Subject: [PATCH] Add an unbounded SPSC queue to substitute for a queue proven SPSC wtc::queue is unbounded and its push never fails, so a substitute has to be too. A fixed ring must either drop an item when full or wait for the consumer, and waiting deadlocks a program in which one thread is both the producer and the consumer, which an SPSC verdict allows. This one is a linked list of fixed-size blocks. Each block counts the slots the producer has published, with release; the consumer frees a block only after draining it and acquiring its successor, and publishing the successor is the producer's last access to the block. Its push and try_pop have the same signatures as wtc::queue's, so the lowering does not change shape. Its annotations are its own, so lifting never mistakes calls to the substitute for calls to the shim. The stress test checks order, count and sum at block sizes 1, 4 and 1024, and that items left at destruction are destroyed exactly once. CI builds the runtime headers standalone and runs the test under ThreadSanitizer. Verified: the stress test passes, unsanitized, with Apple clang on arm64. Not yet verified under TSan: sanitizers are broken in that toolchain (an empty program built with -fsanitize=thread segfaults), so the Linux CI job is the first TSan run. Co-Authored-By: Claude Opus 5.5 --- .github/workflows/ci.yml | 18 ++++ runtime/README.md | 6 +- runtime/spsc_queue.hpp | 156 +++++++++++++++++++++++++++++++++++ runtime/test/spsc_stress.cpp | 86 +++++++++++++++++++ 4 files changed, 265 insertions(+), 1 deletion(-) create mode 100644 runtime/spsc_queue.hpp create mode 100644 runtime/test/spsc_stress.cpp diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index d6e3cf8..b0dc0e5 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -57,6 +57,24 @@ jobs: } CPP clang++ -std=c++17 -Ishim -fsyntax-only -Wall -Werror /tmp/all.cpp + - name: Compile every runtime header on its own + run: | + set -e + for h in runtime/*.hpp; do + echo "== $h" + printf '#include "%s"\nint main(){}\n' "$(basename "$h")" > /tmp/tu.cpp + clang++ -std=c++17 -Iruntime -fsyntax-only -Wall -Werror /tmp/tu.cpp + g++ -std=c++17 -Iruntime -fsyntax-only -Wall \ + -Wno-attributes /tmp/tu.cpp + done + # A lock-free structure is only as good as its race checking. The + # stress test asserts order, count and destruction; TSan asserts that + # getting them right was not luck. + - name: Stress the runtime queues under ThreadSanitizer + run: | + clang++ -std=c++17 -O1 -g -Wall -Wextra -Werror -fsanitize=thread \ + -Iruntime runtime/test/spsc_stress.cpp -o /tmp/spsc_stress + /tmp/spsc_stress # Which toolchain image this commit needs, derived from the submodule pin # rather than configured separately. The two cannot drift apart. diff --git a/runtime/README.md b/runtime/README.md index 4703876..97b4973 100644 --- a/runtime/README.md +++ b/runtime/README.md @@ -15,4 +15,8 @@ Every implementation carries, in its header comment: - what breaks if that fact is false; - the measured numbers against the general-purpose baseline, and on what. -Nothing here yet. +## What is here + +- `spsc_queue.hpp`: an unbounded single-producer, single-consumer queue, the + substitute for a `wtc::queue` proven SPSC. `test/spsc_stress.cpp` + checks order, count and destruction, and CI runs it under ThreadSanitizer. diff --git a/runtime/spsc_queue.hpp b/runtime/spsc_queue.hpp new file mode 100644 index 0000000..3079936 --- /dev/null +++ b/runtime/spsc_queue.hpp @@ -0,0 +1,156 @@ +#pragma once + +// An unbounded single-producer, single-consumer queue, substituted by +// --select-queue-impl for a wtc::queue proven SPSC. +// +// The fact the compiler must prove: over the whole execution, at most one +// thread ever calls push and at most one thread ever calls try_pop. They may +// be the same thread. +// +// What breaks if that is false: two concurrent pushers race on the producer +// cursor and lose or duplicate items, or corrupt a block; two concurrent +// poppers can return one item twice and free a block twice. This is a data +// race, not a slowdown. +// +// It is unbounded, like wtc::queue, so that the substitution is exact: +// push never fails and never blocks. A bounded ring would have to drop an +// item, or wait for the consumer, when full; waiting deadlocks a program in +// which one thread is both the producer and the consumer, which the SPSC +// verdict allows. +// +// Storage is a linked list of fixed-size blocks. Each block counts how many +// of its slots the producer has published. The consumer frees a block only +// once it has drained it and seen the next one; publishing `next` is the +// producer's last access to a block, so nothing touches a freed block. +// +// Measured numbers: not yet. See benchmarks/README.md once they are taken at +// the pinned toolchain. + +#include +#include +#include +#include +#include +#include + +namespace wtc::runtime { + +template +class spsc_queue { + static_assert(BlockSize >= 1, "spsc_queue needs at least one slot per block"); + + struct slot { + alignas(T) unsigned char bytes[sizeof(T)]; + + T *get() { return std::launder(reinterpret_cast(bytes)); } + }; + + struct block { + // Slots [0, written) hold constructed items. Written only by the + // producer, with release, so an acquire read sees those items. + std::atomic written{0}; + std::atomic next{nullptr}; + slot slots[BlockSize]; + }; + + // The consumer's cursor and the producer's cursor are on separate cache + // lines, so neither side's updates invalidate the other's. + struct alignas(64) consumer_side { + block *head; + std::size_t index = 0; + }; + struct alignas(64) producer_side { + block *tail; + std::size_t index = 0; + }; + + consumer_side consumer_; + producer_side producer_; + +public: + [[clang::annotate("wtc_impl_spsc_construction")]] + spsc_queue() { + block *first = new block; + consumer_.head = first; + producer_.tail = first; + } + + spsc_queue(const spsc_queue &) = delete; + spsc_queue &operator=(const spsc_queue &) = delete; + + spsc_queue(spsc_queue &&) = delete; + spsc_queue &operator=(spsc_queue &&) = delete; + + // No other thread may be using the queue by now, so plain loads are + // enough: the joins or whatever ended the threads ordered their writes. + [[clang::annotate("wtc_impl_spsc_dtor")]] + ~spsc_queue() { + block *b = consumer_.head; + std::size_t from = consumer_.index; + while (b) { + std::size_t written = b->written.load(std::memory_order_relaxed); + for (std::size_t i = from; i < written; ++i) + b->slots[i].get()->~T(); + block *next = b->next.load(std::memory_order_relaxed); + delete b; + b = next; + from = 0; + } + } + + [[clang::annotate("wtc_impl_spsc_push")]] + void push(const T &new_value) { + if (producer_.index == BlockSize) { + block *fresh = new block; + producer_.tail->next.store(fresh, std::memory_order_release); + producer_.tail = fresh; + producer_.index = 0; + } + + std::size_t i = producer_.index; + ::new (static_cast(producer_.tail->slots[i].bytes)) T(new_value); + producer_.tail->written.store(i + 1, std::memory_order_release); + producer_.index = i + 1; + } + + // The allocation-free form, for code that names spsc_queue directly. + std::optional try_pop_optional() { + while (true) { + block *b = consumer_.head; + std::size_t i = consumer_.index; + + if (i < BlockSize) { + if (i == b->written.load(std::memory_order_acquire)) + return std::nullopt; + + T *item = b->slots[i].get(); + std::optional result(std::move(*item)); + item->~T(); + consumer_.index = i + 1; + return result; + } + + // This block is drained. The producer moves on only after + // filling it, so if there is no next block, the queue is empty. + block *next = b->next.load(std::memory_order_acquire); + if (!next) + return std::nullopt; + consumer_.head = next; + consumer_.index = 0; + delete b; + } + } + + // The form substituted for wtc::queue::try_pop, whose callers expect a + // shared_ptr, null when the queue is empty. Matching that interface costs + // one allocation per item; try_pop_optional above avoids it. + [[clang::annotate("wtc_impl_spsc_try_pop")]] + std::shared_ptr try_pop() { + std::optional value = try_pop_optional(); + if (!value) + return std::shared_ptr(); + return std::make_shared(std::move(*value)); + } +}; + +} diff --git a/runtime/test/spsc_stress.cpp b/runtime/test/spsc_stress.cpp new file mode 100644 index 0000000..aa6eda1 --- /dev/null +++ b/runtime/test/spsc_stress.cpp @@ -0,0 +1,86 @@ +// Stress test for runtime/spsc_queue.hpp. Build with -fsanitize=thread. +// +// One producer and one consumer move a long sequence through the queue. +// The consumer checks that every item arrives exactly once and in order. A +// small block size makes block hand-over, the delicate part, happen on +// almost every push. A second run leaves items in the queue at destruction +// and checks that every one of them is destroyed. + +#include "spsc_queue.hpp" + +#include +#include +#include +#include + +namespace { + +constexpr long kItems = 10'000'000; + +[[noreturn]] void fail(const char *what, long detail) { + std::fprintf(stderr, "FAIL: %s (%ld)\n", what, detail); + std::exit(1); +} + +template +void order_and_count() { + wtc::runtime::spsc_queue queue; + + std::thread producer([&] { + for (long i = 0; i < kItems; ++i) + queue.push(i); + }); + + long expected = 0; + long sum = 0; + while (expected < kItems) { + if (auto value = queue.try_pop_optional()) { + if (*value != expected) + fail("out of order or duplicated", *value); + sum += *value; + ++expected; + } + } + producer.join(); + + if (queue.try_pop_optional()) + fail("item left after the last one", 0); + if (sum != kItems * (kItems - 1) / 2) + fail("wrong sum", sum); +} + +std::atomic live{0}; + +struct counted { + long value; + explicit counted(long v) : value(v) { live.fetch_add(1, std::memory_order_relaxed); } + counted(const counted &other) : value(other.value) { live.fetch_add(1, std::memory_order_relaxed); } + counted(counted &&other) noexcept : value(other.value) { live.fetch_add(1, std::memory_order_relaxed); } + ~counted() { live.fetch_sub(1, std::memory_order_relaxed); } +}; + +void destroys_what_is_left() { + { + wtc::runtime::spsc_queue queue; + for (long i = 0; i < 103; ++i) + queue.push(counted(i)); + for (long i = 0; i < 37; ++i) { + auto popped = queue.try_pop(); + if (!popped || popped->value != i) + fail("shared_ptr pop returned the wrong item", i); + } + } + if (live.load() != 0) + fail("items leaked or destroyed twice", live.load()); +} + +} + +int main() { + order_and_count<1>(); + order_and_count<4>(); + order_and_count<1024>(); + destroys_what_is_left(); + std::puts("spsc_stress: ok"); + return 0; +}