From d356cc5da500c949498fed2dd1d6885d65ad21af Mon Sep 17 00:00:00 2001 From: Filip Leonarski Date: Sat, 17 Oct 2026 18:13:03 +0200 Subject: [PATCH] ImageBuffer reader: one waiting thread at a time looks at the buffer Every rugnux worker waiting for an image called ImageBuffer::GetImage every millisecond, and GetStatus every 100 polls. Each call takes the buffer's mutex and scans every slot, so sixteen waiting workers contended with the forward threads for the lock they put the images in under (GetImageSlot, ReadyToSend). Now one waiting thread at a time asks the buffer, at most every 2 ms, with a single GetStatus, and notes the images it holds in a per-reader bitmap; the other waiters sleep on a condition variable and are woken when it is refreshed or the collection ends. GetImage is called once, for an image known to be there (or given up on). MAX_IMAGE_LAG and Ended() mean what they did. ImageBuffer itself is unchanged. The test gains eight readers waiting at once. Co-Authored-By: Claude Opus 5.5 --- reader/JFJochImageBufferReader.cpp | 55 +++++++++++++++++++-------- reader/JFJochImageBufferReader.h | 18 ++++++++- tests/JFJochImageBufferReaderTest.cpp | 16 +++++++- 3 files changed, 72 insertions(+), 17 deletions(-) diff --git a/reader/JFJochImageBufferReader.cpp b/reader/JFJochImageBufferReader.cpp index 67d6f687e..d6a9f9b68 100644 --- a/reader/JFJochImageBufferReader.cpp +++ b/reader/JFJochImageBufferReader.cpp @@ -3,8 +3,8 @@ #include "JFJochImageBufferReader.h" +#include #include -#include #include "JFJochStartMessageReader.h" #include "../common/JFJochException.h" @@ -13,13 +13,22 @@ JFJochImageBufferReader::JFJochImageBufferReader(ImageBuffer &buffer) : buffer_(buffer) {} void JFJochImageBufferReader::Open(const StartMessage &start) { - number_of_images_ = start.number_of_images; - ended_ = false; + { + std::unique_lock ul(m_); + number_of_images_ = start.number_of_images; + present_.assign(number_of_images_, false); + max_present_ = -1; + ended_ = false; + } SetStartMessage(DatasetFromStartMessage(start, default_experiment)); } void JFJochImageBufferReader::Ended() { - ended_ = true; + { + std::unique_lock ul(m_); + ended_ = true; + } + cv_.notify_all(); } uint64_t JFJochImageBufferReader::GetNumberOfImages() const { @@ -30,21 +39,37 @@ void JFJochImageBufferReader::Close() { SetStartMessage({}); } +void JFJochImageBufferReader::Poll(std::unique_lock &ul) { + polling_ = true; + next_poll_ = std::chrono::steady_clock::now() + POLL_PERIOD; + ul.unlock(); + const std::vector images = buffer_.GetStatus().images_in_the_buffer; + ul.lock(); + polling_ = false; + for (const int64_t i: images) + if (i >= 0 && static_cast(i) < number_of_images_) { + present_[i] = true; + max_present_ = std::max(max_present_, i); + } + cv_.notify_all(); +} + bool JFJochImageBufferReader::WaitForMessage(int64_t image_number, std::vector &cbor) { if (image_number < 0 || static_cast(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. - 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)); + std::unique_lock ul(m_); + // 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. After the end, one is asked for once more, as it + // may have come in just before. + while (!present_[image_number] && !ended_ && max_present_ < image_number + MAX_IMAGE_LAG) { + const auto now = std::chrono::steady_clock::now(); + if (!polling_ && now >= next_poll_) + Poll(ul); + else + cv_.wait_until(ul, polling_ ? now + POLL_PERIOD : next_poll_); } - return true; + ul.unlock(); + return buffer_.GetImage(cbor, image_number); } bool JFJochImageBufferReader::ReadRawImage(int64_t image_number, JFJochReaderRawImage &ret) { diff --git a/reader/JFJochImageBufferReader.h b/reader/JFJochImageBufferReader.h index 7acf60aa8..29aa87b33 100644 --- a/reader/JFJochImageBufferReader.h +++ b/reader/JFJochImageBufferReader.h @@ -4,6 +4,9 @@ #pragma once #include +#include +#include +#include #include #include @@ -15,10 +18,12 @@ // read of an image not there yet waits for it; after Ended it returns false for an image that never // 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. +// image would hand out that image instead. However many threads wait, one at a time looks at the buffer, +// every POLL_PERIOD, so the waiting stays off the lock the receiver puts the images in under. class JFJochImageBufferReader : public JFJochReader { public: constexpr static int64_t MAX_IMAGE_LAG = 100; + constexpr static auto POLL_PERIOD = std::chrono::milliseconds(2); explicit JFJochImageBufferReader(ImageBuffer &buffer); @@ -37,7 +42,18 @@ private: // The image's CBOR message, as the receiver put it in the buffer; false when it never came. bool WaitForMessage(int64_t image_number, std::vector &cbor); + // Looks at the buffer (GetStatus) and notes the images it holds in present_; under m_, released while + // the buffer is asked. + void Poll(std::unique_lock &ul); + ImageBuffer &buffer_; uint64_t number_of_images_ = 0; std::atomic ended_ = false; + + std::mutex m_; + std::condition_variable cv_; + bool polling_ = false; + std::chrono::steady_clock::time_point next_poll_; + std::vector present_; + int64_t max_present_ = -1; }; diff --git a/tests/JFJochImageBufferReaderTest.cpp b/tests/JFJochImageBufferReaderTest.cpp index 41abe0b31..500123c19 100644 --- a/tests/JFJochImageBufferReaderTest.cpp +++ b/tests/JFJochImageBufferReaderTest.cpp @@ -31,7 +31,7 @@ TEST_CASE("JFJochImageBufferReader", "[JFJochReader]") { DiffractionExperiment experiment(DetJF(1)); experiment.ImagesPerTrigger(200); - ImageBuffer buffer(4 * experiment.GetImageBufferLocationSize()); + ImageBuffer buffer(16 * experiment.GetImageBufferLocationSize()); buffer.StartMeasurement(experiment); StartMessage start{}; @@ -73,6 +73,20 @@ TEST_CASE("JFJochImageBufferReader", "[JFJochReader]") { REQUIRE(pending.wait_for(std::chrono::seconds(5)) == std::future_status::ready); REQUIRE(pending.get()); + // Many readers waiting at once, as rugnux's workers do: each gets its image as it comes. + std::vector> waiting; + for (int64_t i = 2; i < 10; i++) + waiting.push_back(std::async(std::launch::async, [&reader, i] { + JFJochReaderRawImage r; + return reader.ReadRawImage(i, r) && r.image.GetWidth() > 0; + })); + for (int64_t i = 2; i < 10; i++) + PutImage(buffer, experiment, i, pixels); + for (auto &w: waiting) { + REQUIRE(w.wait_for(std::chrono::seconds(5)) == std::future_status::ready); + REQUIRE(w.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);