From 61744c1133d4cb42c39e53bb29cfde6cf0be87b0 Mon Sep 17 00:00:00 2001 From: Filip Leonarski Date: Thu, 5 Nov 2026 20:39:20 +0100 Subject: [PATCH] receiver: live stills report last image -> all written, and Done once written The serial (FPGA + GPU) live path now times the latency from the last image to the end message out and the writer finished, as the rotation receiver does: StillsGPUAnalysis::Written, called by FinalizeMeasurement after the writer is finalised, sets last_image_to_written_s and only then the run's outcome. Until then the status shows a "Writing" step, so the run no longer reads Done (and its elapsed time no longer stops) before the end message and the writer are through. The timeline's latency bracket and the headline latency now show for serial collections. The END message is serialised into a scratch buffer first, as on rotation: merged data that would not fit the END buffer are dropped with a warning instead of failing the collection. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01UqXYdL4qBGYRiSXSGXfCLu --- receiver/JFJochReceiverFPGA.cpp | 25 +++++++++++++++++++++++-- receiver/JFJochReceiverFPGA.h | 1 + receiver/StillsGPUAnalysis.cpp | 19 ++++++++++++++++--- receiver/StillsGPUAnalysis.h | 6 ++++++ tests/JFJochReceiverProcessingTest.cpp | 7 +++++++ tests/StillsGPUAnalysisTest.cpp | 15 +++++++++++++-- 6 files changed, 66 insertions(+), 7 deletions(-) diff --git a/receiver/JFJochReceiverFPGA.cpp b/receiver/JFJochReceiverFPGA.cpp index 818e2e9f8..80c03c117 100644 --- a/receiver/JFJochReceiverFPGA.cpp +++ b/receiver/JFJochReceiverFPGA.cpp @@ -26,7 +26,7 @@ JFJochReceiverFPGA::JFJochReceiverFPGA(const DiffractionExperiment &in_experimen ZMQMetadataSocket *in_zmq_metadata_socket, IndexerThreadPool *indexing_thread_pool, bool in_gpu_analysis, - RugnuxLiveStatus *rugnux_status) + RugnuxLiveStatus *in_rugnux_status) : JFJochReceiver(in_experiment, in_send_buf_ctrl, in_image_sender, @@ -48,7 +48,8 @@ JFJochReceiverFPGA::JFJochReceiverFPGA(const DiffractionExperiment &in_experimen frame_transformation_ready((experiment.GetImageNum() > 0) ? frame_transformation_nthreads : 0), data_acquisition_ready(ndatastreams), gpu_analysis(in_gpu_analysis && experiment.GetImageNum() > 0 - && experiment.GetDetectorMode() == DetectorMode::Standard) { + && experiment.GetDetectorMode() == DetectorMode::Standard), + rugnux_status(in_rugnux_status) { // The FPGA integration core has a fixed bin-count limit. Reject configurations it cannot // handle before starting acquisition, unless azimuthal integration is forced onto the CPU. const auto &azint_settings = experiment.GetAzimuthalIntegrationSettings(); @@ -707,6 +708,9 @@ void JFJochReceiverFPGA::FinalizeMeasurement() { if (push_images_to_writer) writer_error = image_pusher.Finalize(); + // Last image -> all written, as the rugnux receiver reports it + if (gpu) + gpu->Written(); current_status.SetProgress({}); current_status.SetStatus(GetStatus()); @@ -861,4 +865,21 @@ void JFJochReceiverFPGA::AddToEndMessage(EndMessage &message) { if (const auto &m = gpu->GetMerged(); m && gpu->GetMergeExperiment().GetUnitCell()) message.merged = MergedMtzTable(m->merge.merged, *gpu->GetMergeExperiment().GetUnitCell(), gpu->GetMergeExperiment(), OutputProvenance{}); + if (!message.merged) + return; + // Every pusher serialises END into a buffer of MESSAGE_SIZE_FOR_START_END, and an END that does not fit there + // fails the collection. So it is serialised here first, as JFJochReceiverRugnux does; one that does not fit + // goes without the merged data. + const auto scratch = std::make_unique_for_overwrite(MESSAGE_SIZE_FOR_START_END); + CBORStream2Serializer serializer(scratch.get(), MESSAGE_SIZE_FOR_START_END); + try { + serializer.SerializeSequenceEnd(message); + } catch (const JFJochException &e) { + const std::string text = fmt::format("the merged data do not fit in the end message ({} MiB) and are not " + "written", MESSAGE_SIZE_FOR_START_END / (1024 * 1024)); + logger.Error("GPU analysis: {} - {}", text, e.what()); + if (rugnux_status) + rugnux_status->Warning(text); + message.merged.reset(); + } } diff --git a/receiver/JFJochReceiverFPGA.h b/receiver/JFJochReceiverFPGA.h index 6a2028f43..167fe50af 100644 --- a/receiver/JFJochReceiverFPGA.h +++ b/receiver/JFJochReceiverFPGA.h @@ -90,6 +90,7 @@ class JFJochReceiverFPGA : public JFJochReceiver { // Serial data with rugnux on: the converted frames are analysed on the GPU (StillsGPUAnalysis), and // the start message waits for its pre-pass. const bool gpu_analysis; + RugnuxLiveStatus *const rugnux_status; std::unique_ptr gpu; void GPUSink(DataMessage &message, const AzimuthalIntegrationProfile &profile); void AddToEndMessage(EndMessage &message) override; diff --git a/receiver/StillsGPUAnalysis.cpp b/receiver/StillsGPUAnalysis.cpp index c45d603dc..e5ccebe65 100644 --- a/receiver/StillsGPUAnalysis.cpp +++ b/receiver/StillsGPUAnalysis.cpp @@ -44,6 +44,8 @@ StillsGPUAnalysis::~StillsGPUAnalysis() { } catch (const std::exception &e) { logger.Error("GPU analysis: {}", e.what()); } + if (status && outcome) + status->Finish(outcome->first, outcome->second); } void StillsGPUAnalysis::AnalysisError(const std::string &text) { @@ -108,7 +110,7 @@ void StillsGPUAnalysis::Finish() { for (auto &f: analysis_futures) f.get(); if (status) { - status->LastImage(); + last_image = status->LastImage(); Publish(); } if (merge_future.valid()) { @@ -126,17 +128,28 @@ void StillsGPUAnalysis::Finish() { if (setup_error) status->Finish(RugnuxLiveState::Failed, "GPU analysis setup failed"); else { + // Reported by Written, once everything is written - the end message and the writer go on meanwhile. + status->Step("Writing"); std::unique_lock ul(error_mutex); if (analysis_error.empty()) - status->Finish(RugnuxLiveState::Done); + outcome.emplace(RugnuxLiveState::Done, ""); else - status->Finish(RugnuxLiveState::Failed, analysis_error); + outcome.emplace(RugnuxLiveState::Failed, analysis_error); } } if (setup_error) std::rethrow_exception(setup_error); } +void StillsGPUAnalysis::Written() { + if (!status || !outcome) + return; + if (last_image) + status->Latency(std::chrono::duration(std::chrono::steady_clock::now() - last_image.value()).count()); + status->Finish(outcome->first, outcome->second); + outcome.reset(); +} + void StillsGPUAnalysis::PrePassThread() { std::map> sample; { diff --git a/receiver/StillsGPUAnalysis.h b/receiver/StillsGPUAnalysis.h index e87160b82..d6350127f 100644 --- a/receiver/StillsGPUAnalysis.h +++ b/receiver/StillsGPUAnalysis.h @@ -82,6 +82,9 @@ public: // No frame comes after this: the pre-pass runs on what came, if it has not, and every frame is // analysed before it returns. Throws where the setup after the pre-pass failed. void Finish(); + // After Finish, once the end message is out and the writer is done: the status gets the latency from the last + // image and then the run's outcome. Without it the outcome is reported when the analysis goes, with no latency. + void Written(); // After Finish. [[nodiscard]] std::optional GetConsensusUnitCell() const; @@ -109,6 +112,9 @@ private: RugnuxLiveStills live; std::chrono::steady_clock::time_point ready_time; void Publish(); + // Kept by Finish for Written + std::optional last_image; + std::optional> outcome; Setup setup; std::unique_ptr indexer; diff --git a/tests/JFJochReceiverProcessingTest.cpp b/tests/JFJochReceiverProcessingTest.cpp index 8cd8d6bea..0b949f197 100644 --- a/tests/JFJochReceiverProcessingTest.cpp +++ b/tests/JFJochReceiverProcessingTest.cpp @@ -182,6 +182,13 @@ TEST_CASE("JFJochIntegrationTest_ZMQ_crystal_spot_and_index_GPUStills", "[JFJoch CHECK(receiver_out.status.images_sent == experiment.GetImageNum()); CHECK(!receiver_out.status.cancelled); + // Done once everything is written, with the latency from the last image to then + const auto live = service.GetRugnuxStatus(); + CHECK(live.state == RugnuxLiveState::Done); + REQUIRE(live.last_image_to_written_s); + REQUIRE(live.last_image_s); + CHECK(live.elapsed_s >= live.last_image_s.value() + live.last_image_to_written_s.value() - 1e-6); + // No progress value at the end of measurement REQUIRE(!service.GetProgress().has_value()); diff --git a/tests/StillsGPUAnalysisTest.cpp b/tests/StillsGPUAnalysisTest.cpp index 8329f31b9..ef803a2e7 100644 --- a/tests/StillsGPUAnalysisTest.cpp +++ b/tests/StillsGPUAnalysisTest.cpp @@ -110,7 +110,18 @@ TEST_CASE("StillsGPUAnalysis_FailingFrameDoesNotStopAnalysis", "[StillsGPUAnalys REQUIRE(gpu.Hold(Frame(buffer, i))); REQUIRE_NOTHROW(gpu.Finish()); CHECK(sunk == 3); - CHECK(status.Get().state == RugnuxLiveState::Failed); - CHECK(!status.Get().message.empty()); + // The outcome waits for everything to be written, and comes after the latency. + CHECK(status.Get().state == RugnuxLiveState::Running); + CHECK(!status.Get().last_image_to_written_s); + gpu.Written(); + const auto s = status.Get(); + CHECK(s.state == RugnuxLiveState::Failed); + CHECK(!s.message.empty()); + REQUIRE(s.last_image_to_written_s); + REQUIRE(s.last_image_s); + CHECK(s.elapsed_s >= s.last_image_s.value() + s.last_image_to_written_s.value() - 1e-6); + REQUIRE(!s.steps.empty()); + CHECK(s.steps.back().title == "Writing"); + CHECK(s.steps.back().duration_s); CHECK(buffer.GetStatus().preparation_slots == 0); }