diff --git a/reader/JFJochImageBufferReader.cpp b/reader/JFJochImageBufferReader.cpp index a7ca4e2cc..67d6f687e 100644 --- a/reader/JFJochImageBufferReader.cpp +++ b/reader/JFJochImageBufferReader.cpp @@ -34,9 +34,14 @@ bool JFJochImageBufferReader::WaitForMessage(int64_t image_number, std::vector(image_number) >= number_of_images_) throw JFJochException(JFJochExceptionCategory::ArrayOutOfBounds, "Image out of bounds"); // The buffer has no notification of its own; at 100 Hz a millisecond's poll costs nothing. - while (!buffer_.GetImage(cbor, image_number)) { + for (int poll = 1; !buffer_.GetImage(cbor, image_number); poll++) { if (ended_) return buffer_.GetImage(cbor, image_number); // one that came in just before the end + // Images arrive in order, give or take a few: one still missing when the buffer already holds + // an image MAX_IMAGE_LAG past it is lost, not late. GetStatus scans every slot under the + // buffer's lock, so it is asked only every 100 polls. + if (poll % 100 == 0 && buffer_.GetStatus().max_image_number >= image_number + MAX_IMAGE_LAG) + return buffer_.GetImage(cbor, image_number); std::this_thread::sleep_for(std::chrono::milliseconds(1)); } return true; diff --git a/reader/JFJochImageBufferReader.h b/reader/JFJochImageBufferReader.h index 5a7c5203d..7acf60aa8 100644 --- a/reader/JFJochImageBufferReader.h +++ b/reader/JFJochImageBufferReader.h @@ -13,10 +13,13 @@ // Reads a collection inside the broker while it arrives: the dataset comes from its start message, // the images from the broker's image buffer, where the receiver puts each one as it forwards it. A // read of an image not there yet waits for it; after Ended it returns false for an image that never -// came. The buffer must hold the whole collection (slots >= images) - a slot taken again for a later +// came, and so does a read of an image still missing when one MAX_IMAGE_LAG past it is there. The +// buffer must hold the whole collection (slots >= images) - a slot taken again for a later // image would hand out that image instead. class JFJochImageBufferReader : public JFJochReader { public: + constexpr static int64_t MAX_IMAGE_LAG = 100; + explicit JFJochImageBufferReader(ImageBuffer &buffer); void Open(const StartMessage &start); diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt index 5f5c4b207..0866355a7 100644 --- a/tests/CMakeLists.txt +++ b/tests/CMakeLists.txt @@ -87,6 +87,7 @@ ADD_EXECUTABLE(jfjoch_test SpotTilesTest.cpp LossyFilterTest.cpp ImageBufferTest.cpp + JFJochImageBufferReaderTest.cpp PixelMaskTest.cpp RegressionTest.cpp FPGAHLSModulesTest.cpp diff --git a/tests/JFJochImageBufferReaderTest.cpp b/tests/JFJochImageBufferReaderTest.cpp new file mode 100644 index 000000000..41abe0b31 --- /dev/null +++ b/tests/JFJochImageBufferReaderTest.cpp @@ -0,0 +1,97 @@ +// SPDX-FileCopyrightText: 2026 Filip Leonarski, Paul Scherrer Institute +// SPDX-License-Identifier: GPL-3.0-only + +#include +#include +#include +#include + +#include "../reader/JFJochImageBufferReader.h" +#include "../common/ZeroCopyReturnValue.h" +#include "../frame_serialize/CBORStream2Serializer.h" + +// The receiver's way of putting an image in the buffer: serialize the message into a slot, hand it out. +static void PutImage(ImageBuffer &buffer, const DiffractionExperiment &experiment, int64_t number, + const std::vector &pixels) { + DataMessage msg{}; + msg.number = number; + msg.image = CompressedImage(pixels, experiment.GetXPixelsNum(), experiment.GetYPixelsNum()); + auto loc = buffer.GetImageSlot(); + REQUIRE(loc != nullptr); + CBORStream2Serializer serializer((uint8_t *) loc->GetImage(), experiment.GetImageBufferLocationSize()); + serializer.SerializeImage(msg); + loc->SetImageNumber(number); + loc->SetImageSize(serializer.GetBufferSize()); + loc->SetIndexed(false); + loc->ReadyToSend(); + loc->release(); +} + +TEST_CASE("JFJochImageBufferReader", "[JFJochReader]") { + DiffractionExperiment experiment(DetJF(1)); + experiment.ImagesPerTrigger(200); + + ImageBuffer buffer(4 * experiment.GetImageBufferLocationSize()); + buffer.StartMeasurement(experiment); + + StartMessage start{}; + experiment.FillMessage(start); + REQUIRE(start.number_of_images == 200); + + JFJochImageBufferReader reader(buffer); + reader.Open(start); + REQUIRE(reader.GetNumberOfImages() == 200); + REQUIRE(reader.GetDataset()); + + const size_t npixel = experiment.GetXPixelsNum() * experiment.GetYPixelsNum(); + std::vector pixels(npixel, 7); + pixels[123] = 4567; + PutImage(buffer, experiment, 0, pixels); + + JFJochReaderRawImage raw; + REQUIRE(reader.ReadRawImage(0, raw)); + REQUIRE(raw.image.GetWidth() == experiment.GetXPixelsNum()); + REQUIRE(raw.image.GetHeight() == experiment.GetYPixelsNum()); + REQUIRE(raw.image_buffer.size() == npixel * sizeof(int32_t)); + int32_t value; + memcpy(&value, raw.image.GetCompressed() + 123 * sizeof(int32_t), sizeof(int32_t)); + REQUIRE(value == 4567); + + auto image = reader.LoadImage(0); + REQUIRE(image); + REQUIRE(image->ImageData().number == 0); + + REQUIRE_THROWS(reader.ReadRawImage(200, raw)); + + // An image not there yet is waited for: the read returns once the receiver puts it in. + auto pending = std::async(std::launch::async, [&] { + JFJochReaderRawImage r; + return reader.ReadRawImage(1, r); + }); + REQUIRE(pending.wait_for(std::chrono::milliseconds(50)) == std::future_status::timeout); + PutImage(buffer, experiment, 1, pixels); + REQUIRE(pending.wait_for(std::chrono::seconds(5)) == std::future_status::ready); + REQUIRE(pending.get()); + + // Images arrive in order, give or take a few: one still missing when an image MAX_IMAGE_LAG past + // it is there is lost, while one nearer may still come. + PutImage(buffer, experiment, 150, pixels); + auto lost = std::async(std::launch::async, [&] { + JFJochReaderRawImage r; + return reader.ReadRawImage(150 - JFJochImageBufferReader::MAX_IMAGE_LAG, r); + }); + REQUIRE(lost.wait_for(std::chrono::seconds(5)) == std::future_status::ready); + REQUIRE(!lost.get()); + + // After the end, an image that never came is not waited for. + auto missing = std::async(std::launch::async, [&] { + JFJochReaderRawImage r; + return reader.ReadRawImage(150 - JFJochImageBufferReader::MAX_IMAGE_LAG + 1, r); + }); + REQUIRE(missing.wait_for(std::chrono::milliseconds(300)) == std::future_status::timeout); + reader.Ended(); + REQUIRE(missing.wait_for(std::chrono::seconds(5)) == std::future_status::ready); + REQUIRE(!missing.get()); + + REQUIRE(buffer.Finalize(std::chrono::microseconds(1))); +}