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