From f71cb43e850f2ef3b7326cd2cd6bdbd55422119a Mon Sep 17 00:00:00 2001 From: Jorijn van der Graaf Date: Tue, 20 Oct 2026 19:06:37 +0200 Subject: [PATCH] fix(threadpool): batches run every task once, and Stop ends the whole pool - 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 --- implementations/Crafter.Thread-ThreadPool.cpp | 264 ++++++++++-------- interfaces/Crafter.Thread-ThreadPool.cppm | 67 +++-- project.cpp | 6 + tests/EnqueueBatch/main.cpp | 77 +++++ tests/StopJoinPool/main.cpp | 43 +++ 5 files changed, 308 insertions(+), 149 deletions(-) create mode 100644 tests/EnqueueBatch/main.cpp create mode 100644 tests/StopJoinPool/main.cpp diff --git a/implementations/Crafter.Thread-ThreadPool.cpp b/implementations/Crafter.Thread-ThreadPool.cpp index 7f0d341..c908b00 100755 --- a/implementations/Crafter.Thread-ThreadPool.cpp +++ b/implementations/Crafter.Thread-ThreadPool.cpp @@ -1,119 +1,147 @@ -//SPDX-License-Identifier: LGPL-3.0-only -//SPDX-FileCopyrightText: Copyright (C) 2026 Catcrafts® - -module Crafter.Thread:ThreadPool_impl; -import std; -import :ThreadPool; - -using namespace Crafter; - - -void ThreadPool::Start(int maxThreads) -{ - stopped = false; - if (maxThreads > 0) { - threadCount = maxThreads; - } - else { - threadCount = std::thread::hardware_concurrency(); - } - threadStorage = new ThreadStorage[threadCount]; - for (unsigned int i = 0; i < threadCount - 1; i++) - { - threadStorage[i].thread = std::thread(ThreadPool::WaitAndExecuteTask, threadStorage + i); - threadIdMap[threadStorage[i].thread.get_id()] = &threadStorage[i]; - } - threadIdMap[std::this_thread::get_id()] = &threadStorage[threadCount - 1]; - activeThreads = threadCount - 1; -} - -void ThreadPool::Stop() -{ - stopped = true; - for (unsigned int i = 0; i < threadCount - 1; i++) - { - threadStorage[i].cv.notify_one(); - threadStorage[i].thread.join(); - } -} - -void ThreadPool::Enqueue(std::function task) -{ - unsigned int min = std::numeric_limits::max(); - unsigned int minIndex = 0; - for (unsigned int i = 0; i < activeThreads; i++) - { - unsigned int taskCount = threadStorage[i].taskCount; - if (taskCount == 0) { - minIndex = i; - break; - } - if (taskCount < min) { - min = taskCount; - minIndex = i; - } - } - { - std::lock_guard lk(threadStorage[minIndex].mutex); - threadStorage[minIndex].buffer.push_back(task); - threadStorage[minIndex].taskCount++; - } - threadStorage[minIndex].cv.notify_one(); -} - -void ThreadPool::Enqueue(std::vector> tasks) -{ - unsigned int size = tasks.size(); - unsigned int tasksPerThread = size / activeThreads; - int remainder = size % activeThreads; - - if(remainder) { - for (unsigned int i = 0; i < activeThreads - 1; 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; - } - threadStorage[i].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; - } - threadStorage[activeThreads - 1].cv.notify_one(); - } else{ - for (unsigned int i = 0; i < activeThreads; 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; - } - threadStorage[i].cv.notify_one(); - } - } -} - -void ThreadPool::JoinPool() -{ - activeThreads++; - ThreadPool::WaitAndExecuteTask(threadStorage + threadCount - 1); -} - -void ThreadPool::WaitAndExecuteTask(ThreadStorage* threadStorage) -{ - while (!stopped || !threadStorage->buffer.empty()) { - std::unique_lock lock(threadStorage->mutex); - threadStorage->cv.wait(lock, [threadStorage] { return !threadStorage->buffer.empty() || stopped; }); - threadStorage->buffer.swap(threadStorage->tasks); - lock.unlock(); - for (std::function task : threadStorage->tasks) - { - task(); - threadStorage->taskCount--; - } - threadStorage->tasks.clear(); - } +//SPDX-License-Identifier: LGPL-3.0-only +//SPDX-FileCopyrightText: Copyright (C) 2026 Catcrafts® + +module Crafter.Thread:ThreadPool_impl; +import std; +import :ThreadPool; + +using namespace Crafter; + + +void ThreadPool::Start(int maxThreads) +{ + stopped = false; + if (maxThreads > 0) { + threadCount = maxThreads; + } + else { + threadCount = std::thread::hardware_concurrency(); + } + threadStorage = new ThreadStorage[threadCount]; + for (unsigned int i = 0; i < threadCount - 1; i++) + { + threadStorage[i].thread = std::thread(ThreadPool::WaitAndExecuteTask, threadStorage + i); + threadIdMap[threadStorage[i].thread.get_id()] = &threadStorage[i]; + } + threadIdMap[std::this_thread::get_id()] = &threadStorage[threadCount - 1]; + activeThreads = threadCount - 1; +} + +void ThreadPool::Stop() +{ + // This pool's own: a JoinPool caller may Start another as soon as it + // returns, replacing threadStorage. + ThreadStorage* const storage = threadStorage; + const unsigned int count = threadCount; + 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 + // checking the condition and going to sleep misses the notify, and never + // wakes. + auto wake = [](ThreadStorage& slot) { + { + std::lock_guard lk(slot.mutex); + } + slot.cv.notify_one(); + }; + for (unsigned int i = 0; i < count - 1; i++) wake(storage[i]); + for (unsigned int i = 0; i < count - 1; i++) + { + if (storage[i].thread.joinable()) storage[i].thread.join(); + } + // 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]); +} + +void ThreadPool::Enqueue(std::function task) +{ + unsigned int min = std::numeric_limits::max(); + unsigned int minIndex = 0; + for (unsigned int i = 0; i < activeThreads; i++) + { + unsigned int taskCount = threadStorage[i].taskCount; + if (taskCount == 0) { + minIndex = i; + break; + } + if (taskCount < min) { + min = taskCount; + minIndex = i; + } + } + { + std::lock_guard lk(threadStorage[minIndex].mutex); + threadStorage[minIndex].buffer.push_back(task); + threadStorage[minIndex].taskCount++; + } + threadStorage[minIndex].cv.notify_one(); +} + +void ThreadPool::Enqueue(std::vector> tasks) +{ + const unsigned int workers = activeThreads; + if (workers == 0) { + // Start(1) runs no worker threads; like a single Enqueue, the batch + // waits for the caller's JoinPool. + { + std::lock_guard lk(threadStorage[0].mutex); + threadStorage[0].taskCount += static_cast(tasks.size()); + threadStorage[0].buffer.insert(threadStorage[0].buffer.end(), + std::make_move_iterator(tasks.begin()), std::make_move_iterator(tasks.end())); + } + threadStorage[0].cv.notify_one(); + return; + } + + // 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 + // worker's taskCount grows by what it was given, which the single-task + // 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; + std::size_t first = 0; + for (unsigned int i = 0; i < workers && first < tasks.size(); i++) + { + const std::size_t count = perWorker + (i < remainder ? 1 : 0); + { + std::lock_guard lk(threadStorage[i].mutex); + threadStorage[i].taskCount += static_cast(count); + threadStorage[i].buffer.insert(threadStorage[i].buffer.end(), + std::make_move_iterator(tasks.begin() + static_cast(first)), + std::make_move_iterator(tasks.begin() + static_cast(first + count))); + } + threadStorage[i].cv.notify_one(); + first += count; + } +} + +void ThreadPool::JoinPool() +{ + activeThreads++; + ThreadPool::WaitAndExecuteTask(threadStorage + threadCount - 1); +} + +void ThreadPool::WaitAndExecuteTask(ThreadStorage* threadStorage) +{ + while (true) { + std::unique_lock 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& task : threadStorage->tasks) + { + task(); + threadStorage->taskCount--; + } + threadStorage->tasks.clear(); + } } \ No newline at end of file diff --git a/interfaces/Crafter.Thread-ThreadPool.cppm b/interfaces/Crafter.Thread-ThreadPool.cppm index 61de12d..9d82213 100755 --- a/interfaces/Crafter.Thread-ThreadPool.cppm +++ b/interfaces/Crafter.Thread-ThreadPool.cppm @@ -1,32 +1,37 @@ -//SPDX-License-Identifier: LGPL-3.0-only -//SPDX-FileCopyrightText: Copyright (C) 2026 Catcrafts® - -export module Crafter.Thread:ThreadPool; -import std; - -namespace Crafter { - struct ThreadStorage { - std::vector> tasks; - std::vector> buffer; - std::condition_variable cv; - std::mutex mutex; - std::atomic taskCount; - std::thread thread; - }; - - export class ThreadPool { - public: - static inline std::atomic stopped; - static void Start(int maxThreads = -1); - static void Stop(); - static void Enqueue(std::function task); - static void Enqueue(std::vector> tasks); - static void JoinPool(); - static inline unsigned int threadCount; - static inline std::atomic activeThreads; - private: - static inline std::unordered_map threadIdMap; - static inline ThreadStorage* threadStorage; - static void WaitAndExecuteTask(ThreadStorage* threadStorage); - }; +//SPDX-License-Identifier: LGPL-3.0-only +//SPDX-FileCopyrightText: Copyright (C) 2026 Catcrafts® + +export module Crafter.Thread:ThreadPool; +import std; + +namespace Crafter { + struct ThreadStorage { + std::vector> tasks; + std::vector> buffer; + std::condition_variable cv; + std::mutex mutex; + std::atomic taskCount; + std::thread thread; + }; + + export class ThreadPool { + public: + static inline std::atomic stopped; + static void Start(int maxThreads = -1); + // Stops the workers, each once it has run what it was given, and ends + // a JoinPool; a task enqueued once Stop has begun may not run. From + // one thread at a time, not the pool's own: a task can't stop the + // pool it runs in. Again, or before Start, it does nothing. JoinPool + // can return before Stop has: Start a new pool once Stop returned. + static void Stop(); + static void Enqueue(std::function task); + static void Enqueue(std::vector> tasks); + static void JoinPool(); + static inline unsigned int threadCount; + static inline std::atomic activeThreads; + private: + static inline std::unordered_map threadIdMap; + static inline ThreadStorage* threadStorage; + static void WaitAndExecuteTask(ThreadStorage* threadStorage); + }; } \ No newline at end of file diff --git a/project.cpp b/project.cpp index d6b67ed..72f932b 100644 --- a/project.cpp +++ b/project.cpp @@ -23,5 +23,11 @@ extern "C" Configuration CrafterBuildProject(std::span a }; 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; } diff --git a/tests/EnqueueBatch/main.cpp b/tests/EnqueueBatch/main.cpp new file mode 100644 index 0000000..3387e64 --- /dev/null +++ b/tests/EnqueueBatch/main.cpp @@ -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 +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& 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 done = 0; + std::vector> runs(count); + std::vector> 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& 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 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; +} diff --git a/tests/StopJoinPool/main.cpp b/tests/StopJoinPool/main.cpp new file mode 100644 index 0000000..313b439 --- /dev/null +++ b/tests/StopJoinPool/main.cpp @@ -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 +#include + +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 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; +}