Catcrafts/Crafter.Thread
Watch
2
0
Fork
You've already forked Crafter.Thread
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

fix(threadpool): batches run every task once, and Stop ends the whole pool

Browse source 
- Enqueue(vector) dropped tasks when the count didn't divide by the
  workers: the last one got only the remainder. Tasks are now split as
  evenly as the count allows, each worker's taskCount grows by what it
  was given (the single-task Enqueue picks by it), and Start(1), which
  has no workers, leaves the batch for the caller's JoinPool.
- Stop wakes every slot, JoinPool's too, after joining the workers: a
  thread that had joined the pool slept on. It takes each slot's mutex
  before notifying, so a waiter between its check and its sleep can't miss
  the wake-up, and works on the pool it started with. Again, or before
  Start, it does nothing; joining a thread twice threw. Its documentation
  says what it doesn't do: stop from a pool task, or from two threads at
  once, run tasks enqueued once it has begun, or have JoinPool wait for it.
- Tests: EnqueueBatch, StopJoinPool.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Jorijn van der Graaf 2026-10-20 19:06:37 +02:00
commit f71cb43e85
Signed by: jorijnvdgraaf
GPG key ID: 2937E59CDCC1BCFB
5 changed files with 308 additions and 149 deletions
Download patch file Download diff file Expand all files Collapse all files

264
implementations/Crafter.Thread-ThreadPool.cpp
View file

