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/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/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/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/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/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/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); + } +} 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])); + } +} 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); + } +}