From 91f6ac674911f4f7ab5dcbb58285a91ee69222bb Mon Sep 17 00:00:00 2001 From: Filip Leonarski Date: Sat, 17 Oct 2026 20:38:54 +0200 Subject: [PATCH] broker rugnux live: the ParallelFor pool is made at the broker's priority The pool is made by the first thread that uses it and its workers inherit that thread's nice value. That thread was the nice-10 rugnux thread, so without CAP_SYS_NICE / RLIMIT_NICE the workers stayed at nice 10 for the life of the broker - and the comment in RugnuxThread said the opposite. JFJochReceiverRugnux's constructor now makes the pool first, at the broker's priority, so the existing "lower it only where it can be raised back" logic is what decides. The pool constructor also waits for every worker to record its thread id, so a SetNice right after construction reaches all of them. Co-Authored-By: Claude Opus 5.5 --- common/ParallelFor.h | 6 ++++++ receiver/JFJochReceiverRugnux.cpp | 7 ++++++- 2 files changed, 12 insertions(+), 1 deletion(-) diff --git a/common/ParallelFor.h b/common/ParallelFor.h index e8dbb4e4d..be3241991 100644 --- a/common/ParallelFor.h +++ b/common/ParallelFor.h @@ -91,6 +91,7 @@ namespace parallel_detail { std::lock_guard lock(m); tids.push_back(static_cast(syscall(SYS_gettid))); } + cv.notify_all(); #endif Loop(); }); @@ -101,6 +102,11 @@ namespace parallel_detail { StopAndJoin(); throw; } +#ifdef __linux__ + // Every worker is in tids once the pool exists, so a SetNice right after reaches all of them. + std::unique_lock lock(m); + cv.wait(lock, [this] { return tids.size() == workers.size(); }); +#endif } ~WorkerPool() { StopAndJoin(); } diff --git a/receiver/JFJochReceiverRugnux.cpp b/receiver/JFJochReceiverRugnux.cpp index 25594a95a..e4337f7c4 100644 --- a/receiver/JFJochReceiverRugnux.cpp +++ b/receiver/JFJochReceiverRugnux.cpp @@ -123,6 +123,10 @@ JFJochReceiverRugnux::JFJochReceiverRugnux(const DiffractionExperiment &in_exper live_status(in_live_status), measurement_started(1), forward_started(forward_nthreads) { + // The ParallelFor pool is made by the first thread to use it and its workers take that thread's + // priority. Made here, at the broker's own priority, before the rugnux thread lowers its own - so + // the pool is lowered for the analysis only where it can be raised back (see RugnuxThread). + parallel_detail::WorkerPool::Instance(); logger.Info("Starting {} forward threads; rugnux runs on {} threads while the images arrive", forward_nthreads, rugnux_nthreads); for (int i = 0; i < forward_nthreads; i++) @@ -264,7 +268,8 @@ void JFJochReceiverRugnux::StartRugnux(std::future &rugnux_done, std::futu void JFJochReceiverRugnux::RugnuxThread() { // Below the forward threads: every thread rugnux starts inherits this, so an image arriving finds a // core however busy rugnux keeps the machine. - // The process-wide ParallelFor pool is not started by this thread, so it is lowered separately. + // The process-wide ParallelFor pool was made before (in the constructor), so it does not inherit + // this and is lowered separately. setpriority(PRIO_PROCESS, static_cast(syscall(SYS_gettid)), 10); // Only where it can be raised back afterwards: the pool outlives the collection, and left at nice 10 // it would slow every analysis the broker runs after it.