Build Packages / build:jfjoch:ubuntu2204:nocuda (push) Successful in 21m25s
Build Packages / Create release (push) Successful in 8s
Build Packages / build:portable:linux-x86_64:nocuda (push) Failing after 8m11s
Build Packages / build:portable:linux-x86_64:cuda (push) Failing after 10m22s
Build Packages / build:jfjoch:rocky8:nocuda (push) Successful in 22m3s
Build Packages / build:portable:windows-x86_64:cuda (push) Failing after 24m19s
Build Packages / Build documentation (push) Successful in 44s
Build Packages / build:rugnux:linux-aarch64:cuda (push) Successful in 9m33s
Build Packages / build:portable:macos-arm64:nocuda (push) Failing after 2m54s
Build Packages / build:jfjoch:rocky9:nocuda (push) Successful in 25m42s
Build Packages / Generate python client (push) Successful in 25s
Build Packages / build:portable:windows-x86_64:nocuda (push) Failing after 22m30s
Build Packages / HDF5 consumer tests (DIALS, XDS) (push) Successful in 24m36s
Build Packages / build:jfjoch:ubuntu2204:cuda (push) Failing after 1m16s
Build Packages / build:jfjoch:ubuntu2404:nocuda (push) Successful in 26m16s
Build Packages / build:jfjoch:rocky8:cuda-sls9 (push) Successful in 28m46s
Build Packages / build:jfjoch:rocky9:cuda-sls9 (push) Successful in 31m40s
Build Packages / build:jfjoch:rocky8:cuda (push) Successful in 29m47s
Build Packages / build:jfjoch:rocky9:cuda (push) Successful in 28m29s
Build Packages / build:jfjoch:ubuntu2404:cuda (push) Successful in 24m27s
Build Packages / Unit tests (push) Successful in 1h36m2s
Build Packages / Validation Battery (push) Failing after 1h29m55s
The FPGA receiver held every frame for the GPU analysis in a heap vector sized to a full buffer slot and then shrunk, which keeps the capacity: ~25 MB per JUNGFRAU 8M frame instead of its compressed size, in a queue with no bound. During the pre-pass (~1000 frames at 100 Hz), or whenever the analysis runs slower than the frames arrive, that grew the broker's heap without limit on top of the configured image buffer. A held frame is now compressed straight into a slot of the receiver's ImageBuffer, the configured buffer the frames are written from. Two buckets per frame: the slot the compressed image waits in (in preparation, so never previewed or sent), and the slot GPUSink writes the CBOR message, analysis and image to, which is sent as before. The waiting slot goes back to the buffer, unsent, when the last copy of the HeldImage is gone. Nothing is copied to the heap and no frame memory is allocated per frame. The two buckets are also two pools: at most half the buffer's slots wait for analysis (the hold queue is bounded to total_slots / 2), so the analysed frames always find slots to be sent from. A frame that finds no free slot, or the hold queue full, is dropped and writer_queue_full is set, as the receiver does for any frame that finds the image buffer full; the frame thread never blocks. A full hold queue also starts the pre-pass on what it has. Analysis slower than the arrival rate therefore costs frames, not memory. jfjoch_stills_gpu_replay holds its frames in an ImageBuffer too (-b MiB, default 8192), and waits for a slot rather than drop a frame. Measured on the HLS-simulated CrMX JUNGFRAU 8M broker (40 frames, image buffer 8 GiB): RSS grew by ~26 MB per held frame before, linearly until the pre-pass; now it is flat once the receiver threads have warmed up (+1.05 GB less at 40 frames, ~25 GB less for a 1000-frame pre-pass). The stills replay of a stored serial collection gives the same beam centre, distance, mask and indexed count as before the change. Test: StillsGPUAnalysis_HoldIsBounded; the other StillsGPUAnalysis tests check every slot is returned. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01UqXYdJKPUFLoeAchjzv3x4
117 lines
5.0 KiB
C++
117 lines
5.0 KiB
C++
// SPDX-FileCopyrightText: 2026 Filip Leonarski, Paul Scherrer Institute <filip.leonarski@psi.ch>
|
|
// SPDX-License-Identifier: GPL-3.0-only
|
|
|
|
#include <catch2/catch_all.hpp>
|
|
|
|
#include <atomic>
|
|
|
|
#include "../common/ImageBuffer.h"
|
|
#include "../receiver/StillsGPUAnalysis.h"
|
|
|
|
namespace {
|
|
DiffractionExperiment StillsExperiment() {
|
|
DiffractionExperiment x(DetJF(1));
|
|
x.ImagesPerTrigger(200).NumTriggers(1).DetectorDistance_mm(100).BeamX_pxl(500).BeamY_pxl(250)
|
|
.IncidentEnergy_keV(12.4);
|
|
return x;
|
|
}
|
|
|
|
// A frame after the pre-pass's sample (numbers from PREPASS_IMAGES on are never in it), in a slot of `buffer`.
|
|
std::shared_ptr<HeldImage> Frame(ImageBuffer &buffer, int64_t number) {
|
|
auto held = std::make_shared<HeldImage>();
|
|
held->slot = buffer.GetImageSlot();
|
|
REQUIRE(held->slot != nullptr);
|
|
held->message.number = StillsGPUAnalysis::PREPASS_IMAGES + number;
|
|
held->message.image_collection_efficiency = 1.0f;
|
|
return held;
|
|
}
|
|
|
|
StartMessage Start(const DiffractionExperiment &x) {
|
|
StartMessage start;
|
|
x.FillMessage(start);
|
|
return start;
|
|
}
|
|
}
|
|
|
|
TEST_CASE("StillsGPUAnalysis_StartFailureReleasesThreads", "[StillsGPUAnalysis]") {
|
|
Logger logger(Catch::getResultCapture().getCurrentTestName());
|
|
const DiffractionExperiment x = StillsExperiment();
|
|
const PixelMask mask(x);
|
|
ImageBuffer buffer(64 * 1024 * 1024);
|
|
buffer.StartMeasurement(x);
|
|
RugnuxLiveStatus status;
|
|
|
|
std::atomic<int> failed = 0, sunk = 0;
|
|
{
|
|
StillsGPUAnalysis gpu(x, mask, Start(x), nullptr, 4, 100, logger,
|
|
[](const StillsGPUAnalysis::Setup &) {
|
|
throw JFJochException(JFJochExceptionCategory::ZeroMQ, "Timeout on pushing start message");
|
|
},
|
|
[&](DataMessage &, const AzimuthalIntegrationProfile &) { ++sunk; },
|
|
[] { return DiffractionExperiment::DefaultDataProcessingSettings(); },
|
|
[&](const JFJochException &) { ++failed; }, &status);
|
|
for (int i = 0; i < 3; i++)
|
|
REQUIRE(gpu.Hold(Frame(buffer, i)));
|
|
REQUIRE_THROWS_AS(gpu.Finish(), JFJochException);
|
|
}
|
|
CHECK(failed == 1);
|
|
CHECK(sunk == 0);
|
|
CHECK(status.Get().state == RugnuxLiveState::Failed);
|
|
// Every slot the frames were held in is back.
|
|
CHECK(buffer.GetAvailSlots() == buffer.GetStatus().total_slots);
|
|
}
|
|
|
|
TEST_CASE("StillsGPUAnalysis_HoldIsBounded", "[StillsGPUAnalysis]") {
|
|
Logger logger(Catch::getResultCapture().getCurrentTestName());
|
|
const DiffractionExperiment x = StillsExperiment();
|
|
const PixelMask mask(x);
|
|
ImageBuffer buffer(64 * 1024 * 1024);
|
|
buffer.StartMeasurement(x);
|
|
REQUIRE(buffer.GetStatus().total_slots >= 4);
|
|
|
|
StillsGPUAnalysis gpu(x, mask, Start(x), nullptr, 2, 3, logger,
|
|
[](const StillsGPUAnalysis::Setup &) {
|
|
throw JFJochException(JFJochExceptionCategory::ZeroMQ, "no writer");
|
|
},
|
|
[](DataMessage &, const AzimuthalIntegrationProfile &) {},
|
|
[] { return DiffractionExperiment::DefaultDataProcessingSettings(); });
|
|
// The analysis waits for the pre-pass, which waits for its sample: nothing is taken off the queue.
|
|
for (int i = 0; i < 3; i++)
|
|
REQUIRE(gpu.Hold(Frame(buffer, i)));
|
|
{
|
|
const auto refused = Frame(buffer, 3);
|
|
CHECK(!gpu.Hold(refused));
|
|
}
|
|
CHECK(buffer.GetStatus().preparation_slots == 3);
|
|
REQUIRE_THROWS(gpu.Finish());
|
|
CHECK(buffer.GetStatus().preparation_slots == 0);
|
|
}
|
|
|
|
TEST_CASE("StillsGPUAnalysis_FailingFrameDoesNotStopAnalysis", "[StillsGPUAnalysis]") {
|
|
Logger logger(Catch::getResultCapture().getCurrentTestName());
|
|
const DiffractionExperiment x = StillsExperiment();
|
|
const PixelMask mask(x);
|
|
ImageBuffer buffer(64 * 1024 * 1024);
|
|
buffer.StartMeasurement(x);
|
|
RugnuxLiveStatus status;
|
|
|
|
std::atomic<int> sunk = 0;
|
|
StillsGPUAnalysis gpu(x, mask, Start(x), nullptr, 1, 100, logger,
|
|
[](const StillsGPUAnalysis::Setup &) {},
|
|
[&](DataMessage &m, const AzimuthalIntegrationProfile &) {
|
|
++sunk;
|
|
if (m.number == StillsGPUAnalysis::PREPASS_IMAGES)
|
|
throw std::runtime_error("not a JFJochException");
|
|
},
|
|
[] { return DiffractionExperiment::DefaultDataProcessingSettings(); },
|
|
{}, &status);
|
|
// The frames carry no image, so their analysis fails too - they are passed on all the same.
|
|
for (int i = 0; i < 3; i++)
|
|
REQUIRE(gpu.Hold(Frame(buffer, i)));
|
|
REQUIRE_NOTHROW(gpu.Finish());
|
|
CHECK(sunk == 3);
|
|
CHECK(status.Get().state == RugnuxLiveState::Failed);
|
|
CHECK(!status.Get().message.empty());
|
|
CHECK(buffer.GetStatus().preparation_slots == 0);
|
|
}
|