From 5d1fb1a48c75efac25d0c85d525e2cc8330465d9 Mon Sep 17 00:00:00 2001 From: Filip Leonarski Date: Sat, 7 Nov 2026 19:03:38 +0100 Subject: [PATCH] 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 Claude-Session: https://claude.ai/code/session_01HVeDQsU5a8AJUdJPgWevAK --- receiver/CMakeLists.txt | 1 + receiver/JFJochReceiverDarkMask.cpp | 74 +++++++++++++++++++++++++++++ receiver/JFJochReceiverDarkMask.h | 34 +++++++++++++ receiver/JFJochReceiverLite.cpp | 54 +-------------------- receiver/JFJochReceiverLite.h | 8 ---- receiver/JFJochReceiverService.cpp | 19 +++++++- tests/JFJochReceiverRugnuxTest.cpp | 50 ++++++++++++++++++- 7 files changed, 177 insertions(+), 63 deletions(-) create mode 100644 receiver/JFJochReceiverDarkMask.cpp create mode 100644 receiver/JFJochReceiverDarkMask.h diff --git a/receiver/CMakeLists.txt b/receiver/CMakeLists.txt index 416ee815e..5e4f4a787 100644 --- a/receiver/CMakeLists.txt +++ b/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 diff --git a/receiver/JFJochReceiverDarkMask.cpp b/receiver/JFJochReceiverDarkMask.cpp new file mode 100644 index 000000000..a156f1cca --- /dev/null +++ b/receiver/JFJochReceiverDarkMask.cpp @@ -0,0 +1,74 @@ +// SPDX-FileCopyrightText: 2026 Filip Leonarski, Paul Scherrer Institute +// 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; +} diff --git a/receiver/JFJochReceiverDarkMask.h b/receiver/JFJochReceiverDarkMask.h new file mode 100644 index 000000000..064a2580b --- /dev/null +++ b/receiver/JFJochReceiverDarkMask.h @@ -0,0 +1,34 @@ +// SPDX-FileCopyrightText: 2026 Filip Leonarski, Paul Scherrer Institute +// SPDX-License-Identifier: GPL-3.0-only + +#pragma once + +#include + +#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> 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 ¤t_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; +}; diff --git a/receiver/JFJochReceiverLite.cpp b/receiver/JFJochReceiverLite.cpp index 4c6bc4093..4b7ea1c9e 100644 --- a/receiver/JFJochReceiverLite.cpp +++ b/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(collection_efficiency_sum / static_cast(ExpectedImages())); } - -JFJochReceiverOutput JFJochReceiverLite::GetFinalStatistics() const { - JFJochReceiverOutput ret = JFJochReceiver::GetFinalStatistics(); - if (experiment.GetDetectorMode() == DetectorMode::DarkMask) - ret.dark_mask_result = dark_mask_analysis.GetMask(); - return ret; -} diff --git a/receiver/JFJochReceiverLite.h b/receiver/JFJochReceiverLite.h index 02431de91..ea477ffe7 100644 --- a/receiver/JFJochReceiverLite.h +++ b/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> analysis; // per image thread - std::vector> 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 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; }; diff --git a/receiver/JFJochReceiverService.cpp b/receiver/JFJochReceiverService.cpp index 033a3f7b9..24dfc333d 100644 --- a/receiver/JFJochReceiverService.cpp +++ b/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(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(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( experiment, pixel_mask, *image_puller, image_pusher, logger, nthreads_local, std::max(1, static_cast(AvailableCpus()) - rugnux_reserved_cpus), diff --git a/tests/JFJochReceiverRugnuxTest.cpp b/tests/JFJochReceiverRugnuxTest.cpp index 90c5dd599..01bef2b0a 100644 --- a/tests/JFJochReceiverRugnuxTest.cpp +++ b/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());