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 <noreply@anthropic.com>

common/ParallelFor.h

@@ -91,6 +91,7 @@ namespace parallel_detail {
                             std::lock_guard lock(m);
                             tids.push_back(static_cast<int>(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(); }

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<void> &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<id_t>(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.