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; +}