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 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01HVeDQsU5a8AJUdJPgWevAK

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<int64_t>(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<int64_t>(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,

receiver/JFJochReceiverLite.h

@@ -12,6 +12,9 @@
 class JFJochReceiverLite : public JFJochReceiverStream {
     std::vector<std::unique_ptr<MXAnalysisWithoutFPGA>> 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:

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<FPGAImagePuller>(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<JFJochReceiverFPGA>(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<JFJochReceiverRugnuxRotation>(
-                        experiment, pixel_mask, *image_puller, image_pusher, logger, nthreads_local,
+                        experiment, pixel_mask, *image_puller, image_pusher, logger, nthreads,
                         std::max<int>(1, static_cast<int>(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<JFJochReceiverRugnuxStills>(
-                        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,

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<int64_t>(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<int64_t>(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) {
 }

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 &current_status,
                          JFJochReceiverPlots &plots,