From 15266b21eeb3cc462a495045034fd5226d539a0c Mon Sep 17 00:00:00 2001 From: Filip Leonarski Date: Sat, 7 Nov 2026 14:19:52 +0100 Subject: [PATCH] Receiver: prototype FPGAImagePuller, PSI detector frames through the stream receivers A prototype puller over the acquisition devices: one thread per module takes the converted frames off the card and writes them into their image, worker threads compress, and the images leave in image-number order as a filled DataMessage between a start and an end message. JFJochReceiverLite and JFJochReceiverRugnux then take them as they take a DECTRIS stream. Off by default and not wired to the broker: JFJochReceiverService::UseFPGAImagePuller() selects it, the FPGA receiver stays the default, and pedestal runs and summation on the host stay with the FPGA receiver. No module waits for another and nothing waits for the consumer. Images are held in a bounded pool of slots (no ring, so a slow image pins nothing behind it). An image finding no free slot is dropped whole and counted in images_dropped and in the efficiency. A module or a card behind the fastest by the device's own threshold is taken as missing, and its late frames go back to the card. What Lite assumed of a DECTRIS stream, now changed: ApplyStartMessage no longer applies to a PSI detector (it forced pixel_mask_applied false). Calibration goes out right after the start message, as a calibration message could reach the writer after the first image and be refused. Efficiency counts images expected and each image's collection efficiency, not frames. The ADU histogram comes through the end message. An image in raw module layout is forwarded without analysis, as the preprocessor needs the assembled geometry. Measured against the FPGA receiver on the HLS simulation, JUNGFRAU and PSI EIGER, conversion and raw: pixels and source metadata identical. Open: - The azimuthal profile of a PSI EIGER differs between the card and the host (cosine 0.91 on synthetic frames, 0.999996 on JUNGFRAU int16). Cause not known yet. - Unsigned images: the card writes saturated and error pixels as the same marker, which the host preprocessor counts as error (saturated 8 to 0, error 8 to 16 on the same frames). - Not run on a real card or the broker. Missing: per-module packet statistics, plots of empty images, the card's delay in the messages. Above the card's summation limit the run falls back to the FPGA receiver. - JFJochReceiverRugnux: forwarding tested, analysis results not. It has no raw layout guard, its efficiency still divides by frames, and it does not forward the ADU histogram. Co-Authored-By: Claude Opus 5.5 --- image_puller/ImagePuller.h | 7 + receiver/CMakeLists.txt | 1 + receiver/FPGAImagePuller.cpp | 344 ++++++++++++ receiver/FPGAImagePuller.h | 108 ++++ receiver/JFJochReceiver.cpp | 1 + receiver/JFJochReceiver.h | 1 + receiver/JFJochReceiverCurrentStatus.h | 1 + receiver/JFJochReceiverLite.cpp | 38 +- receiver/JFJochReceiverLite.h | 7 + receiver/JFJochReceiverRugnux.cpp | 3 + receiver/JFJochReceiverRugnux.h | 1 + receiver/JFJochReceiverService.cpp | 40 +- receiver/JFJochReceiverService.h | 8 + tests/CMakeLists.txt | 2 + tests/FPGAImagePullerPerfTest.cpp | 412 +++++++++++++++ tests/FPGAImagePullerTest.cpp | 705 +++++++++++++++++++++++++ 16 files changed, 1670 insertions(+), 9 deletions(-) create mode 100644 receiver/FPGAImagePuller.cpp create mode 100644 receiver/FPGAImagePuller.h create mode 100644 tests/FPGAImagePullerPerfTest.cpp create mode 100644 tests/FPGAImagePullerTest.cpp diff --git a/image_puller/ImagePuller.h b/image_puller/ImagePuller.h index b399fb63e..a7e679ecb 100644 --- a/image_puller/ImagePuller.h +++ b/image_puller/ImagePuller.h @@ -7,6 +7,7 @@ #include #include #include +#include #include "../common/ZMQWrappers.h" #include "../frame_serialize/CBORStream2Deserializer.h" @@ -81,6 +82,12 @@ public: virtual bool SendAck(const PullerAckMessage &) { return false; } virtual void Disconnect() = 0; + + // To go to the writer right after the start message, before any image: what a detector stream sends + // as calibration messages can reach the writer after the first image, where it is refused. + virtual std::vector GetCalibration() const { return {}; } + // Images the puller had no room for, which the receiver therefore never gets. + virtual size_t GetDroppedImages() const { return 0; } }; diff --git a/receiver/CMakeLists.txt b/receiver/CMakeLists.txt index 9096e604c..e69cb093f 100644 --- a/receiver/CMakeLists.txt +++ b/receiver/CMakeLists.txt @@ -11,6 +11,7 @@ ADD_LIBRARY(JFJochReceiver STATIC JFJochReceiverCurrentStatus.h JFJochReceiverOutput.h JFJochReceiverLite.h JFJochReceiverLite.cpp + FPGAImagePuller.h FPGAImagePuller.cpp JFJochReceiverRugnux.h JFJochReceiverRugnux.cpp StillsPrePass.h StillsPrePass.cpp StillsGPUAnalysis.h StillsGPUAnalysis.cpp diff --git a/receiver/FPGAImagePuller.cpp b/receiver/FPGAImagePuller.cpp new file mode 100644 index 000000000..4a4d81a22 --- /dev/null +++ b/receiver/FPGAImagePuller.cpp @@ -0,0 +1,344 @@ +// SPDX-FileCopyrightText: 2026 Filip Leonarski, Paul Scherrer Institute +// SPDX-License-Identifier: GPL-3.0-only + +#include "FPGAImagePuller.h" + +#include +#include + +#include "JFJochCompressor.h" + +FPGAImagePuller::FPGAImagePuller(const DiffractionExperiment &in_experiment, const PixelMask &mask, + const JFCalibration *calibration, AcquisitionDeviceGroup &in_devices, + Logger &in_logger, int64_t compression_threads, size_t in_max_slots) + : ImagePuller(in_max_slots + 2), // every slot, the start and the end message: Put never finds it full + experiment(in_experiment), + logger(in_logger), + devices(in_devices), + max_slots(in_max_slots), + image_dropped(experiment.GetImageNum(), 0) { + const int ndatastreams = experiment.GetDataStreamsNum(); + if (devices.size() < ndatastreams) + throw JFJochException(JFJochExceptionCategory::InputParameterInvalid, + "Number of acquisition devices has to match data streams"); + + for (int m = 0; m < experiment.GetModulesNum(); m++) + adu_histogram.emplace_back(std::make_unique()); + + for (int d = 0; d < ndatastreams; d++) { + devices[d].PrepareAction(experiment); + expected_packets_per_image += devices[d].Counters().GetExpectedPacketsPerImage(); + } + + if (experiment.GetDetectorType() != DetectorType::JUNGFRAU) + calibration = nullptr; + if (calibration != nullptr && experiment.GetSaveCalibration()) + PrepareCalibration(*calibration); + + // Making a slot takes tens of milliseconds on a large detector, under the lock all modules go through: + // the first ones are ready when the frames start. The rest only if the receiver falls behind. + for (size_t i = 0; i < std::min(max_slots, InitialSlots); i++) { + free_slots.push_back(std::make_shared(experiment)); + slots_created++; + } + + // The receiver's own start message is built from its experiment; this one only says the run has begun. + StartMessage start{}; + experiment.FillMessage(start); + outside_fifo.Put({.cbor = std::make_shared(start), + .received = std::chrono::steady_clock::now()}); + + for (int i = 0; i < compression_threads; i++) + compress_futures.emplace_back(std::async(std::launch::async, &FPGAImagePuller::CompressImages, this)); + + for (int d = 0; d < ndatastreams; d++) + for (int m = 0; m < experiment.GetModulesNum(d); m++) + module_futures.emplace_back(std::async(std::launch::async, &FPGAImagePuller::ReadModule, this, d, m)); + + std::latch ready(ndatastreams); + for (int d = 0; d < ndatastreams; d++) + acquire_futures.emplace_back(std::async(std::launch::async, &FPGAImagePuller::Acquire, this, + static_cast(d), std::cref(mask), calibration, + std::ref(ready))); + ready.wait(); + + finish_future = std::async(std::launch::async, &FPGAImagePuller::Finish, this); + + if (failed) { + try { + Stop(); + } catch (...) { + outside_fifo.Clear(); // the destructor does not run + throw; + } + } +} + +FPGAImagePuller::~FPGAImagePuller() { + try { + Stop(); + } catch (const std::exception &e) { + logger.Error("FPGA image puller teardown failed: {}", e.what()); + } + // These hold slots, which go back to this object + outside_fifo.Clear(); +} + +void FPGAImagePuller::SetError(std::exception_ptr e) { + std::unique_lock ul(error_mutex); + if (!error) + error = e; + failed = true; +} + +void FPGAImagePuller::Cancel() { + for (int d = 0; d < experiment.GetDataStreamsNum(); d++) + devices[d].Cancel(); +} + +void FPGAImagePuller::Stop() { + Cancel(); + std::call_once(finish_once, [this] { finish_future.get(); }); + std::unique_lock ul(error_mutex); + if (error) + std::rethrow_exception(std::exchange(error, nullptr)); +} + +void FPGAImagePuller::PrepareCalibration(const JFCalibration &calibration) { + JFJochBitShuffleCompressor compressor(CompressionAlgorithm::BSHUF_LZ4); + + const auto add = [&](const std::string &channel, std::vector v) { + calibration_data.emplace_back(std::move(v)); + calibration_images.emplace_back(calibration_data.back().data(), calibration_data.back().size(), + RAW_MODULE_COLS, experiment.GetModulesNum() * RAW_MODULE_LINES, + CompressedImageMode::Uint16, CompressionAlgorithm::BSHUF_LZ4, channel); + }; + for (int sc = 0; sc < experiment.GetStorageCellNumber(); sc++) { + for (int gain = 0; gain < 3; gain++) { + if (experiment.IsFixedGainG1() && (gain != 1)) + continue; + const std::string suffix = (experiment.GetStorageCellNumber() > 1) + ? fmt::format("_g{:d}_sc{:d}", gain, sc) + : fmt::format("_g{:d}", gain); + add("pedestal" + suffix, compressor.Compress(calibration.GetPedestal(gain, sc))); + add("pedestal_rms" + suffix, compressor.Compress(calibration.GetPedestalRMS(gain, sc))); + } + } +} + +void FPGAImagePuller::Acquire(uint16_t data_stream, const PixelMask &mask, const JFCalibration *calibration, + std::latch &ready) { + auto &device = devices[data_stream]; + try { + // Only what the card needs for the pixels: its analysis results are not read. + if (calibration != nullptr) + device.InitializeCalibration(experiment, *calibration); + device.InitializePixelMask(experiment, mask); + device.StartAction(experiment); + } catch (...) { + SetError(std::current_exception()); + Cancel(); // the module threads of the other cards have to come out as well + ready.count_down(); + return; + } + ready.count_down(); + + try { + device.WaitForActionComplete(); + } catch (...) { + SetError(std::current_exception()); + Cancel(); + } +} + +int64_t FPGAImagePuller::FastestFrame() const { + int64_t ret = 0; + for (int d = 0; d < experiment.GetDataStreamsNum(); d++) + ret = std::max(ret, static_cast(devices[d].Counters().GetFastestFrameNumber())); + return ret; +} + +// Polled rather than waited on: a dead card has to be noticed from the other cards' frames, and the +// counters wake every waiter on every completion. +void FPGAImagePuller::WaitForFrame(uint16_t data_stream, uint16_t module_number, int64_t frame) const { + const auto &counters = devices[data_stream].Counters(); + // A frame is complete once the module is two further on, as in JFJochReceiverFPGA + while ((static_cast(counters.GetCurrFrameNumber(module_number)) < frame + 2) + && !counters.IsAcquisitionFinished() && !failed + && (FastestFrame() < frame + 2 + MissingAfterFrames)) + std::this_thread::sleep_for(std::chrono::milliseconds(1)); +} + +// The last frame any card got, for a module whose own card has finished. Waits for the other cards to +// finish too, so every module of the detector stops at the same image. +int64_t FPGAImagePuller::LastFrame() const { + for (int d = 0; d < experiment.GetDataStreamsNum(); d++) + while (!devices[d].Counters().IsAcquisitionFinished() && !failed) + std::this_thread::sleep_for(std::chrono::milliseconds(1)); + return FastestFrame(); +} + +std::shared_ptr FPGAImagePuller::GetSlot(int64_t image_number) { + std::unique_lock ul(images_mutex); + if (image_dropped[image_number]) + return nullptr; + if (const auto it = images_in_progress.find(image_number); it != images_in_progress.end()) + return it->second; + + std::shared_ptr slot; + if (!free_slots.empty()) { + slot = free_slots.back(); + free_slots.pop_back(); + } else if (slots_created < max_slots) { + slots_created++; + slot = std::make_shared(experiment); + } else { + image_dropped[image_number] = 1; + dropped_images++; + Emit(image_number, std::nullopt); + return nullptr; + } + slot->image_number = image_number; + slot->metadata = std::make_unique(experiment); + slot->modules_left = static_cast(experiment.GetModulesNum()); + images_in_progress[image_number] = slot; + return slot; +} + +void FPGAImagePuller::ReturnSlot(const std::shared_ptr &slot) { + std::unique_lock ul(images_mutex); + free_slots.push_back(slot); +} + +void FPGAImagePuller::ReadModule(uint16_t data_stream, uint16_t module_number) { + try { + auto &device = devices[data_stream]; + const auto &counters = device.Counters(); + const size_t module_abs = experiment.GetFirstModuleOfDataStream(data_stream) + module_number; + int64_t oldest_held = 0; + + for (int64_t image = 0; image < experiment.GetImageNum(); image++) { + WaitForFrame(data_stream, module_number, image); + + if (failed) + break; + + if (counters.IsAcquisitionFinished() + && (image > static_cast(counters.GetFastestFrameNumber())) + && (image > LastFrame())) + break; + + // A frame given up on can still arrive, late: its buffer goes back to the card. + while ((oldest_held < image) + && (oldest_held < static_cast(counters.GetCurrFrameNumber(module_number)))) + device.FrameBufferRelease(oldest_held++, module_number); + + const bool collected = counters.IsAnyPacketCollected(image, module_number); + if (const auto slot = GetSlot(image)) { + if (collected) { + const DeviceOutput *output = device.GetDeviceOutput(image, module_number); + slot->metadata->Process(output); + adu_histogram[module_abs]->Add(*output); + slot->transformation.ProcessModule(output, data_stream); + } else + slot->transformation.FillNotCollectedModule(module_number, data_stream); + + if (--slot->modules_left == 0) { + { + std::unique_lock ul(images_mutex); + images_in_progress.erase(image); + } + images_to_compress.Put(slot); + } + } + if (collected) + device.FrameBufferRelease(image, module_number); + } + } catch (...) { + SetError(std::current_exception()); + Cancel(); + } +} + +void FPGAImagePuller::CompressImages() { + while (const auto slot = images_to_compress.GetBlocking()) { + const int64_t image_number = slot->image_number; + try { + DataMessage message{}; + message.number = image_number; + message.original_number = image_number; + slot->metadata->Export(message, expected_packets_per_image); + // The card's spot finder was not set up for this run; the receiver reports its own. + message.strong_pixel_count.reset(); + + if (message.image_collection_efficiency == 0.0f) { // no module of it came + ReturnSlot(slot); + Emit(image_number, std::nullopt); + continue; + } + + message.image = slot->transformation.GetCompressedImage(); + + ImagePullerOutput output; + output.cbor = std::make_shared(message); + // The image lives in the slot: it goes back when the last copy of this output is gone. + output.raw = std::shared_ptr( + new ReceiveBufferRegion{const_cast(message.image.GetCompressed()), + message.image.GetCompressedSize()}, + [this, slot](ReceiveBufferRegion *region) { + delete region; + ReturnSlot(slot); + }); + Emit(image_number, std::move(output)); + } catch (...) { + SetError(std::current_exception()); + Cancel(); + ReturnSlot(slot); + Emit(image_number, std::nullopt); + } + } +} + +// Whichever thread finishes the image next in line lets it, and those waiting behind it, out. +void FPGAImagePuller::Emit(int64_t image_number, std::optional output) { + std::unique_lock ul(order_mutex); + images_done.emplace(image_number, std::move(output)); + for (auto it = images_done.begin(); (it != images_done.end()) && (it->first == next_image); + it = images_done.erase(it), next_image++) { + if (it->second) { + it->second->received = std::chrono::steady_clock::now(); + outside_fifo.Put(std::move(*it->second)); + } + } +} + +void FPGAImagePuller::Finish() { + for (auto &f: module_futures) + f.get(); + for (size_t i = 0; i < compress_futures.size(); i++) + images_to_compress.Put(nullptr); + for (auto &f: compress_futures) + f.get(); + + Cancel(); + for (auto &f: acquire_futures) + f.get(); + + // Only an error leaves images unfinished, holding back those behind them + { + std::unique_lock ul(order_mutex); + for (auto &[n, output]: images_done) + if (output) + outside_fifo.Put(std::move(*output)); + images_done.clear(); + } + + if (dropped_images > 0) + logger.Warning("FPGA image puller: {} images dropped, no free slot", dropped_images.load()); + + EndMessage end{}; + for (size_t i = 0; i < adu_histogram.size(); i++) + end.adu_histogram["module" + std::to_string(i)] = adu_histogram[i]->GetHistogram(); + outside_fifo.Put({.cbor = std::make_shared(end), + .received = std::chrono::steady_clock::now()}); +} diff --git a/receiver/FPGAImagePuller.h b/receiver/FPGAImagePuller.h new file mode 100644 index 000000000..711b36cf4 --- /dev/null +++ b/receiver/FPGAImagePuller.h @@ -0,0 +1,108 @@ +// SPDX-FileCopyrightText: 2026 Filip Leonarski, Paul Scherrer Institute +// SPDX-License-Identifier: GPL-3.0-only + +#pragma once + +#include +#include +#include +#include +#include +#include +#include +#include + +#include "FrameTransformation.h" +#include "ImageMetadata.h" +#include "../acquisition_device/AcquisitionDeviceGroup.h" +#include "../common/ADUHistogram.h" +#include "../image_puller/ImagePuller.h" +#include "../jungfrau/JFCalibration.h" + +// The images of a collection from the FPGA cards, as a detector stream for JFJochReceiverLite: start +// message, images assembled and compressed here in image-number order, end message. The card converts +// the pixels; its analysis results are not used, the receiver analyses the assembled image. +// +// One thread per module takes that module's frames off the card and writes them into the image they +// belong to, so no module ever waits for another and none waits for the consumer. Images are held in a +// bounded pool of slots, from the first module to reach them until the receiver is done with them. An +// image finding no free slot is dropped whole, and its frames go back to the card unread. +class FPGAImagePuller : public ImagePuller { +public: + static constexpr size_t DefaultSlots = 128; + static constexpr size_t InitialSlots = 32; + // A module this far behind the fastest one is taken as missing from that frame. It is what the + // device itself allows before it starts handing late frames straight back to the card. + static constexpr int64_t MissingAfterFrames = AcquisitionCounters::ThresholdFramesLost + 2; +private: + struct Slot { + explicit Slot(const DiffractionExperiment &experiment) : transformation(experiment) {} + FrameTransformation transformation; + std::unique_ptr metadata; + std::atomic modules_left = 0; + int64_t image_number = 0; + }; + + const DiffractionExperiment experiment; + Logger &logger; + AcquisitionDeviceGroup &devices; + const size_t max_slots; + uint64_t expected_packets_per_image = 0; + + std::mutex images_mutex; + std::vector> free_slots; + size_t slots_created = 0; + std::map> images_in_progress; + std::vector image_dropped; + std::atomic dropped_images = 0; + + // What is done, waiting for the images before it. No value: that image will not come. + std::mutex order_mutex; + std::map> images_done; + int64_t next_image = 0; + + ThreadSafeFIFO> images_to_compress; + std::vector> adu_histogram; + + std::vector> calibration_data; + std::vector calibration_images; + + std::mutex error_mutex; + std::exception_ptr error; + std::atomic failed = false; // a failed card may never report finished: nothing waits for it then + void SetError(std::exception_ptr e); + + std::vector> acquire_futures; + std::vector> module_futures; + std::vector> compress_futures; + std::future finish_future; + std::once_flag finish_once; + + void Acquire(uint16_t data_stream, const PixelMask &mask, const JFCalibration *calibration, std::latch &ready); + void ReadModule(uint16_t data_stream, uint16_t module_number); + void CompressImages(); + void Finish(); + + void PrepareCalibration(const JFCalibration &calibration); + void WaitForFrame(uint16_t data_stream, uint16_t module_number, int64_t frame) const; + int64_t FastestFrame() const; + int64_t LastFrame() const; + std::shared_ptr GetSlot(int64_t image_number); + void ReturnSlot(const std::shared_ptr &slot); + void Emit(int64_t image_number, std::optional output); +public: + // Arms the cards; when it returns the detector can be started. + FPGAImagePuller(const DiffractionExperiment &experiment, const PixelMask &mask, const JFCalibration *calibration, + AcquisitionDeviceGroup &devices, Logger &logger, int64_t compression_threads, + size_t max_slots = DefaultSlots); + ~FPGAImagePuller() override; + + void Disconnect() override {} + std::vector GetCalibration() const override { return calibration_images; } + size_t GetDroppedImages() const override { return dropped_images; } + + // The collection is over, whatever the cards have got: what is on its way still comes, then the end message. + void Cancel(); + // After the receiver is done: stops the cards, waits for the threads, throws what went wrong with the cards. + void Stop(); +}; diff --git a/receiver/JFJochReceiver.cpp b/receiver/JFJochReceiver.cpp index 9ffa7badc..46183780f 100644 --- a/receiver/JFJochReceiver.cpp +++ b/receiver/JFJochReceiver.cpp @@ -100,6 +100,7 @@ JFJochReceiverStatus JFJochReceiver::GetStatus() const { ret.images_collected = images_collected; ret.images_sent = images_sent; ret.images_skipped = images_skipped; + ret.images_dropped = GetImagesDropped(); ret.images_written = image_pusher.GetImagesWritten(); ret.cancelled = cancelled; ret.efficiency = GetEfficiency(); diff --git a/receiver/JFJochReceiver.h b/receiver/JFJochReceiver.h index f0f76db20..416aab14a 100644 --- a/receiver/JFJochReceiver.h +++ b/receiver/JFJochReceiver.h @@ -103,6 +103,7 @@ protected: // The last word on the end message before it is sent: a receiver whose analysis is not the // per-image one fills in what it found. virtual void AddToEndMessage(EndMessage &message) {} + virtual uint64_t GetImagesDropped() const { return 0; } virtual void Cancel(const JFJochException &e); public: JFJochReceiver(const DiffractionExperiment& experiment, diff --git a/receiver/JFJochReceiverCurrentStatus.h b/receiver/JFJochReceiverCurrentStatus.h index 030bbf110..0ef2160a6 100644 --- a/receiver/JFJochReceiverCurrentStatus.h +++ b/receiver/JFJochReceiverCurrentStatus.h @@ -23,6 +23,7 @@ struct JFJochReceiverStatus { uint64_t images_collected; uint64_t images_sent; uint64_t images_skipped; + uint64_t images_dropped = 0; // no room for them before the receiver: they never reached it std::optional images_written; bool cancelled; }; diff --git a/receiver/JFJochReceiverLite.cpp b/receiver/JFJochReceiverLite.cpp index c105e1216..1a7dd726d 100644 --- a/receiver/JFJochReceiverLite.cpp +++ b/receiver/JFJochReceiverLite.cpp @@ -156,6 +156,9 @@ void JFJochReceiverLite::MeasurementThread() { start_time = std::chrono::system_clock::now(); // Send new start message out SendStartMessage(); + if (push_images_to_writer) + for (const auto &c: image_puller.GetCalibration()) + image_pusher.SendCalibration(c); measurement_started.count_down(); } catch (const JFJochException &e) { @@ -203,6 +206,10 @@ void JFJochReceiverLite::MeasurementThread() { } void JFJochReceiverLite::ApplyStartMessage(DiffractionExperiment &experiment, const StartMessage &msg) { + // A PSI detector's images are what the experiment set the card up to produce: nothing to learn here. + if (experiment.GetDetectorType() != DetectorType::DECTRIS) + return; + if ((experiment.GetXPixelsNum() != msg.image_size_x) || (experiment.GetYPixelsNum() != msg.image_size_y)) throw JFJochException(JFJochExceptionCategory::InputParameterInvalid, @@ -298,8 +305,10 @@ void JFJochReceiverLite::DataAnalysisThread(uint32_t id) { measurement_started.wait(); try { - analysis = std::make_unique(experiment, *az_int_mapping, pixel_mask, indexer, - /*enable_fused_adaptive_gpu=*/true); + // The analysis works on the assembled image; an image in raw module layout is only forwarded. + if (experiment.IsGeometryTransformed()) + analysis = std::make_unique(experiment, *az_int_mapping, pixel_mask, indexer, + /*enable_fused_adaptive_gpu=*/true); } catch (const JFJochException &e) { Cancel(e); return; @@ -314,6 +323,7 @@ void JFJochReceiverLite::DataAnalysisThread(uint32_t id) { continue; } else if (msg->cbor->end_message) { // Message is end message + end_adu_histogram = msg->cbor->end_message->adu_histogram; end_message_received = true; logger.Debug("Thread {} End message received in JFJochReceiverLite", id); } else if (msg->cbor->calibration) { @@ -325,6 +335,7 @@ void JFJochReceiverLite::DataAnalysisThread(uint32_t id) { DataMessage data_msg = msg->cbor->data_message.value(); ++images_collected; + collection_efficiency_sum += data_msg.image_collection_efficiency.value_or(1.0f); auto compressed_size_img = data_msg.image.GetCompressedSize(); auto uncompressed_size_img = data_msg.image.GetUncompressedSize(); @@ -337,7 +348,8 @@ void JFJochReceiverLite::DataAnalysisThread(uint32_t id) { auto image_start_time = std::chrono::high_resolution_clock::now(); AzimuthalIntegrationProfile profile(*az_int_mapping); - analysis->Analyze(data_msg, profile, GetSpotFindingSettings()); + if (analysis) + analysis->Analyze(data_msg, profile, GetSpotFindingSettings()); auto image_end_time = std::chrono::high_resolution_clock::now(); std::chrono::duration image_duration = image_end_time - image_start_time; @@ -434,16 +446,28 @@ void JFJochReceiverLite::StopReceiver() { } } +void JFJochReceiverLite::AddToEndMessage(EndMessage &message) { + if (!end_adu_histogram.empty()) + message.adu_histogram = end_adu_histogram; +} + +// With the card summing, an image is many frames. A dark mask run has frames and no images. +int64_t JFJochReceiverLite::ExpectedImages() const { + return (experiment.GetDetectorMode() == DetectorMode::DarkMask) ? experiment.GetFrameNum() : experiment.GetImageNum(); +} + float JFJochReceiverLite::GetEfficiency() const { - if (experiment.GetFrameNum() == 0) + if (ExpectedImages() == 0) return 0; - return static_cast(images_collected) / static_cast(experiment.GetFrameNum()); + if (experiment.GetDetectorMode() == DetectorMode::DarkMask) + return static_cast(images_collected) / static_cast(ExpectedImages()); + return static_cast(collection_efficiency_sum / static_cast(ExpectedImages())); } float JFJochReceiverLite::GetProgress() const { - if (experiment.GetFrameNum() == 0) + if (ExpectedImages() == 0) return 0.0; - return static_cast(max_image_number_received) / static_cast(experiment.GetFrameNum()); + return static_cast(max_image_number_received) / static_cast(ExpectedImages()); } void JFJochReceiverLite::SetSpotFindingSettings(const SpotFindingSettings &in_spot_finding_settings) { diff --git a/receiver/JFJochReceiverLite.h b/receiver/JFJochReceiverLite.h index 2b4248310..f33fadd4c 100644 --- a/receiver/JFJochReceiverLite.h +++ b/receiver/JFJochReceiverLite.h @@ -53,6 +53,13 @@ class JFJochReceiverLite : public JFJochReceiver { DarkMaskAnalysis dark_mask_analysis; + // Over the images received, as a detector stream without it counts each as whole + std::atomic collection_efficiency_sum{0.0}; + std::map> end_adu_histogram; + int64_t ExpectedImages() const; + void AddToEndMessage(EndMessage &message) override; + uint64_t GetImagesDropped() const override { return image_puller.GetDroppedImages(); } + void MaskThread(uint32_t id); void MeasurementThread(); void DataAnalysisThread(uint32_t id); diff --git a/receiver/JFJochReceiverRugnux.cpp b/receiver/JFJochReceiverRugnux.cpp index 8a1384bb3..01feb229e 100644 --- a/receiver/JFJochReceiverRugnux.cpp +++ b/receiver/JFJochReceiverRugnux.cpp @@ -281,6 +281,9 @@ void JFJochReceiverRugnux::MeasurementThread() { StartStills(); else { SendStartMessage(); + if (push_images_to_writer) + for (const auto &c: image_puller.GetCalibration()) + image_pusher.SendCalibration(c); // Rugnux reads the images back from the buffer, so it needs one slot per image: a slot taken // again would hand it a later image. Without that the collection is only forwarded. diff --git a/receiver/JFJochReceiverRugnux.h b/receiver/JFJochReceiverRugnux.h index c316012a5..c6f226477 100644 --- a/receiver/JFJochReceiverRugnux.h +++ b/receiver/JFJochReceiverRugnux.h @@ -86,6 +86,7 @@ class JFJochReceiverRugnux : public JFJochReceiver { // the end message goes out with that. Reason: "deadline" or "cancel". void Cut(const std::string &reason); void AddToEndMessage(EndMessage &message) override; + uint64_t GetImagesDropped() const override { return image_puller.GetDroppedImages(); } public: JFJochReceiverRugnux(const DiffractionExperiment &experiment, const PixelMask &pixel_mask, diff --git a/receiver/JFJochReceiverService.cpp b/receiver/JFJochReceiverService.cpp index 61c619f4c..a37c73ca1 100644 --- a/receiver/JFJochReceiverService.cpp +++ b/receiver/JFJochReceiverService.cpp @@ -32,6 +32,16 @@ JFJochReceiverService &JFJochReceiverService::RugnuxLive(bool enable, std::share return *this; } +JFJochReceiverService &JFJochReceiverService::UseFPGAImagePuller(bool enable, size_t slots) { + std::unique_lock ul_state(state_mutex); + if (state != ReceiverState::Idle) + throw JFJochException(JFJochExceptionCategory::WrongDAQState, + "Cannot change the receiver path during data collection"); + fpga_image_puller = enable; + fpga_image_puller_slots = slots; + return *this; +} + RugnuxLiveSnapshot JFJochReceiverService::GetRugnuxStatus() const { return rugnux_status.Get(); } @@ -52,6 +62,8 @@ void JFJochReceiverService::FinalizeMeasurementChangeState() { void JFJochReceiverService::FinalizeMeasurement() { try { receiver->StopReceiver(); + if (fpga_puller) + fpga_puller->Stop(); } catch (...) { FinalizeMeasurementChangeState(); throw; @@ -96,10 +108,27 @@ void JFJochReceiverService::Start(const DiffractionExperiment &experiment, // First clean-up old measurement receiver.reset(); + if (fpga_puller) { + image_puller.reset(); + fpga_puller.reset(); + } preview_image.Configure(experiment, pixel_mask); - switch (experiment.GetDetectorType()) { + // Pedestal runs and host summation stay with JFJochReceiverFPGA + auto detector_type = experiment.GetDetectorType(); + if (fpga_image_puller && (detector_type != DetectorType::DECTRIS) && (experiment.GetImageNum() > 0)) { + if (experiment.IsCPUSummation()) + logger.Warning("Summation above what the card can do needs JFJochReceiverFPGA: FPGA image puller not used"); + else { + fpga_puller = std::make_shared(experiment, pixel_mask, calibration, aq_devices, logger, + nthreads_local, fpga_image_puller_slots); + puller = fpga_puller; + detector_type = DetectorType::DECTRIS; // from here on, one stream receiver input like any other + } + } + + switch (detector_type) { case DetectorType::EIGER: case DetectorType::JUNGFRAU: receiver = std::make_unique(experiment, pixel_mask, @@ -155,6 +184,10 @@ void JFJochReceiverService::Start(const DiffractionExperiment &experiment, // until the next start overwrites it. receiver_status.Clear(); receiver_status.SetProgress({}); + if (fpga_puller) { + image_puller.reset(); + fpga_puller.reset(); + } logger.ErrorException(e); throw; } @@ -162,8 +195,11 @@ void JFJochReceiverService::Start(const DiffractionExperiment &experiment, void JFJochReceiverService::Cancel(bool silent) { std::unique_lock ul(state_mutex); - if (state == ReceiverState::Running) + if (state == ReceiverState::Running) { receiver->Cancel(silent); + if (fpga_puller) + fpga_puller->Cancel(); + } } JFJochReceiverOutput JFJochReceiverService::Stop() { diff --git a/receiver/JFJochReceiverService.h b/receiver/JFJochReceiverService.h index d0c07e8c8..9a875e7ba 100644 --- a/receiver/JFJochReceiverService.h +++ b/receiver/JFJochReceiverService.h @@ -9,6 +9,7 @@ #include #include "JFJochReceiver.h" +#include "FPGAImagePuller.h" #include "../acquisition_device/AcquisitionDeviceGroup.h" #include "../preview/ZMQMetadataSocket.h" @@ -33,6 +34,12 @@ class JFJochReceiverService { double rugnux_deadline_s = 60; // the analysis is cut this long after the last image (0 = never) RugnuxLiveStatus rugnux_status; // of the last collection, kept after it ends + // A PSI detector's frames go through FPGAImagePuller and the stream receivers, instead of + // JFJochReceiverFPGA. Off unless asked for. + bool fpga_image_puller = false; + size_t fpga_image_puller_slots = FPGAImagePuller::DefaultSlots; + std::shared_ptr fpga_puller; + enum class ReceiverState {Idle, Running}; ReceiverState state = ReceiverState::Idle; mutable std::mutex state_mutex; @@ -58,6 +65,7 @@ public: JFJochReceiverService& RugnuxLive(bool enable, std::shared_ptr model, int reserved_cpus, double deadline_s); RugnuxLiveSnapshot GetRugnuxStatus() const; + JFJochReceiverService& UseFPGAImagePuller(bool enable, size_t slots = FPGAImagePuller::DefaultSlots); JFJochReceiverService& PreviewSocket(const std::string &addr, const std::optional &watermark = {}); JFJochReceiverService& MetadataSocket(const std::string &addr); diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt index bd4271384..15e5a031f 100644 --- a/tests/CMakeLists.txt +++ b/tests/CMakeLists.txt @@ -150,6 +150,8 @@ ADD_EXECUTABLE(jfjoch_test ImageMetadataTest.cpp JFJochReceiverLiteTest.cpp JFJochReceiverRugnuxTest.cpp + FPGAImagePullerTest.cpp + FPGAImagePullerPerfTest.cpp GridScanSettingsTest.cpp JFJochReceiverPlotsTest.cpp GoniometerAxisTest.cpp diff --git a/tests/FPGAImagePullerPerfTest.cpp b/tests/FPGAImagePullerPerfTest.cpp new file mode 100644 index 000000000..07421516e --- /dev/null +++ b/tests/FPGAImagePullerPerfTest.cpp @@ -0,0 +1,412 @@ +// SPDX-FileCopyrightText: 2026 Filip Leonarski, Paul Scherrer Institute +// SPDX-License-Identifier: GPL-3.0-only + +// Host-side measurements of FPGAImagePuller without the HLS simulation: a device that hands out already +// converted module frames at a chosen rate. Hidden: run by hand, with FPGA_PULLER_* in the environment. + +#include + +#include + +#include +#include +#include +#include +#include + +#include "../acquisition_device/AcquisitionDevice.h" +#include "../compression/JFJochDecompress.h" +#include "../frame_serialize/CBORStream2Deserializer.h" +#include "../image_pusher/ImagePusher.h" +#include "../receiver/FPGAImagePuller.h" +#include "../receiver/JFJochReceiverService.h" + +namespace { + int64_t Env(const char *name, int64_t def) { + const char *v = std::getenv(name); + return v ? std::atoll(v) : def; + } + + // Converted frame of a module: sparse Poisson photons plus a few spots + std::vector PhotonFrame(uint32_t seed, double photons_per_pixel) { + std::vector ret(RAW_MODULE_SIZE); + std::mt19937 g(seed); + std::poisson_distribution p(photons_per_pixel); + for (auto &v: ret) + v = static_cast(p(g)); + std::uniform_int_distribution where(0, RAW_MODULE_SIZE - 1); + for (int i = 0; i < 2000; i++) + ret[where(g)] = 500; + return ret; + } + + // Like the card, it writes the frame into a buffer it was given and reports the handle; with no buffer + // given back in time the frame is lost. The pixels are put in the buffers beforehand, so that the + // device itself costs nothing: the rate is limited by what is measured, not by this thread. + class ReplayDevice : public AcquisitionDevice { + std::vector> storage; + std::thread generator; + std::atomic stop{false}; + const int64_t rate_hz; + int64_t nimages = 0, nmodules = 0; + std::function present; + + void HW_RunInternalGenerator(const FrameGeneratorConfig &) override {} + void HW_SetSpotFinderParameters(const SpotFinderParameters &) override {} + + void Run() { + const auto t0 = std::chrono::steady_clock::now(); + for (int64_t f = 0; (f < nimages) && !stop; f++) { + std::this_thread::sleep_until(t0 + std::chrono::nanoseconds(f * 1000000000 / rate_hz)); + for (int m = 0; m < nmodules; m++) { + if (!present(f, m)) + continue; + WorkRequest wr{}; + if (work_request_queue.Get(wr) == 0) { + lost++; + continue; + } + auto &s = GetDeviceOutput(static_cast(wr.handle))->module_statistics; + s.frame_number = f; + s.module_number = m; + s.packet_count = 512; + s.timestamp = f * 100; + s.exptime = 100; + Completion c{}; + c.type = Completion::Type::Image; + c.handle = wr.handle; + work_completion_queue.Put(c); + } + } + Completion c{}; + c.type = Completion::Type::End; + work_completion_queue.Put(c); + } + + void Start(const DiffractionExperiment &x, uint32_t) override { + if (generator.joinable()) + generator.join(); + nimages = x.GetImageNum(); + nmodules = x.GetModulesNum(data_stream); + stop = false; + lost = 0; + Completion c{}; + c.type = Completion::Type::Start; + work_completion_queue.Put(c); + generator = std::thread(&ReplayDevice::Run, this); + } + public: + std::atomic lost{0}; + + ReplayDevice(uint16_t stream, size_t nbuffers, int64_t in_rate_hz, double photons_per_pixel, + std::function in_present = [](int64_t, int) { return true; }) + : AcquisitionDevice(stream), rate_hz(in_rate_hz), present(std::move(in_present)) { + max_modules = 32; + std::vector> patterns; + for (int i = 0; i < 16; i++) + patterns.push_back(PhotonFrame(i + 1, photons_per_pixel)); + for (size_t i = 0; i < nbuffers; i++) { + storage.emplace_back(new uint8_t[FPGA_BUFFER_LOCATION_SIZE]); + memset(storage.back().get(), 0, FPGA_BUFFER_LOCATION_SIZE); + buffer_device.push_back(reinterpret_cast(storage.back().get())); + memcpy(buffer_device.back()->pixels, patterns[i % patterns.size()].data(), RAW_MODULE_SIZE * 2); + } + } + ~ReplayDevice() override { + stop = true; + if (generator.joinable()) + generator.join(); + } + void Cancel() override { stop = true; } + uint32_t GetExpectedDescriptorsPerModule() const override { return 1; } + }; + + double CpuSeconds() { + rusage r{}; + getrusage(RUSAGE_SELF, &r); + return static_cast(r.ru_utime.tv_sec + r.ru_stime.tv_sec) + + static_cast(r.ru_utime.tv_usec + r.ru_stime.tv_usec) * 1e-6; + } + + DiffractionExperiment Detector(int64_t modules, int64_t images, CompressionAlgorithm algorithm) { + DiffractionExperiment x(DetJF(modules)); + x.Conversion(); + x.PedestalG0Frames(0).NumTriggers(1).UseInternalPacketGenerator(false).ImagesPerTrigger(images) + .ImagesPerFile(1000).IncidentEnergy_keV(12.4).DetectorDistance_mm(75) + .BeamX_pxl(500).BeamY_pxl(600).Compression(algorithm); + x.DataStreams(1); + return x; + } + + CompressionAlgorithm Algorithm() { + switch (Env("FPGA_PULLER_ALG", 0)) { + case 1: return CompressionAlgorithm::BSHUF_ZSTD; + case 2: return CompressionAlgorithm::NO_COMPRESSION; + default: return CompressionAlgorithm::BSHUF_LZ4; + } + } + + class TimingPusher : public ImagePusher { + public: + std::mutex m; + std::vector decode_s, preprocess_s; + void StartDataCollection(StartMessage &) override {} + bool EndDataCollection(const EndMessage &) override { return true; } + bool SendCalibration(const CompressedImage &) override { return true; } + bool SendImage(const uint8_t *data, size_t size, int64_t) override { + auto d = CBORStream2Deserialize(data, size); + std::unique_lock ul(m); + decode_s.push_back(d->data_message->compression_time_s.value_or(NAN)); + preprocess_s.push_back(d->data_message->preprocessing_time_s.value_or(NAN)); + return true; + } + std::string PrintSetup() const override { return "timing"; } + ImagePusherType GetType() const override { return ImagePusherType::Test; } + }; + + float Median(std::vector v) { + if (v.empty()) + return NAN; + std::sort(v.begin(), v.end()); + return v[v.size() / 2]; + } +} + +// One thread, one image: what assembly and compression cost per image. +TEST_CASE("FPGAImagePullerPerf_SingleThread", "[.FPGAImagePullerPerf]") { + for (const double photons: {0.1, 1.0}) { + for (const int64_t modules: {2, 18}) { + for (const auto alg: {CompressionAlgorithm::BSHUF_LZ4, CompressionAlgorithm::BSHUF_ZSTD}) { + auto x = Detector(modules, 1, alg); + std::vector> frames; + for (int64_t m = 0; m < modules; m++) + frames.push_back(PhotonFrame(m + 1, photons)); + + FrameTransformation t(x); + std::vector out(MaxCompressedSize(alg, x.GetPixelsNum(), 2)); + const int rep = 10; + double assemble = 0, compress = 0, decompress = 0; + size_t size = 0; + std::vector decoded; + for (int r = 0; r < rep; r++) { + const auto t0 = std::chrono::steady_clock::now(); + for (int64_t m = 0; m < modules; m++) + t.ProcessModule(frames[m].data(), m, 0); + const auto t1 = std::chrono::steady_clock::now(); + size = t.CompressImage(out.data(), out.size()); + const auto t2 = std::chrono::steady_clock::now(); + CompressedImage(out.data(), size, x.GetXPixelsNum(), x.GetYPixelsNum(), x.GetImageMode(), alg) + .GetUncompressed(decoded); + const auto t3 = std::chrono::steady_clock::now(); + assemble += std::chrono::duration(t1 - t0).count(); + compress += std::chrono::duration(t2 - t1).count(); + decompress += std::chrono::duration(t3 - t2).count(); + } + std::cout << "single thread: " << modules << " modules, " << photons << " ph/pixel, " + << (alg == CompressionAlgorithm::BSHUF_LZ4 ? "BSHUF_LZ4" : "BSHUF_ZSTD") + << ": raw " << x.GetPixelsNum() * 2 / 1000000.0 << " MB -> " << size / 1000000.0 << " MB" + << "; assemble " << assemble / rep * 1e3 << " ms, compress " << compress / rep * 1e3 + << " ms, host decompress " << decompress / rep * 1e3 << " ms" << std::endl; + } + } + } +} + +// The whole puller at a rate, the consumer letting go of each image at once. +TEST_CASE("FPGAImagePullerPerf_Rate", "[.FPGAImagePullerPerf]") { + const int64_t modules = Env("FPGA_PULLER_MODULES", 2); + const int64_t rate = Env("FPGA_PULLER_RATE", 100); + const int64_t images = Env("FPGA_PULLER_IMAGES", 500); + const int64_t threads = Env("FPGA_PULLER_THREADS", 8); + const double photons = static_cast(Env("FPGA_PULLER_MILLIPHOTONS", 100)) * 1e-3; + + auto x = Detector(modules, images, Algorithm()); + AcquisitionDeviceGroup devices; + auto device = std::make_unique(0, 512, rate, photons); + auto *replay = device.get(); + devices.Add(std::move(device)); + + Logger logger("FPGAImagePullerPerf"); + PixelMask mask(x); + const double cpu0 = CpuSeconds(); + const auto t0 = std::chrono::steady_clock::now(); + FPGAImagePuller puller(x, mask, nullptr, devices, logger, threads); + + int64_t received = 0, last = -1; + bool in_order = true, end = false; + size_t compressed = 0; + while (!end) { + const auto m = puller.PollImage(std::chrono::milliseconds(10)); + if (!m) + continue; + if (m->cbor->data_message) { + received++; + in_order = in_order && (m->cbor->data_message->number > last); + last = m->cbor->data_message->number; + compressed += m->cbor->data_message->image.GetCompressedSize(); + } + end = m->cbor->end_message.has_value(); + } + const double wall = std::chrono::duration(std::chrono::steady_clock::now() - t0).count(); + puller.Stop(); + const double cpu = CpuSeconds() - cpu0; + std::cout << "rate: " << modules << " modules, " << rate << " Hz asked, " << threads << " compression threads, " + << photons << " ph/pixel: " << received << "/" << images << " images in " << wall << " s = " + << received / wall << " Hz, dropped " << puller.GetDroppedImages() << ", lost at the device " + << replay->lost.load() << " module frames, in order " << in_order << ", " + << compressed / std::max(received, 1) / 1000000.0 << " MB/image, CPU " << cpu / wall + << " cores" << std::endl; +} + +// Through Lite, on whatever device the analysis picks: what decoding (or just getting the image across) costs. +TEST_CASE("FPGAImagePullerPerf_Lite", "[.FPGAImagePullerPerf]") { + const int64_t modules = Env("FPGA_PULLER_MODULES", 2); + const int64_t rate = Env("FPGA_PULLER_RATE", 20); + const int64_t images = Env("FPGA_PULLER_IMAGES", 100); + const int64_t threads = Env("FPGA_PULLER_THREADS", 2); + const double photons = static_cast(Env("FPGA_PULLER_MILLIPHOTONS", 100)) * 1e-3; + + auto x = Detector(modules, images, Algorithm()); + AcquisitionDeviceGroup devices; + devices.Add(std::make_unique(0, 512, rate, photons)); + + Logger logger("FPGAImagePullerPerf"); + TimingPusher pusher; + JFJochReceiverService service(devices, logger, pusher, 2048); + service.NumThreads(threads); + SpotFindingSettings s = DiffractionExperiment::DefaultDataProcessingSettings(); + s.enable = false; + s.indexing = false; + service.SetSpotFindingSettings(s); + service.UseFPGAImagePuller(true); + + PixelMask mask(x); + service.Start(x, mask, nullptr); + const auto out = service.Stop(); + std::cout << "lite: " << modules << " modules, " << rate << " Hz, " << threads << " threads, algorithm " + << Env("FPGA_PULLER_ALG", 0) << ", " << photons << " ph/pixel: collected " << out.status.images_collected + << " dropped " << out.status.images_dropped << " sent " << out.status.images_sent + << "; ratio " << out.status.compressed_ratio.value_or(0) + << "; median compression_time_s (decode on the analysis device) " << Median(pusher.decode_s) * 1e3 + << " ms, preprocessing_time_s (rest of getting it into the preprocessed buffer) " + << Median(pusher.preprocess_s) * 1e3 << " ms" << std::endl; +} + +namespace { + struct Stream { + std::vector numbers; + std::vector efficiency; + bool start_first = false, end = false; + }; + + // Reads until the end message, or until the deadline when `until_image` has not come by then + Stream ReadAll(FPGAImagePuller &puller) { + Stream ret; + bool first = true; + const auto t0 = std::chrono::steady_clock::now(); + while (!ret.end && (std::chrono::steady_clock::now() - t0 < std::chrono::seconds(120))) { + const auto m = puller.PollImage(std::chrono::milliseconds(10)); + if (!m) + continue; + if (first) + ret.start_first = m->cbor->start_message.has_value(); + first = false; + if (m->cbor->data_message) { + ret.numbers.push_back(m->cbor->data_message->number); + ret.efficiency.push_back(m->cbor->data_message->image_collection_efficiency.value_or(-1)); + } + ret.end = m->cbor->end_message.has_value(); + } + return ret; + } + + DiffractionExperiment TwoCards(int64_t images) { + auto x = Detector(4, images, CompressionAlgorithm::BSHUF_LZ4); + x.DataStreams(2); + return x; + } +} + +// Replay rather than simulation: the HLS simulation keeps its state in statics, so only one card at a time. +TEST_CASE("FPGAImagePuller_Replay_TwoCards_InOrder", "[FPGAImagePuller]") { + const auto x = TwoCards(300); + REQUIRE(x.GetDataStreamsNum() == 2); + AcquisitionDeviceGroup devices; + devices.Add(std::make_unique(0, 64, 500, 0.1)); + devices.Add(std::make_unique(1, 64, 500, 0.1)); + + Logger logger("FPGAImagePuller_TwoCards"); + PixelMask mask(x); + FPGAImagePuller puller(x, mask, nullptr, devices, logger, 8); + const auto s = ReadAll(puller); + REQUIRE_NOTHROW(puller.Stop()); + + CHECK(s.start_first); + CHECK(s.end); + REQUIRE(s.numbers.size() == 300); + for (int i = 0; i < 300; i++) + CHECK(s.numbers[i] == i); + CHECK(puller.GetDroppedImages() == 0); +} + +// A module that dies: nobody stops the run, the images behind it keep coming in order, with the module +// taken as missing once the others are far enough ahead. +TEST_CASE("FPGAImagePuller_Replay_DeadModule", "[FPGAImagePuller]") { + const auto x = Detector(2, 300, CompressionAlgorithm::BSHUF_LZ4); + AcquisitionDeviceGroup devices; + devices.Add(std::make_unique(0, 64, 200, 0.1, [](int64_t f, int m) { return (m == 0) || (f < 100); })); + + Logger logger("FPGAImagePuller_DeadModule"); + PixelMask mask(x); + FPGAImagePuller puller(x, mask, nullptr, devices, logger, 4); + const auto s = ReadAll(puller); + REQUIRE_NOTHROW(puller.Stop()); + + CHECK(s.end); + REQUIRE(s.numbers.size() == 300); + for (int i = 0; i < 300; i++) { + CHECK(s.numbers[i] == i); + CHECK(s.efficiency[i] == ((i < 100) ? 1.0f : 0.5f)); + } + CHECK(puller.GetDroppedImages() == 0); +} + +// A whole card dies, the other one goes on: as above, with no module of that card ever reporting again, +// until the card is told to stop. +TEST_CASE("FPGAImagePuller_Replay_DeadCard", "[FPGAImagePuller]") { + const auto x = TwoCards(200); + AcquisitionDeviceGroup devices; + devices.Add(std::make_unique(0, 64, 200, 0.1)); + devices.Add(std::make_unique(1, 64, 200, 0.1, [](int64_t f, int) { return f < 20; })); + + Logger logger("FPGAImagePuller_DeadCard"); + PixelMask mask(x); + FPGAImagePuller puller(x, mask, nullptr, devices, logger, 4); + const auto s = ReadAll(puller); + REQUIRE_NOTHROW(puller.Stop()); + + CHECK(s.end); + REQUIRE(s.numbers.size() == 200); + for (int i = 0; i < 200; i++) + CHECK(s.numbers[i] == i); + CHECK(s.efficiency[150] == 0.5f); + CHECK(puller.GetDroppedImages() == 0); +} + +TEST_CASE("FPGAImagePuller_Replay_9M", "[FPGAImagePuller]") { + const auto x = Detector(18, 100, CompressionAlgorithm::BSHUF_LZ4); + AcquisitionDeviceGroup devices; + auto device = std::make_unique(0, 512, 100, 0.1); + auto *replay = device.get(); + devices.Add(std::move(device)); + + Logger logger("FPGAImagePuller_FewBuffers"); + PixelMask mask(x); + FPGAImagePuller puller(x, mask, nullptr, devices, logger, 8); + const auto s = ReadAll(puller); + REQUIRE_NOTHROW(puller.Stop()); + std::cout << "9M replay at 100 Hz: " << s.numbers.size() << " images, dropped " << puller.GetDroppedImages() + << ", module frames lost at the device " << replay->lost.load() << std::endl; + CHECK(s.end); + CHECK(s.numbers.size() + puller.GetDroppedImages() == 100); +} diff --git a/tests/FPGAImagePullerTest.cpp b/tests/FPGAImagePullerTest.cpp new file mode 100644 index 000000000..da2aff5ce --- /dev/null +++ b/tests/FPGAImagePullerTest.cpp @@ -0,0 +1,705 @@ +// SPDX-FileCopyrightText: 2026 Filip Leonarski, Paul Scherrer Institute +// SPDX-License-Identifier: GPL-3.0-only + +// The same frames through JFJochReceiverFPGA and through FPGAImagePuller + JFJochReceiverLite, on the HLS +// simulation of the card. The pixels must be the same; the analysis differs by design, and its numbers are printed. + +#include + +#include +#include +#include +#include +#include +#include + +#include "../acquisition_device/HLSSimulatedDevice.h" +#include "../compression/JFJochDecompress.h" +#include "../frame_serialize/CBORStream2Deserializer.h" +#include "../image_pusher/HDF5FilePusher.h" +#include "../image_pusher/ImagePusher.h" +#include "../writer/HDF5Objects.h" +#include "../writer/StreamWriter.h" +#include "../receiver/FPGAImagePuller.h" +#include "../receiver/JFJochReceiverService.h" +#include "../rugnux/RugnuxDefaults.h" + +using namespace std::chrono_literals; + +namespace { + struct Received { + DataMessage message; + std::vector compressed; + }; + + class CollectingPusher : public ImagePusher { + public: + std::mutex m; + std::map images; + std::vector arrival; + std::vector sequence; + EndMessage end; + std::chrono::milliseconds delay{0}; + + void StartDataCollection(StartMessage &) override { + std::unique_lock ul(m); + sequence.emplace_back("start"); + } + bool EndDataCollection(const EndMessage &message) override { + std::unique_lock ul(m); + sequence.emplace_back("end"); + end = message; + return true; + } + bool SendCalibration(const CompressedImage &c) override { + std::unique_lock ul(m); + sequence.emplace_back("calibration " + c.GetChannel()); + return true; + } + bool SendImage(const uint8_t *image_data, size_t image_size, int64_t image_number) override { + std::this_thread::sleep_for(delay); + auto d = CBORStream2Deserialize(image_data, image_size); + Received r; + r.message = d->data_message.value(); + const uint8_t *p = r.message.image.GetCompressed(); + r.compressed.assign(p, p + r.message.image.GetCompressedSize()); + r.message.image = CompressedImage(r.compressed.data(), r.compressed.size(), r.message.image.GetWidth(), + r.message.image.GetHeight(), r.message.image.GetMode(), + r.message.image.GetCompressionAlgorithm()); + std::unique_lock ul(m); + sequence.emplace_back("image"); + arrival.push_back(image_number); + images.emplace(image_number, std::move(r)); + return true; + } + std::string PrintSetup() const override { return "collecting"; } + ImagePusherType GetType() const override { return ImagePusherType::Test; } + }; + + float Pedestal(int module, int pixel) { + return 3000 + 10 * module + pixel % 50; + } + + JFCalibration Calibration(const DiffractionExperiment &x) { + JFCalibration ret(x); + std::vector g0(RAW_MODULE_SIZE), g1(RAW_MODULE_SIZE), g2(RAW_MODULE_SIZE); + for (int s = 0; s < x.GetStorageCellNumber(); s++) + for (int m = 0; m < x.GetModulesNum(); m++) { + for (int i = 0; i < RAW_MODULE_SIZE; i++) { + g0[i] = Pedestal(m, i); + g1[i] = 15000 + 10 * m + i % 50; + g2[i] = 14000 + 10 * m + i % 50; + } + ret.Pedestal(m, 0, s).LoadPedestal(g0); + ret.Pedestal(m, 1, s).LoadPedestal(g1); + ret.Pedestal(m, 2, s).LoadPedestal(g2); + } + return ret; + } + + // Raw detector frame of a module: background of a few photons, some spots, one saturated and one bad + // pixel, different for each frame. + std::vector Frame(int64_t frame, int module, bool jungfrau) { + std::vector ret(RAW_MODULE_SIZE); + std::mt19937 g(static_cast(frame * 100 + module)); + std::uniform_int_distribution background(0, jungfrau ? 300 : 3); + for (int i = 0; i < RAW_MODULE_SIZE; i++) + ret[i] = (jungfrau ? static_cast(Pedestal(module, i)) : 0) + background(g); + + std::uniform_int_distribution where_x(20, RAW_MODULE_COLS - 20), where_y(20, RAW_MODULE_LINES - 20); + for (int s = 0; s < 40; s++) { + const int x0 = where_x(g), y0 = where_y(g); + for (int y = y0; y < y0 + 3; y++) + for (int x = x0; x < x0 + 3; x++) + ret[y * RAW_MODULE_COLS + x] = jungfrau ? 12000 : 2000; + } + ret[50 * RAW_MODULE_COLS + 500] = jungfrau ? 0xC000 : 0xFFFE; + ret[60 * RAW_MODULE_COLS + 600] = 0xFFFF; + return ret; + } + + SpotFindingSettings Spots() { + SpotFindingSettings s = DiffractionExperiment::DefaultDataProcessingSettings(); + s.signal_to_noise_threshold = 3; + s.photon_count_threshold = 2; + s.min_pix_per_spot = 1; + s.indexing = false; + return s; + } + + AcquisitionDeviceGroup Devices(const DiffractionExperiment &x, size_t buffers = 64) { + AcquisitionDeviceGroup ret; + for (int d = 0; d < x.GetDataStreamsNum(); d++) + ret.Add(std::make_unique(d, buffers)); + return ret; + } + + struct PathResult { + std::map images; + std::vector arrival; + std::vector sequence; + EndMessage end; + JFJochReceiverOutput output; + }; + + PathResult Run(const DiffractionExperiment &x, bool puller, int64_t distinct_frames) { + auto devices = Devices(x); + Logger logger("FPGAImagePullerTest"); + CollectingPusher pusher; + JFJochReceiverService service(devices, logger, pusher, 256); + service.NumThreads(4); + service.SetSpotFindingSettings(Spots()); + service.UseFPGAImagePuller(puller); + + const bool jungfrau = x.GetDetectorType() == DetectorType::JUNGFRAU; + for (int64_t f = 0; f < distinct_frames; f++) { + std::vector raw; + for (int m = 0; m < x.GetModulesNum(); m++) { + const auto frame = Frame(f, m, jungfrau); + raw.insert(raw.end(), frame.begin(), frame.end()); + } + service.LoadInternalGeneratorImage(x, raw, f); + } + + PixelMask mask(x); + const JFCalibration calib = Calibration(x); + service.Start(x, mask, &calib); + PathResult ret; + ret.output = service.Stop(); + ret.images = std::move(pusher.images); + ret.arrival = std::move(pusher.arrival); + ret.sequence = std::move(pusher.sequence); + ret.end = pusher.end; + return ret; + } + + std::vector Pixels(const CompressedImage &image) { + std::vector ret; + image.GetUncompressed(ret); + return ret; + } + + template + void Row(const std::string &name, const std::optional &a, const std::optional &b) { + if (a == b) + return; + std::cout << " differs " << name << ": FPGA "; + if (a) std::cout << *a; else std::cout << "-"; + std::cout << " puller "; + if (b) std::cout << *b; else std::cout << "-"; + std::cout << std::endl; + } + + // Must be the same: asserted. Analysis: printed for the first image and summed over all. + void Compare(const std::string &name, const PathResult &a, const PathResult &b) { + std::cout << "=== " << name << std::endl; + REQUIRE(a.images.size() == b.images.size()); + REQUIRE(!a.images.empty()); + + int64_t spots_a = 0, spots_b = 0, sat_a = 0, sat_b = 0, err_a = 0, err_b = 0, strong_a = 0, strong_b = 0; + double dot = 0, norm_a = 0, norm_b = 0, ratio_sum = 0; + size_t ratio_n = 0, bins_a = 0, bins_b = 0; + + for (const auto &[n, ia]: a.images) { + INFO("image " << n); + REQUIRE(b.images.contains(n)); + const auto &ib = b.images.at(n); + const auto &ma = ia.message; + const auto &mb = ib.message; + + CHECK(ma.image.GetMode() == mb.image.GetMode()); + CHECK(ma.image.GetCompressionAlgorithm() == mb.image.GetCompressionAlgorithm()); + CHECK(ma.image.GetWidth() == mb.image.GetWidth()); + CHECK(ma.image.GetHeight() == mb.image.GetHeight()); + CHECK(Pixels(ma.image) == Pixels(mb.image)); + CHECK(ia.compressed.size() == ib.compressed.size()); + + // From the source + CHECK(ma.number == mb.number); + CHECK(ma.original_number == mb.original_number); + CHECK(ma.packets_expected == mb.packets_expected); + CHECK(ma.packets_received == mb.packets_received); + CHECK(ma.image_collection_efficiency == mb.image_collection_efficiency); + CHECK(ma.timestamp == mb.timestamp); + CHECK(ma.timestamp_base == mb.timestamp_base); + CHECK(ma.exptime == mb.exptime); + CHECK(ma.exptime_base == mb.exptime_base); + CHECK(ma.jf_info == mb.jf_info); + CHECK(ma.storage_cell == mb.storage_cell); + CHECK(ma.xfel_pulse_id == mb.xfel_pulse_id); + CHECK(ma.xfel_event_code == mb.xfel_event_code); + CHECK(ma.pixel_sum == mb.pixel_sum); + CHECK(ma.user_data == mb.user_data); + CHECK(ma.run_number == mb.run_number); + CHECK(ma.run_name == mb.run_name); + + if (n == a.images.begin()->first) { + Row("saturated_pixel_count", ma.saturated_pixel_count, mb.saturated_pixel_count); + Row("error_pixel_count", ma.error_pixel_count, mb.error_pixel_count); + Row("min_viable_pixel_value", ma.min_viable_pixel_value, mb.min_viable_pixel_value); + Row("max_viable_pixel_value", ma.max_viable_pixel_value, mb.max_viable_pixel_value); + Row("strong_pixel_count", ma.strong_pixel_count, mb.strong_pixel_count); + Row("spot_count", ma.spot_count, mb.spot_count); + Row("bkg_estimate", ma.bkg_estimate, mb.bkg_estimate); + Row("ice_ring_score", ma.ice_ring_score, mb.ice_ring_score); + Row("receiver_aq_dev_delay", ma.receiver_aq_dev_delay, mb.receiver_aq_dev_delay); + Row("receiver_buf_in_preparation", ma.receiver_buf_in_preparation, mb.receiver_buf_in_preparation); + Row("receiver_buf_in_sending", ma.receiver_buf_in_sending, mb.receiver_buf_in_sending); + std::cout << " compression_ratio FPGA " << ma.compression_ratio.value_or(0) + << " puller " << mb.compression_ratio.value_or(0) << std::endl; + } + spots_a += ma.spots.size(); + spots_b += mb.spots.size(); + sat_a += ma.saturated_pixel_count.value_or(0); + sat_b += mb.saturated_pixel_count.value_or(0); + err_a += ma.error_pixel_count.value_or(0); + err_b += mb.error_pixel_count.value_or(0); + strong_a += ma.strong_pixel_count.value_or(0); + strong_b += mb.strong_pixel_count.value_or(0); + + bins_a = ma.az_int_profile.size(); + bins_b = mb.az_int_profile.size(); + for (size_t i = 0; i < std::min(bins_a, bins_b); i++) { + if (std::isfinite(ma.az_int_profile[i]) && std::isfinite(mb.az_int_profile[i])) { + dot += ma.az_int_profile[i] * mb.az_int_profile[i]; + norm_a += ma.az_int_profile[i] * ma.az_int_profile[i]; + norm_b += mb.az_int_profile[i] * mb.az_int_profile[i]; + if (ma.az_int_profile[i] != 0) { + ratio_sum += mb.az_int_profile[i] / ma.az_int_profile[i]; + ratio_n++; + } + } + } + } + std::cout << " over " << a.images.size() << " images:" + << " spots FPGA " << spots_a << " puller " << spots_b + << " | strong " << strong_a << " / " << strong_b + << " | saturated " << sat_a << " / " << sat_b + << " | error " << err_a << " / " << err_b + << " | az bins " << bins_a << " / " << bins_b + << " cosine " << (norm_a * norm_b > 0 ? dot / std::sqrt(norm_a * norm_b) : 0) + << " mean ratio puller/FPGA " << (ratio_n > 0 ? ratio_sum / ratio_n : 0) << std::endl; + + const auto &sa = a.output.status; + const auto &sb = b.output.status; + std::cout << " status: efficiency " << a.output.efficiency << " / " << b.output.efficiency + << " | collected " << sa.images_collected << " / " << sb.images_collected + << " | sent " << sa.images_sent << " / " << sb.images_sent + << " | dropped " << sa.images_dropped << " / " << sb.images_dropped + << " | max_image_sent " << sa.max_image_number_sent << " / " << sb.max_image_number_sent + << " | compressed_size " << sa.compressed_size << " / " << sb.compressed_size + << " | ratio " << sa.compressed_ratio.value_or(0) << " / " << sb.compressed_ratio.value_or(0) + << " | max_delay " << sa.max_receive_delay.value_or(0) << " / " << sb.max_receive_delay.value_or(0) + << std::endl; + + CHECK(a.output.efficiency == b.output.efficiency); + CHECK(sa.images_collected == sb.images_collected); + CHECK(sa.images_sent == sb.images_sent); + CHECK(sb.images_dropped == 0); + CHECK(sa.max_image_number_sent == sb.max_image_number_sent); + CHECK(sa.compressed_size == sb.compressed_size); + CHECK(a.end.images_collected_count == b.end.images_collected_count); + CHECK(a.end.images_sent_to_write_count == b.end.images_sent_to_write_count); + CHECK(a.end.max_image_number == b.end.max_image_number); + CHECK(a.end.adu_histogram == b.end.adu_histogram); + + // Calibration between start and the first image, in the order the FPGA path sends it + CHECK(a.sequence.front() == "start"); + CHECK(b.sequence.front() == "start"); + CHECK(a.sequence.back() == "end"); + CHECK(b.sequence.back() == "end"); + CHECK(a.sequence == b.sequence); + + int64_t displacement = 0; + for (size_t i = 0; i < b.arrival.size(); i++) + displacement = std::max(displacement, std::abs(b.arrival[i] - static_cast(i))); + std::cout << " max displacement at the pusher (Lite's threads): " << displacement << " images" << std::endl; + } + + void CompareBoth(const std::string &name, const DiffractionExperiment &x, int64_t distinct_frames) { + Compare(name, Run(x, false, distinct_frames), Run(x, true, distinct_frames)); + } + + DiffractionExperiment Base(const DetectorSetup &detector, int64_t images, int64_t distinct_frames) { + DiffractionExperiment x(detector); + x.PedestalG0Frames(0).NumTriggers(1).UseInternalPacketGenerator(true).InternalPacketGeneratorImages(distinct_frames) + .ImagesPerTrigger(images).ImagesPerFile(10).IncidentEnergy_keV(12.4) + .DetectorDistance_mm(75).BeamX_pxl(500).BeamY_pxl(600); + return x; + } +} + +TEST_CASE("FPGAImagePuller_SameAsFPGA_JF_Conversion", "[FPGAImagePuller]") { + auto x = Base(DetJF(2), 6, 3); + x.Conversion(); + x.Compression(CompressionAlgorithm::BSHUF_LZ4); + CompareBoth("JF conversion, int16, BSHUF_LZ4", x, 3); +} + +TEST_CASE("FPGAImagePuller_SameAsFPGA_JF_Conversion_ApplyMask_ROI", "[FPGAImagePuller]") { + auto x = Base(DetJF(2), 4, 2); + x.Conversion(); + x.Compression(CompressionAlgorithm::BSHUF_ZSTD).ApplyPixelMask(true); + x.ROI().SetROI(ROIDefinition{.boxes = {ROIBox("box", 380, 450, 280, 320)}, .circles = {}, .azimuthal = {}}); + const auto a = Run(x, false, 2); + const auto b = Run(x, true, 2); + Compare("JF conversion, applied mask, ROI, BSHUF_ZSTD", a, b); + const auto &ra = a.images.begin()->second.message.roi.at("box"); + const auto &rb = b.images.begin()->second.message.roi.at("box"); + std::cout << " ROI sum " << ra.sum << " / " << rb.sum << " pixels " << ra.pixels << " / " << rb.pixels + << " max " << ra.max_count << " / " << rb.max_count << std::endl; +} + +TEST_CASE("FPGAImagePuller_SameAsFPGA_JF_Conversion_Unsigned", "[FPGAImagePuller]") { + auto x = Base(DetJF(2), 4, 2); + x.Conversion(); + x.Compression(CompressionAlgorithm::BSHUF_LZ4).BitDepthImage(16).PixelSigned(false); + CompareBoth("JF conversion, uint16, BSHUF_LZ4", x, 2); +} + +TEST_CASE("FPGAImagePuller_SameAsFPGA_JF_Conversion_Summation", "[FPGAImagePuller]") { + auto x = Base(DetJF(1), 3, 2); + x.Conversion(); + x.Summation(2).Compression(CompressionAlgorithm::BSHUF_LZ4); + CompareBoth("JF conversion, card sums 2 frames, int32", x, 2); +} + +TEST_CASE("FPGAImagePuller_SameAsFPGA_JF_Conversion_NoGeometry", "[FPGAImagePuller]") { + auto x = Base(DetJF(2), 4, 2); + x.Conversion(); + x.GeometryTransformation(false).Compression(CompressionAlgorithm::BSHUF_LZ4); + CompareBoth("JF conversion, no geometry transform", x, 2); +} + +TEST_CASE("FPGAImagePuller_SameAsFPGA_JF_Raw", "[FPGAImagePuller]") { + auto x = Base(DetJF(2), 4, 2); + x.Raw(); + x.Compression(CompressionAlgorithm::BSHUF_LZ4); + CompareBoth("JF raw, BSHUF_LZ4", x, 2); +} + +TEST_CASE("FPGAImagePuller_SameAsFPGA_JF_Raw_NoCompression", "[FPGAImagePuller]") { + auto x = Base(DetJF(1), 3, 1); + x.Raw(); + x.Compression(CompressionAlgorithm::NO_COMPRESSION); + CompareBoth("JF raw, no compression", x, 1); +} + +TEST_CASE("FPGAImagePuller_SameAsFPGA_EIGER_Conversion", "[FPGAImagePuller]") { + auto x = Base(DetEIGER(2), 4, 2); + x.Conversion(); + x.Compression(CompressionAlgorithm::BSHUF_LZ4); + CompareBoth("EIGER assembled, 16 bit, BSHUF_LZ4", x, 2); +} + +TEST_CASE("FPGAImagePuller_SameAsFPGA_EIGER_Raw", "[FPGAImagePuller]") { + auto x = Base(DetEIGER(2), 4, 2); + x.Raw(); + x.Compression(CompressionAlgorithm::BSHUF_ZSTD); + CompareBoth("EIGER raw layout, 16 bit, BSHUF_ZSTD", x, 2); +} + +TEST_CASE("FPGAImagePuller_SameAsFPGA_EIGER_32bit", "[FPGAImagePuller]") { + auto x = Base(DetEIGER(2), 3, 1); + x.Compression(CompressionAlgorithm::BSHUF_LZ4); + x.EigerBitDepth(32); + // Pattern from the generator's memory, as the 16-bit raw frames do not map onto a 32-bit stream + auto devices = Devices(x); + Logger logger("FPGAImagePuller_EIGER32"); + for (const bool puller: {false, true}) { + CollectingPusher pusher; + JFJochReceiverService service(devices, logger, pusher, 256); + service.NumThreads(4).SetSpotFindingSettings(Spots()); + service.UseFPGAImagePuller(puller); + PixelMask mask(x); + service.Start(x, mask, nullptr); + const auto out = service.Stop(); + CHECK(out.efficiency == 1.0f); + CHECK(pusher.images.size() == 3); + CHECK(pusher.images.at(0).message.image.GetMode() == x.GetImageMode()); + } +} + +// Nobody takes the images: the slots run out and later images are dropped whole, without any waiting. +TEST_CASE("FPGAImagePuller_SlowConsumer_Puller", "[FPGAImagePuller]") { + auto x = Base(DetJF(1), 20, 1); + x.Conversion(); + x.Compression(CompressionAlgorithm::BSHUF_LZ4); + + auto devices = Devices(x); + Logger logger("FPGAImagePuller_SlowConsumer"); + PixelMask mask(x); + const JFCalibration calib = Calibration(x); + FPGAImagePuller puller(x, mask, &calib, devices, logger, 2, 4); + + // Not read until the card is through, however long the simulation takes + while (!devices[0].Counters().IsAcquisitionFinished()) + std::this_thread::sleep_for(10ms); + + std::vector numbers; + bool end = false; + const auto t0 = std::chrono::steady_clock::now(); + while (!end && std::chrono::steady_clock::now() - t0 < 60s) { + const auto m = puller.PollImage(10ms); + if (!m) + continue; + if (m->cbor->data_message) + numbers.push_back(m->cbor->data_message->number); + end = m->cbor->end_message.has_value(); + } + CHECK(end); + puller.Stop(); + std::cout << "slow consumer, puller alone: " << numbers.size() << " images, " + << puller.GetDroppedImages() << " dropped" << std::endl; + CHECK(numbers.size() + puller.GetDroppedImages() == 20); + CHECK(puller.GetDroppedImages() > 0); + for (size_t i = 1; i < numbers.size(); i++) + CHECK(numbers[i] > numbers[i - 1]); +} + +// The same behind Lite: a writer that takes its time keeps Lite's only thread, and the drops reach the statistics. +TEST_CASE("FPGAImagePuller_SlowConsumer_Receiver", "[FPGAImagePuller]") { + auto x = Base(DetJF(1), 24, 1); + x.Conversion(); + x.Compression(CompressionAlgorithm::BSHUF_LZ4); + + auto devices = Devices(x); + Logger logger("FPGAImagePuller_SlowReceiver"); + CollectingPusher pusher; + pusher.delay = 3s; + JFJochReceiverService service(devices, logger, pusher, 256); + service.NumThreads(1).SetSpotFindingSettings(Spots()); + service.UseFPGAImagePuller(true, 2); + + PixelMask mask(x); + const JFCalibration calib = Calibration(x); + service.Start(x, mask, &calib); + const auto out = service.Stop(); + + std::cout << "slow consumer, Lite: collected " << out.status.images_collected << " dropped " + << out.status.images_dropped << " efficiency " << out.efficiency << std::endl; + CHECK(out.status.images_dropped > 0); + CHECK(out.status.images_collected + out.status.images_dropped == 24); + CHECK(out.efficiency == Catch::Approx(static_cast(out.status.images_collected) / 24.0f)); + CHECK(pusher.sequence.back() == "end"); +} + +// A module frame missing in the middle, and a module that dies: the images keep coming with no help +// from the end of the run, and the missing module is filled, as in JFJochReceiverFPGA. +TEST_CASE("FPGAImagePuller_MissingModule", "[FPGAImagePuller]") { + constexpr int64_t images = 62; + constexpr int64_t dies_at = 3; + auto x = Base(DetJF(2), images, 1); + x.UseInternalPacketGenerator(false).Conversion(); + x.Compression(CompressionAlgorithm::BSHUF_LZ4); + + auto device = std::make_unique(0, 64); + for (int64_t f = 0; f < images; f++) + for (int m = 0; m < 2; m++) { + if ((m == 1) && (f == 1 || f >= dies_at)) + continue; + const auto frame = Frame(f, m, true); + for (int p = 0; p < 128; p++) + device->CreateJFPacket(x, f + 1, p, m, frame.data() + p * 4096); + } + AcquisitionDeviceGroup devices; + devices.Add(std::move(device)); + + Logger logger("FPGAImagePuller_MissingModule"); + PixelMask mask(x); + const JFCalibration calib = Calibration(x); + FPGAImagePuller puller(x, mask, &calib, devices, logger, 2); + + // No final packet and no cancel yet: images must come through anyway, up to the lag behind the card + std::vector numbers; + std::vector efficiency; + bool stop_requested = false, end = false; + size_t images_before_cancel = 0; + const auto t0 = std::chrono::steady_clock::now(); + auto last = t0; + while (!end && std::chrono::steady_clock::now() - t0 < 900s) { + const auto m = puller.PollImage(10ms); + if (!m) { + if (!stop_requested && std::chrono::steady_clock::now() - last > 60s) { + stop_requested = true; + std::cout << "dead module, before cancel: fastest " << devices[0].Counters().GetFastestFrameNumber() + << " module 0 at " << devices[0].Counters().GetCurrFrameNumber(0) + << " module 1 at " << devices[0].Counters().GetCurrFrameNumber(1) + << " images so far " << numbers.size() << std::endl; + puller.Cancel(); + } + continue; + } + last = std::chrono::steady_clock::now(); + if (m->cbor->data_message) { + const auto &d = *m->cbor->data_message; + numbers.push_back(d.number); + efficiency.push_back(d.image_collection_efficiency.value_or(-1)); + if (!stop_requested) + images_before_cancel = numbers.size(); + + if (d.number == 1) { + // Module 1 absent from this frame: its pixels are the fill value + std::vector tmp; + const auto *p = reinterpret_cast(d.image.GetUncompressedPtr(tmp)); + CHECK(p[x.GetPixel0OfModuleConv(1)] == x.GetImageFillValue()); + CHECK(p[x.GetPixel0OfModuleConv(0)] != x.GetImageFillValue()); + } + } + end = m->cbor->end_message.has_value(); + } + CHECK(end); + REQUIRE_NOTHROW(puller.Stop()); + std::cout << "dead module: " << numbers.size() << " images, before any cancel: " + << images_before_cancel << std::endl; + // Image i is let go once the card reports frame i + 2 + MissingAfterFrames of the other module + CHECK(images_before_cancel >= 4); + REQUIRE(numbers.size() >= 4); + for (size_t i = 0; i < numbers.size(); i++) + CHECK(numbers[i] == static_cast(i)); + CHECK(efficiency[0] == 1.0f); + CHECK(efficiency[1] == 0.5f); + // Frame 2 of module 1 is sent, but the card holds a module's last frames until newer ones push them + // out: given up on, it comes with the cancel. Its buffer went back. + CHECK(efficiency[3] == 0.5f); + CHECK(efficiency.back() == 0.5f); +} + +TEST_CASE("FPGAImagePuller_Cancel", "[FPGAImagePuller]") { + for (const bool silent: {false, true}) { + auto x = Base(DetJF(2), 6, 1); + x.UseInternalPacketGenerator(false).Conversion(); + x.Compression(CompressionAlgorithm::BSHUF_LZ4); + + auto device = std::make_unique(0, 64); + for (int64_t f = 0; f < 3; f++) + for (int m = 0; m < 2; m++) { + const auto frame = Frame(f, m, true); + for (int p = 0; p < 128; p++) + device->CreateJFPacket(x, f + 1, p, m, frame.data() + p * 4096); + } + AcquisitionDeviceGroup devices; + devices.Add(std::move(device)); + + Logger logger("FPGAImagePuller_Cancel"); + CollectingPusher pusher; + JFJochReceiverService service(devices, logger, pusher, 256); + service.NumThreads(2).SetSpotFindingSettings(Spots()); + service.UseFPGAImagePuller(true); + PixelMask mask(x); + const JFCalibration calib = Calibration(x); + service.Start(x, mask, &calib); + + std::this_thread::sleep_for(30s); + service.Cancel(silent); + const auto out = service.Stop(); + + std::cout << "cancel silent=" << silent << ": cancelled " << out.status.cancelled << " collected " + << out.status.images_collected << " sent " << out.status.images_sent << " efficiency " + << out.efficiency << std::endl; + CHECK(out.status.cancelled == !silent); + CHECK(pusher.sequence.front() == "start"); + CHECK(pusher.sequence.back() == "end"); + CHECK(out.status.images_collected <= 3); + if (silent) { + CHECK(out.status.images_collected == 3); + CHECK(out.efficiency == Catch::Approx(0.5f)); + } + } +} + +// Default stays the FPGA receiver; pedestal and summation above the card's limit never take the puller +TEST_CASE("FPGAImagePuller_NotForPedestalAndHostSummation", "[FPGAImagePuller]") { + auto x = Base(DetJF(1), 2, 1); + x.Conversion(); + x.Summation(2).CPUSummation(true).Compression(CompressionAlgorithm::BSHUF_LZ4); + REQUIRE(x.IsCPUSummation()); + + auto devices = Devices(x); + Logger logger("FPGAImagePuller_HostSummation"); + CollectingPusher pusher; + JFJochReceiverService service(devices, logger, pusher, 256); + service.NumThreads(4).SetSpotFindingSettings(Spots()); + service.UseFPGAImagePuller(true); + PixelMask mask(x); + const JFCalibration calib = Calibration(x); + service.Start(x, mask, &calib); + const auto out = service.Stop(); + CHECK(out.efficiency == 1.0f); + CHECK(!out.received_packets.empty()); // only the FPGA receiver fills these + CHECK(out.status.images_collected == 2); +} + +// JFJochReceiverRugnux takes the same stream, with its own forwarding thread. +TEST_CASE("FPGAImagePuller_RugnuxReceiver", "[FPGAImagePuller]") { + auto x = Base(DetJF(2), 6, 3); + x.Conversion(); + x.Compression(CompressionAlgorithm::BSHUF_LZ4); + x.Goniometer(GoniometerAxis("omega", 0, 0.1, Coord(-1, 0, 0), std::nullopt)); + x.FilePrefix("fpga_puller_rugnux"); + REQUIRE(RugnuxRotationData(x)); + + auto devices = Devices(x); + Logger logger("FPGAImagePuller_Rugnux"); + CollectingPusher pusher; + JFJochReceiverService service(devices, logger, pusher, 256); + service.NumThreads(2).SetSpotFindingSettings(Spots()); + service.UseFPGAImagePuller(true); + service.RugnuxLive(true, nullptr, 2, 30); + + for (int64_t f = 0; f < 3; f++) { + std::vector raw; + for (int m = 0; m < 2; m++) { + const auto frame = Frame(f, m, true); + raw.insert(raw.end(), frame.begin(), frame.end()); + } + service.LoadInternalGeneratorImage(x, raw, f); + } + + PixelMask mask(x); + const JFCalibration calib = Calibration(x); + service.Start(x, mask, &calib); + const auto out = service.Stop(); + std::cout << "rugnux receiver: efficiency " << out.efficiency << " sent " << out.status.images_sent + << " warning '" << out.analysis_warning << "'" << std::endl; + CHECK(out.efficiency == 1.0f); + CHECK(out.status.images_sent == 6); + CHECK(pusher.images.size() == 6); + CHECK(pusher.sequence.front() == "start"); + CHECK(pusher.sequence.back() == "end"); + CHECK(pusher.sequence[1].starts_with("calibration")); + const auto a = Pixels(pusher.images.at(0).message.image); + const auto b = Pixels(pusher.images.at(3).message.image); + CHECK(a == b); // the third frame in the generator's memory repeats +} + +// The real writer: the master file states what the card did, not what a DECTRIS stream implies. +TEST_CASE("FPGAImagePuller_Writer", "[FPGAImagePuller]") { + RegisterHDF5Filter(); + auto x = Base(DetJF(2), 5, 2); + x.Conversion(); + x.Compression(CompressionAlgorithm::BSHUF_LZ4).ApplyPixelMask(true).ImagesPerFile(2) + .FilePrefix("fpga_image_puller_writer").SetFileWriterFormat(FileWriterFormat::NXmxVDS) + .OverwriteExistingFiles(true).SaveCalibration(true); + + auto devices = Devices(x); + Logger logger("FPGAImagePuller_Writer"); + HDF5FilePusher pusher; + JFJochReceiverService service(devices, logger, pusher, 256); + service.NumThreads(2).SetSpotFindingSettings(Spots()); + service.UseFPGAImagePuller(true); + PixelMask mask(x); + const JFCalibration calib = Calibration(x); + service.Start(x, mask, &calib); + const auto out = service.Stop(); + CHECK(out.efficiency == 1.0f); + CHECK(out.writer_err.empty()); + CHECK(out.status.images_sent == 5); + + HDF5ReadOnlyFile master("fpga_image_puller_writer_master.h5"); + CHECK(master.GetBool("/entry/instrument/detector/pixel_mask_applied")); + HDF5DataSet dataset(master, "/entry/data/data"); + HDF5DataSpace space(dataset); + CHECK(space.GetDimensions()[0] == 5); + CHECK(space.GetDimensions()[1] == x.GetYPixelsNum()); +}