@ -1,119 +1,147 @@
//SPDX-License-Identifier: LGPL-3.0-only //SPDX-License-Identifier: LGPL-3.0-only
//SPDX-FileCopyrightText: Copyright (C) 2026 Catcrafts® //SPDX-FileCopyrightText: Copyright (C) 2026 Catcrafts®
module Crafter.Thread:ThreadPool_impl; module Crafter.Thread:ThreadPool_impl;
import std; import std;
import :ThreadPool; import :ThreadPool;
using namespace Crafter; using namespace Crafter;
void ThreadPool::Start(int maxThreads) void ThreadPool::Start(int maxThreads)
{ {
stopped = false; stopped = false;
if (maxThreads > 0) { if (maxThreads > 0) {
threadCount = maxThreads; threadCount = maxThreads;
} }
else { else {
threadCount = std::thread::hardware_concurrency(); threadCount = std::thread::hardware_concurrency();
} }
threadStorage = new ThreadStorage[threadCount]; threadStorage = new ThreadStorage[threadCount];
for (unsigned int i = 0; i < threadCount - 1; i++) for (unsigned int i = 0; i < threadCount - 1; i++)
{ {
threadStorage[i].thread = std::thread(ThreadPool::WaitAndExecuteTask, threadStorage + i); threadStorage[i].thread = std::thread(ThreadPool::WaitAndExecuteTask, threadStorage + i);
threadIdMap[threadStorage[i].thread.get_id()] = &threadStorage[i]; threadIdMap[threadStorage[i].thread.get_id()] = &threadStorage[i];
} }
threadIdMap[std::this_thread::get_id()] = &threadStorage[threadCount - 1]; threadIdMap[std::this_thread::get_id()] = &threadStorage[threadCount - 1];
activeThreads = threadCount - 1; activeThreads = threadCount - 1;
} }
void ThreadPool::Stop() void ThreadPool::Stop()
{ {
stopped = true; // This pool's own: a JoinPool caller may Start another as soon as it
for (unsigned int i = 0; i < threadCount - 1; i++) // returns, replacing threadStorage.
{ ThreadStorage* const storage = threadStorage;
threadStorage[i].cv.notify_one(); const unsigned int count = threadCount;
threadStorage[i].thread.join(); if (storage == nullptr) return;
} stopped = true;
} // Taking the slot's mutex orders the store above before the next check
// of `stopped` by whoever waits on the slot: without it one between
void ThreadPool::Enqueue(std::function<void(void)> task) // checking the condition and going to sleep misses the notify, and never
{ // wakes.
unsigned int min = std::numeric_limits<unsigned int>::max(); auto wake = [](ThreadStorage& slot) {
unsigned int minIndex = 0; {
for (unsigned int i = 0; i < activeThreads; i++) std::lock_guard lk(slot.mutex);
{ }
unsigned int taskCount = threadStorage[i].taskCount; slot.cv.notify_one();
if (taskCount == 0) { };
minIndex = i; for (unsigned int i = 0; i < count - 1; i++) wake(storage[i]);
break; for (unsigned int i = 0; i < count - 1; i++)
} {
if (taskCount < min) { if (storage[i].thread.joinable()) storage[i].thread.join();
min = taskCount; }
minIndex = i; // The JoinPool caller's slot last: a JoinPool waiting there returns once
} // the workers have stopped (one busy with a task sees `stopped` itself
} // when it's done).
{ wake(storage[count - 1]);
std::lock_guard lk(threadStorage[minIndex].mutex); }
threadStorage[minIndex].buffer.push_back(task);
threadStorage[minIndex].taskCount++; void ThreadPool::Enqueue(std::function<void(void)> task)
} {
threadStorage[minIndex].cv.notify_one(); unsigned int min = std::numeric_limits<unsigned int>::max();
} unsigned int minIndex = 0;
for (unsigned int i = 0; i < activeThreads; i++)
void ThreadPool::Enqueue(std::vector<std::function<void(void)>> tasks) {
{ unsigned int taskCount = threadStorage[i].taskCount;
unsigned int size = tasks.size(); if (taskCount == 0) {
unsigned int tasksPerThread = size / activeThreads; minIndex = i;
int remainder = size % activeThreads; break;
}
if(remainder) { if (taskCount < min) {
for (unsigned int i = 0; i < activeThreads - 1; i++) min = taskCount;
{ minIndex = i;
{ }
std::lock_guard lk(threadStorage[i].mutex); }
threadStorage[i].buffer.insert(threadStorage[i].buffer.end(), tasks.begin() + tasksPerThread * i, tasks.begin() + (tasksPerThread * i + tasksPerThread)); {
threadStorage[i].taskCount += tasksPerThread * i; std::lock_guard lk(threadStorage[minIndex].mutex);
} threadStorage[minIndex].buffer.push_back(task);
threadStorage[i].cv.notify_one(); threadStorage[minIndex].taskCount++;
} }
{ threadStorage[minIndex].cv.notify_one();
std::lock_guard lk(threadStorage[activeThreads - 1].mutex); }
threadStorage[activeThreads - 1].buffer.insert(threadStorage[activeThreads - 1].buffer.end(), tasks.begin() + tasksPerThread * (activeThreads - 1), tasks.begin() + (tasksPerThread * (activeThreads - 1) + remainder));
threadStorage[activeThreads - 1].taskCount += tasksPerThread + remainder; void ThreadPool::Enqueue(std::vector<std::function<void(void)>> tasks)
} {
threadStorage[activeThreads - 1].cv.notify_one(); const unsigned int workers = activeThreads;
} else{ if (workers == 0) {
for (unsigned int i = 0; i < activeThreads; i++) // Start(1) runs no worker threads; like a single Enqueue, the batch
{ // waits for the caller's JoinPool.
{ {
std::lock_guard lk(threadStorage[i].mutex); std::lock_guard lk(threadStorage[0].mutex);
threadStorage[i].buffer.insert(threadStorage[i].buffer.end(), tasks.begin() + tasksPerThread * i, tasks.begin() + (tasksPerThread * i + tasksPerThread)); threadStorage[0].taskCount += static_cast<unsigned int>(tasks.size());
threadStorage[i].taskCount += tasksPerThread * i; threadStorage[0].buffer.insert(threadStorage[0].buffer.end(),
} std::make_move_iterator(tasks.begin()), std::make_move_iterator(tasks.end()));
threadStorage[i].cv.notify_one(); }
} threadStorage[0].cv.notify_one();
} return;
} }
void ThreadPool::JoinPool() // As even a split as the count allows: the first `remainder` workers
{ // take one task more. Every task goes to exactly one worker, and each
activeThreads++; // worker's taskCount grows by what it was given, which the single-task
ThreadPool::WaitAndExecuteTask(threadStorage + threadCount - 1); // Enqueue relies on to pick the least loaded worker.
} const std::size_t perWorker = tasks.size() / workers;
const std::size_t remainder = tasks.size() % workers;
void ThreadPool::WaitAndExecuteTask(ThreadStorage* threadStorage) std::size_t first = 0;
{ for (unsigned int i = 0; i < workers && first < tasks.size(); i++)
while (!stopped || !threadStorage->buffer.empty()) { {
std::unique_lock<std::mutex> lock(threadStorage->mutex); const std::size_t count = perWorker + (i < remainder ? 1 : 0);
threadStorage->cv.wait(lock, [threadStorage] { return !threadStorage->buffer.empty() || stopped; }); {
threadStorage->buffer.swap(threadStorage->tasks); std::lock_guard lk(threadStorage[i].mutex);
lock.unlock(); threadStorage[i].taskCount += static_cast<unsigned int>(count);
for (std::function<void(void)> task : threadStorage->tasks) threadStorage[i].buffer.insert(threadStorage[i].buffer.end(),
{ std::make_move_iterator(tasks.begin() + static_cast<std::ptrdiff_t>(first)),
task(); std::make_move_iterator(tasks.begin() + static_cast<std::ptrdiff_t>(first + count)));
threadStorage->taskCount--; }
} threadStorage[i].cv.notify_one();
threadStorage->tasks.clear(); first += count;
} }
}
void ThreadPool::JoinPool()
{
activeThreads++;
ThreadPool::WaitAndExecuteTask(threadStorage + threadCount - 1);
}
void ThreadPool::WaitAndExecuteTask(ThreadStorage* threadStorage)
{
while (true) {
std::unique_lock<std::mutex> lock(threadStorage->mutex);
threadStorage->cv.wait(lock, [threadStorage] { return !threadStorage->buffer.empty() || stopped; });
// The buffer is only read under the lock: the loop used to test it
// unlocked while Enqueue appended to it. Stopping still waits for
// everything queued to run.
if (threadStorage->buffer.empty()) {
return;
}
threadStorage->buffer.swap(threadStorage->tasks);
lock.unlock();
for (std::function<void(void)>& task : threadStorage->tasks)
{
task();
threadStorage->taskCount--;
}
threadStorage->tasks.clear();
}
} }

