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