From d87508c6b877e30cbb2cb0273999b2b0dfb9ced3 Mon Sep 17 00:00:00 2001 From: Filip Leonarski Date: Sat, 7 Nov 2026 18:45:32 +0100 Subject: [PATCH] JFJochReceiverStream: one parent for the receivers that take a collection from an ImagePuller The START wait with its grace after the detector finished, the poll loop, the finish tail, the thread start and join, StopReceiver, GetProgress and the puller's dropped images move out of JFJochReceiverLite and JFJochReceiverRugnux into a parent. A child says what happens with an image (OnImage), what START does (OnStart) and, if it differs, how the collection ends (FinishCollection). Rugnux is still one class here. The threads start at the end of the most-derived constructor and are joined first thing in its destructor; the parent's own constructor and destructor neither start nor join. JoinThreads waits for the image threads as well, since the measurement thread leaves them behind when it leaves with an error and Lite's per-thread analysis now lives in the child, which goes first. What changes: - Rugnux's efficiency and progress count images rather than frames, as Lite does. With the card summing, a complete run reported efficiency 1/s and never reached progress 1. On DECTRIS the two are the same. - Rugnux takes the END's ADU histogram as Lite does, so a card run through it writes one. - The poll loop and the START wait take Rugnux's second catch: in Lite a std::exception now cancels the collection, where before it ended one thread and came out of the measurement thread without an END. - The dark mask's status now also reports the puller's maximum delay, as every other path did. New tests: JFJochReceiverRugnux_Rotation_Cancel (before START, detector finished without START, during the collection, silent) passed on the old shape first; FPGAImagePuller_RugnuxReceiver_Summation reported 0.5 and 0.5 there. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01HVeDQsU5a8AJUdJPgWevAK --- receiver/CMakeLists.txt | 1 + receiver/JFJochReceiverLite.cpp | 385 ++++++----------------------- receiver/JFJochReceiverLite.h | 73 +----- receiver/JFJochReceiverRugnux.cpp | 200 ++++----------- receiver/JFJochReceiverRugnux.h | 39 ++- receiver/JFJochReceiverStream.cpp | 280 +++++++++++++++++++++ receiver/JFJochReceiverStream.h | 83 +++++++ tests/FPGAImagePullerTest.cpp | 32 +++ tests/JFJochReceiverRugnuxTest.cpp | 60 +++++ 9 files changed, 610 insertions(+), 543 deletions(-) create mode 100644 receiver/JFJochReceiverStream.cpp create mode 100644 receiver/JFJochReceiverStream.h diff --git a/receiver/CMakeLists.txt b/receiver/CMakeLists.txt index 88800a486..1089f1c9d 100644 --- a/receiver/CMakeLists.txt +++ b/receiver/CMakeLists.txt @@ -11,6 +11,7 @@ ADD_LIBRARY(JFJochReceiver STATIC JFJochReceiverCurrentStatus.cpp JFJochReceiverCurrentStatus.h JFJochReceiverOutput.h + JFJochReceiverStream.h JFJochReceiverStream.cpp JFJochReceiverLite.h JFJochReceiverLite.cpp FPGAImagePuller.h FPGAImagePuller.cpp JFJochReceiverRugnux.h JFJochReceiverRugnux.cpp diff --git a/receiver/JFJochReceiverLite.cpp b/receiver/JFJochReceiverLite.cpp index 8f1abc134..4c6bc4093 100644 --- a/receiver/JFJochReceiverLite.cpp +++ b/receiver/JFJochReceiverLite.cpp @@ -5,30 +5,6 @@ #include "../image_analysis/indexing/IndexerFactory.h" #include "../common/CUDAWrapper.h" -using namespace std::chrono_literals; - -int64_t JFJochReceiverLite::NumberOfDataAnalysisThreads(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, @@ -42,216 +18,35 @@ JFJochReceiverLite::JFJochReceiverLite(const DiffractionExperiment &in_experimen ZMQPreviewSocket *in_zmq_preview_socket, ZMQMetadataSocket *in_zmq_metadata_socket, IndexerThreadPool *indexing_thread_pool) - : JFJochReceiver(in_experiment, - in_image_buffer, - in_image_pusher, - in_current_status, - in_plots, - in_spot_finding_settings, - in_logger, - in_pixel_mask, - in_zmq_preview_socket, - in_zmq_metadata_socket, - indexing_thread_pool), - image_puller(in_image_puller), - data_analysis_nthreads(NumberOfDataAnalysisThreads(forward_and_sum_nthreads, in_experiment)), - data_analysis_started(data_analysis_nthreads), - measurement_started(1), + : 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), + analysis(nthreads), + mask_buffer(nthreads), dark_mask_analysis(in_experiment.GetDarkMaskSettings(), in_experiment.GetPixelsNum()) { - logger.Debug("Starting {} data analysis threads", data_analysis_nthreads); - - if (experiment.GetDetectorMode() == DetectorMode::DarkMask) { - for (int i = 0; i < data_analysis_nthreads; i++) - data_analysis_futures.emplace_back( - std::async(std::launch::async, &JFJochReceiverLite::MaskThread, this, i) - ); - } else { - // Start frame transformation threads - for (int i = 0; i < data_analysis_nthreads; i++) - data_analysis_futures.emplace_back( - std::async(std::launch::async, &JFJochReceiverLite::DataAnalysisThread, this, i) - ); - } - - measurement = std::async(std::launch::async, &JFJochReceiverLite::MeasurementThread, this); - - data_analysis_started.wait(); - logger.Debug("Data analysis threads ready"); + StartThreads(); } JFJochReceiverLite::~JFJochReceiverLite() { cancelled = true; - try { - if (measurement.valid()) - measurement.get(); - } catch (const std::exception& e) { - logger.Error("ReceiverLite teardown failed: {}", e.what()); - } catch (...) { - logger.Error("ReceiverLite teardown failed with unknown exception"); - } + JoinThreads(); } -void JFJochReceiverLite::MeasurementThread() { - try { - // Wait for start message to arrive. Once the detector has reported it is done, nothing - // more is coming: allow a start message that is still in flight to land, then give up. - // Without this the poll runs for ever - a detector that never streamed leaves Stop() - // waiting on this thread with no way out other than a cancel from the operator. - std::optional give_up_at; - auto msg = image_puller.PollImage(); - - while (!cancelled && (!msg.has_value() || !msg->cbor || !msg->cbor->start_message)) { - if (detector_finished && !give_up_at) - give_up_at = std::chrono::steady_clock::now() + StartMessageGrace; - if (give_up_at && (std::chrono::steady_clock::now() > *give_up_at)) { - logger.Error("Detector finished, but no start message arrived - abandoning measurement"); - cancelled = true; - break; - } - msg = image_puller.PollImage(); - } - - if (cancelled) { - current_status.SetProgress({}); - current_status.SetStatus(GetStatus()); - measurement_started.count_down(); - return; // just quit the function - no measurement, no results - } - - ApplyStartMessage(experiment, msg->cbor->start_message.value()); - image_buffer.StartMeasurement(experiment); // Only at this point we know bit-depth of the images - start_time = std::chrono::system_clock::now(); - // Send new start message out - SendStartMessage(); - measurement_started.count_down(); - - } catch (const JFJochException &e) { - logger.ErrorException(e); - Cancel(e); - measurement_started.count_down(); - throw; - } - - logger.Info("Receiving started"); - - // Analysis is running - // ... - // Till it is done. - - // Combine frame transformation threads - for (auto &f: data_analysis_futures) - f.get(); - - logger.Debug("Data analysis threads finished"); - - current_status.SetProgress(1.0); - current_status.SetStatus(GetStatus()); - - // Send end message out - SendEndMessage(); - - if (!image_buffer.CheckIfBufferReturned(SendFinalizeTimeout)) { - logger.Error("Send commands not finalized in {} seconds", SendFinalizeTimeout.count()); - throw JFJochException(JFJochExceptionCategory::ZeroMQ, "Send commands not finalized in " - + std::to_string(SendFinalizeTimeout.count()) + " seconds"); - } - - logger.Debug("All images sent through ZeroMQ"); - - if (push_images_to_writer) - writer_error = image_pusher.Finalize(); - - logger.Debug("Writing process finalized"); - - end_time = std::chrono::system_clock::now(); - - current_status.SetProgress({}); - current_status.SetStatus(GetStatus()); -} - -void JFJochReceiverLite::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) +void JFJochReceiverLite::ImageThread(uint32_t id) { + if (experiment.GetDetectorMode() == DetectorMode::DarkMask) { + JFJochReceiverStream::ImageThread(id); return; - - if ((experiment.GetXPixelsNum() != msg.image_size_x) - || (experiment.GetYPixelsNum() != msg.image_size_y)) - throw JFJochException(JFJochExceptionCategory::InputParameterInvalid, - "Mismatch in detector size"); - if (fabs(experiment.GetPixelSize_mm() - msg.pixel_size_x * 1e3) > 1e-7) - throw JFJochException(JFJochExceptionCategory::InputParameterInvalid, - "Mismatch in pixel size"); - - experiment.Detector().Description(msg.detector_description); - experiment.Detector().SerialNumber(msg.detector_serial_number); - experiment.Detector().SensorMaterial(msg.sensor_material); - experiment.Detector().SensorThickness_um(msg.sensor_thickness * 1e6); - experiment.Detector().SaturationLimit(SaturationLimitFromValue(msg.saturation_value)); - // The detector decides the corrections applied to the pixels (DECTRIS enables them by default); - // the outgoing start message is rebuilt from the experiment, so without copying them NXmx would - // record them as not applied. - experiment.Detector().CountRateCorrectionApplied(msg.countrate_correction_enabled); - experiment.Detector().CountRateCorrectionLookupTable(msg.countrate_correction_lookup_table); - experiment.Detector().FlatfieldApplied(msg.flatfield_enabled); - experiment.Detector().VirtualPixelInterpolationApplied(msg.virtual_pixel_interpolation_enabled); - // The mask written with the data is ours (the detector's plus user masking); it is never - // uploaded to the detector, so the pixels arrive without it applied. - experiment.ApplyPixelMask(false); - // Images are forwarded byte-for-byte, so the stream's own image_dtype - not anything configured - // locally - decides both the width and the sign the outgoing metadata must declare. Taking only - // the width used to leave a signed stream declared unsigned in NXmx. - experiment.Detector().BitDepthImage(msg.bit_depth_image); - experiment.PixelSigned(msg.pixel_signed); -} - -void JFJochReceiverLite::MaskThread(uint32_t id) { - std::vector buffer; - data_analysis_started.count_down(); - measurement_started.wait(); - - while (!cancelled && !end_message_received) { - try { - auto msg = image_puller.PollImage(std::chrono::milliseconds(5)); - - if (!msg.has_value() || !msg->cbor) { - // Message not received or not parsed - continue; - } else if (msg->cbor->end_message) { - end_message_received = true; - logger.Debug("Thread {} end message received in JFJochReceiverLite", id); - } else if (msg->cbor->data_message) { - DataMessage data_msg = msg->cbor->data_message.value(); - CountImage(data_msg); - - dark_mask_analysis.AnalyzeImage(data_msg, buffer); - - 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(); - } - } - } - current_status.SetProgress(GetProgress()); - current_status.SetStatus(GetStatus()); - } catch (const JFJochException &e) { - logger.ErrorException(e); - Cancel(e); - } } -} - -void JFJochReceiverLite::DataAnalysisThread(uint32_t id) { - std::unique_ptr analysis; logger.Debug("Thread {} started", id); @@ -261,114 +56,98 @@ void JFJochReceiverLite::DataAnalysisThread(uint32_t id) { logger.Warning("Error pinning GPU {}", e.what()); } - data_analysis_started.count_down(); - measurement_started.wait(); + WaitForImages(); try { // The analysis works on the assembled image; an image in raw module layout is only forwarded. if (experiment.IsGeometryTransformed()) - analysis = std::make_unique(experiment, *az_int_mapping, pixel_mask, indexer, - /*enable_fused_adaptive_gpu=*/true); + analysis[id] = std::make_unique(experiment, *az_int_mapping, pixel_mask, indexer, + /*enable_fused_adaptive_gpu=*/true); } catch (const JFJochException &e) { Cancel(e); return; } - while (!cancelled && !end_message_received) { - try { - auto msg = image_puller.PollImage(std::chrono::milliseconds(5)); - - if (!msg.has_value() || !msg->cbor) { - // Message not received or not parsed - continue; - } else if (msg->cbor->end_message) { - // Message is end message - end_adu_histogram = msg->cbor->end_message->adu_histogram; - end_message_received = true; - logger.Debug("Thread {} End message received in JFJochReceiverLite", id); - } else if (msg->cbor->data_message) { - auto start = std::chrono::high_resolution_clock::now(); - DataMessage data_msg = msg->cbor->data_message.value(); - - CountImage(data_msg); - collection_efficiency_sum += data_msg.image_collection_efficiency.value_or(1.0f); - - auto image_start_time = std::chrono::high_resolution_clock::now(); - - AzimuthalIntegrationProfile profile(*az_int_mapping); - if (analysis) - analysis->AnalyzeStreamImage(data_msg, profile, GetSpotFindingSettings()); - - auto image_end_time = std::chrono::high_resolution_clock::now(); - std::chrono::duration image_duration = image_end_time - image_start_time; - - data_msg.processing_time_s = image_duration.count(); - StampImage(data_msg, image_puller.GetCurrentFifoUtilization()); - AddToAverages(data_msg); - - plots.Add(data_msg, profile); - scan_result.Add(data_msg); - - if (!serialmx_filter.ApplyFilter(data_msg)) - ++images_skipped; - else - ForwardImage(data_msg, image_buffer.GetImageSlot()); - - auto end = std::chrono::high_resolution_clock::now(); - std::chrono::duration duration_total = end - start; - logger.Debug("Thread {} Image {:6d} processing {:8.04f} s analysis {:8.04f} s", - id, data_msg.number, duration_total.count(), image_duration.count()); - } - UpdateMaxDelay(image_puller.GetCurrentFifoUtilization()); - current_status.SetProgress(GetProgress()); - current_status.SetStatus(GetStatus()); - } catch (const JFJochException &e) { - logger.ErrorException(e); - Cancel(e); - } - } + PollImages(id); logger.Debug("Thread {} finished", id); } -void JFJochReceiverLite::Cancel(bool silent) { - JFJochReceiver::Cancel(silent); - // A silent cancel is not a request to abort - it says the detector has finished - so it must - // not set cancelled. It does bound the wait for the start message: see MeasurementThread. - detector_finished = true; +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::StopReceiver() { - if (measurement.valid()) { - measurement.get(); - logger.Debug("Receiver stopped"); +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::AddToEndMessage(EndMessage &message) { - if (!end_adu_histogram.empty()) - message.adu_histogram = end_adu_histogram; +void JFJochReceiverLite::AnalyseImage(uint32_t id, DataMessage &data_msg) { + auto start = std::chrono::high_resolution_clock::now(); + + CountImage(data_msg); + collection_efficiency_sum += data_msg.image_collection_efficiency.value_or(1.0f); + + auto image_start_time = std::chrono::high_resolution_clock::now(); + + AzimuthalIntegrationProfile profile(*az_int_mapping); + if (analysis[id]) + analysis[id]->AnalyzeStreamImage(data_msg, profile, GetSpotFindingSettings()); + + auto image_end_time = std::chrono::high_resolution_clock::now(); + std::chrono::duration image_duration = image_end_time - image_start_time; + + data_msg.processing_time_s = image_duration.count(); + StampImage(data_msg, image_puller.GetCurrentFifoUtilization()); + AddToAverages(data_msg); + + plots.Add(data_msg, profile); + scan_result.Add(data_msg); + + if (!serialmx_filter.ApplyFilter(data_msg)) + ++images_skipped; + else + ForwardImage(data_msg, image_buffer.GetImageSlot()); + + auto end = std::chrono::high_resolution_clock::now(); + std::chrono::duration duration_total = end - start; + logger.Debug("Thread {} Image {:6d} processing {:8.04f} s analysis {:8.04f} s", + id, data_msg.number, duration_total.count(), image_duration.count()); } -// With the card summing, an image is many frames. A dark mask run has frames and no images. +// 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; - if (experiment.GetDetectorMode() == DetectorMode::DarkMask) - return static_cast(images_collected) / static_cast(ExpectedImages()); return static_cast(collection_efficiency_sum / static_cast(ExpectedImages())); } -float JFJochReceiverLite::GetProgress() const { - if (ExpectedImages() == 0) - return 0.0; - return static_cast(max_image_number_received) / static_cast(ExpectedImages()); -} - JFJochReceiverOutput JFJochReceiverLite::GetFinalStatistics() const { JFJochReceiverOutput ret = JFJochReceiver::GetFinalStatistics(); if (experiment.GetDetectorMode() == DetectorMode::DarkMask) diff --git a/receiver/JFJochReceiverLite.h b/receiver/JFJochReceiverLite.h index fdaf32291..02431de91 100644 --- a/receiver/JFJochReceiverLite.h +++ b/receiver/JFJochReceiverLite.h @@ -3,72 +3,27 @@ #pragma once -#include -#include +#include +#include -#include "JFJochReceiver.h" -#include "../common/DiffractionExperiment.h" -#include "../common/PixelMask.h" -#include "../common/ImageBuffer.h" -#include "../common/Logger.h" -#include "../image_pusher/ImagePusher.h" -#include "../preview/PreviewImage.h" -#include "../preview/ZMQPreviewSocket.h" -#include "../preview/ZMQMetadataSocket.h" -#include "LossyFilter.h" -#include "../common/MovingAverage.h" -#include "JFJochReceiverCurrentStatus.h" -#include "../common/JFJochReceiverPlots.h" -#include "JFJochReceiverOutput.h" -#include "../image_puller/ZMQImagePuller.h" +#include "JFJochReceiverStream.h" #include "../image_analysis/MXAnalysisWithoutFPGA.h" #include "../image_analysis/dark_mask_analysis/DarkMaskAnalysis.h" -class JFJochReceiverLite : public JFJochReceiver { - // Overriding Cancel(bool) below would otherwise hide the Cancel(const JFJochException &) - // the analysis threads call. - using JFJochReceiver::Cancel; - - ImagePuller &image_puller; - - std::atomic end_message_received{false}; - - // Set when the caller reports the run is over - JFJochServices::Stop cancels the receiver - // silently once the detector reads back idle. Only consulted while waiting for the start - // message; a silent cancel must not interrupt images that are still being processed. - std::atomic detector_finished{false}; - - // How long a start message may still be in flight after the detector reported it had finished. - // The detector can also read back as idle for a moment just after it was armed, so a run is not - // abandoned the instant the silent cancel arrives. - constexpr static auto StartMessageGrace = std::chrono::seconds(5); - - const int64_t data_analysis_nthreads; - - std::vector> data_analysis_futures; - std::future measurement; - - std::latch measurement_started; - std::latch data_analysis_started; - +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}; - std::map> end_adu_histogram; - int64_t ExpectedImages() const; - void AddToEndMessage(EndMessage &message) override; - uint64_t GetImagesDropped() const override { return image_puller.GetDroppedImages(); } - void MaskThread(uint32_t id); - void MeasurementThread(); - void DataAnalysisThread(uint32_t id); + 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: - // Takes what the detector's start message decides into the experiment. - static void ApplyStartMessage(DiffractionExperiment &experiment, const StartMessage &message); - - static int64_t NumberOfDataAnalysisThreads(int64_t requested_thread_number, const DiffractionExperiment& experiment); - JFJochReceiverLite(const DiffractionExperiment& experiment, const PixelMask &pixel_mask, ImagePuller &image_puller, @@ -83,13 +38,7 @@ public: IndexerThreadPool *indexing_thread_pool = nullptr); ~JFJochReceiverLite() override; - void StopReceiver() override; - void Cancel(bool silent) override; - float GetEfficiency() const override; - float GetProgress() const override; JFJochReceiverOutput GetFinalStatistics() const override; }; - - diff --git a/receiver/JFJochReceiverRugnux.cpp b/receiver/JFJochReceiverRugnux.cpp index e9941b1eb..55e24474e 100644 --- a/receiver/JFJochReceiverRugnux.cpp +++ b/receiver/JFJochReceiverRugnux.cpp @@ -9,7 +9,6 @@ #include "spdlog/fmt/fmt.h" -#include "JFJochReceiverLite.h" #include "../common/HostMemory.h" #include "../common/time_utc.h" #include "../frame_serialize/CBORStream2Deserializer.h" @@ -112,19 +111,15 @@ JFJochReceiverRugnux::JFJochReceiverRugnux(const DiffractionExperiment &in_exper ZMQPreviewSocket *in_zmq_preview_socket, ZMQMetadataSocket *in_zmq_metadata_socket, IndexerThreadPool *in_indexing_thread_pool) - : JFJochReceiver(in_experiment, in_image_buffer, in_image_pusher, in_current_status, in_plots, - in_spot_finding_settings, in_logger, in_pixel_mask, in_zmq_preview_socket, - in_zmq_metadata_socket, nullptr), - image_puller(in_image_puller), + : JFJochReceiverStream(in_experiment, in_pixel_mask, in_image_puller, in_image_pusher, in_logger, + in_forward_nthreads, in_spot_finding_settings, in_current_status, in_plots, + in_image_buffer, in_zmq_preview_socket, in_zmq_metadata_socket, nullptr), reader(in_image_buffer), - forward_nthreads(JFJochReceiverLite::NumberOfDataAnalysisThreads(in_forward_nthreads, in_experiment)), rugnux_nthreads(in_rugnux_nthreads), host_budget(RugnuxHostBudget(in_image_buffer)), model(std::move(in_model)), deadline_s(in_deadline_s), live_status(in_live_status), - measurement_started(1), - forward_started(forward_nthreads), stills_run(!RugnuxRotationData(in_experiment)), indexing_thread_pool(in_indexing_thread_pool), stills_nthreads(std::min(16, in_forward_nthreads)) { @@ -136,13 +131,10 @@ JFJochReceiverRugnux::JFJochReceiverRugnux(const DiffractionExperiment &in_exper [[maybe_unused]] static const FFTIndexerReuse *fft_indexers = new FFTIndexerReuse; [[maybe_unused]] static const FFTPlanReuse *fft_plans = new FFTPlanReuse; logger.Info("rugnux live: starting {} forward threads; rugnux runs on {} threads while the images arrive, " - "planning against {:.1f} GB of host memory", forward_nthreads, rugnux_nthreads, + "planning against {:.1f} GB of host memory", nthreads, rugnux_nthreads, static_cast(host_budget) / 1e9); } - for (int i = 0; i < forward_nthreads; i++) - forward_futures.emplace_back(std::async(std::launch::async, &JFJochReceiverRugnux::ForwardThread, this, i)); - measurement = std::async(std::launch::async, &JFJochReceiverRugnux::MeasurementThread, this); - forward_started.wait(); + StartThreads(); } JFJochReceiverRugnux::~JFJochReceiverRugnux() { @@ -155,83 +147,39 @@ JFJochReceiverRugnux::~JFJochReceiverRugnux() { if (probe != nullptr) probe->Cancel(); } - try { - if (measurement.valid()) - measurement.get(); - } catch (const std::exception &e) { - logger.Error("ReceiverRugnux teardown failed: {}", e.what()); - } + JoinThreads(); } -void JFJochReceiverRugnux::MeasurementThread() { - std::future rugnux_done, rugnux_out, probe_done; - try { - // Waiting for the start message, as JFJochReceiverLite::MeasurementThread. - std::optional give_up_at; - auto msg = image_puller.PollImage(); - while (!cancelled && (!msg.has_value() || !msg->cbor || !msg->cbor->start_message)) { - if (detector_finished && !give_up_at) - give_up_at = std::chrono::steady_clock::now() + StartMessageGrace; - if (give_up_at && (std::chrono::steady_clock::now() > *give_up_at)) { - logger.Error("Detector finished, but no start message arrived - abandoning measurement"); - cancelled = true; - break; - } - msg = image_puller.PollImage(); - } - if (cancelled) { - current_status.SetProgress({}); - current_status.SetStatus(GetStatus()); - measurement_started.count_down(); - return; - } - - JFJochReceiverLite::ApplyStartMessage(experiment, msg->cbor->start_message.value()); - image_buffer.StartMeasurement(experiment); - start_time = std::chrono::system_clock::now(); - if (stills_run) - StartStills(); - else { - SendStartMessage(); - - // Rugnux reads the images back from the buffer, so it needs one slot per image: a slot taken - // again would hand it a later image. Without that the collection is only forwarded. - const auto slots = image_buffer.GetStatus().total_slots; - // The image time as the detector produces images, summation included - const double image_time_s = std::chrono::duration(experiment.GetImageTime()).count(); - std::optional expected_collection_s; - if (image_time_s > 0) - expected_collection_s = static_cast(experiment.GetImageNum()) * image_time_s; - live_status.Start(experiment.GetImageNum(), expected_collection_s, msg->received); - if (slots < experiment.GetImageNum()) { - skipped_warning = fmt::format("no rugnux analysis: the image buffer holds {} images, the collection " - "has {}", slots, experiment.GetImageNum()); - logger.Error("{}", skipped_warning); - live_status.Finish(RugnuxLiveState::Skipped, "the image buffer cannot hold the whole collection"); - } else - StartRugnux(rugnux_done, rugnux_out, probe_done); - } - measurement_started.count_down(); - } catch (const JFJochException &e) { - logger.ErrorException(e); - Cancel(e); - measurement_started.count_down(); - throw; - } catch (const std::exception &e) { - logger.Error("Cancelling data collection due to exception: {}", e.what()); - cancelled = true; - measurement_started.count_down(); - throw; - } - +void JFJochReceiverRugnux::OnStart(std::chrono::steady_clock::time_point received) { if (stills_run) { - StillsMeasurement(); + StartStills(); return; } + SendStartMessage(); - logger.Info("Receiving started"); - for (auto &f: forward_futures) - f.get(); + // Rugnux reads the images back from the buffer, so it needs one slot per image: a slot taken + // again would hand it a later image. Without that the collection is only forwarded. + const auto slots = image_buffer.GetStatus().total_slots; + // The image time as the detector produces images, summation included + const double image_time_s = std::chrono::duration(experiment.GetImageTime()).count(); + std::optional expected_collection_s; + if (image_time_s > 0) + expected_collection_s = static_cast(experiment.GetImageNum()) * image_time_s; + live_status.Start(experiment.GetImageNum(), expected_collection_s, received); + if (slots < experiment.GetImageNum()) { + skipped_warning = fmt::format("no rugnux analysis: the image buffer holds {} images, the collection " + "has {}", slots, experiment.GetImageNum()); + logger.Error("{}", skipped_warning); + live_status.Finish(RugnuxLiveState::Skipped, "the image buffer cannot hold the whole collection"); + } else + StartRugnux(); +} + +void JFJochReceiverRugnux::FinishCollection() { + if (stills_run) { + FinishStills(); + return; + } reader.Ended(); images_done = true; // The latency of the demonstration runs from the last image in (as the puller took it off the socket) @@ -258,13 +206,8 @@ void JFJochReceiverRugnux::MeasurementThread() { SendEndMessage(); const double end_sent_s = last_image ? SecondsSince(last_image.value()) : 0.0; - if (!image_buffer.CheckIfBufferReturned(SendFinalizeTimeout)) { - logger.Error("Send commands not finalized in {} seconds", SendFinalizeTimeout.count()); - throw JFJochException(JFJochExceptionCategory::ZeroMQ, "Send commands not finalized in " - + std::to_string(SendFinalizeTimeout.count()) + " seconds"); - } - if (push_images_to_writer) - writer_error = image_pusher.Finalize(); + CheckIfImagesSent(); + WriterFinalize(); if (rugnux_out.valid()) { // The run's outcome, in one line: its results are in the live status, the end message and the files. const auto live = live_status.Get(); @@ -328,10 +271,7 @@ void JFJochReceiverRugnux::StartStills() { stills->Cut("cancel"); } -void JFJochReceiverRugnux::StillsMeasurement() { - logger.Info("Receiving started"); - for (auto &f: forward_futures) - f.get(); +void JFJochReceiverRugnux::FinishStills() { images_done = true; logger.Info("Receiving finished, {} images", images_collected.load()); @@ -352,18 +292,13 @@ void JFJochReceiverRugnux::StillsMeasurement() { end_time = std::chrono::system_clock::now(); - if (!image_buffer.CheckIfBufferReturned(SendFinalizeTimeout)) { - logger.Error("Send commands not finalized in {} seconds", SendFinalizeTimeout.count()); - throw JFJochException(JFJochExceptionCategory::ZeroMQ, "Send commands not finalized in " - + std::to_string(SendFinalizeTimeout.count()) + " seconds"); - } + CheckIfImagesSent(); if (error) { current_status.SetProgress({}); current_status.SetStatus(GetStatus()); std::rethrow_exception(error); } - if (push_images_to_writer) - writer_error = image_pusher.Finalize(); + WriterFinalize(); stills->Written(); current_status.SetProgress({}); @@ -444,8 +379,7 @@ void JFJochReceiverRugnux::SendAnalysedImage(DataMessage &message, const Azimuth } } -void JFJochReceiverRugnux::StartRugnux(std::future &rugnux_done, std::future &rugnux_out, - std::future &probe_done) { +void JFJochReceiverRugnux::StartRugnux() { // A failure here costs the analysis only: the images go on to the writer as they would without it. try { // The start message as the writer gets it, so rugnux sees the dataset an HDF5 master states. @@ -582,6 +516,7 @@ void JFJochReceiverRugnux::ProbeThread() { } void JFJochReceiverRugnux::AddToEndMessage(EndMessage &message) { + JFJochReceiverStream::AddToEndMessage(message); if (stills) { stills->FillEndMessage(message); return; @@ -639,37 +574,15 @@ void JFJochReceiverRugnux::AddToEndMessage(EndMessage &message) { } } -void JFJochReceiverRugnux::ForwardThread(uint32_t id) { - forward_started.count_down(); - measurement_started.wait(); - - while (!cancelled && !end_message_received) { - try { - auto msg = image_puller.PollImage(std::chrono::milliseconds(5)); - if (!msg.has_value() || !msg->cbor) - continue; - if (msg->cbor->end_message) { - end_message_received = true; - } else if (stills_run && msg->cbor->data_message) { - HoldStillsImage(msg->cbor->data_message.value()); - } else if (msg->cbor->data_message) { - live_status.ImageReceived(msg->received); - DataMessage data_msg = msg->cbor->data_message.value(); - CountImage(data_msg); - StampImage(data_msg, image_puller.GetCurrentFifoUtilization()); - ForwardImage(data_msg, image_buffer.GetImageSlot()); - } - UpdateMaxDelay(image_puller.GetCurrentFifoUtilization()); - current_status.SetProgress(GetProgress()); - current_status.SetStatus(GetStatus()); - } catch (const JFJochException &e) { - logger.ErrorException(e); - Cancel(e); - } catch (const std::exception &e) { - logger.Error("Cancelling data collection due to exception: {}", e.what()); - cancelled = true; - } +void JFJochReceiverRugnux::OnImage(uint32_t id, DataMessage &message, std::chrono::steady_clock::time_point received) { + if (stills_run) { + HoldStillsImage(message); + return; } + live_status.ImageReceived(received); + CountImage(message); + StampImage(message, image_puller.GetCurrentFifoUtilization()); + ForwardImage(message, image_buffer.GetImageSlot()); } void JFJochReceiverRugnux::Cancel(bool silent) { @@ -703,25 +616,12 @@ void JFJochReceiverRugnux::Cut(const std::string &reason) { logger.Debug("rugnux live: analysis cut ({}) - the end message carries what is established", reason); } -void JFJochReceiverRugnux::StopReceiver() { - if (measurement.valid()) { - measurement.get(); - logger.Debug("Receiver stopped"); - } -} - void JFJochReceiverRugnux::SetSpotFindingSettings(const SpotFindingSettings &in_spot_finding_settings) { // A rotation run takes rugnux's own spot finding; the frames of a serial run are analysed with these. if (stills_run) JFJochReceiver::SetSpotFindingSettings(in_spot_finding_settings); } -float JFJochReceiverRugnux::GetEfficiency() const { - if (experiment.GetFrameNum() == 0) - return 0; - return static_cast(images_collected) / static_cast(experiment.GetFrameNum()); -} - JFJochReceiverOutput JFJochReceiverRugnux::GetFinalStatistics() const { JFJochReceiverOutput ret = JFJochReceiver::GetFinalStatistics(); ret.analysis_warning = skipped_warning; @@ -737,9 +637,3 @@ JFJochReceiverOutput JFJochReceiverRugnux::GetFinalStatistics() const { } return ret; } - -float JFJochReceiverRugnux::GetProgress() const { - if (experiment.GetFrameNum() == 0) - return 0.0; - return static_cast(max_image_number_received) / static_cast(experiment.GetFrameNum()); -} diff --git a/receiver/JFJochReceiverRugnux.h b/receiver/JFJochReceiverRugnux.h index c1020f1a7..d30c6724d 100644 --- a/receiver/JFJochReceiverRugnux.h +++ b/receiver/JFJochReceiverRugnux.h @@ -6,8 +6,7 @@ #include #include -#include "JFJochReceiver.h" -#include "../image_puller/ImagePuller.h" +#include "JFJochReceiverStream.h" #include "../reader/JFJochImageBufferReader.h" #include "../rugnux/Rugnux.h" #include "RugnuxLiveStatus.h" @@ -23,17 +22,11 @@ // goes to the writer with its analysis; the start message waits for the pre-pass. The merges and the model step // run beside it, and the end message carries the last of them. Only what the puller delivers is used: a // DataMessage with a compressed image, whatever produced it. -class JFJochReceiverRugnux : public JFJochReceiver { - using JFJochReceiver::Cancel; +class JFJochReceiverRugnux : public JFJochReceiverStream { + using JFJochReceiverStream::Cancel; - ImagePuller &image_puller; JFJochImageBufferReader reader; - std::atomic end_message_received{false}; - std::atomic detector_finished{false}; // see JFJochReceiverLite - constexpr static auto StartMessageGrace = std::chrono::seconds(5); - - const int64_t forward_nthreads; const int rugnux_nthreads; const size_t host_budget; // what the main run and the probe plan their host memory against const std::shared_ptr model; @@ -41,11 +34,6 @@ class JFJochReceiverRugnux : public JFJochReceiver { RugnuxLiveStatus &live_status; std::string skipped_warning; // why the collection was not analysed, for the broker's status - std::vector> forward_futures; - std::future measurement; - std::latch measurement_started; - std::latch forward_started; - mutable std::mutex rugnux_m; Rugnux *rugnux = nullptr; // while it runs, for Cut Rugnux *probe = nullptr; // the same for the lattice probe @@ -60,25 +48,29 @@ class JFJochReceiverRugnux : public JFJochReceiver { int64_t stills_keep_free = 0; // slots frames do not take, for the analysis threads to send with std::unique_ptr stills; // under rugnux_m until the frames are in, for Cut + // rugnux_out is ready once the run's result is in place, rugnux_done once the run is gone as well. + std::future rugnux_out; + std::future probe_done; + std::future rugnux_done; + void StartStills(); - void StillsMeasurement(); + void FinishStills(); void HoldStillsImage(const DataMessage &message); ZeroCopyReturnValue *AnalysedImageSlot(); void SendAnalysedImage(DataMessage &message, const AzimuthalIntegrationProfile &profile); - void MeasurementThread(); + void OnStart(std::chrono::steady_clock::time_point received) override; + void OnImage(uint32_t id, DataMessage &message, std::chrono::steady_clock::time_point received) override; + void FinishCollection() override; // Opens the reader on the collection and starts the rugnux and probe threads; a failure is the - // analysis's alone (live status Failed), never the collection's. rugnux_out is ready once the run's - // result is in place, rugnux_done once the run is gone as well. - void StartRugnux(std::future &rugnux_done, std::future &rugnux_out, std::future &probe_done); - void ForwardThread(uint32_t id); + // analysis's alone (live status Failed), never the collection's. + void StartRugnux(); void RugnuxThread(); void ProbeThread(); // Stops the analysis where it is: rugnux returns what it has established (see Rugnux::Cancel), and // the end message goes out with that. Reason: "deadline" or "cancel". void Cut(const std::string &reason); void AddToEndMessage(EndMessage &message) override; - uint64_t GetImagesDropped() const override { return image_puller.GetDroppedImages(); } public: JFJochReceiverRugnux(const DiffractionExperiment &experiment, const PixelMask &pixel_mask, @@ -99,11 +91,8 @@ public: IndexerThreadPool *indexing_thread_pool = nullptr); ~JFJochReceiverRugnux() override; - void StopReceiver() override; void Cancel(bool silent) override; - float GetEfficiency() const override; - float GetProgress() const override; JFJochReceiverOutput GetFinalStatistics() const override; void SetSpotFindingSettings(const SpotFindingSettings &spot_finding_settings) override; diff --git a/receiver/JFJochReceiverStream.cpp b/receiver/JFJochReceiverStream.cpp new file mode 100644 index 000000000..c659dc02b --- /dev/null +++ b/receiver/JFJochReceiverStream.cpp @@ -0,0 +1,280 @@ +// SPDX-FileCopyrightText: 2026 Filip Leonarski, Paul Scherrer Institute +// SPDX-License-Identifier: GPL-3.0-only + +#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) + return; + + if ((experiment.GetXPixelsNum() != msg.image_size_x) + || (experiment.GetYPixelsNum() != msg.image_size_y)) + throw JFJochException(JFJochExceptionCategory::InputParameterInvalid, + "Mismatch in detector size"); + if (fabs(experiment.GetPixelSize_mm() - msg.pixel_size_x * 1e3) > 1e-7) + throw JFJochException(JFJochExceptionCategory::InputParameterInvalid, + "Mismatch in pixel size"); + + experiment.Detector().Description(msg.detector_description); + experiment.Detector().SerialNumber(msg.detector_serial_number); + experiment.Detector().SensorMaterial(msg.sensor_material); + experiment.Detector().SensorThickness_um(msg.sensor_thickness * 1e6); + experiment.Detector().SaturationLimit(SaturationLimitFromValue(msg.saturation_value)); + // The detector decides the corrections applied to the pixels (DECTRIS enables them by default); + // the outgoing start message is rebuilt from the experiment, so without copying them NXmx would + // record them as not applied. + experiment.Detector().CountRateCorrectionApplied(msg.countrate_correction_enabled); + experiment.Detector().CountRateCorrectionLookupTable(msg.countrate_correction_lookup_table); + experiment.Detector().FlatfieldApplied(msg.flatfield_enabled); + experiment.Detector().VirtualPixelInterpolationApplied(msg.virtual_pixel_interpolation_enabled); + // The mask written with the data is ours (the detector's plus user masking); it is never + // uploaded to the detector, so the pixels arrive without it applied. + experiment.ApplyPixelMask(false); + // Images are forwarded byte-for-byte, so the stream's own image_dtype - not anything configured + // locally - decides both the width and the sign the outgoing metadata must declare. Taking only + // the width used to leave a signed stream declared unsigned in NXmx. + experiment.Detector().BitDepthImage(msg.bit_depth_image); + experiment.PixelSigned(msg.pixel_signed); +} + +JFJochReceiverStream::JFJochReceiverStream(const DiffractionExperiment &in_experiment, + const PixelMask &in_pixel_mask, + ImagePuller &in_image_puller, + ImagePusher &in_image_pusher, + Logger &in_logger, + int64_t requested_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) + : JFJochReceiver(in_experiment, + in_image_buffer, + in_image_pusher, + in_current_status, + in_plots, + in_spot_finding_settings, + in_logger, + in_pixel_mask, + in_zmq_preview_socket, + in_zmq_metadata_socket, + indexing_thread_pool), + image_puller(in_image_puller), + nthreads(NumberOfThreads(requested_nthreads, in_experiment)), + measurement_started(1), + images_started(nthreads) { +} + +void JFJochReceiverStream::StartThreads() { + logger.Debug("Starting {} image threads", nthreads); + for (int i = 0; i < nthreads; i++) + image_futures.emplace_back(std::async(std::launch::async, &JFJochReceiverStream::ImageThread, this, i)); + measurement = std::async(std::launch::async, &JFJochReceiverStream::MeasurementThread, this); + images_started.wait(); + logger.Debug("Image threads ready"); +} + +void JFJochReceiverStream::JoinFuture(std::future &future) { + try { + if (future.valid()) + future.get(); + } catch (const std::exception &e) { + logger.Error("Receiver teardown failed: {}", e.what()); + } catch (...) { + logger.Error("Receiver teardown failed with unknown exception"); + } +} + +void JFJochReceiverStream::JoinThreads() { + JoinFuture(measurement); + // The measurement thread joins them, unless it left with an error + for (auto &f: image_futures) + JoinFuture(f); +} + +void JFJochReceiverStream::MeasurementThread() { + try { + // Wait for start message to arrive. Once the detector has reported it is done, nothing + // more is coming: allow a start message that is still in flight to land, then give up. + std::optional give_up_at; + auto msg = image_puller.PollImage(); + + while (!cancelled && (!msg.has_value() || !msg->cbor || !msg->cbor->start_message)) { + if (detector_finished && !give_up_at) + give_up_at = std::chrono::steady_clock::now() + StartMessageGrace; + if (give_up_at && (std::chrono::steady_clock::now() > *give_up_at)) { + logger.Error("Detector finished, but no start message arrived - abandoning measurement"); + cancelled = true; + break; + } + msg = image_puller.PollImage(); + } + + if (cancelled) { + current_status.SetProgress({}); + current_status.SetStatus(GetStatus()); + measurement_started.count_down(); + return; // just quit the function - no measurement, no results + } + + ApplyStartMessage(experiment, msg->cbor->start_message.value()); + image_buffer.StartMeasurement(experiment); // Only at this point we know bit-depth of the images + start_time = std::chrono::system_clock::now(); + OnStart(msg->received); + measurement_started.count_down(); + } catch (const JFJochException &e) { + logger.ErrorException(e); + Cancel(e); + measurement_started.count_down(); + throw; + } catch (const std::exception &e) { + logger.Error("Cancelling data collection due to exception: {}", e.what()); + cancelled = true; + measurement_started.count_down(); + throw; + } + + logger.Info("Receiving started"); + + for (auto &f: image_futures) + f.get(); + + logger.Debug("Image threads finished"); + + FinishCollection(); +} + +void JFJochReceiverStream::OnStart(std::chrono::steady_clock::time_point received) { + SendStartMessage(); +} + +void JFJochReceiverStream::ImageThread(uint32_t id) { + WaitForImages(); + PollImages(id); +} + +void JFJochReceiverStream::WaitForImages() { + images_started.count_down(); + measurement_started.wait(); +} + +void JFJochReceiverStream::PollImages(uint32_t id) { + while (!cancelled && !end_message_received) { + try { + auto msg = image_puller.PollImage(std::chrono::milliseconds(5)); + + if (!msg.has_value() || !msg->cbor) { + // Message not received or not parsed + continue; + } else if (msg->cbor->end_message) { + end_adu_histogram = msg->cbor->end_message->adu_histogram; + end_message_received = true; + logger.Debug("Thread {} end message received", id); + } else if (msg->cbor->data_message) { + DataMessage message = msg->cbor->data_message.value(); + OnImage(id, message, msg->received); + } + UpdateMaxDelay(image_puller.GetCurrentFifoUtilization()); + current_status.SetProgress(GetProgress()); + current_status.SetStatus(GetStatus()); + } catch (const JFJochException &e) { + logger.ErrorException(e); + Cancel(e); + } catch (const std::exception &e) { + logger.Error("Cancelling data collection due to exception: {}", e.what()); + cancelled = true; + } + } +} + +void JFJochReceiverStream::CheckIfImagesSent() { + if (!image_buffer.CheckIfBufferReturned(SendFinalizeTimeout)) { + logger.Error("Send commands not finalized in {} seconds", SendFinalizeTimeout.count()); + throw JFJochException(JFJochExceptionCategory::ZeroMQ, "Send commands not finalized in " + + std::to_string(SendFinalizeTimeout.count()) + " seconds"); + } +} + +void JFJochReceiverStream::WriterFinalize() { + if (push_images_to_writer) + writer_error = image_pusher.Finalize(); +} + +void JFJochReceiverStream::FinishCollection() { + current_status.SetProgress(1.0); + current_status.SetStatus(GetStatus()); + + SendEndMessage(); + CheckIfImagesSent(); + logger.Debug("All images sent"); + + WriterFinalize(); + logger.Debug("Writing process finalized"); + + end_time = std::chrono::system_clock::now(); + + current_status.SetProgress({}); + current_status.SetStatus(GetStatus()); +} + +void JFJochReceiverStream::StopReceiver() { + if (measurement.valid()) { + measurement.get(); + logger.Debug("Receiver stopped"); + } +} + +void JFJochReceiverStream::Cancel(bool silent) { + JFJochReceiver::Cancel(silent); + // A silent cancel is not a request to abort, it says the detector has finished. It bounds the wait for the start message. + detector_finished = true; +} + +void JFJochReceiverStream::AddToEndMessage(EndMessage &message) { + if (!end_adu_histogram.empty()) + message.adu_histogram = end_adu_histogram; +} + +// With the card summing, an image is many frames. +int64_t JFJochReceiverStream::ExpectedImages() const { + return experiment.GetImageNum(); +} + +float JFJochReceiverStream::GetEfficiency() const { + if (ExpectedImages() == 0) + return 0; + return static_cast(images_collected) / static_cast(ExpectedImages()); +} + +float JFJochReceiverStream::GetProgress() const { + if (ExpectedImages() == 0) + return 0.0; + return static_cast(max_image_number_received) / static_cast(ExpectedImages()); +} diff --git a/receiver/JFJochReceiverStream.h b/receiver/JFJochReceiverStream.h new file mode 100644 index 000000000..70e768a36 --- /dev/null +++ b/receiver/JFJochReceiverStream.h @@ -0,0 +1,83 @@ +// SPDX-FileCopyrightText: 2026 Filip Leonarski, Paul Scherrer Institute +// SPDX-License-Identifier: GPL-3.0-only + +#pragma once + +#include +#include +#include +#include +#include + +#include "JFJochReceiver.h" +#include "../image_puller/ImagePuller.h" + +// A collection coming from an ImagePuller: waits for the start message, takes the images on a number of +// threads until the end message, then finishes the collection. What is done with an image is the child's. +// The most-derived class starts the threads as the last statement of its constructor and joins them as the +// first of its destructor, nothing here does. +class JFJochReceiverStream : public JFJochReceiver { +protected: + using JFJochReceiver::Cancel; + + ImagePuller &image_puller; + std::atomic end_message_received{false}; + + // Set when the caller reports the run is over - JFJochServices::Stop cancels the receiver silently once + // the detector reads back idle. A silent cancel does not interrupt images that are still being processed. + std::atomic detector_finished{false}; + + // How long a start message may still be in flight after the detector reported it had finished + constexpr static auto StartMessageGrace = std::chrono::seconds(5); + + const int64_t nthreads; + std::latch measurement_started; +private: + std::latch images_started; + std::vector> image_futures; + std::future measurement; + std::map> end_adu_histogram; + + void JoinFuture(std::future &future); + void MeasurementThread(); +protected: + void StartThreads(); + void JoinThreads(); + void WaitForImages(); // what an image thread does before it takes images + void PollImages(uint32_t id); // the rest of an image thread, until the end message + void CheckIfImagesSent(); // throws if the pusher did not return every image buffer it held + void WriterFinalize(); + + virtual void ImageThread(uint32_t id); + virtual void OnStart(std::chrono::steady_clock::time_point received); + virtual void OnImage(uint32_t id, DataMessage &message, std::chrono::steady_clock::time_point received) = 0; + virtual void FinishCollection(); // all images are in: end message, writer, status + virtual int64_t ExpectedImages() const; + + void AddToEndMessage(EndMessage &message) override; + uint64_t GetImagesDropped() const override { return image_puller.GetDroppedImages(); } +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, + 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); + + void StopReceiver() override; + void Cancel(bool silent) override; + + float GetEfficiency() const override; + float GetProgress() const override; +}; diff --git a/tests/FPGAImagePullerTest.cpp b/tests/FPGAImagePullerTest.cpp index 5d5ee48b0..2528828ba 100644 --- a/tests/FPGAImagePullerTest.cpp +++ b/tests/FPGAImagePullerTest.cpp @@ -21,6 +21,7 @@ #include "../writer/HDF5Objects.h" #include "../writer/StreamWriter.h" #include "../receiver/FPGAImagePuller.h" +#include "../receiver/JFJochReceiverRugnux.h" #include "../receiver/JFJochReceiverService.h" #include "../rugnux/RugnuxDefaults.h" @@ -670,6 +671,7 @@ TEST_CASE("FPGAImagePuller_RugnuxReceiver", "[FPGAImagePuller]") { CHECK(pusher.images.size() == 6); CHECK(pusher.sequence.front() == "start"); CHECK(pusher.sequence.back() == "end"); + CHECK(pusher.end.adu_histogram.contains("module0")); CHECK_FALSE(pusher.sequence[1].starts_with("calibration")); // the writer gets calibration only from the FPGA receiver const auto a = Pixels(pusher.images.at(0).message.image); const auto b = Pixels(pusher.images.at(3).message.image); @@ -736,6 +738,36 @@ TEST_CASE("FPGAImagePuller_NoStaleROI", "[FPGAImagePuller]") { CHECK(image.message.roi.empty()); } +// An image is many frames: efficiency and progress count images. +TEST_CASE("FPGAImagePuller_RugnuxReceiver_Summation", "[FPGAImagePuller]") { + auto x = Base(DetJF(1), 3, 2); + x.Conversion(); + x.Summation(2).Compression(CompressionAlgorithm::BSHUF_LZ4); + x.Goniometer(GoniometerAxis("omega", 0, 0.1, Coord(-1, 0, 0), std::nullopt)); + x.FilePrefix("fpga_puller_rugnux_sum"); + REQUIRE(x.GetImageNum() == 3); + REQUIRE(x.GetFrameNum() == 6); + + auto devices = Devices(x); + Logger logger("FPGAImagePuller_RugnuxSum"); + CollectingPusher pusher; + ImageBuffer buffer(256 * 1024 * 1024); + JFJochReceiverCurrentStatus status; + JFJochReceiverPlots plots; + RugnuxLiveStatus live; + PixelMask mask(x); + const JFCalibration calib = Calibration(x); + + FPGAImagePuller puller(x, mask, &calib, devices, logger, 2); + JFJochReceiverRugnux receiver(x, mask, puller, pusher, logger, 2, 2, nullptr, 30, live, Spots(), status, plots, + buffer); + receiver.StopReceiver(); + puller.Stop(); + CHECK(receiver.GetEfficiency() == 1.0f); + CHECK(receiver.GetProgress() == 1.0f); + CHECK(pusher.images.size() == 3); +} + // The real writer: the master file states what the card did, not what a DECTRIS stream implies. TEST_CASE("FPGAImagePuller_Writer", "[FPGAImagePuller]") { RegisterHDF5Filter(); diff --git a/tests/JFJochReceiverRugnuxTest.cpp b/tests/JFJochReceiverRugnuxTest.cpp index 316b5a9da..d0c591705 100644 --- a/tests/JFJochReceiverRugnuxTest.cpp +++ b/tests/JFJochReceiverRugnuxTest.cpp @@ -464,6 +464,66 @@ TEST_CASE("JFJochReceiverRugnux_Rotation", "[JFJochReceiverRugnux]") { CHECK(live.state != RugnuxLiveState::Running); } +TEST_CASE("JFJochReceiverRugnux_Rotation_Cancel", "[JFJochReceiverRugnux]") { + Logger logger(Catch::getResultCapture().getCurrentTestName()); + RegisterHDF5Filter(); + + const int64_t images = 5; + DiffractionExperiment x = StillsExperiment(images); + x.Goniometer(GoniometerAxis("omega", 0.0f, 0.1f, {1, 0, 0}, {})); + PixelMask mask(x); + + RecordingPusher pusher; + AcquisitionDeviceGroup group; + JFJochReceiverService service(group, logger, pusher); + service.NumThreads(4); + service.Indexing(x.GetIndexingSettings()); + service.RugnuxLive(true, nullptr, 4, 60); + + SECTION("Before the start message") { + service.Start(x, mask, nullptr, std::make_shared()); + service.Cancel(false); + const auto out = service.Stop(); + CHECK(out.status.cancelled); + CHECK(pusher.Events().empty()); + } + SECTION("Detector finished, no start message") { + service.Start(x, mask, nullptr, std::make_shared()); + service.Cancel(true); + const auto out = service.Stop(); + CHECK(out.status.cancelled); + CHECK(pusher.Events().empty()); + } + SECTION("During the collection") { + Stream stream(x, images); + stream.Frames(0, 2); + service.Start(x, mask, nullptr, stream.puller); + WaitForStatus(service, [](const RugnuxLiveSnapshot &s) { return s.images_received == 2; }); + service.Cancel(false); + const auto out = service.Stop(); + CHECK(out.status.cancelled); + CHECK(out.status.images_sent == 2); + CHECK(pusher.Events() == std::vector{"start", "image", "end"}); + const auto live = service.GetRugnuxStatus(); + CHECK(live.cut == "cancel"); + CHECK(live.state != RugnuxLiveState::Running); + } + SECTION("A silent cancel lets the collection finish") { + Stream stream(x, images); + stream.Frames(0, 2); + service.Start(x, mask, nullptr, stream.puller); + WaitForStatus(service, [](const RugnuxLiveSnapshot &s) { return s.images_received == 2; }); + service.Cancel(true); + stream.Frames(2, images - 2); + stream.End(); + const auto out = service.Stop(); + CHECK(!out.status.cancelled); + CHECK(out.status.images_sent == images); + CHECK(pusher.Events() == std::vector{"start", "image", "end"}); + CHECK(service.GetRugnuxStatus().cut.empty()); + } +} + 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());