weaselab/weaseldb
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

StaticThreadPipeline

Browse Source
This commit is contained in:
Andrew Noyes
2025-08-26 15:13:16 -04:00
parent 0b63e24b98
commit 6dbf29d1e1
4 changed files with 244 additions and 321 deletions
Download Patch File Download Diff File Expand all files Collapse all files

33
benchmarks/bench_thread_pipeline.cpp
View File

@@ -3,7 +3,6 @@
#include <latch> #include <latch>
#include <nanobench.h> #include <nanobench.h>
#include <thread> #include <thread>
#include <vector>
int main() { int main() {
{ {
@@ -25,18 +24,15 @@ int main() {
} }
}); });
std::vector<int> threads_per_stage = {1}; StaticThreadPipeline<std::latch *, WaitStrategy::WaitIfStageEmpty, 1>
ThreadPipeline<std::latch *> pipeline(LOG_PIPELINE_SIZE, threads_per_stage); pipeline(LOG_PIPELINE_SIZE);
std::latch done{0}; std::latch done{0};
// Stage 0 consumer thread // Stage 0 consumer thread
std::thread stage0_thread([&pipeline, &done]() { std::thread stage0_thread([&pipeline, &done]() {
const int stage = 0;
const int thread_id = 0;
for (;;) { for (;;) {
auto guard = pipeline.acquire(stage, thread_id); auto guard = pipeline.acquire<0, 0>();
for (auto &item : guard.batch) { for (auto &item : guard.batch) {
for (volatile int i = 0; i < BUSY_ITERS; i = i + 1) { for (volatile int i = 0; i < BUSY_ITERS; i = i + 1) {
@@ -94,19 +90,15 @@ int main() {
.warmup(100); .warmup(100);
for (int batch_size : {1, 4, 16, 64, 256}) { for (int batch_size : {1, 4, 16, 64, 256}) {
std::vector<int> threads_per_stage = {1}; StaticThreadPipeline<std::latch *, WaitStrategy::WaitIfStageEmpty, 1>
ThreadPipeline<std::latch *> pipeline(LOG_PIPELINE_SIZE, pipeline(LOG_PIPELINE_SIZE);
threads_per_stage);
std::latch done{0}; std::latch done{0};
// Stage 0 consumer thread // Stage 0 consumer thread
std::thread stage0_thread([&pipeline, &done]() { std::thread stage0_thread([&pipeline, &done]() {
const int stage = 0;
const int thread_id = 0;
for (;;) { for (;;) {
auto guard = pipeline.acquire(stage, thread_id); auto guard = pipeline.acquire<0, 0>();
for (auto &item : guard.batch) { for (auto &item : guard.batch) {
for (volatile int i = 0; i < BUSY_ITERS; i = i + 1) { for (volatile int i = 0; i < BUSY_ITERS; i = i + 1) {
@@ -162,18 +154,15 @@ int main() {
constexpr int BUSY_ITERS = constexpr int BUSY_ITERS =
10; // Light work to emphasize coordination overhead 10; // Light work to emphasize coordination overhead
std::vector<int> threads_per_stage = {1, 1}; // Two stages StaticThreadPipeline<std::latch *, strategy, 1, 1> pipeline(
ThreadPipeline<std::latch *, strategy> pipeline(LOG_PIPELINE_SIZE, LOG_PIPELINE_SIZE);
threads_per_stage);
std::latch done{0}; std::latch done{0};
// Stage 0 worker // Stage 0 worker
std::thread stage0_thread([&pipeline, &done]() { std::thread stage0_thread([&pipeline, &done]() {
const int stage = 0;
const int thread_id = 0;
for (;;) { for (;;) {
auto guard = pipeline.acquire(stage, thread_id); auto guard = pipeline.template acquire<0, 0>();
for (auto &item : guard.batch) { for (auto &item : guard.batch) {
for (volatile int i = 0; i < BUSY_ITERS; i = i + 1) { for (volatile int i = 0; i < BUSY_ITERS; i = i + 1) {
} }
@@ -185,10 +174,8 @@ int main() {
// Stage 1 worker (final stage - always calls futex wake) // Stage 1 worker (final stage - always calls futex wake)
std::thread stage1_thread([&pipeline, &done]() { std::thread stage1_thread([&pipeline, &done]() {
const int stage = 1;
const int thread_id = 0;
for (;;) { for (;;) {
auto guard = pipeline.acquire(stage, thread_id); auto guard = pipeline.template acquire<1, 0>();
for (auto &item : guard.batch) { for (auto &item : guard.batch) {
for (volatile int i = 0; i < BUSY_ITERS; i = i + 1) { for (volatile int i = 0; i < BUSY_ITERS; i = i + 1) {
} }

41
src/http_handler.hpp
View File

@@ -64,33 +64,47 @@ struct HttpConnectionState {
*/ */
struct HttpHandler : ConnectionHandler { struct HttpHandler : ConnectionHandler {
HttpHandler() { HttpHandler() {
for (int threadId = 0; threadId < kFinalStageThreads; ++threadId) { finalStageThreads.emplace_back([this]() {
finalStageThreads.emplace_back([this, threadId]() { pthread_setname_np(pthread_self(), "stage-1-0");
pthread_setname_np(pthread_self(),
("stage-1-" + std::to_string(threadId)).c_str());
for (;;) { for (;;) {
auto guard = pipeline.acquire(1, threadId); auto guard = pipeline.acquire<1, 0>();
for (auto it = guard.batch.begin(); it != guard.batch.end(); ++it) { for (auto it = guard.batch.begin(); it != guard.batch.end(); ++it) {
if ((it.index() % kFinalStageThreads) == threadId) { if ((it.index() % 2) == 0) { // Thread 0 handles even indices
auto &c = *it; auto &c = *it;
if (!c) { if (!c) {
return; return;
} }
auto *state = static_cast<HttpConnectionState *>(c->user_data); auto *state = static_cast<HttpConnectionState *>(c->user_data);
TRACE_EVENT("http", "pipeline thread", TRACE_EVENT("http", "release",
perfetto::Flow::Global(state->request_id)); perfetto::Flow::Global(state->request_id));
Server::release_back_to_server(std::move(c)); Server::release_back_to_server(std::move(c));
} }
} }
} }
}); });
finalStageThreads.emplace_back([this]() {
pthread_setname_np(pthread_self(), "stage-1-1");
for (;;) {
auto guard = pipeline.acquire<1, 1>();
for (auto it = guard.batch.begin(); it != guard.batch.end(); ++it) {
if ((it.index() % 2) == 1) { // Thread 1 handles odd indices
auto &c = *it;
if (!c) {
return;
} }
auto *state = static_cast<HttpConnectionState *>(c->user_data);
TRACE_EVENT("http", "release",
perfetto::Flow::Global(state->request_id));
Server::release_back_to_server(std::move(c));
}
}
}
});
stage0Thread = std::thread{[this]() { stage0Thread = std::thread{[this]() {
pthread_setname_np(pthread_self(), "stage-0"); pthread_setname_np(pthread_self(), "stage-0");
for (;;) { for (;;) {
auto guard = pipeline.acquire(0, 0, 0, false); auto guard = pipeline.acquire<0, 0>();
for (auto it = guard.batch.begin(); it != guard.batch.end(); ++it) { for (auto &c : guard.batch) {
auto &c = *it;
if (!c) { if (!c) {
return; return;
} }
@@ -102,7 +116,7 @@ struct HttpHandler : ConnectionHandler {
} }
~HttpHandler() { ~HttpHandler() {
{ {
auto guard = pipeline.push(kFinalStageThreads, true); auto guard = pipeline.push(2, true);
for (auto &c : guard.batch) { for (auto &c : guard.batch) {
c = {}; c = {};
} }
@@ -137,8 +151,9 @@ struct HttpHandler : ConnectionHandler {
private: private:
static constexpr int kFinalStageThreads = 2; static constexpr int kFinalStageThreads = 2;
static constexpr int kLogSize = 12; static constexpr int kLogSize = 12;
ThreadPipeline<std::unique_ptr<Connection>> pipeline{ StaticThreadPipeline<std::unique_ptr<Connection>,
kLogSize, {/*noop serial thread*/ 1, kFinalStageThreads}}; WaitStrategy::WaitIfStageEmpty, 1, 2>
pipeline{kLogSize};
std::thread stage0Thread; std::thread stage0Thread;
std::vector<std::thread> finalStageThreads; std::vector<std::thread> finalStageThreads;

423
src/thread_pipeline.hpp
View File

@@ -1,117 +1,113 @@
#pragma once #pragma once
#include <array>
#include <atomic> #include <atomic>
#include <cassert> #include <cassert>
#include <cstddef> #include <cstddef>
#include <cstdint> #include <cstdint>
#include <cstdio> #include <cstdio>
#include <cstdlib> #include <cstdlib>
#include <functional>
#include <iterator> #include <iterator>
#include <numeric>
#include <utility> #include <utility>
#include <vector> #include <vector>
// Topology configuration - separates stage layout from pipeline logic
struct PipelineTopology {
std::vector<int> threads_per_stage;
int num_stages;
int total_threads;
PipelineTopology(const std::vector<int> &threads)
: threads_per_stage(threads), num_stages(threads.size()) {
assert(!threads.empty());
assert(std::all_of(threads.begin(), threads.end(),
[](int t) { return t > 0; }));
total_threads = std::accumulate(threads.begin(), threads.end(), 0);
}
// Get the flat array offset for the first thread of a stage
int stage_offset(int stage) const {
assert(stage >= 0 && stage < num_stages);
int offset = 0;
for (int i = 0; i < stage; ++i) {
offset += threads_per_stage[i];
}
return offset;
}
// Get flat array index for a specific thread in a stage
int thread_index(int stage, int thread) const {
assert(stage >= 0 && stage < num_stages);
assert(thread >= 0 && thread < threads_per_stage[stage]);
return stage_offset(stage) + thread;
}
// Get number of threads in the previous stage (for initialization)
int prev_stage_thread_count(int stage) const {
return (stage == 0) ? 1 : threads_per_stage[stage - 1];
}
};
// Wait strategies for controlling thread blocking behavior when no work is // Wait strategies for controlling thread blocking behavior when no work is
// available // available
enum class WaitStrategy { enum class WaitStrategy {
// Never block - threads busy-wait (spin) when no work available.
// Stage threads will always use 100% CPU even when idle.
// Requires dedicated CPU cores to avoid scheduler thrashing.
// Use when: latency is critical and you have spare cores.
Never, Never,
// Block only when all upstream stages are idle (no new work entering
// pipeline).
// Downstream threads busy-wait if upstream has work but not for their stage.
// Eliminates futex notifications between stages, reduces to 0% CPU when idle.
// Requires dedicated cores to avoid priority inversion when pipeline has
// work.
// Use when: high throughput with spare cores and sustained workloads.
WaitIfUpstreamIdle, WaitIfUpstreamIdle,
// Block when individual stages are empty (original behavior).
// Each stage waits independently on its input sources.
// Safe for shared CPU environments, works well with variable workloads.
// Use when: general purpose, shared cores, or unpredictable workloads.
WaitIfStageEmpty, WaitIfStageEmpty,
}; };
// Core thread state - extracted from pipeline concerns // Core thread state
struct ThreadState { struct ThreadState {
// Where this thread has published up to
alignas(128) std::atomic<uint32_t> pops{0}; alignas(128) std::atomic<uint32_t> pops{0};
// Where this thread will publish to the next time it publishes, or if idle
// where it has published to
uint32_t local_pops{0}; uint32_t local_pops{0};
// Where the previous stage's threads have published up to last we checked
std::vector<uint32_t> last_push_read; std::vector<uint32_t> last_push_read;
bool last_stage; bool last_stage;
}; };
// Core pipeline algorithms - independent of storage layout // Compile-time topology configuration
namespace PipelineAlgorithms { template <int... ThreadsPerStage> struct StaticPipelineTopology {
static_assert(sizeof...(ThreadsPerStage) > 0,
"Must specify at least one stage");
static_assert(((ThreadsPerStage > 0) && ...),
"All stages must have at least one thread");
// Calculate how many items can be safely acquired from a stage static constexpr int num_stages = sizeof...(ThreadsPerStage);
template <WaitStrategy wait_strategy> static constexpr std::array<int, num_stages> threads_per_stage = {
uint32_t calculate_safe_len(const PipelineTopology &topology, ThreadsPerStage...};
std::vector<ThreadState> &all_threads, static constexpr int total_threads = (ThreadsPerStage + ...);
std::atomic<uint32_t> &pushes, int stage,
int thread_in_stage, bool may_block) { // Compile-time stage offset calculation
int thread_idx = topology.thread_index(stage, thread_in_stage); template <int Stage> static constexpr int stage_offset() {
static_assert(Stage >= 0 && Stage < num_stages,
"Stage index out of bounds");
if constexpr (Stage == 0) {
return 0;
} else {
return stage_offset<Stage - 1>() + threads_per_stage[Stage - 1];
}
}
// Compile-time thread index calculation
template <int Stage, int Thread> static constexpr int thread_index() {
static_assert(Stage >= 0 && Stage < num_stages,
"Stage index out of bounds");
static_assert(Thread >= 0 && Thread < threads_per_stage[Stage],
"Thread index out of bounds");
return stage_offset<Stage>() + Thread;
}
// Compile-time previous stage thread count
template <int Stage> static constexpr int prev_stage_thread_count() {
static_assert(Stage >= 0 && Stage < num_stages,
"Stage index out of bounds");
if constexpr (Stage == 0) {
return 1;
} else {
return threads_per_stage[Stage - 1];
}
}
};
// Static pipeline algorithms - compile-time specialized versions
namespace StaticPipelineAlgorithms {
template <WaitStrategy wait_strategy, typename Topology, int Stage,
int ThreadInStage>
uint32_t calculate_safe_len(
std::array<ThreadState, Topology::total_threads> &all_threads,
std::atomic<uint32_t> &pushes, bool may_block) {
constexpr int thread_idx =
Topology::template thread_index<Stage, ThreadInStage>();
auto &thread = all_threads[thread_idx]; auto &thread = all_threads[thread_idx];
uint32_t safe_len = UINT32_MAX; uint32_t safe_len = UINT32_MAX;
// Check all threads from the previous stage (or pushes for stage 0) constexpr int prev_stage_threads =
int prev_stage_threads = topology.prev_stage_thread_count(stage); Topology::template prev_stage_thread_count<Stage>();
for (int i = 0; i < prev_stage_threads; ++i) {
auto &last_push =
(stage == 0) ? pushes
: all_threads[topology.thread_index(stage - 1, i)].pops;
if (thread.last_push_read[i] == thread.local_pops) { // Compile-time loop over previous stage threads
// Re-read with memory order and try again [&]<std::size_t... Is>(std::index_sequence<Is...>) {
thread.last_push_read[i] = last_push.load(std::memory_order_acquire); (
if (thread.last_push_read[i] == thread.local_pops) { [&] {
auto &last_push = [&]() -> std::atomic<uint32_t> & {
if constexpr (Stage == 0) {
return pushes;
} else {
constexpr int prev_thread_idx =
Topology::template thread_index<Stage - 1, Is>();
return all_threads[prev_thread_idx].pops;
}
}();
if (thread.last_push_read[Is] == thread.local_pops) {
thread.last_push_read[Is] =
last_push.load(std::memory_order_acquire);
if (thread.last_push_read[Is] == thread.local_pops) {
if (!may_block) { if (!may_block) {
return 0; safe_len = 0;
return;
} }
if constexpr (wait_strategy == WaitStrategy::Never) { if constexpr (wait_strategy == WaitStrategy::Never) {
@@ -124,29 +120,36 @@ uint32_t calculate_safe_len(const PipelineTopology &topology,
} }
} else { } else {
static_assert(wait_strategy == WaitStrategy::WaitIfStageEmpty); static_assert(wait_strategy == WaitStrategy::WaitIfStageEmpty);
last_push.wait(thread.last_push_read[i], std::memory_order_relaxed); last_push.wait(thread.last_push_read[Is],
std::memory_order_relaxed);
} }
thread.last_push_read[i] = last_push.load(std::memory_order_acquire); thread.last_push_read[Is] =
last_push.load(std::memory_order_acquire);
} }
} }
safe_len = std::min(safe_len, thread.last_push_read[i] - thread.local_pops); safe_len =
} std::min(safe_len, thread.last_push_read[Is] - thread.local_pops);
}(),
...);
}(std::make_index_sequence<prev_stage_threads>{});
return safe_len; return safe_len;
} }
// Update thread pops counter after processing template <WaitStrategy wait_strategy, typename Topology, int Stage,
template <WaitStrategy wait_strategy> int ThreadInStage>
void update_thread_pops(const PipelineTopology &topology, void update_thread_pops(
std::vector<ThreadState> &all_threads, int stage, std::array<ThreadState, Topology::total_threads> &all_threads,
int thread_in_stage, uint32_t local_pops) { uint32_t local_pops) {
int thread_idx = topology.thread_index(stage, thread_in_stage); constexpr int thread_idx =
Topology::template thread_index<Stage, ThreadInStage>();
auto &thread_state = all_threads[thread_idx]; auto &thread_state = all_threads[thread_idx];
if constexpr (wait_strategy == WaitStrategy::WaitIfStageEmpty) { if constexpr (wait_strategy == WaitStrategy::WaitIfStageEmpty) {
thread_state.pops.store(local_pops, std::memory_order_seq_cst); thread_state.pops.store(local_pops, std::memory_order_seq_cst);
thread_state.pops.notify_all(); thread_state.pops.notify_all();
} else if (thread_state.last_stage) { } else if constexpr (Stage == Topology::num_stages - 1) { // last stage
thread_state.pops.store(local_pops, std::memory_order_seq_cst); thread_state.pops.store(local_pops, std::memory_order_seq_cst);
thread_state.pops.notify_all(); thread_state.pops.notify_all();
} else { } else {
@@ -154,24 +157,22 @@ void update_thread_pops(const PipelineTopology &topology,
} }
} }
// Check if producer can proceed without stomping the ring buffer template <typename Topology>
// Returns: 0 = can proceed, 1 = should retry, 2 = cannot proceed (return empty int check_producer_capacity(
// guard) std::array<ThreadState, Topology::total_threads> &all_threads,
inline int check_producer_capacity(const PipelineTopology &topology, uint32_t slot, uint32_t size, uint32_t slot_count, bool block) {
std::vector<ThreadState> &all_threads, constexpr int last_stage = Topology::num_stages - 1;
uint32_t slot, uint32_t size, constexpr int last_stage_offset =
uint32_t slot_count, bool block) { Topology::template stage_offset<last_stage>();
// Check against last stage threads constexpr int last_stage_thread_count =
int last_stage = topology.num_stages - 1; Topology::threads_per_stage[last_stage];
int last_stage_offset = topology.stage_offset(last_stage);
int last_stage_thread_count = topology.threads_per_stage[last_stage];
for (int i = 0; i < last_stage_thread_count; ++i) { for (int i = 0; i < last_stage_thread_count; ++i) {
auto &thread = all_threads[last_stage_offset + i]; auto &thread = all_threads[last_stage_offset + i];
uint32_t pops = thread.pops.load(std::memory_order_acquire); uint32_t pops = thread.pops.load(std::memory_order_acquire);
if (slot + size - pops > slot_count) { if (slot + size - pops > slot_count) {
if (!block) { if (!block) {
return 2; // Cannot proceed, caller should return empty guard return 2; // Cannot proceed
} }
thread.pops.wait(pops, std::memory_order_relaxed); thread.pops.wait(pops, std::memory_order_relaxed);
return 1; // Should retry return 1; // Should retry
@@ -179,77 +180,24 @@ inline int check_producer_capacity(const PipelineTopology &topology,
} }
return 0; // Can proceed return 0; // Can proceed
} }
} // namespace PipelineAlgorithms } // namespace StaticPipelineAlgorithms
// Multi-stage lock-free pipeline for high-throughput inter-thread // Static multi-stage lock-free pipeline
// communication. template <class T, WaitStrategy wait_strategy, int... ThreadsPerStage>
// struct StaticThreadPipeline {
// Overview: using Topology = StaticPipelineTopology<ThreadsPerStage...>;
// - Items flow from producers through multiple processing stages (stage 0 ->
// stage 1 -> ... -> final stage)
// - Each stage can have multiple worker threads processing items in parallel
// - Uses a shared ring buffer with atomic counters for lock-free coordination
// - Supports batch processing for efficiency
//
// Architecture:
// - Producers: External threads that add items to the pipeline via push()
// - Stages: Processing stages numbered 0, 1, 2, ... that consume items via
// acquire()
// - Items flow: Producers -> Stage 0 -> Stage 1 -> ... -> Final Stage
//
// Usage Pattern:
// // Producer threads (external to pipeline stages - add items for stage 0 to
// consume): auto guard = pipeline.push(batchSize, /*block=*/true); for (auto&
// item : guard.batch) {
// // Initialize item data
// }
// // Guard destructor publishes batch to stage 0 consumers
//
// // Stage worker threads (process items and pass to next stage):
// auto guard = pipeline.acquire(stageNum, threadId, maxBatch,
// /*mayBlock=*/true); for (auto& item : guard.batch) {
// // Process item
// }
// // Guard destructor marks items as consumed and available to next stage
//
// Memory Model:
// - Ring buffer size must be power of 2 for efficient masking
// - Actual ring slots accessed via: index & (slotCount - 1)
// - 128-byte aligned atomics prevent false sharing between CPU cache lines
//
// Thread Safety:
// - Fully lock-free using atomic operations with acquire/release memory
// ordering
// - Uses C++20 atomic wait/notify for efficient blocking when no work available
// - RAII guards ensure proper cleanup even with exceptions
//
// Refactored Design:
// - Topology configuration separated from pipeline logic
// - Flattened thread storage replaces nested vectors
// - Core algorithms extracted into pure functions
// - Ready for compile-time static version implementation
template <class T, WaitStrategy wait_strategy = WaitStrategy::WaitIfStageEmpty>
struct ThreadPipeline {
// Constructor now takes topology configuration explicit StaticThreadPipeline(int lgSlotCount)
ThreadPipeline(int lgSlotCount, const PipelineTopology &topo) : slot_count(1 << lgSlotCount), slot_count_mask(slot_count - 1),
: topology(topo), slot_count(1 << lgSlotCount),
slot_count_mask(slot_count - 1), all_threads(topology.total_threads),
ring(slot_count) { ring(slot_count) {
// Otherwise we can't tell the difference between full and empty
assert(!(slot_count_mask & 0x80000000)); assert(!(slot_count_mask & 0x80000000));
initialize_all_threads(); initialize_all_threads();
} }
// Legacy constructor for backward compatibility StaticThreadPipeline(StaticThreadPipeline const &) = delete;
ThreadPipeline(int lgSlotCount, const std::vector<int> &threadsPerStage) StaticThreadPipeline &operator=(StaticThreadPipeline const &) = delete;
: ThreadPipeline(lgSlotCount, PipelineTopology(threadsPerStage)) {} StaticThreadPipeline(StaticThreadPipeline &&) = delete;
StaticThreadPipeline &operator=(StaticThreadPipeline &&) = delete;
ThreadPipeline(ThreadPipeline const &) = delete;
ThreadPipeline &operator=(ThreadPipeline const &) = delete;
ThreadPipeline(ThreadPipeline &&) = delete;
ThreadPipeline &operator=(ThreadPipeline &&) = delete;
struct Batch { struct Batch {
Batch() : ring(), begin_(), end_() {} Batch() : ring(), begin_(), end_() {}
@@ -317,7 +265,6 @@ struct ThreadPipeline {
} }
friend bool operator<(const Iterator &lhs, const Iterator &rhs) { friend bool operator<(const Iterator &lhs, const Iterator &rhs) {
assert(lhs.ring == rhs.ring); assert(lhs.ring == rhs.ring);
// Handle potential uint32_t wraparound by using signed difference
return static_cast<int32_t>(lhs.index_ - rhs.index_) < 0; return static_cast<int32_t>(lhs.index_ - rhs.index_) < 0;
} }
friend bool operator<=(const Iterator &lhs, const Iterator &rhs) { friend bool operator<=(const Iterator &lhs, const Iterator &rhs) {
@@ -333,9 +280,6 @@ struct ThreadPipeline {
return static_cast<int32_t>(lhs.index_ - rhs.index_) >= 0; return static_cast<int32_t>(lhs.index_ - rhs.index_) >= 0;
} }
/// Returns the ring buffer index (0 to ring->size()-1) for this iterator
/// position. Useful for distributing work across multiple threads by
/// using modulo operations.
uint32_t index() const { return index_ & (ring->size() - 1); } uint32_t index() const { return index_ & (ring->size() - 1); }
private: private:
@@ -356,7 +300,7 @@ struct ThreadPipeline {
} }
private: private:
friend struct ThreadPipeline; friend struct StaticThreadPipeline;
Batch(std::vector<T> *const ring, uint32_t begin_, uint32_t end_) Batch(std::vector<T> *const ring, uint32_t begin_, uint32_t end_)
: ring(ring), begin_(begin_), end_(end_) {} : ring(ring), begin_(begin_), end_(end_) {}
std::vector<T> *const ring; std::vector<T> *const ring;
@@ -364,28 +308,29 @@ struct ThreadPipeline {
uint32_t end_; uint32_t end_;
}; };
private: // Static thread storage - fixed size array
// Pipeline configuration std::array<ThreadState, Topology::total_threads> all_threads;
PipelineTopology topology;
// Core state private:
alignas(128) std::atomic<uint32_t> slots{0}; alignas(128) std::atomic<uint32_t> slots{0};
alignas(128) std::atomic<uint32_t> pushes{0}; alignas(128) std::atomic<uint32_t> pushes{0};
const uint32_t slot_count; const uint32_t slot_count;
const uint32_t slot_count_mask; const uint32_t slot_count_mask;
// Flattened thread storage - single array instead of nested vectors
std::vector<ThreadState> all_threads;
// Ring buffer
std::vector<T> ring; std::vector<T> ring;
void initialize_all_threads() { void initialize_all_threads() {
for (int stage = 0; stage < topology.num_stages; ++stage) { [&]<std::size_t... StageIndices>(std::index_sequence<StageIndices...>) {
int stage_offset = topology.stage_offset(stage); (init_stage_threads<StageIndices>(), ...);
int stage_thread_count = topology.threads_per_stage[stage]; }(std::make_index_sequence<Topology::num_stages>{});
int prev_stage_threads = topology.prev_stage_thread_count(stage); }
bool is_last_stage = (stage == topology.num_stages - 1);
template <int Stage> void init_stage_threads() {
constexpr int stage_offset = Topology::template stage_offset<Stage>();
constexpr int stage_thread_count = Topology::threads_per_stage[Stage];
constexpr int prev_stage_threads =
Topology::template prev_stage_thread_count<Stage>();
constexpr bool is_last_stage = (Stage == Topology::num_stages - 1);
for (int thread = 0; thread < stage_thread_count; ++thread) { for (int thread = 0; thread < stage_thread_count; ++thread) {
auto &thread_state = all_threads[stage_offset + thread]; auto &thread_state = all_threads[stage_offset + thread];
@@ -393,16 +338,15 @@ private:
thread_state.last_push_read = std::vector<uint32_t>(prev_stage_threads); thread_state.last_push_read = std::vector<uint32_t>(prev_stage_threads);
} }
} }
}
Batch acquire_helper(int stage, int thread, uint32_t maxBatch, template <int Stage, int Thread>
bool may_block) { Batch acquire_helper(uint32_t maxBatch, bool mayBlock) {
int thread_idx = topology.thread_index(stage, thread); constexpr int thread_idx = Topology::template thread_index<Stage, Thread>();
auto &thread_state = all_threads[thread_idx]; auto &thread_state = all_threads[thread_idx];
uint32_t begin = thread_state.local_pops & slot_count_mask; uint32_t begin = thread_state.local_pops & slot_count_mask;
uint32_t len = PipelineAlgorithms::calculate_safe_len<wait_strategy>( uint32_t len = StaticPipelineAlgorithms::calculate_safe_len<
topology, all_threads, pushes, stage, thread, may_block); wait_strategy, Topology, Stage, Thread>(all_threads, pushes, mayBlock);
if (maxBatch != 0) { if (maxBatch != 0) {
len = std::min(len, maxBatch); len = std::min(len, maxBatch);
@@ -417,32 +361,37 @@ private:
} }
public: public:
struct StageGuard { template <int Stage, int Thread> struct StageGuard {
Batch batch; Batch batch;
~StageGuard() { ~StageGuard() {
if (cleanup_func) { if (!batch.empty()) {
cleanup_func(); StaticPipelineAlgorithms::update_thread_pops<wait_strategy, Topology,
Stage, Thread>(
pipeline->all_threads, local_pops);
} }
} }
StageGuard(StageGuard const &) = delete; StageGuard(StageGuard const &) = delete;
StageGuard &operator=(StageGuard const &) = delete; StageGuard &operator=(StageGuard const &) = delete;
StageGuard(StageGuard &&other) noexcept StageGuard(StageGuard &&other) noexcept
: batch(other.batch), : batch(other.batch), local_pops(other.local_pops),
cleanup_func(std::exchange(other.cleanup_func, nullptr)) {} pipeline(std::exchange(other.pipeline, nullptr)) {}
StageGuard &operator=(StageGuard &&other) noexcept { StageGuard &operator=(StageGuard &&other) noexcept {
batch = other.batch; batch = other.batch;
cleanup_func = std::exchange(other.cleanup_func, nullptr); local_pops = other.local_pops;
pipeline = std::exchange(other.pipeline, nullptr);
return *this; return *this;
} }
private: private:
friend struct ThreadPipeline; friend struct StaticThreadPipeline;
std::function<void()> cleanup_func; uint32_t local_pops;
StaticThreadPipeline *pipeline;
StageGuard(Batch batch, std::function<void()> cleanup) StageGuard(Batch batch, uint32_t local_pops, StaticThreadPipeline *pipeline)
: batch(batch), cleanup_func(batch.empty() ? nullptr : cleanup) {} : batch(batch), local_pops(local_pops),
pipeline(batch.empty() ? nullptr : pipeline) {}
}; };
struct ProducerGuard { struct ProducerGuard {
@@ -452,8 +401,6 @@ public:
if (tp == nullptr) { if (tp == nullptr) {
return; return;
} }
// Wait for earlier slots to finish being published, since publishing
// implies that all previous slots were also published.
for (;;) { for (;;) {
uint32_t p = tp->pushes.load(std::memory_order_acquire); uint32_t p = tp->pushes.load(std::memory_order_acquire);
if (p == old_slot) { if (p == old_slot) {
@@ -461,65 +408,38 @@ public:
} }
tp->pushes.wait(p, std::memory_order_relaxed); tp->pushes.wait(p, std::memory_order_relaxed);
} }
// Publish. seq_cst so that the notify can't be ordered before the store
tp->pushes.store(new_slot, std::memory_order_seq_cst); tp->pushes.store(new_slot, std::memory_order_seq_cst);
// We have to notify every time, since we don't know if this is the last
// push ever
tp->pushes.notify_all(); tp->pushes.notify_all();
} }
private: private:
friend struct ThreadPipeline; friend struct StaticThreadPipeline;
ProducerGuard() : batch(), tp() {} ProducerGuard() : batch(), tp() {}
ProducerGuard(Batch batch, ThreadPipeline<T, wait_strategy> *tp, ProducerGuard(Batch batch, StaticThreadPipeline *tp, uint32_t old_slot,
uint32_t old_slot, uint32_t new_slot) uint32_t new_slot)
: batch(batch), tp(tp), old_slot(old_slot), new_slot(new_slot) {} : batch(batch), tp(tp), old_slot(old_slot), new_slot(new_slot) {}
ThreadPipeline<T, wait_strategy> *const tp; StaticThreadPipeline *const tp;
uint32_t old_slot; uint32_t old_slot;
uint32_t new_slot; uint32_t new_slot;
}; };
// Acquire a batch of items for processing by a consumer thread. // Static acquire - Stage and Thread are compile-time parameters
// stage: which processing stage (0 = first consumer stage after producers) template <int Stage, int Thread>
// thread: thread ID within the stage (0 to threadsPerStage[stage]-1) [[nodiscard]] StageGuard<Stage, Thread> acquire(int maxBatch = 0,
// maxBatch: maximum items to acquire (0 = no limit) bool may_block = true) {
// mayBlock: whether to block waiting for items (false = return empty batch if static_assert(Stage >= 0 && Stage < Topology::num_stages,
// none available) Returns: StageGuard with batch of items to process "Stage index out of bounds");
[[nodiscard]] StageGuard acquire(int stage, int thread, int maxBatch = 0, static_assert(Thread >= 0 && Thread < Topology::threads_per_stage[Stage],
bool mayBlock = true) { "Thread index out of bounds");
assert(stage >= 0 && stage < topology.num_stages);
assert(thread >= 0 && thread < topology.threads_per_stage[stage]);
auto batch = acquire_helper(stage, thread, maxBatch, mayBlock); auto batch = acquire_helper<Stage, Thread>(maxBatch, may_block);
// Create cleanup function that will update thread state on destruction constexpr int thread_idx = Topology::template thread_index<Stage, Thread>();
int thread_idx = topology.thread_index(stage, thread);
uint32_t local_pops = all_threads[thread_idx].local_pops; uint32_t local_pops = all_threads[thread_idx].local_pops;
auto cleanup = [this, stage, thread, local_pops]() { return StageGuard<Stage, Thread>{std::move(batch), local_pops, this};
PipelineAlgorithms::update_thread_pops<wait_strategy>(
topology, all_threads, stage, thread, local_pops);
};
return StageGuard{std::move(batch), cleanup};
} }
// Reserve slots in the ring buffer for a producer thread to fill with items.
// This is used by producer threads to add new items to stage 0 of the
// pipeline.
//
// size: number of slots to reserve (must be > 0 and <= ring buffer capacity)
// block: if true, blocks when ring buffer is full; if false, returns empty
// guard Returns: ProducerGuard with exclusive access to reserved slots
//
// Usage: Fill items in the returned batch, then let guard destructor publish
// them. The guard destructor ensures items are published in the correct
// order.
//
// Preconditions:
// - size > 0 (must request at least one slot)
// - size <= slotCount (cannot request more slots than ring buffer capacity)
// Violating preconditions results in program termination via abort().
[[nodiscard]] ProducerGuard push(uint32_t const size, bool block) { [[nodiscard]] ProducerGuard push(uint32_t const size, bool block) {
if (size == 0) { if (size == 0) {
std::abort(); std::abort();
@@ -534,16 +454,15 @@ public:
slot = slots.load(std::memory_order_relaxed); slot = slots.load(std::memory_order_relaxed);
begin = slot & slot_count_mask; begin = slot & slot_count_mask;
// Use algorithm function to check capacity int capacity_result =
int capacity_result = PipelineAlgorithms::check_producer_capacity( StaticPipelineAlgorithms::check_producer_capacity<Topology>(
topology, all_threads, slot, size, slot_count, block); all_threads, slot, size, slot_count, block);
if (capacity_result == 1) { if (capacity_result == 1) {
continue; // Retry continue;
} }
if (capacity_result == 2) { if (capacity_result == 2) {
return ProducerGuard{}; // Cannot proceed, return empty guard return ProducerGuard{};
} }
// capacity_result == 0, can proceed
if (slots.compare_exchange_weak(slot, slot + size, if (slots.compare_exchange_weak(slot, slot + size,
std::memory_order_relaxed, std::memory_order_relaxed,

10
tests/test_server_connection_return.cpp
View File

@@ -18,10 +18,12 @@ struct Message {
struct EchoHandler : public ConnectionHandler { struct EchoHandler : public ConnectionHandler {
private: private:
ThreadPipeline<Message> &pipeline; StaticThreadPipeline<Message, WaitStrategy::WaitIfStageEmpty, 1> &pipeline;
public: public:
explicit EchoHandler(ThreadPipeline<Message> &pipeline) explicit EchoHandler(
StaticThreadPipeline<Message, WaitStrategy::WaitIfStageEmpty, 1>
&pipeline)
: pipeline(pipeline) {} : pipeline(pipeline) {}
void on_data_arrived(std::string_view data, void on_data_arrived(std::string_view data,
@@ -42,11 +44,11 @@ TEST_CASE(
config.server.io_threads = 1; config.server.io_threads = 1;
config.server.epoll_instances = 1; config.server.epoll_instances = 1;
ThreadPipeline<Message> pipeline{10, {1}}; StaticThreadPipeline<Message, WaitStrategy::WaitIfStageEmpty, 1> pipeline{10};
EchoHandler handler{pipeline}; EchoHandler handler{pipeline};
auto echoThread = std::thread{[&]() { auto echoThread = std::thread{[&]() {
for (;;) { for (;;) {
auto guard = pipeline.acquire(0, 0); auto guard = pipeline.acquire<0, 0>();
for (auto &message : guard.batch) { for (auto &message : guard.batch) {
bool done = message.done; bool done = message.done;
if (done) { if (done) {