From 90901f4bdd611e79191eb1ec6c9a47ca10a96204 Mon Sep 17 00:00:00 2001 From: GagaLP Date: Tue, 14 Jul 2026 14:48:29 +0200 Subject: [PATCH] Enable preliminary thread affinity/pinning support on Windows --- CHANGELOG.md | 8 ++ CMakeLists.txt | 1 + include/affinity.h | 4 + .../affinity_win32_adapter.h | 121 +++++++++++++++++ src/platform_specific/affinity.unix.cc | 3 + src/platform_specific/affinity.win.cc | 23 ++-- .../affinity_win32_adapter.cc | 110 ++++++++++++++++ test/affinity_tests.cc | 122 +++++++++++++++--- 8 files changed, 358 insertions(+), 34 deletions(-) create mode 100644 include/platform_specific/affinity_win32_adapter.h create mode 100644 src/platform_specific/affinity_win32_adapter.cc diff --git a/CHANGELOG.md b/CHANGELOG.md index 950892e64..19fc4e0b0 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,14 @@ The format is based on [Keep a Changelog](http://keepachangelog.com/en/1.0.0/) and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.html). +## [WIP] + +This release is WIP + +### Added + +- Add preliminary support for thread affinity on Windows by emulating the POSIX interface (#338) + ## [0.7.0] - 2025-08-18 This release includes changes that may require adjustments when upgrading: diff --git a/CMakeLists.txt b/CMakeLists.txt index 5f6a4d3cc..bf8def0ff 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -285,6 +285,7 @@ set(CELERITY_DETAIL_HAS_NAMED_THREADS OFF) if(WIN32) set(SOURCES ${SOURCES} src/platform_specific/affinity.win.cc) + set(SOURCES ${SOURCES} src/platform_specific/affinity_win32_adapter.cc) set(SOURCES ${SOURCES} src/platform_specific/named_threads.win.cc) set(CELERITY_DETAIL_HAS_NAMED_THREADS ON) elseif(UNIX) diff --git a/include/affinity.h b/include/affinity.h index 14f337c2e..769cd861c 100644 --- a/include/affinity.h +++ b/include/affinity.h @@ -6,6 +6,10 @@ #include "named_threads.h" +#ifdef _WIN32 +#include "platform_specific/affinity_win32_adapter.h" +#endif + // The goal of this thread pinning mechanism, when enabled, is to ensure that threads which benefit from fast communication // are pinned to cores that are close to each other in terms of cache hierarchy. // It currently accomplishes this by pinning threads to cores in a round-robin fashion according to their order in the `named_threads::thread_type` enum. diff --git a/include/platform_specific/affinity_win32_adapter.h b/include/platform_specific/affinity_win32_adapter.h new file mode 100644 index 000000000..2285cff7a --- /dev/null +++ b/include/platform_specific/affinity_win32_adapter.h @@ -0,0 +1,121 @@ +#pragma once + +// This file provides an adaptation layer that lets the Celerity runtime use the Windows API for thread affinity through the POSIX-like +// cpu_set_t/sched_*affinity/pthread_*affinity_np interface expected by the rest of the runtime. cpu_set_t covers up to 1024 logical processors, +// spanning multiple Windows processor groups (up to 16 groups of 64 processors each). +// +// Note: Windows' GROUP_AFFINITY API can only bind a thread to a single processor group at a time. If a cpu_set_t's bits span more than one +// group, cpuset_to_group_affinity() cannot represent that as one affinity mask; it logs a warning and fails in that case. +// TODO: If we ever want to support pinning to multiple groups, we would need to find a way around this limitation. + +#include +#include +#include +#include + +#ifdef _WIN32 +// Prevent Windows headers from polluting global namespace with min/max macros. +#ifndef NOMINMAX +#define NOMINMAX +#endif +#ifndef WIN32_LEAN_AND_MEAN +#define WIN32_LEAN_AND_MEAN +#endif + +#include +#endif + +#include "log.h" + +using pthread_t = DWORD; + +struct cpu_set_t { + static constexpr unsigned CPU_SETSIZE = 1024; + static constexpr unsigned WORDS = (CPU_SETSIZE + 63) / 64; + + uint64_t bits[WORDS] = {}; +}; + +inline constexpr unsigned CPU_SETSIZE = cpu_set_t::CPU_SETSIZE; + +void CPU_ZERO(cpu_set_t* set); +void CPU_SET(unsigned cpu, cpu_set_t* set); +void CPU_CLR(unsigned cpu, cpu_set_t* set); +int CPU_ISSET(unsigned cpu, const cpu_set_t* set); + +int CPU_COUNT(const cpu_set_t* set); +int CPU_EQUAL(const cpu_set_t* a, const cpu_set_t* b); + +int sched_getaffinity(int pid, size_t cpusetsize, cpu_set_t* mask); +int sched_setaffinity(int pid, size_t cpusetsize, const cpu_set_t* mask); + +pthread_t pthread_self(); + +int pthread_setaffinity_np(pthread_t thread, size_t cpusetsize, const cpu_set_t* mask); +int pthread_getaffinity_np(pthread_t thread, size_t cpusetsize, cpu_set_t* mask); + +namespace win32_pthread_detail { + +static constexpr unsigned PROCS_PER_GROUP = 64; + +struct cpu_topology_entry { + WORD group; + WORD count; +}; + +using cpu_topology = std::vector; + +struct windows_topology_policy { + static WORD get_group_count() { return GetActiveProcessorGroupCount(); } + + static unsigned get_proc_count(WORD g) { return GetActiveProcessorCount(g); } +}; + +using topology_policy = windows_topology_policy; +template +cpu_topology get_cpu_topology() { + cpu_topology topo; + + const WORD groups = Policy::get_group_count(); + + for(WORD g = 0; g < groups; ++g) { + const unsigned count = Policy::get_proc_count(g); + topo.push_back({g, (WORD)count}); + } + + return topo; +} + +template +bool cpuset_to_group_affinity(const cpu_set_t* set, GROUP_AFFINITY& out) { + const WORD groups = Policy::get_group_count(); + + bool found = false; + + for(WORD g = 0; g < groups; ++g) { + const unsigned procs = Policy::get_proc_count(g); + + KAFFINITY mask = 0; + + for(unsigned p = 0; p < procs && p < PROCS_PER_GROUP; ++p) { + const unsigned global_cpu = g * PROCS_PER_GROUP + p; + + if(CPU_ISSET(global_cpu, set)) { mask |= (KAFFINITY(1) << p); } + } + + if(mask == 0) continue; + + if(found) { + CELERITY_WARN("Affinity mask spans multiple processor groups (not supported on Windows)."); + return false; + } + + found = true; + out.Group = g; + out.Mask = mask; + } + + return found; +} + +} // namespace win32_pthread_detail diff --git a/src/platform_specific/affinity.unix.cc b/src/platform_specific/affinity.unix.cc index 5240a10b3..a98f780b2 100644 --- a/src/platform_specific/affinity.unix.cc +++ b/src/platform_specific/affinity.unix.cc @@ -12,8 +12,11 @@ #include #include + +#if __has_include() && __has_include() #include #include +#endif #include "log.h" #include "named_threads.h" diff --git a/src/platform_specific/affinity.win.cc b/src/platform_specific/affinity.win.cc index 2593ceb17..7a63135df 100644 --- a/src/platform_specific/affinity.win.cc +++ b/src/platform_specific/affinity.win.cc @@ -1,15 +1,8 @@ -#include "affinity.h" - -#include "log.h" - -#include - - -namespace celerity::detail::thread_pinning { - -thread_pinner::thread_pinner(const runtime_configuration& cfg) { - if(cfg.enabled) { CELERITY_WARN("Thread pinning is currently not supported on Windows."); } -} -thread_pinner::~thread_pinner() {} - -} // namespace celerity::detail::thread_pinning +// The Windows thread-pinning logic is identical to the POSIX one in affinity.unix.cc. affinity_win32_adapter.h/.cc +// reimplement the POSIX cpu_set_t / pthread_t / CPU_*() / sched_*affinity() / pthread_*affinity_np() surface on top +// of the Win32 GROUP_AFFINITY API, so rather than duplicating affinity.unix.cc we just pull it in here: its own +// _WIN32 branch picks up the adapter header instead of /, and the shared logic below that +// compiles unchanged. +#include "platform_specific/affinity_win32_adapter.h" + +#include "affinity.unix.cc" diff --git a/src/platform_specific/affinity_win32_adapter.cc b/src/platform_specific/affinity_win32_adapter.cc new file mode 100644 index 000000000..5f288d52d --- /dev/null +++ b/src/platform_specific/affinity_win32_adapter.cc @@ -0,0 +1,110 @@ +#ifdef _WIN32 +#include "platform_specific/affinity_win32_adapter.h" + +#include +#include + +#include "log.h" + +void CPU_ZERO(cpu_set_t* set) { std::memset(set->bits, 0, sizeof(set->bits)); } + +void CPU_SET(unsigned cpu, cpu_set_t* set) { + if(cpu < cpu_set_t::CPU_SETSIZE) set->bits[cpu / 64] |= (uint64_t(1) << (cpu % 64)); +} + +void CPU_CLR(unsigned cpu, cpu_set_t* set) { + if(cpu < cpu_set_t::CPU_SETSIZE) set->bits[cpu / 64] &= ~(uint64_t(1) << (cpu % 64)); +} + +int CPU_ISSET(unsigned cpu, const cpu_set_t* set) { + if(cpu >= cpu_set_t::CPU_SETSIZE) return 0; + return (set->bits[cpu / 64] & (uint64_t(1) << (cpu % 64))) != 0; +} + +int CPU_COUNT(const cpu_set_t* set) { + int n = 0; + for(unsigned i = 0; i < cpu_set_t::WORDS; ++i) + n += std::popcount(set->bits[i]); + return n; +} + +int CPU_EQUAL(const cpu_set_t* a, const cpu_set_t* b) { return std::memcmp(a->bits, b->bits, sizeof(a->bits)) == 0; } + + +namespace win32_pthread_detail {} // namespace win32_pthread_detail + +int sched_getaffinity(int, size_t, cpu_set_t* mask) { + GROUP_AFFINITY ga{}; + + if(!GetThreadGroupAffinity(GetCurrentThread(), &ga)) return -1; + + CPU_ZERO(mask); + + const unsigned base = ga.Group * win32_pthread_detail::PROCS_PER_GROUP; + + for(unsigned p = 0; p < win32_pthread_detail::PROCS_PER_GROUP; ++p) { + if(ga.Mask & (KAFFINITY(1) << p)) CPU_SET(base + p, mask); + } + + return 0; +} + +int sched_setaffinity(int, size_t, const cpu_set_t* mask) { + GROUP_AFFINITY ga{}; + + if(!win32_pthread_detail::cpuset_to_group_affinity(mask, ga)) { + CELERITY_WARN("sched_setaffinity failed (multi-group mask rejected)"); + return -1; + } + + return SetThreadGroupAffinity(GetCurrentThread(), &ga, nullptr) ? 0 : -1; +} + +pthread_t pthread_self() { return GetCurrentThreadId(); } + +int pthread_setaffinity_np(pthread_t thread, size_t, const cpu_set_t* mask) { + const bool self = (thread == GetCurrentThreadId()); + + HANDLE h = self ? GetCurrentThread() : OpenThread(THREAD_SET_INFORMATION | THREAD_QUERY_INFORMATION, FALSE, thread); + + if(!h) return -1; + + GROUP_AFFINITY ga{}; + + if(!win32_pthread_detail::cpuset_to_group_affinity(mask, ga)) { + if(!self) CloseHandle(h); + return -1; + } + + int ok = SetThreadGroupAffinity(h, &ga, nullptr) ? 0 : -1; + + if(!self) CloseHandle(h); + return ok; +} + +int pthread_getaffinity_np(pthread_t thread, size_t, cpu_set_t* mask) { + const bool self = (thread == GetCurrentThreadId()); + + HANDLE h = self ? GetCurrentThread() : OpenThread(THREAD_QUERY_INFORMATION, FALSE, thread); + + if(!h) return -1; + + GROUP_AFFINITY ga{}; + + if(!GetThreadGroupAffinity(h, &ga)) { + if(!self) CloseHandle(h); + return -1; + } + + CPU_ZERO(mask); + + const unsigned base = ga.Group * win32_pthread_detail::PROCS_PER_GROUP; + + for(unsigned p = 0; p < win32_pthread_detail::PROCS_PER_GROUP; ++p) { + if(ga.Mask & (KAFFINITY(1) << p)) CPU_SET(base + p, mask); + } + + if(!self) CloseHandle(h); + return 0; +} +#endif diff --git a/test/affinity_tests.cc b/test/affinity_tests.cc index 5a357fb10..95457d5e2 100644 --- a/test/affinity_tests.cc +++ b/test/affinity_tests.cc @@ -62,12 +62,6 @@ class raii_test_runtime { raii_test_runtime& operator=(raii_test_runtime&&) = delete; }; -#ifdef _WIN32 -#define SKIP_UNSUPPORTED() SKIP("Affinity is not supported on Windows"); -#else -#define SKIP_UNSUPPORTED() -#endif - core_set get_current_cores() { cpu_set_t mask = {}; REQUIRE(sched_getaffinity(0, sizeof(cpu_set_t), &mask) == 0); @@ -138,7 +132,6 @@ TEST_CASE("thread pinning environment parsing error handling", "[affinity][confi } TEST_CASE("a warning is emitted if insufficient cores are available", "[affinity]") { - SKIP_UNSUPPORTED(); test_utils::allow_max_log_level(detail::log_level::warn); raii_affinity_masking mask({0, 1, 2}); @@ -155,7 +148,6 @@ TEST_CASE("a warning is emitted if insufficient cores are available", "[affinity } TEST_CASE("a warning is emitted if hardcoded threads are not available to this process", "[affinity]") { - SKIP_UNSUPPORTED(); test_utils::allow_max_log_level(detail::log_level::warn); raii_affinity_masking mask({0, 1, 2, 3, 4}); @@ -164,7 +156,6 @@ TEST_CASE("a warning is emitted if hardcoded threads are not available to this p } TEST_CASE("do not plan for device submission threads if they are unused", "[affinity]") { - SKIP_UNSUPPORTED(); raii_affinity_masking mask({1, 2, 3, 4}); const detail::thread_pinning::runtime_configuration cfg = {.enabled = true, .num_devices = 10, .use_backend_device_submission_threads = false}; detail::thread_pinning::thread_pinner pinner(cfg); @@ -172,7 +163,6 @@ TEST_CASE("do not plan for device submission threads if they are unused", "[affi } TEST_CASE_METHOD(test_utils::runtime_fixture, "runtime warns on manual core list of wrong size", "[affinity][config]") { - SKIP_UNSUPPORTED(); test_utils::allow_max_log_level(detail::log_level::warn); env::scoped_test_environment ste("CELERITY_THREAD_PINNING", "1,2"); { raii_test_runtime rt(1); } @@ -182,7 +172,6 @@ TEST_CASE_METHOD(test_utils::runtime_fixture, "runtime warns on manual core list // SimSYCL has no backend submission thread support #if !CELERITY_SYCL_IS_SIMSYCL TEST_CASE_METHOD(test_utils::runtime_fixture, "runtime system claims to pin its threads as desired", "[affinity][runtime]") { - SKIP_UNSUPPORTED(); auto test = [](const std::vector& core_ids) { if(!have_cores({core_ids.cbegin(), core_ids.cend()})) { SKIP("Skipping test because not all needed cores are available"); @@ -218,7 +207,6 @@ TEST_CASE_METHOD(test_utils::runtime_fixture, "runtime system claims to pin its #endif // !CELERITY_SYCL_IS_SIMSYCL TEST_CASE_METHOD(test_utils::runtime_fixture, "when pinning disabled: rt warns on insufficient threads and does not pin", "[affinity][runtime]") { - SKIP_UNSUPPORTED(); env::scoped_test_environment ste("CELERITY_THREAD_PINNING", "false"); raii_affinity_masking mask({0}); @@ -230,7 +218,6 @@ TEST_CASE_METHOD(test_utils::runtime_fixture, "when pinning disabled: rt warns o } TEST_CASE_METHOD(test_utils::runtime_fixture, "the application thread is actually pinned when pinning is enabled", "[affinity][runtime]") { - SKIP_UNSUPPORTED(); env::scoped_test_environment ste("CELERITY_THREAD_PINNING", "auto"); // By using a custom mask we also validate that the mechanism which only selects cores that are available to the process works correctly @@ -254,7 +241,6 @@ TEST_CASE_METHOD(test_utils::runtime_fixture, "the application thread is actuall } TEST_CASE_METHOD(test_utils::runtime_fixture, "when pinning is disabled, no threads are pinned", "[affinity][runtime]") { - SKIP_UNSUPPORTED(); env::scoped_test_environment ste("CELERITY_THREAD_PINNING", "false"); const auto initial_core_set = get_current_cores(); @@ -268,7 +254,6 @@ TEST_CASE_METHOD(test_utils::runtime_fixture, "when pinning is disabled, no thre } TEST_CASE_METHOD(test_utils::runtime_fixture, "rt warns when the application thread changes pinning unexpectedly", "[affinity][runtime]") { - SKIP_UNSUPPORTED(); test_utils::allow_max_log_level(detail::log_level::warn); env::scoped_test_environment ste("CELERITY_THREAD_PINNING", "auto"); @@ -295,8 +280,6 @@ TEST_CASE_METHOD(test_utils::runtime_fixture, "rt warns when the application thr } TEST_CASE("multiple subsequent non-overlapping pinner lifetimes are handled correctly", "[affinity]") { - SKIP_UNSUPPORTED(); - const core_set process_mask = {3, 4, 5, 6, 7}; if(!have_cores(process_mask)) { SKIP("Skipping test because not all needed cores are available"); @@ -325,7 +308,6 @@ TEST_CASE("multiple subsequent non-overlapping pinner lifetimes are handled corr } TEST_CASE("trying to initialize two pinning mechanisms with overlapping lifetime is an error", "[affinity]") { - SKIP_UNSUPPORTED(); test_utils::allow_max_log_level(detail::log_level::err); detail::thread_pinning::thread_pinner pinner({.enabled = true, .num_devices = 1}); detail::thread_pinning::thread_pinner another_pinner({.enabled = true, .num_devices = 1}); @@ -333,7 +315,6 @@ TEST_CASE("trying to initialize two pinning mechanisms with overlapping lifetime } TEST_CASE("application threads are not pinned if their affinity mask is modified externally", "[affinity]") { - SKIP_UNSUPPORTED(); test_utils::allow_max_log_level(detail::log_level::warn); const core_set process_mask = {0, 1, 2, 3}; if(!have_cores(process_mask)) { @@ -355,3 +336,106 @@ TEST_CASE("application threads are not pinned if their affinity mask is modified CHECK(get_current_cores() == process_mask); CHECK(test_utils::log_contains_substring(detail::log_level::warn, "Affinity mask for the application thread was modified, will not pin it.")); } + + +#ifdef _WIN32 +struct test_topology_policy { + struct group { + unsigned count; + }; + + static inline const std::vector* groups = nullptr; + + static WORD get_group_count() { return static_cast(groups ? groups->size() : 1); } + + static unsigned get_proc_count(WORD g) { return groups ? (*groups)[g].count : 0; } +}; + +struct scoped_topology { + std::vector m_groups; + + scoped_topology(std::initializer_list entries) { + for(const auto& e : entries) { + m_groups.push_back({e.count}); + } + test_topology_policy::groups = &m_groups; + } + ~scoped_topology() { test_topology_policy::groups = nullptr; } + scoped_topology(const scoped_topology&) = delete; + scoped_topology(scoped_topology&&) = delete; + scoped_topology& operator=(const scoped_topology&) = delete; + scoped_topology& operator=(scoped_topology&&) = delete; +}; + + +TEST_CASE("single group cpuset produces correct mask") { + test_utils::allow_max_log_level(detail::log_level::warn); + + scoped_topology topo({{0, 64}}); + + cpu_set_t set{}; + CPU_ZERO(&set); + CPU_SET(0, &set); + CPU_SET(1, &set); + + GROUP_AFFINITY ga{}; + REQUIRE(win32_pthread_detail::cpuset_to_group_affinity(&set, ga)); + + CHECK(ga.Group == 0); + CHECK(ga.Mask == 0b11); +} + +TEST_CASE("multi-group cpuset triggers warning") { + test_utils::allow_max_log_level(detail::log_level::warn); + + scoped_topology topo({{0, 64}, {1, 64}}); + + cpu_set_t set{}; + CPU_ZERO(&set); + CPU_SET(0, &set); // group 0 + CPU_SET(64, &set); // group 1 + + GROUP_AFFINITY ga{}; + CHECK_FALSE(win32_pthread_detail::cpuset_to_group_affinity(&set, ga)); + CHECK(test_utils::log_contains_substring(detail::log_level::warn, "Affinity mask spans multiple processor groups")); +} + +TEST_CASE("single cpu produces single-bit mask") { + scoped_topology topo({{0, 64}, {1, 64}}); + + cpu_set_t set{}; + CPU_ZERO(&set); + CPU_SET(0, &set); + + GROUP_AFFINITY ga{}; + REQUIRE(win32_pthread_detail::cpuset_to_group_affinity(&set, ga)); + + CHECK(ga.Group == 0); + CHECK(ga.Mask == 1); +} + +TEST_CASE("empty cpuset is rejected") { + scoped_topology topo({{0, 64}}); + + cpu_set_t set{}; + CPU_ZERO(&set); + + GROUP_AFFINITY ga{}; + CHECK_FALSE(win32_pthread_detail::cpuset_to_group_affinity(&set, ga)); +} + +TEST_CASE("cpu in second group maps to correct group and bit") { + scoped_topology topo({{0, 64}, {1, 64}}); + + cpu_set_t set{}; + CPU_ZERO(&set); + CPU_SET(64, &set); // first CPU of group 1 + + GROUP_AFFINITY ga{}; + REQUIRE(win32_pthread_detail::cpuset_to_group_affinity(&set, ga)); + + CHECK(ga.Group == 1); + CHECK(ga.Mask == 1); // bit 0 within the group +} + +#endif // _WIN32