From 3c5ff3f5dffef208fcc55ebc0c98e3a904d8c452 Mon Sep 17 00:00:00 2001 From: Filip Leonarski Date: Sun, 11 Oct 2026 09:27:32 +0200 Subject: [PATCH] rugnux: on a machine of several sockets, keep to the sockets the GPUs are on With -N left to choose, rugnux now confines the process (main thread, the CPUs RestoreThreadAffinity hands back, and what AvailableCpus counts) to the CPUs of the packages its cards hang off, and sizes its threads to them. Nothing changes where the allowed CPUs sit in one package, there is no card or a card's NUMA node is unknown, or the cards' packages already hold every allowed CPU. A mask set from outside (taskset, numactl, a cpuset) is kept: the decision works within it. An explicit -N keeps the old behaviour. On a 48-CPU server with four cards on two of its four NUMA nodes, a rotation run took 24.5-24.7 s confined to those nodes' 24 CPUs against 27.7-27.9 s on all 48 (scaling and merging 8.1 -> 6.2 s), and two larger sets moved 95-99 -> 84-87 s and 131-135 -> 113-115 s. On a single-socket machine this is a no-op. The decision itself (CpusOfGpuPackages) is a pure function, tested on synthetic 4-package, 2-package hyper-threaded, cards-on-both, no-card and restricted-mask topologies. Co-Authored-By: Claude Opus 5.5 (1M context) Claude-Session: https://claude.ai/code/session_01EizBKhTqrAYago9KJG3dA3 --- common/CUDAWrapper.cpp | 2 + common/CUDAWrapper.cu | 7 ++++ common/CUDAWrapper.h | 4 ++ common/ThreadAffinity.cpp | 86 ++++++++++++++++++++++++++++++++++----- common/ThreadAffinity.h | 17 ++++++++ rugnux/rugnux_cli.cpp | 21 +++++++++- tests/ParallelForTest.cpp | 65 +++++++++++++++++++++++++++++ 7 files changed, 190 insertions(+), 12 deletions(-) diff --git a/common/CUDAWrapper.cpp b/common/CUDAWrapper.cpp index dc4d31102..5eebbd99e 100644 --- a/common/CUDAWrapper.cpp +++ b/common/CUDAWrapper.cpp @@ -77,6 +77,8 @@ ScopedGpuPin::~ScopedGpuPin() {} void enable_gpu_numa_binding() {} +std::vector get_gpu_numa_nodes() { return {}; } + void set_gpu_blocking_sync() {} void create_gpu_contexts(int32_t) {} diff --git a/common/CUDAWrapper.cu b/common/CUDAWrapper.cu index 434b7b5ee..f12c984a6 100644 --- a/common/CUDAWrapper.cu +++ b/common/CUDAWrapper.cu @@ -89,6 +89,13 @@ namespace { } } +std::vector get_gpu_numa_nodes() { + std::vector nodes; + for (int32_t i = 0; i < get_gpu_count(); i++) + nodes.push_back(GpuNumaNode(i)); + return nodes; +} + void enable_gpu_numa_binding() { gpu_numa_binding = true; } diff --git a/common/CUDAWrapper.h b/common/CUDAWrapper.h index 83e02bca8..7941dbd60 100644 --- a/common/CUDAWrapper.h +++ b/common/CUDAWrapper.h @@ -35,6 +35,10 @@ void pin_gpu(); // Off unless a program asks for it. void enable_gpu_numa_binding(); +// The NUMA node each visible GPU is attached to, in device order; -1 where it is not known or the +// machine has a single node. Empty without CUDA or a device. +std::vector get_gpu_numa_nodes(); + // pin_gpu() onto a given device: set_gpu(dev_id), and the NUMA binding above where it is enabled. // For a worker that takes its card by index rather than round-robin. void pin_gpu(int32_t dev_id); diff --git a/common/ThreadAffinity.cpp b/common/ThreadAffinity.cpp index 8e59d117d..461a08b49 100644 --- a/common/ThreadAffinity.cpp +++ b/common/ThreadAffinity.cpp @@ -3,9 +3,31 @@ #include "ThreadAffinity.h" +#include +#include + +std::vector CpusOfGpuPackages(const std::vector &allowed, const std::vector &package_of_cpu, + const std::vector &gpu_packages) { + if (gpu_packages.empty() + || std::find(gpu_packages.begin(), gpu_packages.end(), -1) != gpu_packages.end()) + return {}; + const auto package = [&](int c) { + return c < static_cast(package_of_cpu.size()) ? package_of_cpu[c] : -1; + }; + std::set packages; + std::vector out; + for (const int c: allowed) { + packages.insert(package(c)); + if (std::find(gpu_packages.begin(), gpu_packages.end(), package(c)) != gpu_packages.end()) + out.push_back(c); + } + if (packages.size() < 2 || out.empty() || out.size() == allowed.size()) + return {}; + return out; +} + #ifdef __linux__ -#include #include #include #include @@ -26,7 +48,8 @@ namespace { CPU_SET(c, &mask); return mask; } - const cpu_set_t start_mask = StartMask(); + // Narrowed once by ConfineToGpuPackages, before the process starts any thread. + cpu_set_t start_mask = StartMask(); // The CPUs the cgroup's quota allows, rounded up; 0 where none is set. The process's own cgroup // first, then the root of the hierarchy as mounted, which is the container's own cgroup where the @@ -99,6 +122,33 @@ namespace { } return any; } + + int ReadInt(const std::string &path) { + std::ifstream f(path); + int v = -1; + if (!(f >> v)) + return -1; + return v; + } + + int PackageOfCpu(int cpu) { + return ReadInt("/sys/devices/system/cpu/cpu" + std::to_string(cpu) + "/topology/physical_package_id"); + } + + // The package of a NUMA node's first CPU; -1 for a node without CPUs or an unknown node. + int PackageOfNode(int node) { + if (node < 0) + return -1; + std::ifstream f("/sys/devices/system/node/node" + std::to_string(node) + "/cpulist"); + std::string list; + cpu_set_t cpus; + if (!std::getline(f, list) || !ParseCpuList(list, cpus)) + return -1; + for (int c = 0; c < CPU_SETSIZE; c++) + if (CPU_ISSET(c, &cpus)) + return PackageOfCpu(c); + return -1; + } } int NumaNodeOfPciDevice(const std::string &pci_bus_id) { @@ -118,13 +168,29 @@ int NumaNodeOfPciDevice(const std::string &pci_bus_id) { } unsigned AvailableCpus() { - static const unsigned n = [] { - unsigned cpus = static_cast(std::max(1, CPU_COUNT(&start_mask))); - if (const unsigned quota = CgroupCpuQuota(); quota > 0) - cpus = std::min(cpus, quota); - return cpus; - }(); - return n; + static const unsigned quota = CgroupCpuQuota(); + const unsigned cpus = static_cast(std::max(1, CPU_COUNT(&start_mask))); + return quota > 0 ? std::min(cpus, quota) : cpus; +} + +std::vector ConfineToGpuPackages(const std::vector &gpu_nodes) { + std::vector allowed, package_of_cpu, gpu_packages; + for (int c = 0; c < CPU_SETSIZE; c++) + if (CPU_ISSET(c, &start_mask)) { + allowed.push_back(c); + package_of_cpu.resize(c + 1, -1); + package_of_cpu[c] = PackageOfCpu(c); + } + for (const int node: gpu_nodes) + gpu_packages.push_back(PackageOfNode(node)); + const std::vector cpus = CpusOfGpuPackages(allowed, package_of_cpu, gpu_packages); + if (cpus.empty()) + return {}; + CPU_ZERO(&start_mask); + for (const int c: cpus) + CPU_SET(c, &start_mask); + RestoreThreadAffinity(); + return cpus; } void PinThreadToNumaNode(int node) { @@ -168,9 +234,9 @@ void SetThreadCpus(const std::vector &cpus) { #else -#include #include +std::vector ConfineToGpuPackages(const std::vector &) { return {}; } unsigned AvailableCpus() { return std::max(1u, std::thread::hardware_concurrency()); } int NumaNodeOfPciDevice(const std::string &) { return -1; } void PinThreadToNumaNode(int) {} diff --git a/common/ThreadAffinity.h b/common/ThreadAffinity.h index aa7a02a8f..a2cc6fc0d 100644 --- a/common/ThreadAffinity.h +++ b/common/ThreadAffinity.h @@ -22,6 +22,23 @@ int NumaNodeOfPciDevice(const std::string &pci_bus_id); // first call. Outside Linux it is hardware_concurrency(). At least 1. unsigned AvailableCpus(); +// The CPUs a process driving GPUs keeps to on a machine of several packages (sockets): those of the +// packages its cards hang off. `allowed` is what the process may use, `package_of_cpu[c]` the package +// of CPU c (-1 unknown) and `gpu_packages` the package of each card (-1 unknown). Empty where nothing +// changes: the allowed CPUs sit in one package, there is no card or one card's package is unknown, the +// cards' packages hold every allowed CPU, or none of them. +std::vector CpusOfGpuPackages(const std::vector &allowed, const std::vector &package_of_cpu, + const std::vector &gpu_packages); + +// Keep the process to CpusOfGpuPackages for cards on the NUMA nodes `gpu_nodes` (-1 unknown): the +// calling thread, the CPUs RestoreThreadAffinity gives back and what AvailableCpus counts all become +// those, within the CPUs it was started on. Threads, and the memory they first touch, spread over +// sockets the cards are not on make the latency-bound scaling slower: on a 48-CPU server with its cards +// on two of four NUMA nodes, a rotation run kept to those two nodes' 24 CPUs took 12 % less wall time +// than on all 48, and the 12 CPUs of one node without a card still beat all 48. Call it on the main thread before any thread +// is started. Returns the CPUs it keeps to; empty where nothing changed. Linux only. +std::vector ConfineToGpuPackages(const std::vector &gpu_nodes); + // Restrict the calling thread to the CPUs of `node` that the process may run on. Nothing happens for // node < 0, or where that would leave the thread no CPU. void PinThreadToNumaNode(int node); diff --git a/rugnux/rugnux_cli.cpp b/rugnux/rugnux_cli.cpp index 10f78955c..cbfb502f0 100644 --- a/rugnux/rugnux_cli.cpp +++ b/rugnux/rugnux_cli.cpp @@ -1551,9 +1551,14 @@ static int RunRugnux(int argc, char **argv) { // behaves the same. The scale/merge engines expand 0 on their own, but the per-image processing // loop (Rugnux) spawns exactly nthreads workers, so passing 0 there would spawn none and process // nothing - hence resolving it centrally rather than relying on each consumer. + // Left to choose, a run on a machine of several sockets also keeps to the sockets its cards are on, + // and takes their CPUs (ConfineToGpuPackages). const bool nthreads_auto = nthreads <= 0; - if (nthreads <= 0) + std::vector confined_cpus; + if (nthreads_auto) { + confined_cpus = ConfineToGpuPackages(get_gpu_numa_nodes()); nthreads = static_cast(AvailableCpus()); + } // The log file, next to the outputs, and everything logged so far into it. std::string &log_file = g_log_file; @@ -1570,7 +1575,19 @@ static int RunRugnux(int argc, char **argv) { } } log_sinks->remove_sink(early_log); - logger.Info("Threads: {}{}", nthreads, nthreads_auto ? " (the CPUs this run may use)" : " (-N)"); + if (!confined_cpus.empty()) { + std::string list; // "24-35,72-83" + for (size_t i = 0; i < confined_cpus.size();) { + size_t j = i; + while (j + 1 < confined_cpus.size() && confined_cpus[j + 1] == confined_cpus[j] + 1) + j++; + list += (list.empty() ? "" : ",") + std::to_string(confined_cpus[i]) + + (j > i ? "-" + std::to_string(confined_cpus[j]) : ""); + i = j + 1; + } + logger.Info("Threads: {} (the CPUs of the GPUs' sockets: {})", nthreads, list); + } else + logger.Info("Threads: {}{}", nthreads, nthreads_auto ? " (the CPUs this run may use)" : " (-N)"); console->Field("Input", input_file); console->Field("Output", log_file.empty() ? output_prefix + "_*" diff --git a/tests/ParallelForTest.cpp b/tests/ParallelForTest.cpp index 750cf5d99..21328614a 100644 --- a/tests/ParallelForTest.cpp +++ b/tests/ParallelForTest.cpp @@ -212,3 +212,68 @@ TEST_CASE("ScopedGpuPin_GivesTheThreadItsCpusBack", "[ParallelFor]") { CHECK(seen_by_task[t] == all); } #endif + +namespace { + std::vector CpuRange(int lo, int hi) { // [lo, hi] + std::vector v(hi - lo + 1); + std::iota(v.begin(), v.end(), lo); + return v; + } + + std::vector Concat(std::vector a, const std::vector &b) { + a.insert(a.end(), b.begin(), b.end()); + return a; + } +} + +// Synthetic machines: which CPUs a GPU run keeps to, given the CPUs it may use, the package of every +// CPU and the package of every card. +TEST_CASE("CpusOfGpuPackages_KeepsToTheCardsPackages", "[ParallelFor]") { + // Four packages of 12 CPUs, four cards on packages 2 and 3. + std::vector four(48); + for (int c = 0; c < 48; c++) + four[c] = c / 12; + CHECK(CpusOfGpuPackages(CpuRange(0, 47), four, {2, 2, 3, 3}) == CpuRange(24, 47)); + + // Two packages of 12 cores with hyper-threading, numbered as Linux does: CPUs 0-11 and 24-35 on + // package 0, 12-23 and 36-47 on package 1. + std::vector two(48); + for (int c = 0; c < 48; c++) + two[c] = (c / 12) % 2; + const std::vector package1 = Concat(CpuRange(12, 23), CpuRange(36, 47)); + CHECK(CpusOfGpuPackages(CpuRange(0, 47), two, {1}) == package1); + CHECK(CpusOfGpuPackages(CpuRange(0, 47), two, {1, 1}) == package1); + + // Cards on both packages: nothing to keep to. + CHECK(CpusOfGpuPackages(CpuRange(0, 47), two, {0, 1}).empty()); + + // No card (a CPU build, or none visible), or a card whose package is not known. + CHECK(CpusOfGpuPackages(CpuRange(0, 47), two, {}).empty()); + CHECK(CpusOfGpuPackages(CpuRange(0, 47), two, {1, -1}).empty()); + + // One package, as on a desktop. + CHECK(CpusOfGpuPackages(CpuRange(0, 31), std::vector(32, 0), {0}).empty()); +} + +// A process started on fewer CPUs (taskset, numactl, a cpuset) keeps within them. +TEST_CASE("CpusOfGpuPackages_RespectsARestrictedMask", "[ParallelFor]") { + std::vector four(48); + for (int c = 0; c < 48; c++) + four[c] = c / 12; + // Already on one package - with the cards or without them - stays where it is. + CHECK(CpusOfGpuPackages(CpuRange(0, 11), four, {2, 3}).empty()); + CHECK(CpusOfGpuPackages(CpuRange(24, 35), four, {2, 3}).empty()); + // Already on the cards' packages only. + CHECK(CpusOfGpuPackages(CpuRange(24, 47), four, {2, 3}).empty()); + // Over packages 1 and 2, cards on 2 and 3: the part on package 2. + CHECK(CpusOfGpuPackages(CpuRange(18, 29), four, {2, 3}) == CpuRange(24, 29)); +} + +// Without a card nothing is confined, so the process keeps the CPUs it started with. +TEST_CASE("ConfineToGpuPackages_DoesNothingWithoutACard", "[ParallelFor]") { + const unsigned before = AvailableCpus(); + const std::vector cpus = GetThreadCpus(); + CHECK(ConfineToGpuPackages({}).empty()); + CHECK(AvailableCpus() == before); + CHECK(GetThreadCpus() == cpus); +}