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

Move thread local state into stack

Browse Source
This commit is contained in:
Andrew Noyes
2025-09-15 20:33:45 -04:00
parent 5a88047b9f
commit 917066d8c0
3 changed files with 302 additions and 306 deletions
Download Patch File Download Diff File Expand all files Collapse all files

112
src/commit_pipeline.cpp
View File

@@ -2,6 +2,7 @@
#include <cstring>
#include <pthread.h>
#include <unordered_set>
#include "commit_request.hpp"
#include "cpu_work.hpp"
@@ -16,52 +17,35 @@ auto banned_request_ids_memory_gauge =
.create({});
CommitPipeline::CommitPipeline(const weaseldb::Config &config)
: config_(config), banned_request_ids_(ArenaStlAllocator<std::string_view>(
&banned_request_arena_)),
pipeline_(lg_size) {
: config_(config), pipeline_(lg_size) {
// Stage 0: Sequence assignment thread
sequence_thread_ = std::thread{[this]() {
pthread_setname_np(pthread_self(), "txn-sequence");
for (int shutdowns_received = 0; shutdowns_received < 2;) {
auto guard = pipeline_.acquire<0, 0>();
process_sequence_batch(guard.batch, shutdowns_received);
}
run_sequence_stage();
}};
// Stage 1: Precondition resolution thread
resolve_thread_ = std::thread{[this]() {
pthread_setname_np(pthread_self(), "txn-resolve");
for (int shutdowns_received = 0; shutdowns_received < 2;) {
auto guard = pipeline_.acquire<1, 0>(/*maxBatch*/ 1);
process_resolve_batch(guard.batch, shutdowns_received);
}
run_resolve_stage();
}};
// Stage 2: Transaction persistence thread
persist_thread_ = std::thread{[this]() {
pthread_setname_np(pthread_self(), "txn-persist");
for (int shutdowns_received = 0; shutdowns_received < 2;) {
auto guard = pipeline_.acquire<2, 0>();
process_persist_batch(guard.batch, shutdowns_received);
}
run_persist_stage();
}};
// Stage 3: Connection return to server threads (2 threads)
release_thread_1_ = std::thread{[this]() {
pthread_setname_np(pthread_self(), "txn-release-1");
for (int shutdowns_received = 0; shutdowns_received < 1;) {
auto guard = pipeline_.acquire<3, 0>();
process_release_batch(guard.batch, 0, shutdowns_received);
}
run_release_stage<0>();
}};
release_thread_2_ = std::thread{[this]() {
pthread_setname_np(pthread_self(), "txn-release-2");
for (int shutdowns_received = 0; shutdowns_received < 1;) {
auto guard = pipeline_.acquire<3, 1>();
process_release_batch(guard.batch, 1, shutdowns_received);
}
run_release_stage<1>();
}};
}
@@ -95,8 +79,22 @@ void CommitPipeline::submit_batch(std::span<PipelineEntry> entries) {
// Guard destructor publishes batch to stage 0
}
void CommitPipeline::process_sequence_batch(BatchType &batch,
int &shutdowns_received) {
void CommitPipeline::run_sequence_stage() {
int64_t next_version = 1;
// Request ID deduplication (sequence stage only)
Arena banned_request_arena;
using BannedRequestIdSet =
std::unordered_set<std::string_view, std::hash<std::string_view>,
std::equal_to<std::string_view>,
ArenaStlAllocator<std::string_view>>;
BannedRequestIdSet banned_request_ids{
ArenaStlAllocator<std::string_view>(&banned_request_arena)};
for (int shutdowns_received = 0; shutdowns_received < 2;) {
auto guard = pipeline_.acquire<0, 0>();
auto &batch = guard.batch;
// Stage 0: Sequence assignment
// This stage performs ONLY work that requires serial processing:
// - Version/sequence number assignment (must be sequential)
@@ -122,41 +120,34 @@ void CommitPipeline::process_sequence_batch(BatchType &batch,
commit_entry.commit_request->request_id().has_value()) {
auto commit_request_id =
commit_entry.commit_request->request_id().value();
if (banned_request_ids_.contains(commit_request_id)) {
if (banned_request_ids.contains(commit_request_id)) {
// Request ID is banned, this commit should fail
auto conn_ref = commit_entry.connection.lock();
if (!conn_ref) {
// Connection is gone, drop the entry silently
return; // Skip this entry and continue processing
}
conn_ref->send_response(
commit_entry.protocol_context,
R"({"status": "not_committed", "error": "request_id_banned"})",
Arena{});
commit_entry.response_json =
R"({"status": "not_committed", "error": "request_id_banned"})";
return;
}
}
// Assign sequential version number
commit_entry.assigned_version = next_version_++;
commit_entry.assigned_version = next_version++;
} else if constexpr (std::is_same_v<T, StatusEntry>) {
// Process status entry: add request_id to banned list, get version
// upper bound
// Process status entry: add request_id to banned list, get
// version upper bound
auto &status_entry = e;
// Add request_id to banned list - store the string in arena and
// use string_view
std::string_view request_id_view =
banned_request_arena_.copy_string(
banned_request_arena.copy_string(
status_entry.status_request_id);
banned_request_ids_.insert(request_id_view);
banned_request_ids.insert(request_id_view);
// Update memory usage metric
banned_request_ids_memory_gauge.set(
banned_request_arena_.total_allocated());
banned_request_arena.total_allocated());
// Set version upper bound to current highest assigned version
status_entry.version_upper_bound = next_version_ - 1;
status_entry.version_upper_bound = next_version - 1;
} else if constexpr (std::is_same_v<T, HealthCheckEntry>) {
// Process health check entry: noop in sequence stage
}
@@ -164,9 +155,13 @@ void CommitPipeline::process_sequence_batch(BatchType &batch,
entry);
}
}
}
void CommitPipeline::run_resolve_stage() {
for (int shutdowns_received = 0; shutdowns_received < 2;) {
auto guard = pipeline_.acquire<1, 0>(/*maxBatch*/ 1);
auto &batch = guard.batch;
void CommitPipeline::process_resolve_batch(BatchType &batch,
int &shutdowns_received) {
// Stage 1: Precondition resolution
// This stage must be serialized to maintain consistent database state view
// - Validate preconditions against current database state
@@ -197,9 +192,13 @@ void CommitPipeline::process_resolve_batch(BatchType &batch,
entry);
}
}
}
void CommitPipeline::run_persist_stage() {
for (int shutdowns_received = 0; shutdowns_received < 2;) {
auto guard = pipeline_.acquire<2, 0>();
auto &batch = guard.batch;
void CommitPipeline::process_persist_batch(BatchType &batch,
int &shutdowns_received) {
// Stage 2: Transaction persistence
// Mark everything as durable immediately (simplified implementation)
// In real implementation: batch S3 writes, update subscribers, etc.
@@ -218,7 +217,8 @@ void CommitPipeline::process_persist_batch(BatchType &batch,
// Check if connection is still alive first
// Skip if resolve failed or connection is in error state
if (!commit_entry.commit_request || !commit_entry.resolve_success) {
if (!commit_entry.commit_request ||
!commit_entry.resolve_success) {
return;
}
@@ -227,7 +227,8 @@ void CommitPipeline::process_persist_batch(BatchType &batch,
committed_version_.store(commit_entry.assigned_version,
std::memory_order_seq_cst);
const CommitRequest &commit_request = *commit_entry.commit_request;
const CommitRequest &commit_request =
*commit_entry.commit_request;
// Generate success JSON response with actual assigned version
std::string_view response_json;
@@ -235,7 +236,8 @@ void CommitPipeline::process_persist_batch(BatchType &batch,
response_json = format(
commit_entry.request_arena,
R"({"request_id":"%.*s","status":"committed","version":%ld,"leader_id":"leader123"})",
static_cast<int>(commit_request.request_id().value().size()),
static_cast<int>(
commit_request.request_id().value().size()),
commit_request.request_id().value().data(),
commit_entry.assigned_version);
} else {
@@ -283,9 +285,13 @@ void CommitPipeline::process_persist_batch(BatchType &batch,
entry);
}
}
}
template <int thread_index> void CommitPipeline::run_release_stage() {
for (int shutdowns_received = 0; shutdowns_received < 1;) {
auto guard = pipeline_.acquire<3, thread_index>();
auto &batch = guard.batch;
void CommitPipeline::process_release_batch(BatchType &batch, int thread_index,
int &shutdowns_received) {
// Stage 3: Connection release
// Return connections to server for response transmission
@@ -359,7 +365,8 @@ void CommitPipeline::process_release_batch(BatchType &batch, int thread_index,
// Send the response using protocol-agnostic interface
// HTTP formatting will happen in on_preprocess_writes()
conn_ref->send_response(get_version_entry.protocol_context,
conn_ref->send_response(
get_version_entry.protocol_context,
get_version_entry.response_json,
std::move(get_version_entry.request_arena));
}
@@ -367,3 +374,4 @@ void CommitPipeline::process_release_batch(BatchType &batch, int thread_index,
entry);
}
}
}

26
src/commit_pipeline.hpp
View File

@@ -3,9 +3,7 @@
#include <atomic>
#include <span>
#include <thread>
#include <unordered_set>
#include "arena.hpp"
#include "config.hpp"
#include "pipeline_entry.hpp"
#include "thread_pipeline.hpp"
@@ -90,20 +88,9 @@ private:
// Configuration reference
const weaseldb::Config &config_;
// Pipeline state (sequence stage only)
int64_t next_version_ = 1; // Next version to assign (sequence thread only)
// Pipeline state (persist thread writes, other threads read)
std::atomic<int64_t> committed_version_{0}; // Highest committed version
// Request ID deduplication (sequence stage only)
Arena banned_request_arena_;
using BannedRequestIdSet =
std::unordered_set<std::string_view, std::hash<std::string_view>,
std::equal_to<std::string_view>,
ArenaStlAllocator<std::string_view>>;
BannedRequestIdSet banned_request_ids_;
// Lock-free pipeline configuration
static constexpr int lg_size = 16; // Ring buffer size (2^16 slots)
static constexpr auto wait_strategy = WaitStrategy::WaitIfStageEmpty;
@@ -118,14 +105,15 @@ private:
std::thread release_thread_1_;
std::thread release_thread_2_;
// Pipeline stage processing methods (batch-based)
// Pipeline stage main loops
void run_sequence_stage();
void run_resolve_stage();
void run_persist_stage();
template <int thread_index> void run_release_stage();
// Pipeline batch type alias
using BatchType =
StaticThreadPipeline<PipelineEntry, wait_strategy, 1, 1, 1, 2>::Batch;
void process_sequence_batch(BatchType &batch, int &shutdowns_received);
void process_resolve_batch(BatchType &batch, int &shutdowns_received);
void process_persist_batch(BatchType &batch, int &shutdowns_received);
void process_release_batch(BatchType &batch, int thread_index,
int &shutdowns_received);
// Make non-copyable and non-movable
CommitPipeline(const CommitPipeline &) = delete;

2
src/pipeline_entry.hpp
View File

@@ -13,7 +13,7 @@ struct CommitRequest;
*/
struct CommitEntry {
WeakRef<MessageSender> connection;
int64_t assigned_version = 0; // Set by sequence stage
int64_t assigned_version = -1; // Set by sequence stage
bool resolve_success = false; // Set by resolve stage
bool persist_success = false; // Set by persist stage