diff --git a/common/AzimuthalIntegrationMapping.cpp b/common/AzimuthalIntegrationMapping.cpp index efebb322b..f33ba2ca6 100644 --- a/common/AzimuthalIntegrationMapping.cpp +++ b/common/AzimuthalIntegrationMapping.cpp @@ -53,7 +53,7 @@ AzimuthalIntegrationMapping::AzimuthalIntegrationMapping(const DiffractionExperi polarization_factor = experiment.GetPolarizationFactor(); if (in_nthreads == 0) - nthreads = std::thread::hardware_concurrency(); + nthreads = AvailableCpus(); else nthreads = in_nthreads; diff --git a/common/ImageBuffer.cpp b/common/ImageBuffer.cpp index 5b2809b3e..fff84beb2 100644 --- a/common/ImageBuffer.cpp +++ b/common/ImageBuffer.cpp @@ -4,6 +4,7 @@ #include "ImageBuffer.h" #include "JFJochException.h" +#include "ThreadAffinity.h" #include "ZeroCopyReturnValue.h" #include @@ -19,7 +20,7 @@ namespace { // faulting in a 150-200 GB allocation. Replaces numa_alloc_interleaved + a single-threaded // memset, so this file no longer needs libnuma. void parallel_first_touch(uint8_t *buffer, size_t buffer_size) { - const unsigned n = std::max(1u, std::thread::hardware_concurrency()); + const unsigned n = AvailableCpus(); const size_t chunk = (buffer_size + n - 1) / n; std::vector threads; diff --git a/common/ParallelFor.h b/common/ParallelFor.h index 276cbd35a..65b9e262c 100644 --- a/common/ParallelFor.h +++ b/common/ParallelFor.h @@ -58,7 +58,7 @@ namespace parallel_detail { private: WorkerPool() { - const unsigned hw = std::max(1u, std::thread::hardware_concurrency()); + const unsigned hw = AvailableCpus(); workers.reserve(hw - 1); try { for (unsigned i = 0; i + 1 < hw; i++) // the submitting thread takes a share too diff --git a/common/ThreadAffinity.cpp b/common/ThreadAffinity.cpp index 9a85305da..cd4d9d52c 100644 --- a/common/ThreadAffinity.cpp +++ b/common/ThreadAffinity.cpp @@ -5,6 +5,8 @@ #ifdef __linux__ +#include +#include #include #include #include @@ -26,6 +28,45 @@ namespace { } const 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 + // container does not see the host's paths. cgroup v2 (cpu.max: "max 100000" or "200000 100000") + // and v1 (cpu.cfs_quota_us = -1 or the quota, cpu.cfs_period_us). + unsigned CgroupCpuQuota() { + std::ifstream f("/proc/self/cgroup"); + std::string line; + while (std::getline(f, line)) { + const size_t c1 = line.find(':'), c2 = line.find(':', c1 + 1); + if (c1 == std::string::npos || c2 == std::string::npos) + continue; + const std::string controllers = "," + line.substr(c1 + 1, c2 - c1 - 1) + ","; + const std::string path = line.substr(c2 + 1); + const bool v2 = line.compare(0, c2 + 1, "0::") == 0; + if (!v2 && controllers.find(",cpu,") == std::string::npos) + continue; + for (const std::string &dir : {"/sys/fs/cgroup" + std::string(v2 ? "" : "/cpu") + path, + "/sys/fs/cgroup" + std::string(v2 ? "" : "/cpu")}) { + double quota = -1, period = 0; + if (v2) { + std::ifstream q(dir + "/cpu.max"); + std::string max; + if (!(q >> max >> period)) + continue; + if (max != "max") + quota = std::atof(max.c_str()); + } else { + std::ifstream q(dir + "/cpu.cfs_quota_us"), p(dir + "/cpu.cfs_period_us"); + if (!(q >> quota) || !(p >> period)) + continue; + } + if (quota > 0 && period > 0) + return static_cast(std::ceil(quota / period)); + break; // found, and no quota set + } + } + return 0; + } + int NumaNodeCount() { int n = 0; std::error_code ec; @@ -76,6 +117,16 @@ int NumaNodeOfPciDevice(const std::string &pci_bus_id) { return node; } +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; +} + void PinThreadToNumaNode(int node) { if (node < 0) return; @@ -97,6 +148,10 @@ void RestoreThreadAffinity() { #else +#include +#include + +unsigned AvailableCpus() { return std::max(1u, std::thread::hardware_concurrency()); } int NumaNodeOfPciDevice(const std::string &) { return -1; } void PinThreadToNumaNode(int) {} void RestoreThreadAffinity() {} diff --git a/common/ThreadAffinity.h b/common/ThreadAffinity.h index 07a7e2c1c..3af3232a4 100644 --- a/common/ThreadAffinity.h +++ b/common/ThreadAffinity.h @@ -14,6 +14,13 @@ // not known or the machine has a single node. int NumaNodeOfPciDevice(const std::string &pci_bus_id); +// How many CPUs this process may use: the CPUs it was started on (its affinity mask - what taskset, +// a cpuset or a Slurm job gives it), and no more than its cgroup's CPU quota where one is set (a +// container started with --cpus). std::thread::hardware_concurrency() counts every CPU of the +// machine, so a job given 16 of a node's 192 would otherwise start 192 threads. Read once, at the +// first call. Outside Linux it is hardware_concurrency(). At least 1. +unsigned AvailableCpus(); + // 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/image_analysis/beam_stop/ShadowFinder.cpp b/image_analysis/beam_stop/ShadowFinder.cpp index 3d62d6d6d..879d3e6ec 100644 --- a/image_analysis/beam_stop/ShadowFinder.cpp +++ b/image_analysis/beam_stop/ShadowFinder.cpp @@ -442,7 +442,7 @@ ShadowFinder::Projection ShadowFinder::Reduce() const { // A pixel is touched by one worker only and the sums and counts are integers, so the result is // the same as folding on one thread. out.frames += host.frames; - ParallelChunks(static_cast(out.max_value.size()), std::thread::hardware_concurrency(), [&](int lo, int hi) { + ParallelChunks(static_cast(out.max_value.size()), AvailableCpus(), [&](int lo, int hi) { for (int i = lo; i < hi; i++) { if (host.valid_count[i] == 0) continue; @@ -590,7 +590,7 @@ std::vector ShadowFinder::GetMeanProjection() const { const auto &valid_count = p.valid_count; std::vector mean(static_cast(width) * height); - ParallelChunks(static_cast(mean.size()), std::thread::hardware_concurrency(), [&](int lo, int hi) { + ParallelChunks(static_cast(mean.size()), AvailableCpus(), [&](int lo, int hi) { for (int i = lo; i < hi; i++) mean[i] = valid_count[i] > 0 && pixel_mask[i] == 0 ? static_cast(static_cast(sum_value[i]) / valid_count[i]) : NAN; @@ -601,7 +601,7 @@ std::vector ShadowFinder::GetMeanProjection() const { std::vector ShadowFinder::GetMask(size_t nthreads) const { std::unique_lock ul(m); if (nthreads == 0) - nthreads = std::max(1u, std::thread::hardware_concurrency()); + nthreads = AvailableCpus(); #ifdef JFJOCH_USE_CUDA // Where every frame went to the device the mask is made there too, from the projection as it lies. if (Gpu() && gpu->GetFrameCount() > 0 && host.frames == 0) { diff --git a/image_analysis/geom_refinement/BeamCenterFromBackground.cpp b/image_analysis/geom_refinement/BeamCenterFromBackground.cpp index 0b99a030c..3f7512114 100644 --- a/image_analysis/geom_refinement/BeamCenterFromBackground.cpp +++ b/image_analysis/geom_refinement/BeamCenterFromBackground.cpp @@ -91,7 +91,7 @@ FindBeamCenterFromBackground(const DiffractionExperiment &experiment, const Pixe const std::vector &mean, size_t nthreads, std::optional> start, bool allow_device) { if (nthreads == 0) - nthreads = std::max(1u, std::thread::hardware_concurrency()); + nthreads = AvailableCpus(); const auto W = static_cast(experiment.GetXPixelsNumConv()); const auto H = static_cast(experiment.GetYPixelsNumConv()); const size_t n_pixels = static_cast(W) * H; diff --git a/image_analysis/scale_merge/RotationScaleMerge.cpp b/image_analysis/scale_merge/RotationScaleMerge.cpp index bc164861c..d48061fb9 100644 --- a/image_analysis/scale_merge/RotationScaleMerge.cpp +++ b/image_analysis/scale_merge/RotationScaleMerge.cpp @@ -299,7 +299,7 @@ RotationScaleMerge::RotationScaleMerge(const DiffractionExperiment &experiment, size_t nthreads, Logger &logger, std::string observation_dump_path) : x(experiment), partials_out(partial_outcomes), reference_cell(std::move(reference_cell)), - nthreads(nthreads == 0 ? std::thread::hardware_concurrency() : nthreads), logger(&logger), + nthreads(nthreads == 0 ? AvailableCpus() : nthreads), logger(&logger), observation_dump_path(std::move(observation_dump_path)) { const auto s = x.GetScalingSettings(); min_partiality = s.GetMinPartiality(); diff --git a/image_analysis/scale_merge/ScaleOnTheFly.cpp b/image_analysis/scale_merge/ScaleOnTheFly.cpp index 6d4ccf030..823b42706 100644 --- a/image_analysis/scale_merge/ScaleOnTheFly.cpp +++ b/image_analysis/scale_merge/ScaleOnTheFly.cpp @@ -4,6 +4,7 @@ #include "ScaleOnTheFly.h" #include "../../common/Logger.h" +#include "../../common/ThreadAffinity.h" #include #include @@ -195,7 +196,7 @@ void ScaleOnTheFly::RejectCollapsedScales(std::vector &integ void ScaleOnTheFly::Scale(std::vector &integration, size_t nthreads) const { if (nthreads == 0) - nthreads = std::thread::hardware_concurrency(); + nthreads = AvailableCpus(); if (nthreads <= 1) { for (auto & i : integration) diff --git a/image_analysis/scale_merge/StillsPartialityRefine.cpp b/image_analysis/scale_merge/StillsPartialityRefine.cpp index acdf8e81c..44910f6cb 100644 --- a/image_analysis/scale_merge/StillsPartialityRefine.cpp +++ b/image_analysis/scale_merge/StillsPartialityRefine.cpp @@ -13,6 +13,7 @@ #include #include "Merge.h" +#include "../../common/ThreadAffinity.h" namespace { constexpr size_t MIN_FIT_REFLECTIONS = 20; @@ -350,7 +351,7 @@ double StillsPartialityRefine::RefineOne(IntegrationOutcome &outcome, double StillsPartialityRefine::Run(std::vector &outcomes, size_t nthreads) const { if (nthreads == 0) - nthreads = std::thread::hardware_concurrency(); + nthreads = AvailableCpus(); nthreads = std::max(1, nthreads); double last_mean_tilt = 0.0; diff --git a/preview/PreviewImage.cpp b/preview/PreviewImage.cpp index b7f941727..b53b94cef 100644 --- a/preview/PreviewImage.cpp +++ b/preview/PreviewImage.cpp @@ -13,6 +13,7 @@ #include "JFJochTIFF.h" #include "../common/JFJochException.h" #include "../common/JFJochMath.h" +#include "../common/ThreadAffinity.h" #include "../common/DiffractionGeometry.h" #include "../frame_serialize/CBORStream2Deserializer.h" #include "../compression/JFJochDecompress.h" @@ -365,7 +366,7 @@ void PreviewImage::Configure(const DiffractionExperiment &in_experiment, const P mask.resize(experiment.GetPixelsNum(), 0); if (nthreads == 0) - nthreads = std::thread::hardware_concurrency(); + nthreads = AvailableCpus(); nthreads = std::clamp(nthreads, 1, 8); diff --git a/reader/SweepLayout.cpp b/reader/SweepLayout.cpp index d43609df9..343fbd171 100644 --- a/reader/SweepLayout.cpp +++ b/reader/SweepLayout.cpp @@ -14,6 +14,7 @@ #include "../common/JFJochException.h" #include "../common/Logger.h" +#include "../common/ThreadAffinity.h" namespace { @@ -285,7 +286,7 @@ Layout Place(const std::vector &frames, const std::string &logger_name) { } void ForEachInOrder(size_t n, const std::function &fn) { - const size_t nthreads = std::min(std::max(1u, std::thread::hardware_concurrency()), 8); + const size_t nthreads = std::min(AvailableCpus(), 8); std::atomic next{0}; std::vector> futures; for (size_t t = 0; t < nthreads; t++) diff --git a/rugnux/rugnux_cli.cpp b/rugnux/rugnux_cli.cpp index 962c17e45..4545ae1a4 100644 --- a/rugnux/rugnux_cli.cpp +++ b/rugnux/rugnux_cli.cpp @@ -34,6 +34,7 @@ #include #include "../common/Logger.h" +#include "../common/ThreadAffinity.h" #include "../common/GitInfo.h" #include "../common/Definitions.h" #include "../common/DiffractionExperiment.h" @@ -98,7 +99,7 @@ void print_usage() { std::cout << " -o, --output-prefix Output file prefix (default: output). mx and scale runs always write " "_report.txt, the results report; every run writes _log.txt, the full timestamped log " "(debug level); an empty prefix writes nothing at all" << std::endl; - std::cout << " -N, --threads Number of threads (default: all hardware threads, of which the " + std::cout << " -N, --threads Number of threads (default: every CPU the run may use - its affinity and cgroup quota - of which the " "per-image loop uses at most 16 per GPU; an explicit value is used as given)" << std::endl; std::cout << " -s, --start-image Start image number (default: 0)" << std::endl; std::cout << " -e, --end-image End image number (default: all)" << std::endl; @@ -846,7 +847,7 @@ static int RunRugnux(int argc, char **argv) { std::string input_file; std::string output_prefix = "output"; - int nthreads = 0; // 0 = auto: resolved to all hardware threads after parsing (see below) + int nthreads = 0; // 0 = auto: resolved to the CPUs the run may use after parsing (see below) int start_image = 0; int end_image = -1; // -1 indicates process until end int image_stride = 1; @@ -1512,15 +1513,13 @@ static int RunRugnux(int argc, char **argv) { input_file = argv[optind]; - // -N defaults to 0 = "use all hardware threads"; resolve it to a concrete count here so every mode + // -N defaults to 0 = "use every CPU the run may use" (AvailableCpus); resolve it to a concrete count here so every mode // 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. const bool nthreads_auto = nthreads <= 0; - if (nthreads <= 0) { - unsigned int hw = std::thread::hardware_concurrency(); - nthreads = hw > 0 ? static_cast(hw) : 1; - } + if (nthreads <= 0) + 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; @@ -1537,7 +1536,7 @@ static int RunRugnux(int argc, char **argv) { } } log_sinks->remove_sink(early_log); - logger.Info("Threads: {}{}", nthreads, nthreads_auto ? " (all hardware threads)" : " (-N)"); + 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/viewer/windows/JFJochProcessingJobsWindow.cpp b/viewer/windows/JFJochProcessingJobsWindow.cpp index 0c9b849c7..c63f3ff6a 100644 --- a/viewer/windows/JFJochProcessingJobsWindow.cpp +++ b/viewer/windows/JFJochProcessingJobsWindow.cpp @@ -7,6 +7,7 @@ #include "../../rugnux/RugnuxDefaults.h" #include "../widgets/ToolbarIcons.h" #include "../../rugnux/RugnuxCommandLine.h" +#include "../../common/ThreadAffinity.h" #include #include @@ -37,15 +38,13 @@ #include #include -#include namespace { // Status (the progress bar) is column 2 so it stays visible in the narrow dock. enum Column { COL_NAME = 0, COL_STATUS, COL_STARTED, COL_MODE, COL_IMAGES, COL_INDEX, COL_CELL, COL_ACTIONS, COL_COUNT }; int default_threads() { - const unsigned hc = std::thread::hardware_concurrency(); - return hc == 0 ? 4 : static_cast(hc); + return static_cast(AvailableCpus()); } QTableWidgetItem *fixedItem(const QString &text) { // a non-editable cell