From 73205c46008db9c718f5d2017c820ebca72f9047 Mon Sep 17 00:00:00 2001 From: Filip Leonarski Date: Fri, 6 Nov 2026 17:05:17 +0100 Subject: [PATCH] Live stills: the pre-pass holds more frames until the geometry can be fitted With a cell given, a pre-pass whose geometry fit did not reach its 200 frames now runs again on all the frames held each time they have doubled, until the fit has its frames, half the image buffer is full or the collection ends. A low hit rate, or a chip whose first frames are empty, then still gets a fitted geometry where the buffer holds enough frames. The analysis starts when the pre-pass ends and catches up from the held frames; the geometry stays fixed after that. Without a cell the pre-pass stops at the first try, as there is nothing to fit. The replay tool logs when the analysis has caught up. Co-Authored-By: Claude Opus 5.5 (cherry picked from commit ab54c514e556fa21e1a59bffd2b47f277d103d98) --- docs/JFJOCH_BROKER.md | 5 ++ receiver/StillsGPUAnalysis.cpp | 123 +++++++++++++++++------------ receiver/StillsGPUAnalysis.h | 8 +- tools/jfjoch_stills_gpu_replay.cpp | 10 ++- 4 files changed, 89 insertions(+), 57 deletions(-) diff --git a/docs/JFJOCH_BROKER.md b/docs/JFJOCH_BROKER.md index 7fb6bde35..cf65388e0 100644 --- a/docs/JFJOCH_BROKER.md +++ b/docs/JFJOCH_BROKER.md @@ -266,6 +266,11 @@ all: on fewer, too few index for the geometry refinement to run, and the stated buffer has to hold them, and the frames that arrive while the pre-pass runs, in half its slots (for a JUNGFRAU 8M, a slot is the size of an uncompressed frame); a smaller buffer starts the pre-pass on as many as it holds. +Where a unit cell is given and those frames are too few for the geometry fit (200 indexed frames), frames go on +being held and the pre-pass runs again on all of them each time they have doubled, until the fit has its frames, half +the image buffer is full, or the collection ends. The analysis starts when the pre-pass ends and catches up from the +frames held, so it starts later on a low hit rate; a frame finding the buffer full meanwhile is dropped as usual. + The geometry is fixed when the pre-pass ends and does not change during the collection, so every frame is analysed and merged at it and the radial profile has one binning throughout. Where the pre-pass could not refine it (too few of its frames indexed, as with a small image buffer, a low hit rate or a chip whose first frames are empty, or no unit diff --git a/receiver/StillsGPUAnalysis.cpp b/receiver/StillsGPUAnalysis.cpp index 5ebeda6f3..ed363530e 100644 --- a/receiver/StillsGPUAnalysis.cpp +++ b/receiver/StillsGPUAnalysis.cpp @@ -34,7 +34,7 @@ StillsGPUAnalysis::StillsGPUAnalysis(const DiffractionExperiment &in_experiment, 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)); - logger.Info("GPU analysis: {} threads, the first {} images held for the pre-pass", nthreads, PREPASS_IMAGES); + logger.Info("GPU analysis: {} threads, at least the first {} images held for the pre-pass", nthreads, PREPASS_IMAGES); } StillsGPUAnalysis::~StillsGPUAnalysis() { @@ -69,22 +69,16 @@ bool StillsGPUAnalysis::Hold(const std::shared_ptr &image) { if (status) status->ImageReceived(std::chrono::steady_clock::now()); if (!queue.Put(image)) { - // As many frames wait as the buffer is to hold: the pre-pass, if it has not run, runs on what it has. + // As many frames wait as the buffer is to hold: the pre-pass runs on what it has, for the last time. StartPrePass(); return false; } { std::unique_lock ul(prepass_mutex); - if (!prepass_go) { - if (image->message.number < PREPASS_IMAGES) { - prepass_images[image->message.number] = image; - std::unique_lock ul_live(live_mutex); - live.images_held = static_cast(prepass_images.size()); - } - ++images_held; - // Every one of the first images is there - or, where some never come, twice as many are. - prepass_go = (prepass_images.size() == std::min(PREPASS_IMAGES, experiment.GetImageNum())) - || (images_held >= 2 * PREPASS_IMAGES); + if (!prepass_over) { + 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(); @@ -96,7 +90,7 @@ bool StillsGPUAnalysis::Hold(const std::shared_ptr &image) { void StillsGPUAnalysis::StartPrePass() { { std::unique_lock ul(prepass_mutex); - prepass_go = true; + prepass_last = true; } prepass_cv.notify_all(); } @@ -151,65 +145,90 @@ void StillsGPUAnalysis::Written() { } void StillsGPUAnalysis::PrePassThread() { - std::map> sample; - { - std::unique_lock ul(prepass_mutex); - prepass_cv.wait(ul, [this] { return prepass_go; }); - sample.swap(prepass_images); - } - const auto masked = [](const PixelMask &m) { return std::count_if(m.GetMask().begin(), m.GetMask().end(), [](uint32_t v) { return v != 0; }); }; setup.experiment = experiment; setup.mask = mask; - { - std::unique_lock ul(live_mutex); - live.prepass = RugnuxLivePrePass::Running; - } - if (status) - Publish(); - bool failed = sample.empty(); - int geometry_frames = 0; - const auto prepass_start = std::chrono::steady_clock::now(); - if (!sample.empty()) { + const int64_t all_images = experiment.GetImageNum(); + std::optional result; + 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. + for (int64_t needed = std::min(PREPASS_IMAGES, all_images); ;) { + std::map> sample; + bool last; + { + std::unique_lock ul(prepass_mutex); + prepass_cv.wait(ul, [&] { + return prepass_last || static_cast(prepass_images.size()) >= needed; + }); + sample = prepass_images; + last = prepass_last || static_cast(sample.size()) >= all_images; + } + if (sample.empty()) + break; + { + std::unique_lock ul(live_mutex); + live.prepass = RugnuxLivePrePass::Running; + } + if (status) + Publish(); try { const auto t0 = std::chrono::steady_clock::now(); - const auto r = RunStillsPrePass(start, sample, static_cast(std::thread::hardware_concurrency())); - setup.experiment.BeamX_pxl(r.beam_x_pxl).BeamY_pxl(r.beam_y_pxl).DetectorDistance_mm(r.distance_mm); - setup.mask = r.mask; - geometry_frames = r.geometry_frames; - const auto &m = r.mask.GetMask(); - std::unique_lock ul(live_mutex); - live.beam_x_pxl = r.beam_x_pxl; - live.beam_y_pxl = r.beam_y_pxl; - live.distance_mm = r.distance_mm; - live.shadow_pixels = std::count_if(m.begin(), m.end(), [](uint32_t v) { - return (v & (1u << PixelMask::BeamStopPixelBit)) != 0; - }); - live.bad_pixels = std::count_if(m.begin(), m.end(), [](uint32_t v) { - return (v & (1u << PixelMask::HotPixelBit)) != 0; - }); + auto r = RunStillsPrePass(start, sample, static_cast(std::thread::hardware_concurrency())); logger.Info("Pre-pass on {} images ({:.1f} s): beam centre ({:.2f}, {:.2f}) -> ({:.2f}, {:.2f}) px, " - "distance {:.2f} -> {:.2f} mm, masked pixels {} -> {}", r.images, + "distance {:.2f} -> {:.2f} mm on {} frames, masked pixels {} -> {}", r.images, std::chrono::duration(std::chrono::steady_clock::now() - t0).count(), experiment.GetBeamX_pxl(), experiment.GetBeamY_pxl(), r.beam_x_pxl, r.beam_y_pxl, - experiment.GetDetectorDistance_mm(), r.distance_mm, + experiment.GetDetectorDistance_mm(), r.distance_mm, r.geometry_frames, masked(mask), masked(r.mask)); + result = std::move(r); } catch (const std::exception &e) { - failed = true; - logger.Warning("Pre-pass failed, the images are analysed with the stated beam centre and mask - {}", - e.what()); + logger.Warning("Pre-pass on {} images failed - {}", sample.size(), e.what()); } + if (last || !experiment.GetUnitCell() + || (result && result->geometry_frames >= RUGNUX_REFINE_GEOMETRY_FRAMES)) + break; + needed = std::min(2 * static_cast(sample.size()), all_images); + std::unique_lock ul(live_mutex); + live.prepass = RugnuxLivePrePass::Holding; + live.images_needed = needed; } - sample.clear(); + { + // 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_images.clear(); + } + const bool failed = !result; + int geometry_frames = 0; + if (result) { + const auto &r = *result; + setup.experiment.BeamX_pxl(r.beam_x_pxl).BeamY_pxl(r.beam_y_pxl).DetectorDistance_mm(r.distance_mm); + setup.mask = r.mask; + geometry_frames = r.geometry_frames; + const auto &m = r.mask.GetMask(); + std::unique_lock ul(live_mutex); + live.beam_x_pxl = r.beam_x_pxl; + live.beam_y_pxl = r.beam_y_pxl; + live.distance_mm = r.distance_mm; + live.shadow_pixels = std::count_if(m.begin(), m.end(), [](uint32_t v) { + return (v & (1u << PixelMask::BeamStopPixelBit)) != 0; + }); + live.bad_pixels = std::count_if(m.begin(), m.end(), [](uint32_t v) { + return (v & (1u << PixelMask::HotPixelBit)) != 0; + }); + } else + logger.Warning("Pre-pass failed, the images are analysed with the stated beam centre and mask"); try { setup.mapping = std::make_unique(setup.experiment, setup.mask); // Retained for the merge, which takes each image's reflections as soon as it is analysed. indexer = std::make_unique(setup.experiment, indexing_thread_pool, merge_frames, true); if (geometry_frames == 0) { const std::string text = fmt::format( - "The pre-pass could not refine the geometry (too few frames indexed, or no unit cell given): the " + "The pre-pass could not refine the geometry (too few frames indexed in what the buffer holds, or no unit cell given): the " "whole run is analysed at the distance in the header, {:.2f} mm, beam centre ({:.2f}, {:.2f}) px", setup.experiment.GetDetectorDistance_mm(), setup.experiment.GetBeamX_pxl(), setup.experiment.GetBeamY_pxl()); diff --git a/receiver/StillsGPUAnalysis.h b/receiver/StillsGPUAnalysis.h index 071662810..7c6e08b16 100644 --- a/receiver/StillsGPUAnalysis.h +++ b/receiver/StillsGPUAnalysis.h @@ -41,7 +41,9 @@ // before Finish returns (GetMerged). The cell is the run's consensus at the first merge, as rugnux merges in // it (the collection's cell picks the frames merged); the space group the collection's, else P1. // merge = false leaves it out. -// The geometry is fixed when the pre-pass ends: every frame is analysed and merged at it. Where the pre-pass could +// 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 // 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. @@ -115,8 +117,8 @@ private: std::mutex prepass_mutex; std::condition_variable prepass_cv; - bool prepass_go = false; - int64_t images_held = 0; + bool prepass_last = false; // what is held is all there will be + bool prepass_over = false; std::map> prepass_images; std::latch prepass_done{1}; std::exception_ptr setup_error; // set before prepass_done diff --git a/tools/jfjoch_stills_gpu_replay.cpp b/tools/jfjoch_stills_gpu_replay.cpp index 0e4c723df..2af39ef5b 100644 --- a/tools/jfjoch_stills_gpu_replay.cpp +++ b/tools/jfjoch_stills_gpu_replay.cpp @@ -138,7 +138,8 @@ int main(int argc, char **argv) { ImageBuffer buffer(buffer_MiB * 1024 * 1024); buffer.StartMeasurement(x); - std::atomic analysed = 0, indexed = 0, spots = 0; + std::atomic analysed = 0, indexed = 0, spots = 0, given = 0; + std::atomic caught_up = false; std::atomic processing_s = 0.0; std::chrono::steady_clock::time_point t_ready; const auto t0 = std::chrono::steady_clock::now(); @@ -163,7 +164,11 @@ int main(int argc, char **argv) { std::unique_lock ul(run_profile_mutex); *run_profile += profile; } - ++analysed; + // 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)) @@ -195,6 +200,7 @@ int main(int argc, char **argv) { held->message.image_collection_efficiency = 1.0f; while (!gpu.Hold(held)) std::this_thread::sleep_for(std::chrono::milliseconds(1)); + ++given; } logger.Info("Gave {} images in {:.1f} s", n, std::chrono::duration(std::chrono::steady_clock::now() - t0).count());