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());