Live stills: the escalating pre-pass is bounded, so frames arriving during it are written

With a cell given, the live stills pre-pass ran again on all the frames held each time
their number doubled, until half the image buffer was full. On a chip with a low hit rate
the sample grew to tens of thousands of frames; the last run on them took long enough for
the queue to fill, and every frame arriving until it ended was refused and not written.

The sample now grows to PREPASS_MAX_IMAGES (4 x PREPASS_IMAGES) or half of what the queue
holds, whichever is smaller, and frames that arrived during the previous run do not make the
next one larger than that: the last run is short and the other half of the queue absorbs the
frames arriving meanwhile.

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

receiver/StillsGPUAnalysis.cpp

@@ -12,13 +12,13 @@
 
 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,
+                                     int64_t nthreads, int64_t in_max_held, Logger &in_logger, Ready in_ready,
                                      Sink in_sink, Settings in_settings, Failed in_failed,
                                      RugnuxLiveStatus *in_status, bool in_merge)
     : 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),
-      queue(std::max<int64_t>(1, max_held)) {
+      max_held(std::max<int64_t>(1, in_max_held)), queue(max_held) {
     live.images_needed = std::min(PREPASS_IMAGES, experiment.GetImageNum());
     live.file_beam_x_pxl = experiment.GetBeamX_pxl();
     live.file_beam_y_pxl = experiment.GetBeamY_pxl();
@@ -155,7 +155,9 @@ void StillsGPUAnalysis::PrePassThread() {
     std::chrono::steady_clock::time_point prepass_start = std::chrono::steady_clock::now();
     // At a low hit rate the first frames may not be enough to fit the geometry on, or may not hold a crystal: with
     // a cell to fit, the pre-pass runs again on all that is held once that has doubled, until the fit has its
-    // frames, or half the image buffer is full (the last try), or the collection ended.
+    // frames, or the collection ended, or the sample is as large as it may grow (the last try): bounded so that
+    // the last run is short and the other half of the queue takes the frames arriving meanwhile.
+    const int64_t max_sample = std::min(PREPASS_MAX_IMAGES, max_held / 2);
     for (int64_t needed = std::min(PREPASS_IMAGES, all_images); ;) {
         std::map<int64_t, std::shared_ptr<HeldImage>> sample;
         bool last;
@@ -164,8 +166,13 @@ void StillsGPUAnalysis::PrePassThread() {
             prepass_cv.wait(ul, [&] {
                 return prepass_last || static_cast<int64_t>(prepass_images.size()) >= needed;
             });
-            sample = prepass_images;
-            last = prepass_last || static_cast<int64_t>(sample.size()) >= all_images;
+            // Frames that arrived during the previous run do not make this one larger than the bound.
+            for (const auto &[n, image]: prepass_images) {
+                if (static_cast<int64_t>(sample.size()) >= std::max(needed, max_sample))
+                    break;
+                sample.emplace(n, image);
+            }
+            last = prepass_last || static_cast<int64_t>(sample.size()) >= std::min(all_images, max_sample);
         }
         if (sample.empty())
             break;
@@ -191,7 +198,7 @@ void StillsGPUAnalysis::PrePassThread() {
         if (last || !experiment.GetUnitCell()
             || (result && result->geometry_frames >= RUGNUX_REFINE_GEOMETRY_FRAMES))
             break;
-        needed = std::min(2 * static_cast<int64_t>(sample.size()), all_images);
+        needed = std::min({2 * static_cast<int64_t>(sample.size()), all_images, max_sample});
         std::unique_lock ul(live_mutex);
         live.prepass = RugnuxLivePrePass::Holding;
         live.images_needed = needed;

receiver/StillsGPUAnalysis.h

@@ -42,8 +42,9 @@
 // it (the collection's cell picks the frames merged); the space group the collection's, else P1.
 // merge = false leaves it out.
 // With a cell given, the pre-pass runs again on all the frames held each time they have doubled, until the geometry
-// fit has its frames, half the image buffer is full or the collection ends; the analysis starts after that and
-// catches up from the frames held. The geometry is fixed when the pre-pass ends: every frame is analysed and merged at it. Where the pre-pass could
+// fit has its frames, the collection ends or the sample reaches PREPASS_MAX_IMAGES or half of `max_held`; the
+// analysis starts after that and catches up from the frames held. The bound keeps the last run short and leaves
+// room in the queue for the frames arriving meanwhile, which are otherwise not written. The geometry is fixed when the pre-pass ends: every frame is analysed and merged at it. Where the pre-pass could
 // not refine it (too few of the frames it had indexed: a small image buffer, a serial hit rate, a chip whose first
 // frames are empty; or no cell given), the header's distance stays for the whole run, with a warning in the log
 // and the status.
@@ -53,6 +54,7 @@ public:
     // at 18%, 7 indexed at a distance 2.6% off and the geometry refinement did not run; of 500, 34 did and
     // refined it to 0.4%. They wait in the image buffer, which has to hold them in half its slots.
     constexpr static int64_t PREPASS_IMAGES = 500;
+    constexpr static int64_t PREPASS_MAX_IMAGES = 4 * PREPASS_IMAGES;
     constexpr static std::array<double, 2> CHECKPOINTS = {0.5, 0.75};
     constexpr static int MERGE_THREADS = 8;
 
@@ -127,6 +129,7 @@ private:
     std::string analysis_error; // the first, where an analysis thread failed
     void AnalysisError(const std::string &text);
 
+    const int64_t max_held;
     ThreadSafeFIFO<std::shared_ptr<HeldImage>> queue;
     std::future<void> prepass_future;
     std::vector<std::future<void>> analysis_futures;