From 10f83da9e8da0b027e2a70e2827018351ea16038 Mon Sep 17 00:00:00 2001 From: Filip Leonarski Date: Tue, 3 Nov 2026 13:54:04 +0100 Subject: [PATCH] Live stills: held frames wait in the image buffer, bounded to half of it The FPGA receiver held every frame for the GPU analysis in a heap vector sized to a full buffer slot and then shrunk, which keeps the capacity: ~25 MB per JUNGFRAU 8M frame instead of its compressed size, in a queue with no bound. During the pre-pass (~1000 frames at 100 Hz), or whenever the analysis runs slower than the frames arrive, that grew the broker's heap without limit on top of the configured image buffer. A held frame is now compressed straight into a slot of the receiver's ImageBuffer, the configured buffer the frames are written from. Two buckets per frame: the slot the compressed image waits in (in preparation, so never previewed or sent), and the slot GPUSink writes the CBOR message, analysis and image to, which is sent as before. The waiting slot goes back to the buffer, unsent, when the last copy of the HeldImage is gone. Nothing is copied to the heap and no frame memory is allocated per frame. The two buckets are also two pools: at most half the buffer's slots wait for analysis (the hold queue is bounded to total_slots / 2), so the analysed frames always find slots to be sent from. A frame that finds no free slot, or the hold queue full, is dropped and writer_queue_full is set, as the receiver does for any frame that finds the image buffer full; the frame thread never blocks. A full hold queue also starts the pre-pass on what it has. Analysis slower than the arrival rate therefore costs frames, not memory. jfjoch_stills_gpu_replay holds its frames in an ImageBuffer too (-b MiB, default 8192), and waits for a slot rather than drop a frame. Measured on the HLS-simulated CrMX JUNGFRAU 8M broker (40 frames, image buffer 8 GiB): RSS grew by ~26 MB per held frame before, linearly until the pre-pass; now it is flat once the receiver threads have warmed up (+1.05 GB less at 40 frames, ~25 GB less for a 1000-frame pre-pass). The stills replay of a stored serial collection gives the same beam centre, distance, mask and indexed count as before the change. Test: StillsGPUAnalysis_HoldIsBounded; the other StillsGPUAnalysis tests check every slot is returned. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01UqXYdJKPUFLoeAchjzv3x4 --- receiver/JFJochReceiverFPGA.cpp | 31 +++++++++++++------ receiver/StillsGPUAnalysis.cpp | 16 ++++++---- receiver/StillsGPUAnalysis.h | 6 ++-- receiver/StillsPrePass.h | 20 +++++++++++-- tests/StillsGPUAnalysisTest.cpp | 48 ++++++++++++++++++++++++++---- tools/jfjoch_stills_gpu_replay.cpp | 24 +++++++++++---- 6 files changed, 114 insertions(+), 31 deletions(-) diff --git a/receiver/JFJochReceiverFPGA.cpp b/receiver/JFJochReceiverFPGA.cpp index 3da176215..2f239c16b 100644 --- a/receiver/JFJochReceiverFPGA.cpp +++ b/receiver/JFJochReceiverFPGA.cpp @@ -129,7 +129,9 @@ JFJochReceiverFPGA::JFJochReceiverFPGA(const DiffractionExperiment &in_experimen // The start message and the calibration go out once the pre-pass is done. gpu = std::make_unique( experiment, pixel_mask, MakeStartMessage(experiment, pixel_mask, *az_int_mapping), - indexing_thread_pool, std::min(16, frame_transformation_nthreads), logger, + indexing_thread_pool, std::min(16, frame_transformation_nthreads), + // Half the image buffer for the frames waiting, the other half for those sent. + image_buffer.GetStatus().total_slots / 2, logger, [this](const StillsGPUAnalysis::Setup &s) { // The profiles are binned on the pre-pass's geometry. plots.Setup(s.experiment, *s.mapping); @@ -455,19 +457,30 @@ void JFJochReceiverFPGA::FrameTransformationThread(uint32_t threadid) { CompressionAlgorithm::NO_COMPRESSION); if (gpu_analysis) { - // Compressed here, as it is written, and analysed by GPUAnalysisThread. + // Compressed here into a slot of the image buffer, where it waits for the GPU analysis; GPUSink + // writes it, with its analysis, to a second slot, which is sent. A frame that finds no slot + // free, or as many frames waiting as the analysis is to hold, is dropped, as the receiver drops + // any frame that finds the image buffer full. + ++images_collected; + uncompressed_size += experiment.GetModulesNum() * RAW_MODULE_SIZE * experiment.GetByteDepthImage(); auto held = std::make_shared(); - held->image.resize(experiment.GetImageBufferLocationSize()); - held->image.resize(transformation.CompressImage(held->image.data(), held->image.size())); + held->slot = image_buffer.GetImageSlot(); + if (held->slot == nullptr) { + writer_queue_full = true; + continue; + } + const size_t size = transformation.CompressImage(static_cast(held->slot->GetImage()), + experiment.GetImageBufferLocationSize()); held->message = std::move(message); - held->message.image = CompressedImage(held->image.data(), held->image.size(), + held->message.image = CompressedImage(held->slot->GetImage(), size, experiment.GetXPixelsNum(), experiment.GetYPixelsNum(), experiment.GetImageMode(), experiment.GetCompressionAlgorithm()); - compressed_size += held->image.size(); - uncompressed_size += experiment.GetModulesNum() * RAW_MODULE_SIZE * experiment.GetByteDepthImage(); - ++images_collected; - gpu->Hold(held); + if (!gpu->Hold(held)) { + writer_queue_full = true; + continue; + } + compressed_size += size; continue; } diff --git a/receiver/StillsGPUAnalysis.cpp b/receiver/StillsGPUAnalysis.cpp index 7e84c5e5d..822ef12d1 100644 --- a/receiver/StillsGPUAnalysis.cpp +++ b/receiver/StillsGPUAnalysis.cpp @@ -10,11 +10,12 @@ StillsGPUAnalysis::StillsGPUAnalysis(const DiffractionExperiment &in_experiment, const PixelMask &in_mask, const StartMessage &in_start, IndexerThreadPool *in_indexing_thread_pool, - int64_t nthreads, Logger &in_logger, Ready in_ready, Sink in_sink, - Settings in_settings, Failed in_failed, RugnuxLiveStatus *in_status) + 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) : 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) { + setup_failed(std::move(in_failed)), status(in_status), queue(std::max(1, 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(); @@ -59,9 +60,14 @@ void StillsGPUAnalysis::Publish() { status->Stills(live); } -void StillsGPUAnalysis::Hold(const std::shared_ptr &image) { +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. + StartPrePass(); + return false; + } { std::unique_lock ul(prepass_mutex); if (!prepass_go) { @@ -79,7 +85,7 @@ void StillsGPUAnalysis::Hold(const std::shared_ptr &image) { prepass_cv.notify_all(); if (status) Publish(); - queue.PutBlocking(image); + return true; } void StillsGPUAnalysis::StartPrePass() { diff --git a/receiver/StillsGPUAnalysis.h b/receiver/StillsGPUAnalysis.h index 9223b503f..a414b40d7 100644 --- a/receiver/StillsGPUAnalysis.h +++ b/receiver/StillsGPUAnalysis.h @@ -29,6 +29,7 @@ // 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). +// At most `max_held` frames wait for analysis: Hold refuses one more, so the slots held stay bounded. // Where the setup after the pre-pass fails (`ready` included), `failed` is called 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. @@ -48,12 +49,13 @@ public: // `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, Logger &logger, + IndexerThreadPool *indexing_thread_pool, int64_t nthreads, int64_t max_held, Logger &logger, Ready ready, Sink sink, Settings settings, Failed failed = {}, RugnuxLiveStatus *status = nullptr); ~StillsGPUAnalysis(); - void Hold(const std::shared_ptr &image); + // False where `max_held` frames already wait: the frame is not taken. + bool Hold(const std::shared_ptr &image); // 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(); diff --git a/receiver/StillsPrePass.h b/receiver/StillsPrePass.h index ea5c8c021..b43197f2d 100644 --- a/receiver/StillsPrePass.h +++ b/receiver/StillsPrePass.h @@ -9,12 +9,26 @@ #include "../common/JFJochMessages.h" #include "../common/PixelMask.h" +#include "../common/ZeroCopyReturnValue.h" -// An image the receiver holds until the pre-pass is done: its message, with the image pointing into -// `image`, the compressed frame it owns. +// An image the receiver holds until it is analysed: its message, with the image pointing into `slot`, the +// image buffer slot the compressed frame was written to. The slot goes back to the buffer, unsent, with the +// last copy of the image held. struct HeldImage { DataMessage message; - std::vector image; + ZeroCopyReturnValue *slot = nullptr; + + HeldImage() = default; + HeldImage(const HeldImage &) = delete; + HeldImage &operator=(const HeldImage &) = delete; + ~HeldImage() { + if (slot) { + slot->SetImageNumber(-1); + slot->SetImageSize(0); + slot->SetIndexed(false); + slot->release(); + } + } }; struct StillsPrePassResult { diff --git a/tests/StillsGPUAnalysisTest.cpp b/tests/StillsGPUAnalysisTest.cpp index a78d88dac..6526f4576 100644 --- a/tests/StillsGPUAnalysisTest.cpp +++ b/tests/StillsGPUAnalysisTest.cpp @@ -5,6 +5,7 @@ #include +#include "../common/ImageBuffer.h" #include "../receiver/StillsGPUAnalysis.h" namespace { @@ -15,9 +16,11 @@ namespace { return x; } - // A frame after the pre-pass's sample (numbers from PREPASS_IMAGES on are never in it). - std::shared_ptr Frame(int64_t number) { + // A frame after the pre-pass's sample (numbers from PREPASS_IMAGES on are never in it), in a slot of `buffer`. + std::shared_ptr Frame(ImageBuffer &buffer, int64_t number) { auto held = std::make_shared(); + held->slot = buffer.GetImageSlot(); + REQUIRE(held->slot != nullptr); held->message.number = StillsGPUAnalysis::PREPASS_IMAGES + number; held->message.image_collection_efficiency = 1.0f; return held; @@ -34,11 +37,13 @@ TEST_CASE("StillsGPUAnalysis_StartFailureReleasesThreads", "[StillsGPUAnalysis]" Logger logger(Catch::getResultCapture().getCurrentTestName()); const DiffractionExperiment x = StillsExperiment(); const PixelMask mask(x); + ImageBuffer buffer(64 * 1024 * 1024); + buffer.StartMeasurement(x); RugnuxLiveStatus status; std::atomic failed = 0, sunk = 0; { - StillsGPUAnalysis gpu(x, mask, Start(x), nullptr, 4, logger, + StillsGPUAnalysis gpu(x, mask, Start(x), nullptr, 4, 100, logger, [](const StillsGPUAnalysis::Setup &) { throw JFJochException(JFJochExceptionCategory::ZeroMQ, "Timeout on pushing start message"); }, @@ -46,22 +51,52 @@ TEST_CASE("StillsGPUAnalysis_StartFailureReleasesThreads", "[StillsGPUAnalysis]" [] { return DiffractionExperiment::DefaultDataProcessingSettings(); }, [&](const JFJochException &) { ++failed; }, &status); for (int i = 0; i < 3; i++) - gpu.Hold(Frame(i)); + REQUIRE(gpu.Hold(Frame(buffer, i))); REQUIRE_THROWS_AS(gpu.Finish(), JFJochException); } CHECK(failed == 1); CHECK(sunk == 0); CHECK(status.Get().state == RugnuxLiveState::Failed); + // Every slot the frames were held in is back. + CHECK(buffer.GetAvailSlots() == buffer.GetStatus().total_slots); +} + +TEST_CASE("StillsGPUAnalysis_HoldIsBounded", "[StillsGPUAnalysis]") { + Logger logger(Catch::getResultCapture().getCurrentTestName()); + const DiffractionExperiment x = StillsExperiment(); + const PixelMask mask(x); + ImageBuffer buffer(64 * 1024 * 1024); + 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(); }); + // The analysis waits for the pre-pass, which waits for its sample: nothing is taken off the queue. + for (int i = 0; i < 3; i++) + REQUIRE(gpu.Hold(Frame(buffer, i))); + { + const auto refused = Frame(buffer, 3); + CHECK(!gpu.Hold(refused)); + } + CHECK(buffer.GetStatus().preparation_slots == 3); + REQUIRE_THROWS(gpu.Finish()); + CHECK(buffer.GetStatus().preparation_slots == 0); } TEST_CASE("StillsGPUAnalysis_FailingFrameDoesNotStopAnalysis", "[StillsGPUAnalysis]") { Logger logger(Catch::getResultCapture().getCurrentTestName()); const DiffractionExperiment x = StillsExperiment(); const PixelMask mask(x); + ImageBuffer buffer(64 * 1024 * 1024); + buffer.StartMeasurement(x); RugnuxLiveStatus status; std::atomic sunk = 0; - StillsGPUAnalysis gpu(x, mask, Start(x), nullptr, 1, logger, + StillsGPUAnalysis gpu(x, mask, Start(x), nullptr, 1, 100, logger, [](const StillsGPUAnalysis::Setup &) {}, [&](DataMessage &m, const AzimuthalIntegrationProfile &) { ++sunk; @@ -72,9 +107,10 @@ TEST_CASE("StillsGPUAnalysis_FailingFrameDoesNotStopAnalysis", "[StillsGPUAnalys {}, &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(i)); + REQUIRE(gpu.Hold(Frame(buffer, i))); REQUIRE_NOTHROW(gpu.Finish()); CHECK(sunk == 3); CHECK(status.Get().state == RugnuxLiveState::Failed); CHECK(!status.Get().message.empty()); + CHECK(buffer.GetStatus().preparation_slots == 0); } diff --git a/tools/jfjoch_stills_gpu_replay.cpp b/tools/jfjoch_stills_gpu_replay.cpp index d7161dbec..a251b9870 100644 --- a/tools/jfjoch_stills_gpu_replay.cpp +++ b/tools/jfjoch_stills_gpu_replay.cpp @@ -14,6 +14,7 @@ #include #include +#include "../common/ImageBuffer.h" #include "../common/Logger.h" #include "../image_analysis/indexing/IndexerThreadPool.h" #include "../reader/JFJochHDF5Reader.h" @@ -27,6 +28,7 @@ void print_usage() { std::cout << " -n images to replay (default all)" << std::endl; std::cout << " -r images given per second (default 0 = as fast as they are read)" << std::endl; std::cout << " -N analysis threads (default 16)" << std::endl; + std::cout << " -b image buffer the frames wait in (default 8192)" << std::endl; std::cout << " -x , -y beam centre, overriding the file" << std::endl; std::cout << " -C unit cell of the collection" << std::endl; std::cout << " -F the broker's default fixed-threshold spot finder (default: adaptive, as the receiver)" << std::endl; @@ -42,19 +44,21 @@ int main(int argc, char **argv) { std::optional images; double rate = 0.0; int nthreads = 16; + size_t buffer_MiB = 8192; std::optional beam_x, beam_y; std::optional cell; bool no_cell = false; bool fixed_threshold = false; int opt; - while ((opt = getopt(argc, argv, "o:s:n:r:N:x:y:C:cF")) != -1) { + while ((opt = getopt(argc, argv, "o:s:n:r:N:b:x:y:C:cF")) != -1) { switch (opt) { case 'o': prefix = optarg; break; case 's': first = atoll(optarg); break; case 'n': images = atoll(optarg); break; case 'r': rate = atof(optarg); break; case 'N': nthreads = atoi(optarg); break; + case 'b': buffer_MiB = atoll(optarg); break; case 'x': beam_x = atof(optarg); break; case 'y': beam_y = atof(optarg); break; case 'c': no_cell = true; break; @@ -116,11 +120,15 @@ int main(int argc, char **argv) { IndexerThreadPool pool(x.GetIndexingSettings()); + // The frames wait in an image buffer, as in the receiver; here every slot can hold one. + ImageBuffer buffer(buffer_MiB * 1024 * 1024); + buffer.StartMeasurement(x); + std::atomic analysed = 0, indexed = 0; std::chrono::steady_clock::time_point t_ready; const auto t0 = std::chrono::steady_clock::now(); { - StillsGPUAnalysis gpu(x, mask, start, &pool, nthreads, logger, + StillsGPUAnalysis gpu(x, mask, start, &pool, nthreads, buffer.GetStatus().total_slots, logger, [&](const StillsGPUAnalysis::Setup &s) { t_ready = std::chrono::steady_clock::now(); const auto &m = s.mask.GetMask(); @@ -143,13 +151,17 @@ int main(int argc, char **argv) { std::this_thread::sleep_until(t0 + std::chrono::duration(static_cast(i) / rate)); reader.ReadRawImage(first + i, raw); auto held = std::make_shared(); - held->image.assign(raw.image.GetCompressed(), raw.image.GetCompressed() + raw.image.GetCompressedSize()); + // Unlike the receiver, the replay waits for a slot rather than lose the frame. + while ((held->slot = buffer.GetImageSlot()) == nullptr) + std::this_thread::sleep_for(std::chrono::milliseconds(1)); + memcpy(held->slot->GetImage(), raw.image.GetCompressed(), raw.image.GetCompressedSize()); held->message.number = i; - held->message.image = CompressedImage(held->image.data(), held->image.size(), raw.image.GetWidth(), - raw.image.GetHeight(), raw.image.GetMode(), + held->message.image = CompressedImage(held->slot->GetImage(), raw.image.GetCompressedSize(), + raw.image.GetWidth(), raw.image.GetHeight(), raw.image.GetMode(), raw.image.GetCompressionAlgorithm()); held->message.image_collection_efficiency = 1.0f; - gpu.Hold(held); + while (!gpu.Hold(held)) + std::this_thread::sleep_for(std::chrono::milliseconds(1)); } logger.Info("Gave {} images in {:.1f} s", n, std::chrono::duration(std::chrono::steady_clock::now() - t0).count());