Add Windows thread affinity support

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:

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)

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.

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 <bit>
+#include <cstdint>
+#include <cstring>
+#include <vector>
+
+#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 <windows.h>
+#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<cpu_topology_entry>;
+
+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 <typename Policy = topology_policy>
+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 <typename Policy = windows_topology_policy>
+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

src/platform_specific/affinity.unix.cc

@@ -12,8 +12,11 @@
 
 #include <fmt/format.h>
 #include <fmt/ranges.h>
+
+#if __has_include(<pthread.h>) && __has_include(<sched.h>)
 #include <pthread.h>
 #include <sched.h>
+#endif
 
 #include "log.h"
 #include "named_threads.h"

src/platform_specific/affinity.win.cc

@@ -1,15 +1,8 @@
-#include "affinity.h"
-
-#include "log.h"
-
-#include <cassert>
-
-
-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 <pthread.h>/<sched.h>, and the shared logic below that
+// compiles unchanged.
+#include "platform_specific/affinity_win32_adapter.h"
+
+#include "affinity.unix.cc"

src/platform_specific/affinity_win32_adapter.cc

@@ -0,0 +1,110 @@
+#ifdef _WIN32
+#include "platform_specific/affinity_win32_adapter.h"
+
+#include <algorithm>
+#include <numeric>
+
+#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