diff --git a/receiver/JFJochReceiverRugnuxStills.cpp b/receiver/JFJochReceiverRugnuxStills.cpp index 032a53a14..7f43a5d4a 100644 --- a/receiver/JFJochReceiverRugnuxStills.cpp +++ b/receiver/JFJochReceiverRugnuxStills.cpp @@ -6,6 +6,8 @@ #include #include +#include "../common/time_utc.h" + JFJochReceiverRugnuxStills::JFJochReceiverRugnuxStills(const DiffractionExperiment &in_experiment, const PixelMask &in_pixel_mask, ImagePuller &in_image_puller, @@ -46,25 +48,7 @@ void JFJochReceiverRugnuxStills::OnStart(std::chrono::steady_clock::time_point r indexing_thread_pool, stills_nthreads, // What the pre-pass holds and fits on; the frames beyond it wait in the image buffer as well. slots / 2, logger, - [this](const StillsGPUAnalysis::Setup &s) { - // The profiles are binned on the pre-pass's geometry. - plots.Setup(s.experiment, *s.mapping); - SendStartMessage(MakeStartMessage(s.experiment, s.mask, *s.mapping)); - }, - [this](DataMessage &m, const AzimuthalIntegrationProfile &p) { SendAnalysedImage(m, p); }, - [this] { - // The broker's settings, with the spot finder rugnux uses on stills (the pre-pass decided the - // centre and mask with it): adaptive threshold, single-pixel spots kept. With the broker's - // default fixed threshold 3 points fewer frames index. - SpotFindingSettings settings = GetSpotFindingSettings(); - settings.adaptive_threshold = true; - settings.min_pix_per_spot = std::nullopt; - return settings; - }, - // The start message did not go out (or the analysis could not be set up): nothing can be - // written, so the collection stops. - [this](const JFJochException &e) { Cancel(e); }, - &live_status, true, model, deadline_s); + *this, live_status, true, model, deadline_s); std::unique_lock ul(stills_mutex); stills = std::move(analysis); if (cut) @@ -130,12 +114,34 @@ void JFJochReceiverRugnuxStills::FinishCollection() { std::rethrow_exception(error); } WriterFinalize(); - stills->Written(); + if (const auto last_image = live_status.LastImage()) + live_status.Latency(SecondsSince(last_image.value())); + stills->ReportOutcome(); current_status.SetProgress({}); current_status.SetStatus(GetStatus()); } +void JFJochReceiverRugnuxStills::StillsReady(const StillsSetup &setup) { + // The profiles are binned on the pre-pass's geometry. + plots.Setup(setup.experiment, *setup.mapping); + SendStartMessage(MakeStartMessage(setup.experiment, setup.mask, *setup.mapping)); +} + +SpotFindingSettings JFJochReceiverRugnuxStills::AnalysisSettings() { + // The broker's settings, with the spot finder rugnux uses on stills (the pre-pass decided the centre and mask + // with it): adaptive threshold, single-pixel spots kept. + SpotFindingSettings settings = GetSpotFindingSettings(); + settings.adaptive_threshold = true; + settings.min_pix_per_spot = std::nullopt; + return settings; +} + +// The start message did not go out (or the analysis could not be set up): nothing can be written, so the collection stops. +void JFJochReceiverRugnuxStills::StillsSetupFailed(const JFJochException &error) { + Cancel(error); +} + ZeroCopyReturnValue *JFJochReceiverRugnuxStills::AnalysedImageSlot() { // An analysed frame is not dropped because the writer lags behind the catch-up from a full buffer: the analysis // waits for the slots being sent to come back, and the frames arriving meanwhile find the buffer full. A diff --git a/receiver/JFJochReceiverRugnuxStills.h b/receiver/JFJochReceiverRugnuxStills.h index 23a9c9b45..8541456c6 100644 --- a/receiver/JFJochReceiverRugnuxStills.h +++ b/receiver/JFJochReceiverRugnuxStills.h @@ -14,7 +14,7 @@ // A serial collection analysed frame by frame (StillsGPUAnalysis): each image the puller delivers waits in the // image buffer for the pre-pass, is analysed at the geometry it decides, and goes to the writer with its analysis. // The start message waits for the pre-pass, the end message carries the last merge and model step. -class JFJochReceiverRugnuxStills : public JFJochReceiverStream { +class JFJochReceiverRugnuxStills : public JFJochReceiverStream, public StillsSink { using JFJochReceiverStream::Cancel; const std::shared_ptr model; @@ -33,7 +33,10 @@ class JFJochReceiverRugnuxStills : public JFJochReceiverStream { void OnImage(uint32_t id, DataMessage &message, std::chrono::steady_clock::time_point received) override; void FinishCollection() override; ZeroCopyReturnValue *AnalysedImageSlot(); - void SendAnalysedImage(DataMessage &message, const AzimuthalIntegrationProfile &profile); + void StillsReady(const StillsSetup &setup) override; + void SendAnalysedImage(DataMessage &message, const AzimuthalIntegrationProfile &profile) override; + SpotFindingSettings AnalysisSettings() override; + void StillsSetupFailed(const JFJochException &error) override; void Cut(const std::string &reason); void AddToEndMessage(EndMessage &message) override; public: diff --git a/receiver/StillsGPUAnalysis.cpp b/receiver/StillsGPUAnalysis.cpp index f42350589..fa64b1c37 100644 --- a/receiver/StillsGPUAnalysis.cpp +++ b/receiver/StillsGPUAnalysis.cpp @@ -16,26 +16,22 @@ StillsGPUAnalysis::StillsGPUAnalysis(const DiffractionExperiment &in_experiment, const PixelMask &in_mask, const StartMessage &in_start, IndexerThreadPool *in_indexing_thread_pool, - int64_t nthreads, int64_t max_held, Logger &in_logger, Ready in_ready, - Sink in_sink, Settings in_settings, Failed in_failed, - RugnuxLiveStatus *in_status, bool in_merge, + int64_t nthreads, int64_t max_held, Logger &in_logger, StillsSink &in_sink, + RugnuxLiveStatus &in_status, bool in_merge, std::shared_ptr in_model, double in_deadline_s) : experiment(in_experiment), mask(in_mask), start(in_start), indexing_thread_pool(in_indexing_thread_pool), - logger(in_logger), ready(std::move(in_ready)), sink(std::move(in_sink)), settings(std::move(in_settings)), - setup_failed(std::move(in_failed)), status(in_status), merge_frames(in_merge), + logger(in_logger), sink(in_sink), status(in_status), merge_frames(in_merge), hold_limit(std::max(1, max_held)), model(std::move(in_model)), deadline_s(in_deadline_s) { live.images_needed = std::min({PREPASS_IMAGES, experiment.GetImageNum(), hold_limit}); live.file_beam_x_pxl = experiment.GetBeamX_pxl(); live.file_beam_y_pxl = experiment.GetBeamY_pxl(); live.file_distance_mm = experiment.GetDetectorDistance_mm(); - if (status) { - status->Start(experiment.GetImageNum(), - static_cast(experiment.GetImageNum()) - * std::chrono::duration(experiment.GetImageTime()).count(), - std::chrono::steady_clock::now()); - status->Step("Pre-pass"); - Publish(); - } + status.Start(experiment.GetImageNum(), + static_cast(experiment.GetImageNum()) + * std::chrono::duration(experiment.GetImageTime()).count(), + std::chrono::steady_clock::now()); + status.Step("Pre-pass"); + Publish(); prepass_future = std::async(std::launch::async, &StillsGPUAnalysis::PrePassThread, this); for (int64_t i = 0; i < nthreads; i++) analysis_futures.emplace_back(std::async(std::launch::async, &StillsGPUAnalysis::AnalysisThread, this)); @@ -44,14 +40,13 @@ StillsGPUAnalysis::StillsGPUAnalysis(const DiffractionExperiment &in_experiment, StillsGPUAnalysis::~StillsGPUAnalysis() { try { - if (!finished) + if (prepass_future.valid()) Finish(); } catch (const std::exception &e) { logger.Error("GPU analysis: {}", e.what()); } StopModel(true); - if (status && outcome) - status->Finish(outcome->first, outcome->second); + ReportOutcome(); } void StillsGPUAnalysis::AnalysisError(const std::string &text) { @@ -68,56 +63,51 @@ void StillsGPUAnalysis::Publish() { if (t > 0 && live.images_analysed > 0) live.analysis_rate_hz = static_cast(live.images_analysed) / t; } - status->Stills(live); + status.Stills(live); } void StillsGPUAnalysis::Hold(const std::shared_ptr &image) { - if (status) - status->ImageReceived(std::chrono::steady_clock::now()); + status.ImageReceived(std::chrono::steady_clock::now()); queue.PutBlocking(image); { std::unique_lock ul(prepass_mutex); - if (!prepass_over && static_cast(prepass_images.size()) < hold_limit) { + if (prepass_state != PrePassState::Over && static_cast(prepass_images.size()) < hold_limit) { prepass_images[image->message.number] = image; std::unique_lock ul_live(live_mutex); live.images_held = static_cast(prepass_images.size()); } } prepass_cv.notify_all(); - if (status) - Publish(); + Publish(); } void StillsGPUAnalysis::StartPrePass() { { std::unique_lock ul(prepass_mutex); - prepass_last = true; + if (prepass_state == PrePassState::Holding) + prepass_state = PrePassState::Closing; } prepass_cv.notify_all(); } void StillsGPUAnalysis::Finish() { const auto finish_start = std::chrono::steady_clock::now(); - finished = true; StartPrePass(); prepass_future.get(); for (size_t i = 0; i < analysis_futures.size(); i++) queue.PutBlocking(nullptr); for (auto &f: analysis_futures) f.get(); - if (status) { - last_image = status->LastImage(); - Publish(); - } + const auto last_image = status.LastImage(); + Publish(); if (merge_future.valid()) { - if (status) - status->Step("Merge"); + status.Step("Merge"); merge_queue.Put(-1); merge_future.get(); - if (merged && status) { + if (merged) { PublishMerge(*merged, "the collection"); - status->Headline(merge_experiment.GetSpaceGroupOrP1().xhm(), merged->merge.statistics.overall.d_min, - std::nullopt); + status.Headline(merge_experiment.GetSpaceGroupOrP1().xhm(), merged->merge.statistics.overall.d_min, + std::nullopt); } } if (model_future.valid()) { @@ -126,13 +116,17 @@ void StillsGPUAnalysis::Finish() { if (merged && model_left_out.empty()) { const auto deadline = last_image.value_or(finish_start) + std::chrono::duration_cast< std::chrono::steady_clock::duration>(std::chrono::duration(deadline_s)); - const auto cut = SubmitModel(*merged, "the collection", true); + SubmitModel(*merged, "the collection", true); StopModel(false); if (deadline_s > 0 && model_future.wait_until(deadline) == std::future_status::timeout) { - *cut = true; + { + std::unique_lock ul(model_mutex); + step_cut = true; + if (model_next) + model_next->cut = true; + } logger.Warning("Model step cut at the deadline, {:.1f} s after the last image", deadline_s); - if (status) - status->Cut("deadline"); + status.Cut("deadline"); } model_future.get(); } else @@ -140,29 +134,23 @@ void StillsGPUAnalysis::Finish() { std::unique_lock ul(model_mutex); model_over = true; } - if (status) { - 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()) - outcome.emplace(RugnuxLiveState::Done, ""); - else - outcome.emplace(RugnuxLiveState::Failed, analysis_error); - } - } - if (setup_error) + if (setup_error) { + status.Finish(RugnuxLiveState::Failed, "GPU analysis setup failed"); std::rethrow_exception(setup_error); + } + // Reported by ReportOutcome, 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()) + outcome.emplace(RugnuxLiveState::Done, ""); + else + outcome.emplace(RugnuxLiveState::Failed, analysis_error); } -void StillsGPUAnalysis::Written() { - if (!status || !outcome) +void StillsGPUAnalysis::ReportOutcome() { + if (!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); + status.Finish(outcome->first, outcome->second); outcome.reset(); } @@ -185,10 +173,10 @@ void StillsGPUAnalysis::PrePassThread() { { std::unique_lock ul(prepass_mutex); prepass_cv.wait(ul, [&] { - return prepass_last || static_cast(prepass_images.size()) >= needed; + return prepass_state != PrePassState::Holding || static_cast(prepass_images.size()) >= needed; }); sample = prepass_images; - last = prepass_last || static_cast(sample.size()) >= most_held; + last = prepass_state != PrePassState::Holding || static_cast(sample.size()) >= most_held; } if (sample.empty()) break; @@ -196,8 +184,7 @@ void StillsGPUAnalysis::PrePassThread() { std::unique_lock ul(live_mutex); live.prepass = RugnuxLivePrePass::Running; } - if (status) - Publish(); + Publish(); try { const auto t0 = std::chrono::steady_clock::now(); auto r = RunStillsPrePass(start, sample, static_cast(std::thread::hardware_concurrency())); @@ -222,7 +209,7 @@ void StillsGPUAnalysis::PrePassThread() { { // The frames are analysed from here on: their slots go back to the buffer once they are sent. std::unique_lock ul(prepass_mutex); - prepass_over = true; + prepass_state = PrePassState::Over; prepass_images.clear(); } const bool failed = !result; @@ -256,8 +243,7 @@ void StillsGPUAnalysis::PrePassThread() { setup.experiment.GetDetectorDistance_mm(), setup.experiment.GetBeamX_pxl(), setup.experiment.GetBeamY_pxl()); logger.Warning("{}", text); - if (status) - status->Note(text); + status.Note(text); } if (merge_frames) { merge_experiment = setup.experiment; @@ -271,7 +257,7 @@ void StillsGPUAnalysis::PrePassThread() { model_future = std::async(std::launch::async, &StillsGPUAnalysis::ModelThread, this); merge_future = std::async(std::launch::async, &StillsGPUAnalysis::MergeThread, this); } - ready(setup); + sink.StillsReady(setup); } catch (const std::exception &e) { // Nothing can be written without the start message: the frames are dropped (AnalysisThread), and // the collection is stopped. @@ -280,8 +266,7 @@ void StillsGPUAnalysis::PrePassThread() { std::string("GPU analysis setup failed: ") + e.what()); logger.ErrorException(error); setup_error = j ? std::current_exception() : std::make_exception_ptr(error); - if (setup_failed) - setup_failed(error); + sink.StillsSetupFailed(error); prepass_done.count_down(); return; } @@ -291,10 +276,8 @@ void StillsGPUAnalysis::PrePassThread() { live.prepass_s = std::chrono::duration(std::chrono::steady_clock::now() - prepass_start).count(); ready_time = std::chrono::steady_clock::now(); } - if (status) { - status->Step("GPU analysis"); - Publish(); - } + status.Step("GPU analysis"); + Publish(); prepass_done.count_down(); } @@ -329,7 +312,7 @@ void StillsGPUAnalysis::AnalysisThread() { profile.emplace(*setup.mapping); try { if (analysis) - analysis->AnalyzeStreamImage(message, *profile, settings()); + analysis->AnalyzeStreamImage(message, *profile, sink.AnalysisSettings()); } catch (const JFJochException &e) { // The frame is still written, without its analysis. logger.ErrorException(e); @@ -339,20 +322,18 @@ void StillsGPUAnalysis::AnalysisThread() { message.processing_time_s = std::chrono::duration(std::chrono::steady_clock::now() - start_time).count(); try { - sink(message, *profile); + sink.SendAnalysedImage(message, *profile); if (merge) merge_queue.Put(message.number); - if (status) { - int64_t analysed; - { - std::unique_lock ul(live_mutex); - analysed = ++live.images_analysed; - if (message.indexing_result.value_or(false)) - ++live.images_indexed; - } - status->Progress(analysed, experiment.GetImageNum()); - Publish(); + int64_t analysed; + { + std::unique_lock ul(live_mutex); + analysed = ++live.images_analysed; + if (message.indexing_result.value_or(false)) + ++live.images_indexed; } + status.Progress(analysed, experiment.GetImageNum()); + Publish(); } catch (const std::exception &e) { // The thread goes on with the next frame, so that the queue is drained. AnalysisError(fmt::format("image {} lost - {}", message.number, e.what())); @@ -405,8 +386,7 @@ void StillsGPUAnalysis::MergeThread() { if (!g.space_group) { model_left_out = "the model is left out: " + g.reason; logger.Warning("{}", model_left_out); - if (status) - status->Warning(model_left_out); + status.Warning(model_left_out); } else if (!merge_experiment.GetGemmiSpaceGroup()) { merge_experiment.SetSpaceGroup(*g.space_group); merge->SpaceGroup(*g.space_group); @@ -457,12 +437,10 @@ void StillsGPUAnalysis::PublishMerge(const StillsMergeOutput &out, const std::st ? 100.0 * o.unique_reflections / o.possible_unique_reflections : 0.0, o.mean_i_over_sigma, out.seconds); logger.Info("{}", text); - if (status) - status->Note(text); + status.Note(text); } -std::shared_ptr> StillsGPUAnalysis::SubmitModel(StillsMergeOutput &out, const std::string &title, - bool final) { +void StillsGPUAnalysis::SubmitModel(StillsMergeOutput &out, const std::string &title, bool final) { ModelJob job; // The merge of the collection is kept for the end message job.merged = final ? out.merge.merged : std::move(out.merge.merged); @@ -470,19 +448,14 @@ std::shared_ptr> StillsGPUAnalysis::SubmitModel(StillsMergeOut job.shells.push_back(sh.d_min); job.title = title; job.final = final; - const auto cut = job.cut; { std::unique_lock ul(model_mutex); - if (model_cut) - *job.cut = true; - if (model_running_cut) - *model_running_cut = true; + step_cut = true; if (model_next) logger.Info("Model step on the merge of {} left out: the merge of {} is there", model_next->title, title); model_next = std::move(job); } model_cv.notify_all(); - return cut; } void StillsGPUAnalysis::Cut(const std::string &reason) { @@ -493,14 +466,10 @@ void StillsGPUAnalysis::Cut(const std::string &reason) { if (model_cut || model_over) return; model_cut = true; - if (model_running_cut) - *model_running_cut = true; - if (model_next) - *model_next->cut = true; + step_cut = true; } logger.Info("Model step cut: {}", reason); - if (status) - status->Cut(reason); + status.Cut(reason); } void StillsGPUAnalysis::StopModel(bool cut) { @@ -511,8 +480,7 @@ void StillsGPUAnalysis::StopModel(bool cut) { model_stop = true; if (cut) { model_next.reset(); - if (model_running_cut) - *model_running_cut = true; + step_cut = true; } } model_cv.notify_all(); @@ -535,7 +503,7 @@ void StillsGPUAnalysis::ModelThread() { return; job = std::move(*model_next); model_next.reset(); - model_running_cut = job.cut; + step_cut = model_cut || job.cut; } try { ModelStepSettings settings; @@ -544,9 +512,9 @@ void StillsGPUAnalysis::ModelThread() { // The analysis's indexers stay on the card for the broker's life settings.gpu_resident_bytes = indexing_thread_pool ? indexing_thread_pool->DeviceBytes() : 0; settings.rfree_fraction = merge_experiment.GetScalingSettings().GetRfreeFraction(); - settings.cut = job.cut.get(); - if (job.final && status) - settings.step = [this](const std::string &title) { status->Step(title); }; + settings.cut = &step_cut; + if (job.final) + settings.step = [this](const std::string &title) { status.Step(title); }; const DiffractionExperiment &x = merge_experiment; ModelStepResult r = RunModelStep(job.merged, x.GetUnitCell().value(), x.GetSpaceGroupOrP1(), *model, x.GetWavelength_A(), @@ -580,17 +548,15 @@ void StillsGPUAnalysis::PublishModel(const ModelStepResult &r, const std::string text += "; " + l + " left out"; text += fmt::format(" ({:.1f} s placement, {:.1f} s refinement)", r.placement_s, r.refinement_s); logger.Info("{}", text); - if (!status) - return; - status->Note(text); + status.Note(text); if (const auto m = LiveModel(ModelStepReport(r))) - status->Model(*m); + status.Model(*m); if (!r.placed_model) return; std::optional refined; if (r.refinement && r.refinement->ok && r.refinement->model) refined = PlacedModelOutput{r.refinement->model, r.placed_model->cell, r.placed_model->space_group}; - PublishStructures(*status, *r.placed_model, r.maps.get(), refined ? &*refined : nullptr, + PublishStructures(status, *r.placed_model, r.maps.get(), refined ? &*refined : nullptr, refined && r.refinement->maps ? r.refinement->maps.get() : nullptr); } @@ -620,5 +586,5 @@ void StillsGPUAnalysis::FillEndMessage(EndMessage &message) const { message.merged = MergedMtzTable(m->merge.merged, *GetMergeExperiment().GetUnitCell(), GetMergeExperiment(), OutputProvenance{}); if (message.merged) - DropResultBlocksThatDoNotFit(message, status, logger); + DropResultBlocksThatDoNotFit(message, &status, logger); } diff --git a/receiver/StillsGPUAnalysis.h b/receiver/StillsGPUAnalysis.h index 46b5227d2..42b934d34 100644 --- a/receiver/StillsGPUAnalysis.h +++ b/receiver/StillsGPUAnalysis.h @@ -6,7 +6,6 @@ #include #include #include -#include #include #include #include @@ -26,15 +25,33 @@ #include "../rugnux/StillsLiveMerge.h" #include "../rugnux/ModelStep.h" +struct StillsSetup { + DiffractionExperiment experiment; + PixelMask mask; + std::unique_ptr mapping; +}; + +// What StillsGPUAnalysis needs from its owner (JFJochReceiverRugnuxStills). +class StillsSink { +public: + virtual ~StillsSink() = default; + // The pre-pass is over, no frame is analysed before this returns: the owner sends the start message. + virtual void StillsReady(const StillsSetup &setup) = 0; + // A frame is analysed; called from the analysis threads. + virtual void SendAnalysedImage(DataMessage &message, const AzimuthalIntegrationProfile &profile) = 0; + virtual SpotFindingSettings AnalysisSettings() = 0; + // The setup after the pre-pass failed, StillsReady included. + virtual void StillsSetupFailed(const JFJochException &error) = 0; +}; + // Serial frames analysed on the GPU (MXAnalysisWithoutFPGA), after a pre-pass on // the first PREPASS_IMAGES (RunStillsPrePass) has decided the beam centre and the mask they are analysed -// with. Every frame given before the pre-pass is done is held; `ready` is called once it is, with the -// geometry and mask the frames are then analysed at, before any of them is analysed; `sink` gets each -// frame once analysed, from one of the analysis threads. Where `status` is given, the run is reported there -// (pre-pass, its decisions, the analysis). +// with. Every frame given before the pre-pass is done is held; the sink is told once it is, with the +// geometry and mask the frames are then analysed at, and gets each frame once analysed. The run is reported in +// `status` (pre-pass, its decisions, the analysis). // Every frame given is taken and analysed: the slots of the image buffer are what bounds them. The pre-pass holds // and fits on at most `max_held` of them; those arriving beyond that, or while it runs, wait for the analysis. -// Where the setup after the pre-pass fails (`ready` included), `failed` is called at once, the frames are +// Where the setup after the pre-pass fails, the sink is told at once, the frames are // dropped unanalysed and Finish throws the error. An analysis thread that fails passes its frames on // unanalysed, and the run is reported as failed. // The indexed frames are merged as `rugnux --mode scale` merges them (StillsLiveMerge), beside the analysis: @@ -58,21 +75,10 @@ public: constexpr static std::array CHECKPOINTS = {0.5, 0.75}; constexpr static int MERGE_THREADS = 8; - struct Setup { - DiffractionExperiment experiment; - PixelMask mask; - std::unique_ptr mapping; - }; - using Ready = std::function; - using Sink = std::function; - using Settings = std::function; - using Failed = std::function; - // `start`: the collection's start message, as it would be sent without the pre-pass. StillsGPUAnalysis(const DiffractionExperiment &experiment, const PixelMask &mask, const StartMessage &start, IndexerThreadPool *indexing_thread_pool, int64_t nthreads, int64_t max_held, Logger &logger, - Ready ready, Sink sink, Settings settings, Failed failed = {}, - RugnuxLiveStatus *status = nullptr, bool merge = true, + StillsSink &sink, RugnuxLiveStatus &status, bool merge = true, std::shared_ptr model = nullptr, double deadline_s = 0); ~StillsGPUAnalysis(); @@ -80,9 +86,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, once the end message is out and the writer is done: the status gets the run's outcome. + // Without it the outcome is reported when the analysis goes. + void ReportOutcome(); // POST /cancel: the model step stops where it is, as at the deadline, and the frames are still analysed and merged. void Cut(const std::string &reason); @@ -106,28 +112,23 @@ private: const StartMessage start; IndexerThreadPool *indexing_thread_pool; Logger &logger; - Ready ready; - Sink sink; - Settings settings; - Failed setup_failed; - RugnuxLiveStatus *status; + StillsSink &sink; + RugnuxLiveStatus &status; const bool merge_frames; std::mutex live_mutex; RugnuxLiveStills live; std::chrono::steady_clock::time_point ready_time; void Publish(); - // Kept by Finish for Written - std::optional last_image; - std::optional> outcome; + std::optional> outcome; // kept by Finish for ReportOutcome - Setup setup; + StillsSetup setup; std::unique_ptr indexer; std::mutex prepass_mutex; std::condition_variable prepass_cv; - bool prepass_last = false; // what is held is all there will be - bool prepass_over = false; + enum class PrePassState {Holding, Closing, Over}; // Closing: what is held is all there will be + PrePassState prepass_state = PrePassState::Holding; std::map> prepass_images; std::latch prepass_done{1}; std::exception_ptr setup_error; // set before prepass_done @@ -140,7 +141,6 @@ private: ThreadSafeFIFO> queue; std::future prepass_future; std::vector> analysis_futures; - bool finished = false; // The frames analysed, by number, for the merge (-1 after the last); it takes their outcomes from the indexer. ThreadSafeFIFO merge_queue; @@ -164,20 +164,20 @@ private: std::vector shells; // the merge statistics' shell bounds std::string title; bool final = false; - std::shared_ptr> cut = std::make_shared>(false); + bool cut = false; }; std::mutex model_mutex; std::condition_variable model_cv; std::optional model_next; - std::shared_ptr> model_running_cut; // of the step running + std::atomic step_cut{false}; // the step running is cut bool model_stop = false; bool model_cut = false; // the steps from here on are cut too bool model_over = false; // the step on the merge of the collection has ended: nothing is left to cut std::optional model_placed; // the last step that placed the model, for the next to start from std::optional model_final; std::future model_future; - // The step on `out`, the merge of `title` (its reflections moved out, but for the final one); returns its cut. - std::shared_ptr> SubmitModel(StillsMergeOutput &out, const std::string &title, bool final); + // The step on `out`, the merge of `title` (its reflections moved out, but for the final one). + void SubmitModel(StillsMergeOutput &out, const std::string &title, bool final); void StopModel(bool cut); void ModelThread(); void PublishModel(const ModelStepResult &r, const std::string &title); diff --git a/tests/StillsGPUAnalysisTest.cpp b/tests/StillsGPUAnalysisTest.cpp index d4873afff..23c44e234 100644 --- a/tests/StillsGPUAnalysisTest.cpp +++ b/tests/StillsGPUAnalysisTest.cpp @@ -4,13 +4,26 @@ #include #include +#include #include #include "../common/ImageBuffer.h" +#include "../common/time_utc.h" #include "../image_analysis/structure_refinement/RugnuxModel.h" #include "../receiver/StillsGPUAnalysis.h" namespace { + struct TestSink : public StillsSink { + std::function ready = [](const StillsSetup &) {}; + std::function analysed = [](DataMessage &) {}; + std::function failed = [] {}; + + void StillsReady(const StillsSetup &setup) override { ready(setup); } + void SendAnalysedImage(DataMessage &message, const AzimuthalIntegrationProfile &) override { analysed(message); } + SpotFindingSettings AnalysisSettings() override { return DiffractionExperiment::DefaultDataProcessingSettings(); } + void StillsSetupFailed(const JFJochException &) override { failed(); } + }; + DiffractionExperiment StillsExperiment() { DiffractionExperiment x(DetJF(1)); x.ImagesPerTrigger(StillsGPUAnalysis::PREPASS_IMAGES + 200).NumTriggers(1).DetectorDistance_mm(100).BeamX_pxl(500).BeamY_pxl(250) @@ -44,14 +57,14 @@ TEST_CASE("StillsGPUAnalysis_StartFailureReleasesThreads", "[StillsGPUAnalysis]" RugnuxLiveStatus status; std::atomic failed = 0, sunk = 0; + TestSink sink; + sink.ready = [](const StillsSetup &) { + throw JFJochException(JFJochExceptionCategory::ZeroMQ, "Timeout on pushing start message"); + }; + sink.analysed = [&](DataMessage &) { ++sunk; }; + sink.failed = [&] { ++failed; }; { - StillsGPUAnalysis gpu(x, mask, Start(x), nullptr, 4, 100, logger, - [](const StillsGPUAnalysis::Setup &) { - throw JFJochException(JFJochExceptionCategory::ZeroMQ, "Timeout on pushing start message"); - }, - [&](DataMessage &, const AzimuthalIntegrationProfile &) { ++sunk; }, - [] { return DiffractionExperiment::DefaultDataProcessingSettings(); }, - [&](const JFJochException &) { ++failed; }, &status); + StillsGPUAnalysis gpu(x, mask, Start(x), nullptr, 4, 100, logger, sink, status); for (int i = 0; i < 3; i++) gpu.Hold(Frame(buffer, i)); REQUIRE_THROWS_AS(gpu.Finish(), JFJochException); @@ -71,12 +84,10 @@ TEST_CASE("StillsGPUAnalysis_HoldIsBounded", "[StillsGPUAnalysis]") { buffer.StartMeasurement(x); REQUIRE(buffer.GetStatus().total_slots >= 4); - StillsGPUAnalysis gpu(x, mask, Start(x), nullptr, 2, 3, logger, - [](const StillsGPUAnalysis::Setup &) { - throw JFJochException(JFJochExceptionCategory::ZeroMQ, "no writer"); - }, - [](DataMessage &, const AzimuthalIntegrationProfile &) {}, - [] { return DiffractionExperiment::DefaultDataProcessingSettings(); }); + RugnuxLiveStatus status; + TestSink sink; + sink.ready = [](const StillsSetup &) { throw JFJochException(JFJochExceptionCategory::ZeroMQ, "no writer"); }; + StillsGPUAnalysis gpu(x, mask, Start(x), nullptr, 2, 3, logger, sink, status); // Past what the pre-pass holds, frames are still taken: only the image buffer bounds them. for (int i = 0; i < 4; i++) gpu.Hold(Frame(buffer, i)); @@ -95,15 +106,13 @@ TEST_CASE("StillsGPUAnalysis_FailingFrameDoesNotStopAnalysis", "[StillsGPUAnalys RugnuxLiveStatus status; std::atomic sunk = 0; - StillsGPUAnalysis gpu(x, mask, Start(x), nullptr, 1, 100, logger, - [](const StillsGPUAnalysis::Setup &) {}, - [&](DataMessage &m, const AzimuthalIntegrationProfile &) { - ++sunk; - if (m.number == StillsGPUAnalysis::PREPASS_IMAGES) - throw std::runtime_error("not a JFJochException"); - }, - [] { return DiffractionExperiment::DefaultDataProcessingSettings(); }, - {}, &status); + TestSink sink; + sink.analysed = [&](DataMessage &m) { + ++sunk; + if (m.number == StillsGPUAnalysis::PREPASS_IMAGES) + throw std::runtime_error("not a JFJochException"); + }; + StillsGPUAnalysis gpu(x, mask, Start(x), nullptr, 1, 100, logger, sink, status); // The frames carry no image, so their analysis fails too - they are passed on all the same. for (int i = 0; i < 3; i++) gpu.Hold(Frame(buffer, i)); @@ -112,7 +121,8 @@ TEST_CASE("StillsGPUAnalysis_FailingFrameDoesNotStopAnalysis", "[StillsGPUAnalys // 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(); + status.Latency(SecondsSince(status.LastImage().value())); + gpu.ReportOutcome(); const auto s = status.Get(); CHECK(s.state == RugnuxLiveState::Failed); CHECK(!s.message.empty()); @@ -129,12 +139,10 @@ TEST_CASE("StillsGPUAnalysis_CancelCutsTheModelStep", "[StillsGPUAnalysis]") { Logger logger(Catch::getResultCapture().getCurrentTestName()); const DiffractionExperiment x = StillsExperiment(); const PixelMask mask(x); + TestSink sink; const auto run = [&](RugnuxLiveStatus &status, std::shared_ptr model) { - return std::make_unique( - x, mask, Start(x), nullptr, 1, 100, logger, [](const StillsGPUAnalysis::Setup &) {}, - [](DataMessage &, const AzimuthalIntegrationProfile &) {}, - [] { return DiffractionExperiment::DefaultDataProcessingSettings(); }, - StillsGPUAnalysis::Failed{}, &status, true, std::move(model), 60.0); + return std::make_unique(x, mask, Start(x), nullptr, 1, 100, logger, sink, status, true, + std::move(model), 60.0); }; RugnuxLiveStatus during; @@ -169,12 +177,11 @@ TEST_CASE("StillsGPUAnalysis_FramesDuringThePrePassAreAnalysed", "[StillsGPUAnal std::atomic sunk = 0; std::latch offered(1); RugnuxLiveStatus status; - StillsGPUAnalysis gpu(x, mask, Start(x), nullptr, 2, 3, logger, - // The pre-pass is not over until every frame has been offered. - [&](const StillsGPUAnalysis::Setup &) { offered.wait(); }, - [&](DataMessage &, const AzimuthalIntegrationProfile &) { ++sunk; }, - [] { return DiffractionExperiment::DefaultDataProcessingSettings(); }, - {}, &status); + TestSink sink; + // The pre-pass is not over until every frame has been offered. + sink.ready = [&](const StillsSetup &) { offered.wait(); }; + sink.analysed = [&](DataMessage &) { ++sunk; }; + StillsGPUAnalysis gpu(x, mask, Start(x), nullptr, 2, 3, logger, sink, status); CHECK(status.Get().stills->images_needed == 3); for (int i = 0; i < frames; i++) gpu.Hold(Frame(buffer, i)); diff --git a/tools/jfjoch_stills_gpu_replay.cpp b/tools/jfjoch_stills_gpu_replay.cpp index 208710c23..ed9747660 100644 --- a/tools/jfjoch_stills_gpu_replay.cpp +++ b/tools/jfjoch_stills_gpu_replay.cpp @@ -13,6 +13,7 @@ #include #include #include +#include #include #include #include @@ -31,6 +32,17 @@ #include "../writer/WriteRugnuxFiles.h" #include "../image_analysis/WriteReflections.h" +struct ReplaySink : public StillsSink { + std::function ready; + std::function analysed; + SpotFindingSettings settings; + + void StillsReady(const StillsSetup &setup) override { ready(setup); } + void SendAnalysedImage(DataMessage &message, const AzimuthalIntegrationProfile &profile) override { analysed(message, profile); } + SpotFindingSettings AnalysisSettings() override { return settings; } + void StillsSetupFailed(const JFJochException &) override {} +}; + void print_usage() { std::cout << "Usage: jfjoch_stills_gpu_replay [options] " << std::endl; std::cout << " -o output prefix (default stills_gpu_replay)" << std::endl; @@ -161,34 +173,37 @@ int main(int argc, char **argv) { // As the receiver: half the buffer for the frames waiting, the other half for those sent. std::mutex run_profile_mutex; std::unique_ptr run_profile; - StillsGPUAnalysis gpu(x, mask, start, &pool, nthreads, buffer.GetStatus().total_slots / 2, logger, - [&](const StillsGPUAnalysis::Setup &s) { - t_ready = std::chrono::steady_clock::now(); - run_profile = std::make_unique(*s.mapping); - const auto &m = s.mask.GetMask(); - std::ofstream f(prefix + "_mask.u32", std::ios::binary); - f.write(reinterpret_cast(m.data()), m.size() * sizeof(uint32_t)); - logger.Info("Ready after {:.1f} s: beam centre {:.2f} {:.2f}", - std::chrono::duration(t_ready - t0).count(), - s.experiment.GetBeamX_pxl(), s.experiment.GetBeamY_pxl()); - }, - [&](DataMessage &m, const AzimuthalIntegrationProfile &profile) { - { - // As the receiver's plots do: the frames' profiles add up over the run. - std::unique_lock ul(run_profile_mutex); - *run_profile += profile; - } - // The analysis has caught up with the frames given. - if (++analysed + nthreads >= given && !caught_up.exchange(true)) - logger.Info("Analysis caught up {:.1f} s after it started, {} frames analysed", - std::chrono::duration(std::chrono::steady_clock::now() - - t_ready).count(), analysed.load()); - spots += static_cast(m.spots.size()); - processing_s += m.processing_time_s.value_or(0.0f); - if (m.indexing_result.value_or(false)) - ++indexed; - }, - [&] { return spot_finding; }, {}, nullptr, merge, model, deadline_s); + RugnuxLiveStatus status; // nobody reads it here + ReplaySink sink; + sink.ready = [&](const StillsSetup &setup) { + t_ready = std::chrono::steady_clock::now(); + run_profile = std::make_unique(*setup.mapping); + const auto &m = setup.mask.GetMask(); + std::ofstream f(prefix + "_mask.u32", std::ios::binary); + f.write(reinterpret_cast(m.data()), m.size() * sizeof(uint32_t)); + logger.Info("Ready after {:.1f} s: beam centre {:.2f} {:.2f}", + std::chrono::duration(t_ready - t0).count(), + setup.experiment.GetBeamX_pxl(), setup.experiment.GetBeamY_pxl()); + }; + sink.analysed = [&](DataMessage &m, const AzimuthalIntegrationProfile &profile) { + { + // As the receiver's plots do: the frames' profiles add up over the run. + std::unique_lock ul(run_profile_mutex); + *run_profile += profile; + } + // The analysis has caught up with the frames given. + if (++analysed + nthreads >= given && !caught_up.exchange(true)) + logger.Info("Analysis caught up {:.1f} s after it started, {} frames analysed", + std::chrono::duration(std::chrono::steady_clock::now() + - t_ready).count(), analysed.load()); + spots += static_cast(m.spots.size()); + processing_s += m.processing_time_s.value_or(0.0f); + if (m.indexing_result.value_or(false)) + ++indexed; + }; + sink.settings = spot_finding; + StillsGPUAnalysis gpu(x, mask, start, &pool, nthreads, buffer.GetStatus().total_slots / 2, logger, sink, status, + merge, model, deadline_s); JFJochReaderRawImage raw; std::vector empty;