cpp14-thread-toolkit

A small, dependency-free C++14 concurrency toolkit: reader/writer lock guards, a one-slot producer/consumer handshake, a named worker thread with a registry, a locked FIFO, a sliding-window circular buffer, and a process-wide run state driven by SIGINT.

Namespace threadkit. Header-only except for Thread (src/thread.cpp). Requires only the standard library and pthreads; RunState::printTrace() uses glibc backtrace() and compiles to a no-op elsewhere.

#include "threadkit/threadkit.h"   // or the individual headers below

Components

locks.h — Lock, scoped_read_lock, scoped_write_lock

Lock is std::shared_timed_mutex (the only shared mutex in C++14). The two guards hold a shared or exclusive lock for their scope.

threadkit::Lock lock;
std::map<std::string, int> table;

int get(const std::string& k) {
    threadkit::scoped_read_lock guard(lock);     // many readers at once
    return table.at(k);
}
void put(const std::string& k, int v) {
    threadkit::scoped_write_lock guard(lock);    // exclusive
    table[k] = v;
}

Thread safety: that is the point. Non-copyable; not recursive — taking a write lock while holding a read lock on the same Lock deadlocks.

pc_sem.h — PCSem

A one-slot rendezvous. produce() blocks until the slot is empty, fills it and wakes a consumer; consume() blocks until it is full, empties it and wakes a producer; join() blocks until the slot is empty; disable() releases every waiter and makes further calls no-ops.

threadkit::PCSem sem;
std::thread worker([&] { for (;;) { sem.consume(); do_work(); } });
sem.produce();   // hand one unit of work over
sem.join();      // wait until it has been picked up

Thread safety: all methods take the internal mutex; waits are predicate-based (spurious-wakeup safe). One producer and one consumer is the intended pattern; several of either serialise on the mutex.

thread.h — Thread

A named worker that owns a std::thread and a PCSem, with three modes:

mode	behaviour
BLOCKING_LOOP (default)	runs the functor once per produce(), blocking in between
NON_BLOCKING_LOOP	calls the functor back-to-back until stop() — the functor should block or sleep
ONE_SHOT	calls the functor once and exits

static void* work(void* arg) { /* ... */ return nullptr; }

int ctx = 42;
threadkit::Thread t;
t.init("worker", &work, &ctx);
t.go();                        // BLOCKING_LOOP
t.produce();                   // run work(&ctx) once, on the worker
t.join();                      // wait until that run has finished
std::string s = t.stop();      // "Thread stopped"

The functor is std::function<void*(void*)>; lambdas work. Every mode also exits when RunState reaches EXITSTATE; BLOCKING_LOOP holds while PAUSESTATE and prints a backtrace and exits on DUMPSTATE.

Thread safety: produce(), consume(), join(), unPause() and dumpStack() may be called from any thread. init(), go() and stop() are owner-thread operations. stop() aborts work that has not started: for ONE_SHOT, wait for your own completion signal before stopping. The functor runs with the PCSem mutex held, so produce() from another thread blocks for the duration of a work unit — the handshake is a rendezvous, not a queue. Because RunState is polled, call unPause() (or ThreadRegistry::unPauseAll()) after changing it. The destructor calls stop() if the worker is still running.

thread_registry.h — ThreadRegistry

A std::unordered_map<std::string, Thread*> behind a Lock. Pass it to Thread::init() and the thread registers itself and deregisters in stop()/shutDown(). Does not own the threads.

threadkit::ThreadRegistry reg;
threadkit::Thread a, b;
a.init("a", &work, nullptr, &reg);
b.init("b", &work, nullptr, &reg);
reg.count();          // 2
reg.lookup("a");      // &a
{
    threadkit::scoped_read_lock guard(reg.getLock());
    for (auto& kv : reg) std::cout << kv.first << '\n';
}
reg.unPauseAll();     // wake everyone to re-read RunState

Thread safety: registryInsert, registryErase, lookup, count, unPauseAll lock internally. Direct use of the inherited map interface (iteration, find, size) is not locked — take getLock() yourself.

pc_queue.h — PCQueue<T>

An unbounded FIFO of std::shared_ptr<T> behind a Lock. It does not block; combine it with a PCSem or a Thread to wake a consumer.

struct Job { int id; };
threadkit::PCQueue<Job> q;
q.enqueue(std::make_shared<Job>(Job{1}));
if (auto job = q.dequeue()) handle(*job);   // null when empty

Thread safety: enqueue/dequeue take the write lock, isEmpty/size the read lock; any number of producers and consumers. size() is a snapshot. lock() exposes the internal Lock for composing operations.

circ_buff.h — CircBuff<T = double>

A circular buffer for a window of fullSize() elements, backed by a ring factor (default 16) times larger so the read pointer can lag the write pointer by many windows before data is overwritten.

threadkit::CircBuff<double> cb(256);     // window 256, ring 4096
cb.push_back(sample);
cb.diff();                               // unread elements
double x = cb.pop_front();
cb[0];                                   // relative to the read pointer
if (cb.isFull()) {                       // >= 256 pushed
    const double* w = cb.contiguous();   // 256 elements from the read pointer,
    fft(w, cb.fullSize());               // copied only if the window wraps
}

Thread safety: none. It was owned by one worker thread in its original setting; wrap it in a Lock if you share it. No bounds checks: pop_front() on an empty buffer returns whatever is in the slot.

run_state.h — RunState

A process-global RUNSTATE / PAUSESTATE / EXITSTATE / DUMPSTATE flag that Thread polls, a SIGINT handler that advances it one step per signal, and a demangling backtrace printer.

threadkit::RunState::installSigintHandler();
// ... start threads ...
while (!threadkit::RunState::gotExitSignal()) { /* main loop */ }
reg.unPauseAll();                        // let waiting workers see EXITSTATE
for (auto& kv : reg) kv.second->stop();

Thread safety: the state is a lock-free std::atomic<int>; the signal handler touches nothing else and is async-signal-safe. printTrace() is not (it allocates) and is meant for diagnostics on the way out.

Build and test

cmake -S . -B build -DCMAKE_BUILD_TYPE=Release
cmake --build build
./build/threadkit_tests          # or: ctest --test-dir build

Needs CMake 3.14+, a C++14 compiler and GoogleTest (libgtest-dev on Debian/Ubuntu; find_package(GTest)). -DTHREADKIT_BUILD_TESTS=OFF drops the GoogleTest requirement. -DTHREADKIT_TSAN=ON builds with ThreadSanitizer; on kernels with 32-bit mmap randomisation run the binary under setarch $(uname -m) -R. The suite is clean under TSAN.

To use from another CMake project, add_subdirectory it and link threadkit::threadkit, or install and add include/ plus libthreadkit.a.

Provenance

Written in 2016-2019 as the concurrency layer of a private trading system, where it ran a producer/consumer pipeline in production. Extracted and published in 2026. The extraction removed the application framework the classes were bound to (a component base class, a logging library, a signal-handling console loop) and the domain-specific payload types; it also fixed a double-free in the queue's dequeue() and a pointer-aliasing leak in the circular buffer's contiguous-window path, replaced timed waits with predicate waits, and made the buffer and the queue templates. tests/thread_test.cpp carries the original unit test; the rest of the suite was written for the extraction.

License

MIT. Copyright (c) 2016-2026 Wade Stone. See LICENSE.