Add an unbounded SPSC queue to substitute for a queue proven SPSC

.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.

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<T>` proven SPSC. `test/spsc_stress.cpp`
+  checks order, count and destruction, and CI runs it under ThreadSanitizer.

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<T> 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<T>, 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 <atomic>
+#include <cstddef>
+#include <memory>
+#include <new>
+#include <optional>
+#include <utility>
+
+namespace wtc::runtime {
+
+template <typename T, std::size_t BlockSize = 1024>
+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<T *>(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<std::size_t> written{0};
+        std::atomic<block *> 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<void *>(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<T> 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<T> 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<T>::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<T> try_pop() {
+        std::optional<T> value = try_pop_optional();
+        if (!value)
+            return std::shared_ptr<T>();
+        return std::make_shared<T>(std::move(*value));
+    }
+};
+
+}

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 <atomic>
+#include <cstdio>
+#include <cstdlib>
+#include <thread>
+
+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 <std::size_t BlockSize>
+void order_and_count() {
+    wtc::runtime::spsc_queue<long, BlockSize> 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<long> 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<counted, 4> 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;
+}