From 15b18a8f1f7e29c9e98e87a56032573b7b10472c Mon Sep 17 00:00:00 2001 From: agent Date: Tue, 27 Oct 2026 01:16:26 +0300 Subject: [PATCH 1/3] Threading: cpu topology and thread pinning ThreadPinning reads the sibling groups from the kernel topology and pins the calling thread to a cpu set. It is stateless and per-thread: a caller reads the groups, picks the cpus for a thread and pins it, so any mapping of threads to cpus is expressible. --- CMakeLists.txt | 1 + Threading/CMakeLists.txt | 15 +++++ Threading/private/ThreadPinning.cpp | 88 ++++++++++++++++++++++++++ Threading/public/ThreadPinning.hpp | 13 ++++ Threading/tests/Test.cpp | 3 + Threading/tests/ThreadPinningTests.cpp | 56 ++++++++++++++++ 6 files changed, 176 insertions(+) create mode 100644 Threading/CMakeLists.txt create mode 100644 Threading/private/ThreadPinning.cpp create mode 100644 Threading/public/ThreadPinning.hpp create mode 100644 Threading/tests/Test.cpp create mode 100644 Threading/tests/ThreadPinningTests.cpp diff --git a/CMakeLists.txt b/CMakeLists.txt index a4e3f718..db5b4851 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -19,6 +19,7 @@ include(cmake/FindPortAudio.cmake) add_subdirectory(Externals) add_subdirectory(Modules) +add_subdirectory(Threading) add_subdirectory(Containers) add_subdirectory(Math) add_subdirectory(Callstack) diff --git a/Threading/CMakeLists.txt b/Threading/CMakeLists.txt new file mode 100644 index 00000000..605b5cc8 --- /dev/null +++ b/Threading/CMakeLists.txt @@ -0,0 +1,15 @@ +project(Threading) + +### ---------------------- Static Library --------------------- ### +file(GLOB SOURCES "./private/*.cpp") +file(GLOB HEADERS "./public/*.hpp") +add_library(${PROJECT_NAME} STATIC ${SOURCES} ${HEADERS}) +target_include_directories(${PROJECT_NAME} PUBLIC ./public/) +find_package(Threads REQUIRED) +target_link_libraries(${PROJECT_NAME} PUBLIC Threads::Threads) + +### -------------------------- Tests -------------------------- ### +file(GLOB TEST_SOURCES "./tests/*.cpp") +add_executable(Tests${PROJECT_NAME} ${TEST_SOURCES}) +target_link_libraries(Tests${PROJECT_NAME} ${PROJECT_NAME} UnitTest++) +add_test(NAME Tests${PROJECT_NAME} COMMAND Tests${PROJECT_NAME}) diff --git a/Threading/private/ThreadPinning.cpp b/Threading/private/ThreadPinning.cpp new file mode 100644 index 00000000..1e3516df --- /dev/null +++ b/Threading/private/ThreadPinning.cpp @@ -0,0 +1,88 @@ +#include "ThreadPinning.hpp" + +#include +#include +#include +#include +#include + +#if defined(__linux__) || defined(__ANDROID__) +#include +#define TP_THREAD_PINNING_SUPPORTED +#endif + +namespace tp { + + namespace { + std::vector parseCpuList(const std::string& text) { + std::vector cpus; + std::stringstream stream(text); + std::string range; + while (std::getline(stream, range, ',')) { + if (range.empty()) continue; + const size_t dash = range.find('-'); + if (dash == std::string::npos) { + cpus.push_back(std::stoi(range)); + continue; + } + const int first = std::stoi(range.substr(0, dash)); + const std::string tail = range.substr(dash + 1); + const int last = tail.empty() ? first : std::stoi(tail); + for (int cpu = first; cpu <= last; cpu++) + cpus.push_back(cpu); + } + return cpus; + } + + std::vector siblingsOf(int cpu) { + std::ifstream file("/sys/devices/system/cpu/cpu" + std::to_string(cpu) + "/topology/thread_siblings_list"); + std::string text; + if (!file || !std::getline(file, text)) return { cpu }; + const std::vector siblings = parseCpuList(text); + return siblings.empty() ? std::vector{ cpu } : siblings; + } + } + + std::vector ThreadPinning::allowedCpus() { +#ifdef TP_THREAD_PINNING_SUPPORTED + cpu_set_t set; + CPU_ZERO(&set); + if (sched_getaffinity(0, sizeof(set), &set) != 0) return {}; + std::vector cpus; + for (int cpu = 0; cpu < CPU_SETSIZE; cpu++) + if (CPU_ISSET(cpu, &set)) cpus.push_back(cpu); + return cpus; +#else + return {}; +#endif + } + + std::vector> ThreadPinning::siblingGroups(const std::vector& allowed) { + const std::vector cpus = allowed.empty() ? allowedCpus() : allowed; + std::vector> groups; + std::set seen; + for (int cpu : cpus) { + if (seen.count(cpu)) continue; + std::vector group; + for (int sibling : siblingsOf(cpu)) { + if (!std::count(cpus.begin(), cpus.end(), sibling)) continue; + group.push_back(sibling); + seen.insert(sibling); + } + if (group.empty()) group.push_back(cpu); + groups.push_back(group); + } + return groups; + } + + void ThreadPinning::pinCallingThread(const std::vector& cpus) { +#ifdef TP_THREAD_PINNING_SUPPORTED + if (cpus.empty()) return; + cpu_set_t set; + CPU_ZERO(&set); + for (int cpu : cpus) + CPU_SET(cpu, &set); + sched_setaffinity(0, sizeof(set), &set); +#endif + } +} diff --git a/Threading/public/ThreadPinning.hpp b/Threading/public/ThreadPinning.hpp new file mode 100644 index 00000000..0b88e8f0 --- /dev/null +++ b/Threading/public/ThreadPinning.hpp @@ -0,0 +1,13 @@ +#pragma once + +#include + +namespace tp { + + class ThreadPinning { + public: + [[nodiscard]] static std::vector allowedCpus(); + [[nodiscard]] static std::vector> siblingGroups(const std::vector& allowed = {}); + static void pinCallingThread(const std::vector& cpus); + }; +} diff --git a/Threading/tests/Test.cpp b/Threading/tests/Test.cpp new file mode 100644 index 00000000..7ae121b1 --- /dev/null +++ b/Threading/tests/Test.cpp @@ -0,0 +1,3 @@ +#include "UnitTest++/UnitTest++.h" + +int main() { return UnitTest::RunAllTests(); } diff --git a/Threading/tests/ThreadPinningTests.cpp b/Threading/tests/ThreadPinningTests.cpp new file mode 100644 index 00000000..2242f0a6 --- /dev/null +++ b/Threading/tests/ThreadPinningTests.cpp @@ -0,0 +1,56 @@ +#include "ThreadPinning.hpp" + +#include "UnitTest++/UnitTest++.h" + +#include +#include +#include + +using namespace tp; + +namespace { + std::vector sorted(std::vector values) { + std::sort(values.begin(), values.end()); + return values; + } +} + +SUITE(ThreadPinning) { + TEST(SiblingGroupsCoverEveryAllowedCpuExactlyOnce) { + const std::vector allowed = ThreadPinning::allowedCpus(); + if (allowed.empty()) return; + const auto groups = ThreadPinning::siblingGroups(allowed); + std::vector seen; + for (const auto& group : groups) + seen.insert(seen.end(), group.begin(), group.end()); + CHECK(sorted(seen) == sorted(allowed)); + } + + TEST(PinningRestrictsTheCallingThread) { + const std::vector allowed = ThreadPinning::allowedCpus(); + const auto groups = ThreadPinning::siblingGroups(allowed); + if (groups.size() < 2) return; + ThreadPinning::pinCallingThread(groups.front()); + CHECK(sorted(ThreadPinning::allowedCpus()) == sorted(groups.front())); + } + + TEST(TwoThreadsCanBePinnedToDifferentGroups) { + const auto groups = ThreadPinning::siblingGroups(); + if (groups.size() < 2) return; + + std::vector first, second; + std::thread left([&]() { + ThreadPinning::pinCallingThread(groups[0]); + first = ThreadPinning::allowedCpus(); + }); + std::thread right([&]() { + ThreadPinning::pinCallingThread(groups[1]); + second = ThreadPinning::allowedCpus(); + }); + left.join(); + right.join(); + + CHECK(sorted(first) == sorted(groups[0])); + CHECK(sorted(second) == sorted(groups[1])); + } +} -- 2.49.1 From e9b70045490f8c2203de94919b7f3cd413f4a04c Mon Sep 17 00:00:00 2001 From: agent Date: Tue, 27 Oct 2026 01:16:31 +0300 Subject: [PATCH 2/3] Threading: queue jobs in one lane per worker thread WorkerLanes keeps one job queue per lane, runs each lane on its own thread and can pause every lane to drain the queues on the calling thread. A thread-start hook lets a caller run code once per lane (pin it, name it) without the queue knowing why. --- Threading/public/WorkerLanes.hpp | 219 +++++++++++++++++++++++++++ Threading/tests/WorkerLanesTests.cpp | 121 +++++++++++++++ 2 files changed, 340 insertions(+) create mode 100644 Threading/public/WorkerLanes.hpp create mode 100644 Threading/tests/WorkerLanesTests.cpp diff --git a/Threading/public/WorkerLanes.hpp b/Threading/public/WorkerLanes.hpp new file mode 100644 index 00000000..6c332ab3 --- /dev/null +++ b/Threading/public/WorkerLanes.hpp @@ -0,0 +1,219 @@ +#pragma once + +#include +#include +#include +#include +#include +#include +#include +#include + +namespace tp { + + template + class WorkerLanes { + struct Lane { + std::mutex queueMutex; + std::condition_variable wake; + std::vector queue; + bool stopping = false; + std::mutex workMutex; + std::thread thread; + }; + + public: + using Handler = std::function; + using ThreadStart = std::function; + + class Pause { + public: + explicit Pause(WorkerLanes* owner) : + mOwner(owner) { + if (!mOwner) return; + mOwner->mPaused.store(true, std::memory_order_release); + for (auto& lane : mOwner->mLanes) + lane->workMutex.lock(); + } + + ~Pause() { release(); } + + Pause(const Pause&) = delete; + Pause& operator=(const Pause&) = delete; + Pause(Pause&& other) noexcept : + mOwner(other.mOwner) { + other.mOwner = nullptr; + } + + void release() { + if (!mOwner) return; + for (auto& lane : mOwner->mLanes) + lane->workMutex.unlock(); + { + std::lock_guard lock(mOwner->mGateMutex); + mOwner->mPaused.store(false, std::memory_order_release); + } + mOwner->mGate.notify_all(); + mOwner = nullptr; + } + + private: + WorkerLanes* mOwner; + }; + + WorkerLanes(size_t lanes, Handler handler, ThreadStart onStart = {}) : + mHandler(std::move(handler)), + mOnStart(std::move(onStart)) { + for (size_t index = 0; index < lanes; index++) + mLanes.push_back(std::make_unique()); + for (size_t index = 0; index < mLanes.size(); index++) + mLanes[index]->thread = std::thread([this, index]() { + if (mOnStart) mOnStart(index); + run(index); + }); + } + + ~WorkerLanes() { + for (auto& lane : mLanes) { + { + std::lock_guard lock(lane->queueMutex); + lane->stopping = true; + } + lane->wake.notify_one(); + } + for (auto& lane : mLanes) + lane->thread.join(); + } + + WorkerLanes(const WorkerLanes&) = delete; + WorkerLanes& operator=(const WorkerLanes&) = delete; + + [[nodiscard]] size_t laneCount() const { return mLanes.size(); } + [[nodiscard]] size_t pending() const { return mPending.load(std::memory_order_acquire); } + + void push(size_t lane, Job job) { + if (mLanes.empty()) { + mHandler(0, job); + return; + } + + Lane& target = *mLanes[lane % mLanes.size()]; + mPending.fetch_add(1, std::memory_order_relaxed); + bool wasEmpty = false; + { + std::lock_guard lock(target.queueMutex); + wasEmpty = target.queue.empty(); + target.queue.push_back(std::move(job)); + } + if (wasEmpty) target.wake.notify_one(); + } + + void pushAll(size_t lane, std::vector& jobs) { + if (jobs.empty()) return; + if (mLanes.empty()) { + for (Job& job : jobs) + mHandler(0, job); + jobs.clear(); + return; + } + + Lane& target = *mLanes[lane % mLanes.size()]; + mPending.fetch_add(jobs.size(), std::memory_order_relaxed); + bool wasEmpty = false; + { + std::lock_guard lock(target.queueMutex); + wasEmpty = target.queue.empty(); + for (Job& job : jobs) + target.queue.push_back(std::move(job)); + } + jobs.clear(); + if (wasEmpty) target.wake.notify_one(); + } + + void wait() { + if (pending() == 0) return; + std::unique_lock lock(mIdleMutex); + mIdle.wait(lock, [this]() { return pending() == 0; }); + } + + [[nodiscard]] Pause pause() { return Pause(mLanes.empty() ? nullptr : this); } + + void runQueuedHere() { + for (size_t index = 0; index < mLanes.size(); index++) { + Lane& lane = *mLanes[index]; + std::vector batch; + { + std::lock_guard lock(lane.queueMutex); + batch.swap(lane.queue); + } + for (Job& job : batch) { + mHandler(index, job); + mPending.fetch_sub(1, std::memory_order_acq_rel); + } + } + } + + private: + void waitAtGate() { + if (!mPaused.load(std::memory_order_acquire)) return; + std::unique_lock lock(mGateMutex); + mGate.wait(lock, [this]() { return !mPaused.load(std::memory_order_acquire); }); + } + + void run(size_t index) { + Lane& lane = *mLanes[index]; + std::vector batch; + while (true) { + { + std::unique_lock lock(lane.queueMutex); + lane.wake.wait(lock, [&lane]() { return lane.stopping || !lane.queue.empty(); }); + if (lane.queue.empty()) return; + batch.swap(lane.queue); + } + + for (size_t at = 0; at < batch.size(); at++) { + if (mPaused.load(std::memory_order_acquire)) { + { + std::lock_guard lock(lane.queueMutex); + lane.queue.insert( + lane.queue.begin(), + std::make_move_iterator(batch.begin() + (std::ptrdiff_t) at), + std::make_move_iterator(batch.end()) + ); + } + break; + } + { + std::lock_guard work(lane.workMutex); + if (mPaused.load(std::memory_order_acquire)) { + std::lock_guard lock(lane.queueMutex); + lane.queue.insert( + lane.queue.begin(), + std::make_move_iterator(batch.begin() + (std::ptrdiff_t) at), + std::make_move_iterator(batch.end()) + ); + break; + } + mHandler(index, batch[at]); + } + if (mPending.fetch_sub(1, std::memory_order_acq_rel) == 1) { + std::lock_guard lock(mIdleMutex); + mIdle.notify_all(); + } + } + batch.clear(); + waitAtGate(); + } + } + + Handler mHandler; + ThreadStart mOnStart; + std::vector> mLanes; + std::atomic mPending{ 0 }; + std::mutex mIdleMutex; + std::condition_variable mIdle; + std::atomic mPaused{ false }; + std::mutex mGateMutex; + std::condition_variable mGate; + }; +} diff --git a/Threading/tests/WorkerLanesTests.cpp b/Threading/tests/WorkerLanesTests.cpp new file mode 100644 index 00000000..2e2fc8c0 --- /dev/null +++ b/Threading/tests/WorkerLanesTests.cpp @@ -0,0 +1,121 @@ +#include "WorkerLanes.hpp" + +#include "UnitTest++/UnitTest++.h" + +#include +#include +#include +#include + +using namespace tp; + +namespace { + struct Step { + size_t lane = 0; + int value = 0; + }; +} + +SUITE(WorkerLanes) { + TEST(WaitReturnsAfterEveryJobRan) { + std::atomic sum{ 0 }; + WorkerLanes lanes(4, [&sum](size_t, Step& step) { sum += step.value; }); + for (int value = 1; value <= 1000; value++) + lanes.push((size_t) value, { (size_t) value, value }); + lanes.wait(); + CHECK_EQUAL(500500, sum.load()); + CHECK_EQUAL(0u, lanes.pending()); + } + + TEST(OneLaneRunsItsJobsInOrder) { + std::vector seen[3]; + WorkerLanes lanes(3, [&seen](size_t, Step& step) { seen[step.lane].push_back(step.value); }); + for (int value = 0; value < 300; value++) + lanes.push((size_t) value % 3, { (size_t) value % 3, value }); + lanes.wait(); + for (size_t lane = 0; lane < 3; lane++) { + CHECK_EQUAL(100u, seen[lane].size()); + for (size_t index = 1; index < seen[lane].size(); index++) + CHECK(seen[lane][index - 1] < seen[lane][index]); + } + } + + TEST(JobsRunOffTheCallingThread) { + std::mutex mutex; + std::set threads; + WorkerLanes lanes(2, [&](size_t, Step&) { + std::lock_guard lock(mutex); + threads.insert(std::this_thread::get_id()); + }); + for (int value = 0; value < 64; value++) + lanes.push((size_t) value, {}); + lanes.wait(); + CHECK(threads.count(std::this_thread::get_id()) == 0); + CHECK(!threads.empty()); + } + + TEST(BatchedJobsKeepTheirOrder) { + std::vector seen; + WorkerLanes lanes(2, [&seen](size_t, Step& step) { seen.push_back(step.value); }); + std::vector batch; + for (int value = 0; value < 50; value++) + batch.push_back({ 1, value }); + lanes.pushAll(1, batch); + CHECK(batch.empty()); + lanes.wait(); + CHECK_EQUAL(50u, seen.size()); + for (size_t index = 0; index < seen.size(); index++) + CHECK_EQUAL((int) index, seen[index]); + } + + TEST(PauseHoldsJobsUntilReleased) { + std::atomic ran{ 0 }; + WorkerLanes lanes(2, [&ran](size_t, Step&) { ran++; }); + { + auto pause = lanes.pause(); + for (int value = 0; value < 20; value++) + lanes.push((size_t) value, {}); + std::this_thread::sleep_for(std::chrono::milliseconds(20)); + CHECK_EQUAL(0, ran.load()); + CHECK_EQUAL(20u, lanes.pending()); + } + lanes.wait(); + CHECK_EQUAL(20, ran.load()); + } + + TEST(ZeroLanesRunJobsImmediately) { + int sum = 0; + WorkerLanes lanes(0, [&sum](size_t, Step& step) { sum += step.value; }); + lanes.push(0, { 0, 7 }); + CHECK_EQUAL(7, sum); + CHECK_EQUAL(0u, lanes.pending()); + lanes.wait(); + } + + TEST(DestroyingFinishesQueuedJobs) { + std::atomic ran{ 0 }; + { + WorkerLanes lanes(2, [&ran](size_t, Step&) { ran++; }); + for (int value = 0; value < 500; value++) + lanes.push((size_t) value, {}); + } + CHECK_EQUAL(500, ran.load()); + } + + TEST(OnStartRunsOncePerLaneOffTheCallingThread) { + std::mutex mutex; + std::set lanes; + std::set ids; + std::atomic started{ 0 }; + WorkerLanes pool(3, [](size_t, Step&) {}, [&](size_t index) { + std::lock_guard lock(mutex); + lanes.insert(index); + ids.insert(std::this_thread::get_id()); + started++; + }); + while (started.load() < 3) + std::this_thread::yield(); + CHECK_EQUAL(3u, lanes.size()); + CHECK(ids.count(std::this_thread::get_id()) == 0); + } +} -- 2.49.1 From 40a3535393bc1428e18834bd4f89c219fc416d56 Mon Sep 17 00:00:00 2001 From: agent Date: Tue, 27 Oct 2026 01:16:31 +0300 Subject: [PATCH 3/3] Threading: parallel for over a fixed worker pool ParallelFor runs a counted loop across a fixed set of threads created up front and hands out indices from an atomic counter, so the calling thread shares the work. A thread-start hook runs once per helper thread. --- Threading/private/ParallelFor.cpp | 74 ++++++++++++++++++++++++++++ Threading/public/ParallelFor.hpp | 43 ++++++++++++++++ Threading/tests/ParallelForTests.cpp | 66 +++++++++++++++++++++++++ 3 files changed, 183 insertions(+) create mode 100644 Threading/private/ParallelFor.cpp create mode 100644 Threading/public/ParallelFor.hpp create mode 100644 Threading/tests/ParallelForTests.cpp diff --git a/Threading/private/ParallelFor.cpp b/Threading/private/ParallelFor.cpp new file mode 100644 index 00000000..a81218cb --- /dev/null +++ b/Threading/private/ParallelFor.cpp @@ -0,0 +1,74 @@ +#include "ParallelFor.hpp" + +namespace tp { + + ParallelFor::ParallelFor(size_t threads, ThreadStart onStart) : + mOnStart(std::move(onStart)) { + for (size_t index = 1; index < threads; index++) + mThreads.emplace_back([this, index]() { + if (mOnStart) mOnStart(index); + work(0); + }); + } + + ParallelFor::~ParallelFor() { + { + std::lock_guard lock(mMutex); + mStopping = true; + } + mStart.notify_all(); + for (std::thread& thread : mThreads) + thread.join(); + } + + void ParallelFor::drain() { + while (true) { + const size_t index = mNext.fetch_add(1, std::memory_order_relaxed); + if (index >= mCount) return; + (*mBody)(index); + } + } + + void ParallelFor::work(size_t seen) { + while (true) { + { + std::unique_lock lock(mMutex); + mStart.wait(lock, [&]() { return mStopping || (mGeneration != seen && mBody); }); + if (mStopping) return; + seen = mGeneration; + mBusy++; + } + drain(); + bool last = false; + { + std::lock_guard lock(mMutex); + last = --mBusy == 0; + } + if (last) mDone.notify_one(); + } + } + + void ParallelFor::run(size_t count, const std::function& body) { + if (count == 0) return; + if (mThreads.empty() || count == 1) { + for (size_t index = 0; index < count; index++) + body(index); + return; + } + + { + std::lock_guard lock(mMutex); + mBody = &body; + mCount = count; + mNext.store(0, std::memory_order_relaxed); + mGeneration++; + } + mStart.notify_all(); + drain(); + + std::unique_lock lock(mMutex); + mDone.wait(lock, [this]() { return mBusy == 0; }); + mBody = nullptr; + mCount = 0; + } +} diff --git a/Threading/public/ParallelFor.hpp b/Threading/public/ParallelFor.hpp new file mode 100644 index 00000000..1a4eb8d1 --- /dev/null +++ b/Threading/public/ParallelFor.hpp @@ -0,0 +1,43 @@ +#pragma once + +#include +#include +#include +#include +#include +#include +#include + +namespace tp { + + class ParallelFor { + public: + using ThreadStart = std::function; + + explicit ParallelFor(size_t threads, ThreadStart onStart = {}); + ~ParallelFor(); + + ParallelFor(const ParallelFor&) = delete; + ParallelFor& operator=(const ParallelFor&) = delete; + + [[nodiscard]] size_t threads() const { return mThreads.size() + 1; } + + void run(size_t count, const std::function& body); + + private: + void work(size_t generation); + void drain(); + + std::vector mThreads; + ThreadStart mOnStart; + std::mutex mMutex; + std::condition_variable mStart; + std::condition_variable mDone; + const std::function* mBody = nullptr; + size_t mCount = 0; + std::atomic mNext{ 0 }; + size_t mBusy = 0; + size_t mGeneration = 0; + bool mStopping = false; + }; +} diff --git a/Threading/tests/ParallelForTests.cpp b/Threading/tests/ParallelForTests.cpp new file mode 100644 index 00000000..ece88f6c --- /dev/null +++ b/Threading/tests/ParallelForTests.cpp @@ -0,0 +1,66 @@ +#include "ParallelFor.hpp" + +#include "UnitTest++/UnitTest++.h" + +#include +#include +#include +#include +#include +#include + +using namespace tp; + +SUITE(ParallelFor) { + TEST(EveryIndexRunsOnce) { + ParallelFor parallel(4); + std::vector> hits(1000); + parallel.run(hits.size(), [&hits](size_t index) { hits[index]++; }); + for (const auto& hit : hits) + CHECK_EQUAL(1, hit.load()); + } + + TEST(RunsAreRepeatableAndReturnWhenDone) { + ParallelFor parallel(3); + for (int round = 0; round < 200; round++) { + std::atomic sum{ 0 }; + parallel.run(37, [&sum](size_t index) { sum += (int) index; }); + CHECK_EQUAL(666, sum.load()); + } + } + + TEST(WorkIsSharedWithHelperThreads) { + ParallelFor parallel(4); + CHECK_EQUAL(4u, parallel.threads()); + std::mutex mutex; + std::set threads; + parallel.run(64, [&](size_t) { + std::this_thread::sleep_for(std::chrono::microseconds(500)); + std::lock_guard lock(mutex); + threads.insert(std::this_thread::get_id()); + }); + CHECK(threads.size() > 1); + } + + TEST(SingleThreadRunsInline) { + ParallelFor parallel(1); + const std::thread::id caller = std::this_thread::get_id(); + bool inline_ = true; + parallel.run(10, [&](size_t) { inline_ = inline_ && std::this_thread::get_id() == caller; }); + CHECK(inline_); + parallel.run(0, [](size_t) {}); + } + + TEST(OnStartRunsForEachHelperThread) { + std::mutex mutex; + std::set indices; + ParallelFor parallel(4, [&](size_t index) { + std::lock_guard lock(mutex); + indices.insert(index); + }); + parallel.run(8, [](size_t) {}); + CHECK_EQUAL(3u, indices.size()); + for (size_t index = 1; index < 4; index++) + CHECK(indices.count(index) != 0); + } +} -- 2.49.1