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);