67
interfaces/Crafter.Thread-ThreadPool.cppm
View file

@ -1,32 +1,37 @@
//SPDX-License-Identifier: LGPL-3.0-only //SPDX-License-Identifier: LGPL-3.0-only
//SPDX-FileCopyrightText: Copyright (C) 2026 Catcrafts® //SPDX-FileCopyrightText: Copyright (C) 2026 Catcrafts®
export module Crafter.Thread:ThreadPool; export module Crafter.Thread:ThreadPool;
import std; import std;
namespace Crafter { namespace Crafter {
struct ThreadStorage { struct ThreadStorage {
std::vector<std::function<void(void)>> tasks; std::vector<std::function<void(void)>> tasks;
std::vector<std::function<void(void)>> buffer; std::vector<std::function<void(void)>> buffer;
std::condition_variable cv; std::condition_variable cv;
std::mutex mutex; std::mutex mutex;
std::atomic<unsigned int> taskCount; std::atomic<unsigned int> taskCount;
std::thread thread; std::thread thread;
}; };
export class ThreadPool { export class ThreadPool {
public: public:
static inline std::atomic<bool> stopped; static inline std::atomic<bool> stopped;
static void Start(int maxThreads = -1); static void Start(int maxThreads = -1);
static void Stop(); // Stops the workers, each once it has run what it was given, and ends
static void Enqueue(std::function<void(void)> task); // a JoinPool; a task enqueued once Stop has begun may not run. From
static void Enqueue(std::vector<std::function<void(void)>> tasks); // one thread at a time, not the pool's own: a task can't stop the
static void JoinPool(); // pool it runs in. Again, or before Start, it does nothing. JoinPool
static inline unsigned int threadCount; // can return before Stop has: Start a new pool once Stop returned.
static inline std::atomic<unsigned int> activeThreads; static void Stop();
private: static void Enqueue(std::function<void(void)> task);
static inline std::unordered_map<std::thread::id, ThreadStorage*> threadIdMap; static void Enqueue(std::vector<std::function<void(void)>> tasks);
static inline ThreadStorage* threadStorage; static void JoinPool();
static void WaitAndExecuteTask(ThreadStorage* threadStorage); static inline unsigned int threadCount;
}; static inline std::atomic<unsigned int> activeThreads;
private:
static inline std::unordered_map<std::thread::id, ThreadStorage*> threadIdMap;
static inline ThreadStorage* threadStorage;
static void WaitAndExecuteTask(ThreadStorage* threadStorage);
};
} }

6
project.cpp
View file

@ -23,5 +23,11 @@ extern "C" Configuration CrafterBuildProject(std::span<const std::string_view> a
}; };
cfg.GetInterfacesAndImplementations(ifaces, impls); cfg.GetInterfacesAndImplementations(ifaces, impls);
// Tests link this project's static library; wasm builds run no tests.
if (cfg.target.find("wasm") == std::string::npos) {
cfg.AddTest("EnqueueBatch").Dependencies({ &cfg });
cfg.AddTest("StopJoinPool").Dependencies({ &cfg });
}
return cfg; return cfg;
} }

77
tests/EnqueueBatch/main.cpp Normal file
View file

