diff --git a/UnitTests/UnitTests.vcxproj b/UnitTests/UnitTests.vcxproj index 33eac273d..5bbbf4303 100644 --- a/UnitTests/UnitTests.vcxproj +++ b/UnitTests/UnitTests.vcxproj @@ -124,6 +124,7 @@ Create + diff --git a/UnitTests/UnitTests.vcxproj.filters b/UnitTests/UnitTests.vcxproj.filters index 69db57137..48ea9a45d 100644 --- a/UnitTests/UnitTests.vcxproj.filters +++ b/UnitTests/UnitTests.vcxproj.filters @@ -42,6 +42,9 @@ Source Files + + Source Files + diff --git a/UnitTests/pch.hpp b/UnitTests/pch.hpp index 7e9a3ad0b..830b74348 100644 --- a/UnitTests/pch.hpp +++ b/UnitTests/pch.hpp @@ -16,10 +16,12 @@ #include #include #include +#include #include #include #include #include +#include #include #include #include @@ -32,4 +34,5 @@ #include #include +#include #include diff --git a/UnitTests/pool.cpp b/UnitTests/pool.cpp new file mode 100644 index 000000000..2bda06e1c --- /dev/null +++ b/UnitTests/pool.cpp @@ -0,0 +1,42 @@ +/* + SPDX-License-Identifier: MIT + Copyright © 2023 Amebis +*/ + +#include "pch.hpp" + +using namespace std; +#ifdef _WIN32 +using namespace Microsoft::VisualStudio::CppUnitTestFramework; +#endif + +namespace UnitTests +{ + constexpr size_t capacity = 50; + + TEST_CLASS(pool) + { + public: + TEST_METHOD(test) + { + using worker_t = unique_ptr; + using pool_t = stdex::pool; + pool_t pool; + list workers; + for (auto n = thread::hardware_concurrency(); n--; ) { + workers.push_back(std::move(thread([](_Inout_ pool_t& pool) + { + for (size_t n = 10000; n--; ) { + worker_t el = move(pool.pop()); + if (!el) + el.reset(new int(1)); + pool.push(move(el)); + } + }, ref(pool)))); + } + + for (auto& w : workers) + w.join(); + } + }; +} diff --git a/include/stdex/pool.hpp b/include/stdex/pool.hpp new file mode 100644 index 000000000..51dd57c7c --- /dev/null +++ b/include/stdex/pool.hpp @@ -0,0 +1,91 @@ +/* + SPDX-License-Identifier: MIT + Copyright © 2023 Amebis +*/ + +#pragma once + +#include "compat.hpp" +#include "spinlock.hpp" +#include "windows.h" +#include +#include +#include + +namespace stdex +{ + /// + /// Per-NUMA pool of items + /// + template + class pool + { + public: +#ifdef _WIN32 + using numaid_t = USHORT; +#else + using numaid_t = int; +#endif + + private: + struct numaentry_t { + mutable spinlock lock; + std::list list; + }; + + mutable std::mutex m_mutex; + std::map m_available; + + private: + static inline numaid_t numa_node() + { +#ifdef _WIN32 + PROCESSOR_NUMBER Processor; + GetCurrentProcessorNumberEx(&Processor); + USHORT NodeNumber = 0; + return GetNumaProcessorNodeEx(&Processor, &NodeNumber) ? NodeNumber : 0; +#else + return numa_node_of_cpu(sched_getcpu()); +#endif + } + + inline numaentry_t& numa_entry(numaid_t numa = numa_node()) + { + const std::lock_guard guard(m_mutex); + return m_available[numa]; + } + + public: + /// + /// Removes an item from the pool + /// + /// \param[in] numa NUMA node to identify subpool to remove item from + /// + /// \returns An item from the pool or default value if pool is empty + /// + T pop(_In_ numaid_t numa = numa_node()) + { + auto& ne = numa_entry(numa); + const std::lock_guard guard(ne.lock); + if (!ne.list.empty()) { + auto r = std::move(ne.list.front()); + ne.list.pop_front(); + return r; + } + return T(); + } + + /// + /// Adds an item to the pool + /// + /// \param[in] r Item to add + /// \param[in] numa NUMA node to identify subpool to add item to + /// + void push(_Inout_ T&& r, _In_ numaid_t numa = numa_node()) + { + auto& ne = numa_entry(numa); + const std::lock_guard guard(ne.lock); + ne.list.push_front(std::move(r)); + } + }; +} \ No newline at end of file diff --git a/include/stdex/spinlock.hpp b/include/stdex/spinlock.hpp new file mode 100644 index 000000000..17546bf3a --- /dev/null +++ b/include/stdex/spinlock.hpp @@ -0,0 +1,76 @@ +/* + SPDX-License-Identifier: MIT + Copyright © 2023 Amebis +*/ + +#pragma once + +#ifdef _WIN32 +#include "windows.h" +#include +#endif +#include + +namespace stdex +{ + /// + /// Spin-lock + /// + /// \sa [Correctly implementing a spinlock in C++](https://rigtorp.se/spinlock/) + /// + class spinlock + { + private: + std::atomic m_lock = { false }; + + public: + /// + /// Blocks until a lock can be acquired for the current execution agent (thread, process, task). + /// + void lock() noexcept + { + for (;;) { + // Optimistically assume the lock is free on the first try + if (!m_lock.exchange(true, std::memory_order_acquire)) + return; + + // Wait for lock to be released without generating cache misses + while (m_lock.load(std::memory_order_relaxed)) { + // Issue X86 PAUSE or ARM YIELD instruction to reduce contention between + // hyper-threads +#if _M_ARM || _M_ARM64 + __yield(); +#elif _M_IX86 || _M_X64 + _mm_pause(); +#elif __aarch64__ + __yield(); +#elif __i386__ || __x86_64__ + __builtin_ia32_pause(); +#endif + } + } + } + + /// + /// Attempts to acquire the lock for the current execution agent (thread, process, task) without blocking. + /// + /// \returns true if the lock was acquired, false otherwise + /// + bool try_lock() noexcept + { + // First do a relaxed load to check if lock is free in order to prevent + // unnecessary cache misses if someone does while(!try_lock()) + return + !m_lock.load(std::memory_order_relaxed) && + !m_lock.exchange(true, std::memory_order_acquire); + } + + /// + /// Releases the non-shared lock held by the execution agent. + /// + void unlock() noexcept + { + m_lock.store(false, std::memory_order_release); + } + }; +}