From 89fb963bd9dd87e2536f79f5ad0c2e37f8ce7c6a Mon Sep 17 00:00:00 2001 From: Filip Leonarski Date: Sun, 8 Nov 2026 10:52:20 +0100 Subject: [PATCH] receiver: thread-count rules belong to the receivers they were made for The 100 Hz / 16-thread rule came with the Lite receiver, where each thread builds a GPU analysis; it now lives there again. The dark mask, rugnux rotation and rugnux stills receivers take the requested thread count. For the dark mask receiver this means several threads instead of one: the rule saw a dark mask run as zero images. Its frames are decompressed in parallel and counted into the mask under the analysis' lock. The four threads for summation on the host apply to JFJochReceiverFPGA only, which is the receiver that sums; before, they were applied before the receiver was chosen. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01HVeDQsU5a8AJUdJPgWevAK --- receiver/JFJochReceiverLite.cpp | 26 +++++++++++++++++++++++++- receiver/JFJochReceiverLite.h | 3 +++ receiver/JFJochReceiverService.cpp | 20 ++++++++++---------- receiver/JFJochReceiverStream.cpp | 28 ++-------------------------- receiver/JFJochReceiverStream.h | 3 +-- 5 files changed, 41 insertions(+), 39 deletions(-) diff --git a/receiver/JFJochReceiverLite.cpp b/receiver/JFJochReceiverLite.cpp index a85098e85..f02aff211 100644 --- a/receiver/JFJochReceiverLite.cpp +++ b/receiver/JFJochReceiverLite.cpp @@ -4,6 +4,30 @@ #include "JFJochReceiverLite.h" #include "../common/CUDAWrapper.h" +using namespace std::chrono_literals; + +int64_t JFJochReceiverLite::NumberOfThreads(int64_t requested_thread_number, + const DiffractionExperiment &in_experiment) { + int64_t number_of_images = in_experiment.GetImageNum(); + auto image_time = in_experiment.GetImageTime(); + + if (requested_thread_number <= 0) + return 1; + + // For very small datasets no need to go multithreaded + if (number_of_images < 4) + return 1; + + if (number_of_images < 16) + return std::min(2, requested_thread_number); + + // For 100 Hz, there is no reason to go for a very large number of threads + if (number_of_images < 256 || (image_time >= 10ms)) + return std::min(16, requested_thread_number); + + return requested_thread_number; +} + JFJochReceiverLite::JFJochReceiverLite(const DiffractionExperiment &in_experiment, const PixelMask &in_pixel_mask, ImagePuller &in_image_puller, @@ -22,7 +46,7 @@ JFJochReceiverLite::JFJochReceiverLite(const DiffractionExperiment &in_experimen in_image_puller, in_image_pusher, in_logger, - forward_and_sum_nthreads, + NumberOfThreads(forward_and_sum_nthreads, in_experiment), in_spot_finding_settings, in_current_status, in_plots, diff --git a/receiver/JFJochReceiverLite.h b/receiver/JFJochReceiverLite.h index df2c8fa93..f228d7899 100644 --- a/receiver/JFJochReceiverLite.h +++ b/receiver/JFJochReceiverLite.h @@ -12,6 +12,9 @@ class JFJochReceiverLite : public JFJochReceiverStream { std::vector> analysis; // per image thread + // Each thread builds a GPU analysis of its own + static int64_t NumberOfThreads(int64_t requested_thread_number, const DiffractionExperiment &experiment); + void ImageThread(uint32_t id) override; void OnImage(uint32_t id, DataMessage &message, std::chrono::steady_clock::time_point received) override; public: diff --git a/receiver/JFJochReceiverService.cpp b/receiver/JFJochReceiverService.cpp index abe2e08d2..f792343d3 100644 --- a/receiver/JFJochReceiverService.cpp +++ b/receiver/JFJochReceiverService.cpp @@ -99,10 +99,6 @@ void JFJochReceiverService::Start(const DiffractionExperiment &experiment, throw JFJochException(JFJochExceptionCategory::WrongDAQState, "Receiver not idle, cannot start"); try { - auto nthreads_local = nthreads; - if (experiment.IsCPUSummation()) - nthreads_local = 4; - // First clean-up old measurement receiver.reset(); if (fpga_puller) { @@ -119,7 +115,7 @@ void JFJochReceiverService::Start(const DiffractionExperiment &experiment, logger.Warning("Summation above what the card can do needs JFJochReceiverFPGA: FPGA image puller not used"); else { fpga_puller = std::make_shared(experiment, pixel_mask, calibration, aq_devices, logger, - nthreads_local, fpga_image_puller_slots); + nthreads, fpga_image_puller_slots); puller = fpga_puller; stream_input = true; } @@ -136,11 +132,15 @@ void JFJochReceiverService::Start(const DiffractionExperiment &experiment, zmq_preview_socket.get(), zmq_metadata_socket.get()); } else if (!stream_input) { + // Summation on the host runs threads of its own for each image + auto fpga_nthreads = nthreads; + if (experiment.IsCPUSummation()) + fpga_nthreads = 4; receiver = std::make_unique(experiment, pixel_mask, calibration, aq_devices, image_pusher, logger, - nthreads_local, + fpga_nthreads, spot_finding_settings, receiver_status, plots, @@ -161,7 +161,7 @@ void JFJochReceiverService::Start(const DiffractionExperiment &experiment, *image_puller, image_pusher, logger, - nthreads_local, + nthreads, spot_finding_settings, receiver_status, plots, @@ -170,13 +170,13 @@ void JFJochReceiverService::Start(const DiffractionExperiment &experiment, zmq_metadata_socket.get()); else if (rugnux_enable && (experiment.GetDetectorMode() == DetectorMode::Standard) && RugnuxRotationData(experiment)) receiver = std::make_unique( - experiment, pixel_mask, *image_puller, image_pusher, logger, nthreads_local, + experiment, pixel_mask, *image_puller, image_pusher, logger, nthreads, std::max(1, static_cast(AvailableCpus()) - rugnux_reserved_cpus), rugnux_model, rugnux_deadline_s, rugnux_status, spot_finding_settings, receiver_status, plots, image_buffer, zmq_preview_socket.get(), zmq_metadata_socket.get()); else if (rugnux_enable && (experiment.GetDetectorMode() == DetectorMode::Standard)) receiver = std::make_unique( - experiment, pixel_mask, *image_puller, image_pusher, logger, nthreads_local, + experiment, pixel_mask, *image_puller, image_pusher, logger, nthreads, rugnux_model, rugnux_deadline_s, rugnux_status, spot_finding_settings, receiver_status, plots, image_buffer, zmq_preview_socket.get(), zmq_metadata_socket.get(), indexer_thread_pool.get()); else @@ -185,7 +185,7 @@ void JFJochReceiverService::Start(const DiffractionExperiment &experiment, *image_puller, image_pusher, logger, - nthreads_local, + nthreads, spot_finding_settings, receiver_status, plots, diff --git a/receiver/JFJochReceiverStream.cpp b/receiver/JFJochReceiverStream.cpp index 2ab22a398..6c569fb12 100644 --- a/receiver/JFJochReceiverStream.cpp +++ b/receiver/JFJochReceiverStream.cpp @@ -3,30 +3,6 @@ #include "JFJochReceiverStream.h" -using namespace std::chrono_literals; - -int64_t JFJochReceiverStream::NumberOfThreads(int64_t requested_thread_number, - const DiffractionExperiment &in_experiment) { - int64_t number_of_images = in_experiment.GetImageNum(); - auto image_time = in_experiment.GetImageTime(); - - if (requested_thread_number <= 0) - return 1; - - // For very small datasets no need to go multithreaded - if (number_of_images < 4) - return 1; - - if (number_of_images < 16) - return std::min(2, requested_thread_number); - - // For 100 Hz, there is no reason to go for a very large number of threads - if (number_of_images < 256 || (image_time >= 10ms)) - return std::min(16, requested_thread_number); - - return requested_thread_number; -} - void JFJochReceiverStream::ApplyStartMessage(DiffractionExperiment &experiment, const StartMessage &msg) { // A PSI detector's images are what the experiment set the card up to produce: nothing to learn here. if (experiment.GetDetectorType() != DetectorType::DECTRIS) @@ -62,7 +38,7 @@ JFJochReceiverStream::JFJochReceiverStream(const DiffractionExperiment &in_exper ImagePuller &in_image_puller, ImagePusher &in_image_pusher, Logger &in_logger, - int64_t requested_nthreads, + int64_t in_nthreads, const SpotFindingSettings &in_spot_finding_settings, JFJochReceiverCurrentStatus &in_current_status, JFJochReceiverPlots &in_plots, @@ -82,7 +58,7 @@ JFJochReceiverStream::JFJochReceiverStream(const DiffractionExperiment &in_exper in_zmq_metadata_socket, indexing_thread_pool), image_puller(in_image_puller), - nthreads(NumberOfThreads(requested_nthreads, in_experiment)), + nthreads(in_nthreads), measurement_started(1), images_started(nthreads) { } diff --git a/receiver/JFJochReceiverStream.h b/receiver/JFJochReceiverStream.h index 01b2b8fe0..0617d0fd7 100644 --- a/receiver/JFJochReceiverStream.h +++ b/receiver/JFJochReceiverStream.h @@ -56,14 +56,13 @@ protected: public: // Takes what the detector's start message decides into the experiment. static void ApplyStartMessage(DiffractionExperiment &experiment, const StartMessage &message); - static int64_t NumberOfThreads(int64_t requested_thread_number, const DiffractionExperiment &experiment); JFJochReceiverStream(const DiffractionExperiment &experiment, const PixelMask &pixel_mask, ImagePuller &image_puller, ImagePusher &image_pusher, Logger &logger, - int64_t requested_nthreads, + int64_t nthreads, const SpotFindingSettings &spot_finding_settings, JFJochReceiverCurrentStatus ¤t_status, JFJochReceiverPlots &plots,