JFJochReceiverDarkMask: the mask sequence has a receiver of its own

Lite loses its four mode branches, and no longer builds a DarkMaskAnalysis for runs that do not use one.
The dark mask goes where it went: a DECTRIS mask sequence here with rugnux on or off, a PSI detector's
to JFJochReceiverFPGA.

JFJochReceiverRugnux_DestroyedMidRun destroys each of the four leaves with its threads waiting, since the
start and join moved to the most-derived class.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01HVeDQsU5a8AJUdJPgWevAK

receiver/CMakeLists.txt

@@ -13,6 +13,7 @@ ADD_LIBRARY(JFJochReceiver STATIC
         JFJochReceiverOutput.h
         JFJochReceiverStream.h JFJochReceiverStream.cpp
         JFJochReceiverLite.h JFJochReceiverLite.cpp
+        JFJochReceiverDarkMask.h JFJochReceiverDarkMask.cpp
         FPGAImagePuller.h FPGAImagePuller.cpp
         JFJochReceiverRugnuxRotation.h JFJochReceiverRugnuxRotation.cpp
         JFJochReceiverRugnuxStills.h JFJochReceiverRugnuxStills.cpp

receiver/JFJochReceiverDarkMask.cpp

@@ -0,0 +1,74 @@
+// SPDX-FileCopyrightText: 2026 Filip Leonarski, Paul Scherrer Institute <filip.leonarski@psi.ch>
+// SPDX-License-Identifier: GPL-3.0-only
+
+#include "JFJochReceiverDarkMask.h"
+#include "../frame_serialize/CBORStream2Serializer.h"
+
+JFJochReceiverDarkMask::JFJochReceiverDarkMask(const DiffractionExperiment &in_experiment,
+                                               const PixelMask &in_pixel_mask,
+                                               ImagePuller &in_image_puller,
+                                               ImagePusher &in_image_pusher,
+                                               Logger &in_logger,
+                                               int64_t forward_and_sum_nthreads,
+                                               const SpotFindingSettings &in_spot_finding_settings,
+                                               JFJochReceiverCurrentStatus &in_current_status,
+                                               JFJochReceiverPlots &in_plots,
+                                               ImageBuffer &in_image_buffer,
+                                               ZMQPreviewSocket *in_zmq_preview_socket,
+                                               ZMQMetadataSocket *in_zmq_metadata_socket,
+                                               IndexerThreadPool *indexing_thread_pool)
+    : JFJochReceiverStream(in_experiment,
+                           in_pixel_mask,
+                           in_image_puller,
+                           in_image_pusher,
+                           in_logger,
+                           forward_and_sum_nthreads,
+                           in_spot_finding_settings,
+                           in_current_status,
+                           in_plots,
+                           in_image_buffer,
+                           in_zmq_preview_socket,
+                           in_zmq_metadata_socket,
+                           indexing_thread_pool),
+      mask_buffer(nthreads),
+      dark_mask_analysis(in_experiment.GetDarkMaskSettings(), in_experiment.GetPixelsNum()) {
+    StartThreads();
+}
+
+JFJochReceiverDarkMask::~JFJochReceiverDarkMask() {
+    cancelled = true;
+    JoinThreads();
+}
+
+void JFJochReceiverDarkMask::OnImage(uint32_t id, DataMessage &data_msg, std::chrono::steady_clock::time_point received) {
+    CountImage(data_msg);
+
+    dark_mask_analysis.AnalyzeImage(data_msg, mask_buffer[id]);
+
+    auto loc = image_buffer.GetImageSlot();
+
+    if (loc == nullptr)
+        writer_queue_full = true;
+    else {
+        auto writer_buffer = (uint8_t *) loc->GetImage();
+        CBORStream2Serializer serializer(writer_buffer, experiment.GetImageBufferLocationSize());
+        if (SerializeImageOrDrop(serializer, data_msg, *loc,
+                                 experiment.GetImageBufferLocationSize(), logger)) {
+            loc->SetImageNumber(data_msg.number);
+            loc->SetImageSize(serializer.GetBufferSize());
+            loc->SetIndexed(false);
+            loc->release();
+        }
+    }
+}
+
+// A dark mask run has frames and no images.
+int64_t JFJochReceiverDarkMask::ExpectedImages() const {
+    return experiment.GetFrameNum();
+}
+
+JFJochReceiverOutput JFJochReceiverDarkMask::GetFinalStatistics() const {
+    JFJochReceiverOutput ret = JFJochReceiver::GetFinalStatistics();
+    ret.dark_mask_result = dark_mask_analysis.GetMask();
+    return ret;
+}

receiver/JFJochReceiverDarkMask.h

@@ -0,0 +1,34 @@
+// SPDX-FileCopyrightText: 2026 Filip Leonarski, Paul Scherrer Institute <filip.leonarski@psi.ch>
+// SPDX-License-Identifier: GPL-3.0-only
+
+#pragma once
+
+#include <vector>
+
+#include "JFJochReceiverStream.h"
+#include "../image_analysis/dark_mask_analysis/DarkMaskAnalysis.h"
+
+// The dark frames of the mask sequence: each goes into DarkMaskAnalysis and the image buffer, nothing is sent.
+class JFJochReceiverDarkMask : public JFJochReceiverStream {
+    std::vector<std::vector<uint8_t>> mask_buffer;   // per image thread
+    DarkMaskAnalysis dark_mask_analysis;
+
+    void OnImage(uint32_t id, DataMessage &message, std::chrono::steady_clock::time_point received) override;
+    int64_t ExpectedImages() const override;
+public:
+    JFJochReceiverDarkMask(const DiffractionExperiment &experiment,
+                           const PixelMask &pixel_mask,
+                           ImagePuller &image_puller,
+                           ImagePusher &image_pusher,
+                           Logger &logger, int64_t forward_and_sum_nthreads,
+                           const SpotFindingSettings &spot_finding_settings,
+                           JFJochReceiverCurrentStatus &current_status,
+                           JFJochReceiverPlots &plots,
+                           ImageBuffer &image_buffer,
+                           ZMQPreviewSocket *zmq_preview_socket = nullptr,
+                           ZMQMetadataSocket *zmq_metadata_socket = nullptr,
+                           IndexerThreadPool *indexing_thread_pool = nullptr);
+    ~JFJochReceiverDarkMask() override;
+
+    JFJochReceiverOutput GetFinalStatistics() const override;
+};

receiver/JFJochReceiverLite.cpp

@@ -31,9 +31,7 @@ JFJochReceiverLite::JFJochReceiverLite(const DiffractionExperiment &in_experimen
                            in_zmq_preview_socket,
                            in_zmq_metadata_socket,
                            indexing_thread_pool),
-      analysis(nthreads),
-      mask_buffer(nthreads),
-      dark_mask_analysis(in_experiment.GetDarkMaskSettings(), in_experiment.GetPixelsNum()) {
+      analysis(nthreads) {
     StartThreads();
 }
 
@@ -43,11 +41,6 @@ JFJochReceiverLite::~JFJochReceiverLite() {
 }
 
 void JFJochReceiverLite::ImageThread(uint32_t id) {
-    if (experiment.GetDetectorMode() == DetectorMode::DarkMask) {
-        JFJochReceiverStream::ImageThread(id);
-        return;
-    }
-
     logger.Debug("Thread {} started", id);
 
     try {
@@ -73,36 +66,7 @@ void JFJochReceiverLite::ImageThread(uint32_t id) {
     logger.Debug("Thread {} finished", id);
 }
 
-void JFJochReceiverLite::OnImage(uint32_t id, DataMessage &message, std::chrono::steady_clock::time_point received) {
-    if (experiment.GetDetectorMode() == DetectorMode::DarkMask)
-        MaskImage(id, message);
-    else
-        AnalyseImage(id, message);
-}
-
-void JFJochReceiverLite::MaskImage(uint32_t id, DataMessage &data_msg) {
-    CountImage(data_msg);
-
-    dark_mask_analysis.AnalyzeImage(data_msg, mask_buffer[id]);
-
-    auto loc = image_buffer.GetImageSlot();
-
-    if (loc == nullptr)
-        writer_queue_full = true;
-    else {
-        auto writer_buffer = (uint8_t *) loc->GetImage();
-        CBORStream2Serializer serializer(writer_buffer, experiment.GetImageBufferLocationSize());
-        if (SerializeImageOrDrop(serializer, data_msg, *loc,
-                                 experiment.GetImageBufferLocationSize(), logger)) {
-            loc->SetImageNumber(data_msg.number);
-            loc->SetImageSize(serializer.GetBufferSize());
-            loc->SetIndexed(false);
-            loc->release();
-        }
-    }
-}
-
-void JFJochReceiverLite::AnalyseImage(uint32_t id, DataMessage &data_msg) {
+void JFJochReceiverLite::OnImage(uint32_t id, DataMessage &data_msg, std::chrono::steady_clock::time_point received) {
     auto start = std::chrono::high_resolution_clock::now();
 
     CountImage(data_msg);
@@ -135,22 +99,8 @@ void JFJochReceiverLite::AnalyseImage(uint32_t id, DataMessage &data_msg) {
                  id, data_msg.number, duration_total.count(), image_duration.count());
 }
 
-// A dark mask run has frames and no images.
-int64_t JFJochReceiverLite::ExpectedImages() const {
-    return (experiment.GetDetectorMode() == DetectorMode::DarkMask) ? experiment.GetFrameNum() : experiment.GetImageNum();
-}
-
 float JFJochReceiverLite::GetEfficiency() const {
-    if (experiment.GetDetectorMode() == DetectorMode::DarkMask)
-        return JFJochReceiverStream::GetEfficiency();
     if (ExpectedImages() == 0)
         return 0;
     return static_cast<float>(collection_efficiency_sum / static_cast<double>(ExpectedImages()));
 }
-
-JFJochReceiverOutput JFJochReceiverLite::GetFinalStatistics() const {
-    JFJochReceiverOutput ret = JFJochReceiver::GetFinalStatistics();
-    if (experiment.GetDetectorMode() == DetectorMode::DarkMask)
-        ret.dark_mask_result = dark_mask_analysis.GetMask();
-    return ret;
-}

receiver/JFJochReceiverLite.h

@@ -8,21 +8,15 @@
 
 #include "JFJochReceiverStream.h"
 #include "../image_analysis/MXAnalysisWithoutFPGA.h"
-#include "../image_analysis/dark_mask_analysis/DarkMaskAnalysis.h"
 
 class JFJochReceiverLite : public JFJochReceiverStream {
     std::vector<std::unique_ptr<MXAnalysisWithoutFPGA>> analysis;   // per image thread
-    std::vector<std::vector<uint8_t>> mask_buffer;                  // per image thread
-    DarkMaskAnalysis dark_mask_analysis;
 
     // Over the images received, as a detector stream without it counts each as whole
     std::atomic<double> collection_efficiency_sum{0.0};
 
     void ImageThread(uint32_t id) override;
     void OnImage(uint32_t id, DataMessage &message, std::chrono::steady_clock::time_point received) override;
-    int64_t ExpectedImages() const override;
-    void MaskImage(uint32_t id, DataMessage &message);
-    void AnalyseImage(uint32_t id, DataMessage &message);
 public:
     JFJochReceiverLite(const DiffractionExperiment& experiment,
                        const PixelMask &pixel_mask,
@@ -39,6 +33,4 @@ public:
     ~JFJochReceiverLite() override;
 
     float GetEfficiency() const override;
-
-    JFJochReceiverOutput GetFinalStatistics() const override;
 };

receiver/JFJochReceiverService.cpp

@@ -8,6 +8,7 @@
 #include "../common/ThreadAffinity.h"
 #include "JFJochReceiverFPGA.h"
 #include "JFJochReceiverLite.h"
+#include "JFJochReceiverDarkMask.h"
 #include "JFJochReceiverPedestal.h"
 #include "../preview/JFJochJPEG.h"
 #include "../preview/JFJochTIFF.h"
@@ -161,8 +162,22 @@ void JFJochReceiverService::Start(const DiffractionExperiment &experiment,
             else
                 image_puller = std::make_shared<ZMQImagePuller>(experiment.GetDetectorSetup().GetDECTRISStream2Addr());
 
-            // The dark frames of the mask sequence take Lite's mask path, with rugnux on as well.
-            if (rugnux_enable && (experiment.GetDetectorMode() == DetectorMode::Standard) && RugnuxRotationData(experiment))
+            // The dark frames of the mask sequence have a receiver of their own, with rugnux on as well.
+            if (experiment.GetDetectorMode() == DetectorMode::DarkMask)
+                receiver = std::make_unique<JFJochReceiverDarkMask>(experiment,
+                                                                    pixel_mask,
+                                                                    *image_puller,
+                                                                    image_pusher,
+                                                                    logger,
+                                                                    nthreads_local,
+                                                                    spot_finding_settings,
+                                                                    receiver_status,
+                                                                    plots,
+                                                                    image_buffer,
+                                                                    zmq_preview_socket.get(),
+                                                                    zmq_metadata_socket.get(),
+                                                                    indexer_thread_pool.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,
                         std::max<int>(1, static_cast<int>(AvailableCpus()) - rugnux_reserved_cpus),

tests/JFJochReceiverRugnuxTest.cpp

@@ -10,6 +10,10 @@
 #include "../image_analysis/structure_refinement/RugnuxModel.h"
 #include "../image_puller/TestImagePuller.h"
 #include "../image_pusher/ImagePusher.h"
+#include "../receiver/JFJochReceiverDarkMask.h"
+#include "../receiver/JFJochReceiverLite.h"
+#include "../receiver/JFJochReceiverRugnuxRotation.h"
+#include "../receiver/JFJochReceiverRugnuxStills.h"
 #include "../receiver/JFJochReceiverService.h"
 #include "../writer/HDF5Objects.h"
 
@@ -523,6 +527,50 @@ TEST_CASE("JFJochReceiverRugnux_Rotation_Cancel", "[JFJochReceiverRugnux]") {
     }
 }
 
+// A leaf joins its threads in its own destructor, with no one having asked for the end of the collection.
+TEST_CASE("JFJochReceiverRugnux_DestroyedMidRun", "[JFJochReceiverRugnux]") {
+    Logger logger(Catch::getResultCapture().getCurrentTestName());
+    RegisterHDF5Filter();
+
+    const int64_t images = 5;
+    DiffractionExperiment x = StillsExperiment(images);
+    PixelMask mask(x);
+    Stream stream(x, images);
+    stream.Frames(0, 2);
+
+    RecordingPusher pusher;
+    JFJochReceiverCurrentStatus status;
+    JFJochReceiverPlots plots;
+    RugnuxLiveStatus live;
+    ImageBuffer buffer(512 * 1024 * 1024);
+
+    SECTION("JFJochReceiverLite") {
+        JFJochReceiverLite receiver(x, mask, *stream.puller, pusher, logger, 4, DiffractionExperiment::DefaultDataProcessingSettings(),
+                                    status, plots, buffer);
+        std::this_thread::sleep_for(std::chrono::milliseconds(200));
+    }
+    SECTION("JFJochReceiverDarkMask") {
+        DiffractionExperiment dark = x;
+        dark.Mode(DetectorMode::DarkMask);
+        JFJochReceiverDarkMask receiver(dark, mask, *stream.puller, pusher, logger, 4, DiffractionExperiment::DefaultDataProcessingSettings(),
+                                        status, plots, buffer);
+        std::this_thread::sleep_for(std::chrono::milliseconds(200));
+    }
+    SECTION("JFJochReceiverRugnuxRotation") {
+        DiffractionExperiment rotation = x;
+        rotation.Goniometer(GoniometerAxis("omega", 0.0f, 0.1f, {1, 0, 0}, {}));
+        JFJochReceiverRugnuxRotation receiver(rotation, mask, *stream.puller, pusher, logger, 4, 4, nullptr, 60, live,
+                                              DiffractionExperiment::DefaultDataProcessingSettings(), status, plots, buffer);
+        std::this_thread::sleep_for(std::chrono::milliseconds(200));
+    }
+    SECTION("JFJochReceiverRugnuxStills") {
+        JFJochReceiverRugnuxStills receiver(x, mask, *stream.puller, pusher, logger, 4, nullptr, 60, live,
+                                            DiffractionExperiment::DefaultDataProcessingSettings(), status, plots, buffer);
+        std::this_thread::sleep_for(std::chrono::milliseconds(200));
+    }
+    SUCCEED("the destructor joined the threads");
+}
+
 TEST_CASE("JFJochReceiverFPGA_RugnuxOnStaysPerImage", "[JFJochReceiverRugnux]") {
     // Live stills are this receiver's alone until a puller feeds it: the card's run is as with rugnux off.
     Logger logger(Catch::getResultCapture().getCurrentTestName());
@@ -549,7 +597,7 @@ TEST_CASE("JFJochReceiverFPGA_RugnuxOnStaysPerImage", "[JFJochReceiverRugnux]")
     CHECK(service.GetRugnuxStatus().state == RugnuxLiveState::Idle);
 }
 
-TEST_CASE("JFJochReceiverRugnux_DarkMaskStaysWithLite", "[JFJochReceiverRugnux]") {
+TEST_CASE("JFJochReceiverRugnux_DarkMaskStaysOut", "[JFJochReceiverRugnux]") {
     // The mask sequence of a later POST /pedestal, with rugnux on: the state machine's experiment still has the
     // last /start's goniometer, or none.
     Logger logger(Catch::getResultCapture().getCurrentTestName());