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