StillsGPUAnalysis: a sink instead of four callbacks, fewer flags, the status always there

The analysis holds a StillsSink &, implemented by JFJochReceiverRugnuxStills, the way the receivers hold an
ImagePusher &: StillsReady is what sends START in a stills run, and it is now visible who does.

The status is a reference. The replay tool, the only caller without one, owns one nobody reads, and fifteen
guards go. The latency moves to the receiver, as in rotation; the analysis only keeps the outcome until
ReportOutcome.

State: finished is prepass_future.valid(). prepass_last and prepass_over are one PrePassState. The cut
flag per job, in a shared_ptr, is one atomic step_cut of the running step, taken from model_cut and a plain
job.cut when a job starts, so that a final step still queued at the deadline starts cut, as it did.
model_over stays: without it a cancel during the pre-pass would no longer say "cut" in the status.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01HVeDQsU5a8AJUdJPgWevAK

receiver/JFJochReceiverRugnuxStills.cpp

@@ -6,6 +6,8 @@
 #include <cstring>
 #include <thread>
 
+#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

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<const RugnuxModel> 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:

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<const RugnuxModel> 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<int64_t>(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<double>(experiment.GetImageNum())
-                      * std::chrono::duration<double>(experiment.GetImageTime()).count(),
-                      std::chrono::steady_clock::now());
-        status->Step("Pre-pass");
-        Publish();
-    }
+    status.Start(experiment.GetImageNum(),
+                 static_cast<double>(experiment.GetImageNum())
+                 * std::chrono::duration<double>(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<double>(live.images_analysed) / t;
     }
-    status->Stills(live);
+    status.Stills(live);
 }
 
 void StillsGPUAnalysis::Hold(const std::shared_ptr<HeldImage> &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<int64_t>(prepass_images.size()) < hold_limit) {
+        if (prepass_state != PrePassState::Over && static_cast<int64_t>(prepass_images.size()) < hold_limit) {
             prepass_images[image->message.number] = image;
             std::unique_lock ul_live(live_mutex);
             live.images_held = static_cast<int64_t>(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<double>(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<double>(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<int64_t>(prepass_images.size()) >= needed;
+                return prepass_state != PrePassState::Holding || static_cast<int64_t>(prepass_images.size()) >= needed;
             });
             sample = prepass_images;
-            last = prepass_last || static_cast<int64_t>(sample.size()) >= most_held;
+            last = prepass_state != PrePassState::Holding || static_cast<int64_t>(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<int>(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<double>(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<float>(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<std::atomic<bool>> 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<std::atomic<bool>> 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<PlacedModelOutput> 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);
 }

receiver/StillsGPUAnalysis.h

@@ -6,7 +6,6 @@
 #include <array>
 #include <condition_variable>
 #include <exception>
-#include <functional>
 #include <future>
 #include <latch>
 #include <map>
@@ -26,15 +25,33 @@
 #include "../rugnux/StillsLiveMerge.h"
 #include "../rugnux/ModelStep.h"
 
+struct StillsSetup {
+    DiffractionExperiment experiment;
+    PixelMask mask;
+    std::unique_ptr<AzimuthalIntegrationMapping> 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<double, 2> CHECKPOINTS = {0.5, 0.75};
     constexpr static int MERGE_THREADS = 8;
 
-    struct Setup {
-        DiffractionExperiment experiment;
-        PixelMask mask;
-        std::unique_ptr<AzimuthalIntegrationMapping> mapping;
-    };
-    using Ready = std::function<void(const Setup &)>;
-    using Sink = std::function<void(DataMessage &, const AzimuthalIntegrationProfile &)>;
-    using Settings = std::function<SpotFindingSettings()>;
-    using Failed = std::function<void(const JFJochException &)>;
-
     // `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<const RugnuxModel> 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<std::chrono::steady_clock::time_point> last_image;
-    std::optional<std::pair<RugnuxLiveState, std::string>> outcome;
+    std::optional<std::pair<RugnuxLiveState, std::string>> outcome;   // kept by Finish for ReportOutcome
 
-    Setup setup;
+    StillsSetup setup;
     std::unique_ptr<IndexAndRefine> 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<int64_t, std::shared_ptr<HeldImage>> prepass_images;
     std::latch prepass_done{1};
     std::exception_ptr setup_error; // set before prepass_done
@@ -140,7 +141,6 @@ private:
     ThreadSafeFIFO<std::shared_ptr<HeldImage>> queue;
     std::future<void> prepass_future;
     std::vector<std::future<void>> 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<int64_t> merge_queue;
@@ -164,20 +164,20 @@ private:
         std::vector<float> shells;    // the merge statistics' shell bounds
         std::string title;
         bool final = false;
-        std::shared_ptr<std::atomic<bool>> cut = std::make_shared<std::atomic<bool>>(false);
+        bool cut = false;
     };
     std::mutex model_mutex;
     std::condition_variable model_cv;
     std::optional<ModelJob> model_next;
-    std::shared_ptr<std::atomic<bool>> model_running_cut;   // of the step running
+    std::atomic<bool> 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<ModelStepResult> model_placed;   // the last step that placed the model, for the next to start from
     std::optional<ModelStepResult> model_final;
     std::future<void> 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<std::atomic<bool>> 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);

tests/StillsGPUAnalysisTest.cpp

@@ -4,13 +4,26 @@
 #include <catch2/catch_all.hpp>
 
 #include <atomic>
+#include <functional>
 #include <latch>
 
 #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<void(const StillsSetup &)> ready = [](const StillsSetup &) {};
+        std::function<void(DataMessage &)> analysed = [](DataMessage &) {};
+        std::function<void()> 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<int> 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<int> 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<const RugnuxModel> model) {
-        return std::make_unique<StillsGPUAnalysis>(
-                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<StillsGPUAnalysis>(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<int> 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));

tools/jfjoch_stills_gpu_replay.cpp

@@ -13,6 +13,7 @@
 #include <chrono>
 #include <cstring>
 #include <fstream>
+#include <functional>
 #include <iostream>
 #include <sstream>
 #include <thread>
@@ -31,6 +32,17 @@
 #include "../writer/WriteRugnuxFiles.h"
 #include "../image_analysis/WriteReflections.h"
 
+struct ReplaySink : public StillsSink {
+    std::function<void(const StillsSetup &)> ready;
+    std::function<void(DataMessage &, const AzimuthalIntegrationProfile &)> 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] <input master>" << std::endl;
     std::cout << "   -o <prefix>  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<AzimuthalIntegrationProfile> 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<AzimuthalIntegrationProfile>(*s.mapping);
-                                  const auto &m = s.mask.GetMask();
-                                  std::ofstream f(prefix + "_mask.u32", std::ios::binary);
-                                  f.write(reinterpret_cast<const char *>(m.data()), m.size() * sizeof(uint32_t));
-                                  logger.Info("Ready after {:.1f} s: beam centre {:.2f} {:.2f}",
-                                              std::chrono::duration<double>(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<double>(std::chrono::steady_clock::now()
-                                                                                - t_ready).count(), analysed.load());
-                                  spots += static_cast<int64_t>(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<AzimuthalIntegrationProfile>(*setup.mapping);
+            const auto &m = setup.mask.GetMask();
+            std::ofstream f(prefix + "_mask.u32", std::ios::binary);
+            f.write(reinterpret_cast<const char *>(m.data()), m.size() * sizeof(uint32_t));
+            logger.Info("Ready after {:.1f} s: beam centre {:.2f} {:.2f}",
+                        std::chrono::duration<double>(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<double>(std::chrono::steady_clock::now()
+                                                          - t_ready).count(), analysed.load());
+            spots += static_cast<int64_t>(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<uint8_t> empty;