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 <noreply@anthropic.com>

reader/JFJochImageBufferReader.cpp

@@ -3,8 +3,8 @@
 
 #include "JFJochImageBufferReader.h"
 
+#include <algorithm>
 #include <cstring>
-#include <thread>
 
 #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<std::mutex> &ul) {
+    polling_ = true;
+    next_poll_ = std::chrono::steady_clock::now() + POLL_PERIOD;
+    ul.unlock();
+    const std::vector<int64_t> images = buffer_.GetStatus().images_in_the_buffer;
+    ul.lock();
+    polling_ = false;
+    for (const int64_t i: images)
+        if (i >= 0 && static_cast<uint64_t>(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<uint8_t> &cbor) {
     if (image_number < 0 || static_cast<uint64_t>(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) {