@ -0,0 +1,77 @@
//SPDX-License-Identifier: LGPL-3.0-only
//SPDX-FileCopyrightText: Copyright (C) 2026 Catcrafts®
// ThreadPool::Enqueue(vector) runs every task exactly once, whatever the
// task count. When the count didn't divide by the number of worker threads,
// the last worker got only the remainder and tasksPerThread tasks were
// dropped. The counts here cover fewer tasks than workers, an exact multiple,
// one more and one less than a multiple, and a large uneven batch, then a
// batch of single Enqueue calls, which rely on the per-thread task counts the
// batch path also kept wrong.
#include <cstdlib>
import Crafter.Thread;
import std;
using namespace Crafter;
namespace {
int failures = 0;
void Check(bool ok, std::string_view what) {
std::println("{} {}", ok ? "PASS" : "FAIL", what);
if (!ok) ++failures;
}
// Waits until `done` reaches `expected` or a few seconds pass.
bool WaitFor(const std::atomic<std::uint32_t>& done, std::uint32_t expected) {
const auto deadline = std::chrono::steady_clock::now() + std::chrono::seconds(5);
while (done.load() < expected) {
if (std::chrono::steady_clock::now() > deadline) return false;
std::this_thread::sleep_for(std::chrono::milliseconds(1));
}
return true;
}
void RunBatch(std::uint32_t count) {
std::atomic<std::uint32_t> done = 0;
std::vector<std::atomic<std::uint32_t>> runs(count);
std::vector<std::function<void(void)>> tasks;
tasks.reserve(count);
for (std::uint32_t i = 0; i < count; ++i) {
tasks.emplace_back([&runs, &done, i] {
runs[i]++;
done++;
});
}
ThreadPool::Enqueue(std::move(tasks));
const bool finished = WaitFor(done, count);
// A little longer, in case something runs twice.
std::this_thread::sleep_for(std::chrono::milliseconds(20));
const bool once = std::ranges::all_of(runs, [](const std::atomic<std::uint32_t>& r) { return r.load() == 1; });
Check(finished && once, std::format("a batch of {} tasks runs every task exactly once ({} of {} done)", count, done.load(), count));
}
} // namespace
int main() {
ThreadPool::Start(4);
const std::uint32_t workers = ThreadPool::activeThreads;
std::println(" {} worker threads", workers);
for (std::uint32_t count : {1u, workers - 1, workers, workers + 1, 3 * workers + 2, 1000u}) {
RunBatch(count);
}
// Single Enqueue calls pick the least loaded worker by its task count;
// a count left wrong by a batch would starve or flood one of them.
std::atomic<std::uint32_t> done = 0;
for (std::uint32_t i = 0; i < 200; ++i) {
ThreadPool::Enqueue([&done] { done++; });
}
Check(WaitFor(done, 200), std::format("200 single tasks after the batches all run ({} done)", done.load()));
ThreadPool::Stop();
std::println("{}", failures == 0 ? "all checks passed" : "check(s) failed");
return failures == 0 ? EXIT_SUCCESS : EXIT_FAILURE;
}

43
tests/StopJoinPool/main.cpp Normal file
View file

@ -0,0 +1,43 @@
//SPDX-License-Identifier: LGPL-3.0-only
//SPDX-FileCopyrightText: Copyright (C) 2026 Catcrafts®
// ThreadPool::Stop ends the pool for everyone in it, the thread that joined
// it with JoinPool too. Stop woke and joined the worker threads only, so a
// JoinPool caller slept on, never told. And Stop called again joined threads
// already joined, which throws.
//
// Here another thread stops the pool this one has joined, then stops it
// again. A watchdog fails the test if JoinPool doesn't return.
#include <cstdlib>
#include <unistd.h>
import Crafter.Thread;
import std;
using namespace Crafter;
using namespace std::chrono_literals;
int main() {
std::thread watchdog([] {
std::this_thread::sleep_for(10s);
std::println("FAIL JoinPool didn't return within 10 s of Stop");
_exit(EXIT_FAILURE);
});
watchdog.detach();
ThreadPool::Start(4);
std::atomic<int> ran = 0;
for (int i = 0; i < 100; ++i) ThreadPool::Enqueue([&ran] { ++ran; });
std::thread stopper([] {
std::this_thread::sleep_for(100ms);
ThreadPool::Stop();
});
ThreadPool::JoinPool();
stopper.join();
std::println("{} Stop from another thread ends the JoinPool of the thread that joined, after the queued tasks ({} of 100)",
ran == 100 ? "PASS" : "FAIL", ran.load());
ThreadPool::Stop();
std::println("PASS and stopping it again does nothing");
return ran == 100 ? EXIT_SUCCESS : EXIT_FAILURE;
}