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