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.