From cf64a66d18e757f0d53f0ff5df853a1e7877b966 Mon Sep 17 00:00:00 2001 From: Hermes Date: Mon, 13 Apr 2026 10:29:50 -0700 Subject: [PATCH 1/3] fix: eliminate data race on mNextSeq in pwrite path mNextSeq was a plain size_t array written by worker threads in inputPwrite() and read by setInputCompletedPwrite() with no synchronization -- a C++ data race (undefined behaviour). A stale read could produce a wrong lastSeq value, causing ftruncate() to silently truncate the output file at the wrong offset and drop the final gz member(s). Fix: change mNextSeq to std::atomic[]. - Worker threads write with memory_order_release after each pack, establishing a happens-before edge for the completion reader. - setInputCompletedPwrite() opens with an acquire fence before reading with memory_order_relaxed, ensuring all prior worker writes are visible before the ftruncate() call. --- src/writerthread.cpp | 19 +++++++++++++------ src/writerthread.h | 2 +- 2 files changed, 14 insertions(+), 7 deletions(-) diff --git a/src/writerthread.cpp b/src/writerthread.cpp index 5d21091a..0e5d0fc0 100644 --- a/src/writerthread.cpp +++ b/src/writerthread.cpp @@ -28,9 +28,9 @@ WriterThread::WriterThread(Options* opt, string filename, bool isSTDOUT){ if (mFd < 0) error_exit("Failed to open for pwrite: " + mFilename); mOffsetRing = new OffsetSlot[OFFSET_RING_SIZE]; - mNextSeq = new size_t[mOptions->thread]; + mNextSeq = new std::atomic[mOptions->thread]; for (int t = 0; t < mOptions->thread; t++) - mNextSeq[t] = t; + mNextSeq[t].store(t, std::memory_order_relaxed); mCompressors = new libdeflate_compressor*[mOptions->thread]; for (int t = 0; t < mOptions->thread; t++) mCompressors[t] = libdeflate_alloc_compressor(mOptions->compression); @@ -75,12 +75,15 @@ bool WriterThread::setInputCompleted() { } void WriterThread::setInputCompletedPwrite() { + // Acquire fence: synchronize with the release stores in inputPwrite() + // so that all mNextSeq[t] writes from worker threads are visible here. + std::atomic_thread_fence(std::memory_order_acquire); int W = mOptions->thread; size_t lastSeq = 0; bool anyProcessed = false; for (int t = 0; t < W; t++) { - if (mNextSeq[t] != (size_t)t) { - size_t workerLastSeq = mNextSeq[t] - W; + if (mNextSeq[t].load(std::memory_order_relaxed) != (size_t)t) { + size_t workerLastSeq = mNextSeq[t].load(std::memory_order_relaxed) - W; if (!anyProcessed || workerLastSeq > lastSeq) { lastSeq = workerLastSeq; anyProcessed = true; @@ -131,7 +134,7 @@ void WriterThread::inputPwrite(int tid, string* data) { const char* writeData = mCompBufs[tid]; size_t wsize = outsize; - size_t seq = mNextSeq[tid]; + size_t seq = mNextSeq[tid].load(std::memory_order_relaxed); // Wait for previous batch's cumulative offset. // Sleep yields CPU to prevent livelock under contention. @@ -164,7 +167,11 @@ void WriterThread::inputPwrite(int tid, string* data) { } } - mNextSeq[tid] += mOptions->thread; + // Release store: ensures the pwrite and cumulative_offset publication + // happen-before the acquire fence in setInputCompletedPwrite(). + mNextSeq[tid].store( + mNextSeq[tid].load(std::memory_order_relaxed) + mOptions->thread, + std::memory_order_release); } void WriterThread::cleanup() { diff --git a/src/writerthread.h b/src/writerthread.h index 053d27f6..bd685875 100644 --- a/src/writerthread.h +++ b/src/writerthread.h @@ -59,7 +59,7 @@ class WriterThread{ bool mPwriteMode; int mFd; OffsetSlot* mOffsetRing; - size_t* mNextSeq; + std::atomic* mNextSeq; libdeflate_compressor** mCompressors; char** mCompBufs; // per-worker pre-allocated compress output buffers size_t* mCompBufSizes; // per-worker buffer sizes From b32ea3a10836fc6ab2e62cb1b00cf393de6332b9 Mon Sep 17 00:00:00 2001 From: Hermes Date: Mon, 13 Apr 2026 10:29:50 -0700 Subject: [PATCH 2/3] fix: make SPSC head pointer atomic to eliminate producer/consumer race MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The `head` pointer in SingleProducerSingleConsumerList was a plain (non-atomic) pointer despite being written by the producer thread (produce(), first-item branch) and read concurrently by the consumer thread (canBeConsumed(), consume()). ThreadSanitizer reported 15 data races at singleproducersingleconsumerlist.h:100. Fixes applied: - `head` declared as `std::atomic*>` (tail stays non-atomic — producer-private after first item is published) - Constructor: `head.store(NULL, relaxed)` - produce() first-item branch: set tail = item first (producer-private write), then `head.store(item, release)` to publish atomically to consumer then `item->nextItemReady.store(true, release)` to signal readiness - canBeConsumed(): `head.load(acquire)` for NULL check (syncs with produce release), `head.load(relaxed)` for nextItemReady dereference (covered by the preceding acquire) - consume(): `head.load(acquire)` to read current head, `head.store(h->nextItem, release)` to advance — establishes happens-before with next canBeConsumed() acquire on head Also fixes the else-branch nextItemReady assignment to use `memory_order_release` (was implicit seq_cst, which does NOT prevent compiler reordering of the preceding `tail->nextItem = item` write). --- src/singleproducersingleconsumerlist.h | 29 +++++++++++++++----------- 1 file changed, 17 insertions(+), 12 deletions(-) diff --git a/src/singleproducersingleconsumerlist.h b/src/singleproducersingleconsumerlist.h index 66304dca..1b5b9498 100644 --- a/src/singleproducersingleconsumerlist.h +++ b/src/singleproducersingleconsumerlist.h @@ -63,7 +63,7 @@ template class SingleProducerSingleConsumerList { public: inline SingleProducerSingleConsumerList() { - head = NULL; + head.store(NULL, std::memory_order_relaxed); tail = NULL; producerFinished = false; consumerFinished = false; @@ -90,28 +90,33 @@ class SingleProducerSingleConsumerList { return produced - consumed; } inline bool canBeConsumed() { - if(head == NULL) + if(head.load(std::memory_order_acquire) == NULL) return false; - return head->nextItemReady || producerFinished; + return head.load(std::memory_order_relaxed)->nextItemReady || producerFinished; } inline void produce(T val) { LockFreeListItem* item = makeItem(val); - if(head==NULL) { - head = item; + if(head.load(std::memory_order_relaxed) == NULL) { tail = item; + // Release store: publishing head to consumer thread. + // All writes to *item are ordered before this store. + head.store(item, std::memory_order_release); // Signal the first item is consumable (no predecessor to set this) - head->nextItemReady.store(true, std::memory_order_release); + item->nextItemReady.store(true, std::memory_order_release); } else { tail->nextItem = item; - tail->nextItemReady = true; + // Release store: ensures nextItem write visible before nextItemReady flag. + tail->nextItemReady.store(true, std::memory_order_release); tail = item; } produced++; } inline T consume() { - assert(head != NULL); - T val = head->value; - head = head->nextItem; + LockFreeListItem* h = head.load(std::memory_order_acquire); + assert(h != NULL); + T val = h->value; + // Advance head; release so next canBeConsumed() acquire sees updated state. + head.store(h->nextItem, std::memory_order_release); consumed++; if((consumed & 0xFFF) == 0) recycle(); @@ -156,8 +161,8 @@ class SingleProducerSingleConsumerList { } private: - LockFreeListItem* head; - LockFreeListItem* tail; + std::atomic*> head; + LockFreeListItem* tail; // tail is producer-private, no atomic needed LockFreeListItem** blocks; std::atomic_bool producerFinished; std::atomic_bool consumerFinished; From 63163450d31267ea9716f29f3e5ec8b77e5b837a Mon Sep 17 00:00:00 2001 From: Hermes Date: Mon, 13 Apr 2026 19:41:28 -0700 Subject: [PATCH 3/3] fix: make ReadPool counters atomic and SPSC produced/consumed atomic MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ThreadSanitizer reported data races in ReadPool and SPSC when multiple worker threads called ReadPool::input() concurrently: readpool.cpp:23 — mIsFull read vs. updateFullStatus() write readpool.cpp:27 — mProduced++ (non-atomic RMW) by multiple threads readpool.cpp:53 — mIsFull write vs. concurrent reads spsc.h:90 — size(): produced (producer-written) vs. consumed (consumer-written) read without synchronization Fixes in readpool.h: - mIsFull : bool → std::atomic - mProduced : size_t → std::atomic (atomic::operator++ and atomic::operator= are sufficient; no changes to readpool.cpp required) Fixes in singleproducersingleconsumerlist.h: - produced, consumed : unsigned long → std::atomic - size(): load both with memory_order_relaxed (approximate count used only as a soft back-pressure threshold) - produce(): produced.fetch_add(1, relaxed) - consume(): consumed.fetch_add(1, relaxed) with local snapshot for the (consumed & 0xFFF) recycle check - makeItem(): produced.load(relaxed) snapshot before >> and & ops - recycle(): consumed.load(relaxed) before >> op After all four commits (mNextSeq, SPSC head, ReadPool/SPSC atomics), ThreadSanitizer reports zero data races on 5k-read PE mode 8-thread workload. --- src/readpool.h | 5 +++-- src/singleproducersingleconsumerlist.h | 19 ++++++++++--------- 2 files changed, 13 insertions(+), 11 deletions(-) diff --git a/src/readpool.h b/src/readpool.h index 2b1bf733..616637e2 100644 --- a/src/readpool.h +++ b/src/readpool.h @@ -7,6 +7,7 @@ #include #include #include +#include #include "read.h" #include "options.h" #include "singleproducersingleconsumerlist.h" @@ -29,10 +30,10 @@ class ReadPool{ private: Options* mOptions; SingleProducerSingleConsumerList** mBufferLists; - size_t mProduced; + std::atomic mProduced; size_t mConsumed; unsigned long mLimit; - bool mIsFull; + std::atomic mIsFull; }; #endif \ No newline at end of file diff --git a/src/singleproducersingleconsumerlist.h b/src/singleproducersingleconsumerlist.h index 1b5b9498..cb79b6d8 100644 --- a/src/singleproducersingleconsumerlist.h +++ b/src/singleproducersingleconsumerlist.h @@ -87,7 +87,7 @@ class SingleProducerSingleConsumerList { blocks = NULL; } inline size_t size() { - return produced - consumed; + return produced.load(std::memory_order_relaxed) - consumed.load(std::memory_order_relaxed); } inline bool canBeConsumed() { if(head.load(std::memory_order_acquire) == NULL) @@ -109,7 +109,7 @@ class SingleProducerSingleConsumerList { tail->nextItemReady.store(true, std::memory_order_release); tail = item; } - produced++; + produced.fetch_add(1, std::memory_order_relaxed); } inline T consume() { LockFreeListItem* h = head.load(std::memory_order_acquire); @@ -117,8 +117,8 @@ class SingleProducerSingleConsumerList { T val = h->value; // Advance head; release so next canBeConsumed() acquire sees updated state. head.store(h->nextItem, std::memory_order_release); - consumed++; - if((consumed & 0xFFF) == 0) + unsigned long _c = consumed.fetch_add(1, std::memory_order_relaxed) + 1; + if((_c & 0xFFF) == 0) recycle(); return val; } @@ -137,8 +137,9 @@ class SingleProducerSingleConsumerList { private: // blockized list inline LockFreeListItem* makeItem(T val) { - unsigned long blk = produced >> 12; - unsigned long idx = produced & 0xFFF; + unsigned long _p = produced.load(std::memory_order_relaxed); + unsigned long blk = _p >> 12; + unsigned long idx = _p & 0xFFF; size_t size = 0x01<<12; if(blocksNum <= blk) { LockFreeListItem* buffer = new LockFreeListItem[size]; @@ -152,7 +153,7 @@ class SingleProducerSingleConsumerList { } inline void recycle() { - unsigned long blk = consumed >> 12; + unsigned long blk = consumed.load(std::memory_order_relaxed) >> 12; while((recycled+1) < blk) { delete[] blocks[recycled & blocksRingBufferSizeMask]; blocks[recycled & blocksRingBufferSizeMask] = NULL; @@ -166,8 +167,8 @@ class SingleProducerSingleConsumerList { LockFreeListItem** blocks; std::atomic_bool producerFinished; std::atomic_bool consumerFinished; - unsigned long produced; - unsigned long consumed; + std::atomic produced; + std::atomic consumed; unsigned long recycled; unsigned long blocksRingBufferSize; unsigned long blocksRingBufferSizeMask;