Merge branch 'writer-end-catch' into rc175: writer ends any failure in error state; received images in a byte-bounded ring; CloseOldFiles walks open files only; ZeroMQ pusher logs the switch to dropping

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

docs/JFJOCH_WRITER.md

@@ -21,7 +21,8 @@ Options:
 -d<path>  | --root_dir=<path>          Root directory for file writing (-R is a deprecated alias)
 -r<int>   | --zmq_repub_port=<int>     ZeroMQ port for PUSH socket to republish images
 -f<int>   | --zmq_file_port=<int>      ZeroMQ port for PUB socket for notifications on finalized files
--w<int>   | --rcv_watermark=<int>      Receiving ZeroMQ socket watermark (default = 100)
+-w<int>   | --rcv_watermark=<int>      Receiving ZeroMQ socket watermark in messages (default = 10)
+-m<int>   | --rcv_ring_mb=<int>        Receive ring buffer in MB; holds images until written and republished (default = 2048)
 -W<int>   | --repub_watermark=<int>    Republish ZeroMQ socket watermark (default = 1000)
 -b<int>   | --rcv_buffer=<int>         Receiving TCP socket buffer in bytes (only with -T; disables autotuning, capped by net.core.rmem_max)
 -v        | --verbose                  Verbose output
@@ -32,6 +33,33 @@ for example:
 jfjoch_writer -d /data tcp://dcu-address:5400 
 ```
 
+## Memory
+Every message the writer receives is placed in a ring buffer of a fixed size (`-m`, default 2048 MB,
+per writer process - `-j 8` is eight rings) and stays there until it has been written to HDF5 and, if
+republish is on, republished. The ring is the writer's memory budget for buffered images, and the
+peak use is printed at the end of every run. Size it as the number of images you can afford to have
+in flight times their compressed size: at 2048 MB that is some 250-500 JUNGFRAU 4M images but only
+50-100 JUNGFRAU 9M images, so a 9M site will want more.
+
+When the ring is full the writer stops reading from the socket and the sender has to wait. How long
+it is prepared to wait depends on the stream:
+
+- **ZeroMQ.** The socket queue fills up to its watermark (`-w`, a count of messages, kept small for
+  that reason) and the broker's send blocks for one second. After that the broker drops the image,
+  logs the fact, and sends without waiting for the rest of the run, so every later image that finds
+  the queue full is dropped too. A writer that stalls for longer than its ring covers therefore
+  loses images on ZeroMQ; the ring has to be large enough for the stalls of the filesystem it writes
+  to. (Before the ring the same stall was absorbed by queues of about 1500 messages, whatever their
+  size - a budget nobody had chosen.)
+- **TCP/IP.** The broker sees the writer's BUSY heartbeats and waits, giving up only after a minute
+  in which not a single image was written.
+
+Republish holds its own images in the same ring. Because the ring is reclaimed in arrival order, an
+image kept for a slow republish consumer would hold back everything received after it, so republish
+may hold at most a quarter of the ring; beyond that images are simply not republished, exactly as
+when its queue is full. A single message larger than the whole ring - a START or END message of a
+very long run - is held outside it.
+
 ## Status and cancellation
 When a data collection is finalized, each writer reports its outcome back to `jfjoch_broker` over
 the writer notification socket — a ZeroMQ address the broker passes in the START message

image_puller/CMakeLists.txt

@@ -4,5 +4,7 @@ ADD_LIBRARY(JFJochImagePuller ZMQImagePuller.cpp ZMQImagePuller.h
         TestImagePuller.cpp
         TestImagePuller.h
         TCPImagePuller.cpp
-        TCPImagePuller.h)
+        TCPImagePuller.h
+        ReceiveRingBuffer.cpp
+        ReceiveRingBuffer.h)
 TARGET_LINK_LIBRARIES(JFJochImagePuller JFJochCommon JFJochZMQ JFJochLogger CBORStream2FrameSerialize)

image_puller/ImagePuller.cpp

@@ -3,7 +3,15 @@
 
 #include "ImagePuller.h"
 
-ImagePuller::ImagePuller(size_t fifo_size) : outside_fifo(fifo_size) {}
+ImagePuller::ImagePuller(size_t fifo_size, size_t ring_size) : outside_fifo(fifo_size), ring(ring_size) {}
+
+size_t ImagePuller::GetRingBufferCapacity() const {
+    return ring.GetCapacity();
+}
+
+size_t ImagePuller::GetMaxRingBufferUtilization() const {
+    return ring.GetMaxUtilization();
+}
 
 size_t ImagePuller::GetCurrentFifoUtilization() const {
     return outside_fifo.GetCurrentUtilization();

image_puller/ImagePuller.h

@@ -3,6 +3,7 @@
 
 #pragma once
 
+#include <atomic>
 #include <string>
 #include <memory>
 
@@ -10,6 +11,7 @@
 #include "../frame_serialize/CBORStream2Deserializer.h"
 #include "../common/ThreadSafeFIFO.h"
 #include "../common/JfjochTCP.h"
+#include "ReceiveRingBuffer.h"
 
 struct PullerAckMessage {
     TCPFrameType ack_for = TCPFrameType::DATA;
@@ -25,12 +27,14 @@ struct PullerAckMessage {
 
 struct RawFrame {
     TcpFrameHeader header{};
-    std::vector<uint8_t> payload;
     bool end = false;
 };
 
 struct ImagePullerOutput {
-    std::shared_ptr<ZMQMessage> zmq_msg;
+    // The message as it came off the socket, in the puller's ring buffer. It stays there for as long as
+    // any copy of this struct is alive - the deserialised image in cbor points into it, and republish
+    // sends it as is.
+    std::shared_ptr<ReceiveBufferRegion> raw;
     std::shared_ptr<RawFrame> tcp_msg;
     std::shared_ptr<CBORStream2DeserializerOutput> cbor;
 };
@@ -42,16 +46,32 @@ struct ImagePullerStatistics {
 class ImagePuller {
 protected:
     ThreadSafeFIFO<ImagePullerOutput> outside_fifo; // assuming 1 file is <1000 files, this should be enough to buffer
+    // Every received message is held here until consumed; the queues count messages, this bounds bytes.
+    // Size 0 means no ring: messages are held on the heap, bounded by the queues alone, as they always
+    // were - the broker's own DECTRIS puller runs that way, jfjoch_writer sets a size.
+    ReceiveRingBuffer ring;
+
+    // Bytes republish holds right now. Republish keeps the OLDEST not yet sent images, and the ring is
+    // reclaimed in arrival order, so an image waiting for a slow republish consumer pins everything
+    // received after it - the writer would then run at the republish rate. What republish may hold is
+    // therefore capped at a share of the ring; an image over the cap is not republished, exactly as one
+    // that finds the republish queue full. Without a ring there is no cap.
+    std::atomic<size_t> repub_bytes{0};
+    bool RepubCanHold(size_t bytes) const {
+        return ring.GetCapacity() == 0 || repub_bytes.load() + bytes <= ring.GetCapacity() / 4;
+    }
 public:
     constexpr static size_t DefaultQueueSize = 1000;
     constexpr static auto PollTimeout = std::chrono::milliseconds(100);
 
-    explicit ImagePuller(size_t fifo_size = DefaultQueueSize);
+    explicit ImagePuller(size_t fifo_size = DefaultQueueSize, size_t ring_size = 0);
     virtual ~ImagePuller() = default;
 
     [[nodiscard]] std::optional<ImagePullerOutput> PollImage(std::chrono::microseconds timeout = PollTimeout);
     [[nodiscard]] size_t GetMaxFifoUtilization() const;
     [[nodiscard]] size_t GetCurrentFifoUtilization() const;
+    [[nodiscard]] size_t GetRingBufferCapacity() const;
+    [[nodiscard]] size_t GetMaxRingBufferUtilization() const;
 
     virtual bool SupportsAck() const { return false; }
     virtual bool SendAck(const PullerAckMessage &) { return false; }

image_puller/ReceiveRingBuffer.cpp

@@ -0,0 +1,146 @@
+// SPDX-FileCopyrightText: 2026 Filip Leonarski, Paul Scherrer Institute <filip.leonarski@psi.ch>
+// SPDX-License-Identifier: GPL-3.0-only
+
+#include "ReceiveRingBuffer.h"
+
+#include <algorithm>
+#include <optional>
+
+struct ReceiveRingBuffer::State {
+    // Uninitialised on purpose: a ring that is never filled should not cost its capacity in
+    // resident memory.
+    std::unique_ptr<uint8_t[]> memory;
+    const size_t capacity;
+
+    std::mutex m;
+    std::condition_variable cv;
+
+    // Live regions in arrival order. Only the oldest can give back space, so a region released out of
+    // turn is marked and waits.
+    struct Slot {
+        size_t offset;
+        size_t size;
+        bool released = false;
+    };
+    std::deque<Slot> slots;
+    // Next byte to hand out. The ring is empty when slots is empty; the oldest live byte is then
+    // slots.front().offset.
+    size_t tail = 0;
+
+    size_t utilization = 0;
+    size_t max_utilization = 0;
+    bool stopped = false;
+
+    explicit State(size_t in_capacity)
+            : memory(in_capacity > 0 ? std::make_unique_for_overwrite<uint8_t[]>(in_capacity) : nullptr),
+              capacity(in_capacity) {}
+
+    // Where a region of `size` bytes would start, or nothing if it does not fit right now. Called
+    // under the mutex.
+    std::optional<size_t> FindSpace(size_t size) const {
+        if (slots.empty())
+            return 0;
+        const size_t head = slots.front().offset;
+        // tail has come round to head: treated as full. (After a wrap the piece between the old tail
+        // and the end is free, but it is behind the live data and not worth tracking.)
+        if (tail == head)
+            return std::nullopt;
+        if (tail > head) {
+            // Live data is [head, tail): free space is behind it to the end, or before it at the start.
+            if (capacity - tail >= size)
+                return tail;
+            if (head >= size)
+                return 0;
+            return std::nullopt;
+        }
+        // Wrapped: live data is [head, capacity) + [0, tail), free space is [tail, head).
+        if (head - tail >= size)
+            return tail;
+        return std::nullopt;
+    }
+
+    void Release(size_t offset) {
+        std::unique_lock ul(m);
+        for (auto &s: slots) {
+            if (s.offset == offset) {
+                s.released = true;
+                break;
+            }
+        }
+        while (!slots.empty() && slots.front().released) {
+            utilization -= slots.front().size;
+            slots.pop_front();
+        }
+        if (slots.empty())
+            tail = 0;
+        cv.notify_all();
+    }
+
+    void ReleaseHeap(size_t size) {
+        std::unique_lock ul(m);
+        utilization -= size;
+    }
+};
+
+ReceiveRingBuffer::ReceiveRingBuffer(size_t capacity) : state(std::make_shared<State>(capacity)) {}
+
+std::shared_ptr<ReceiveBufferRegion> ReceiveRingBuffer::Allocate(size_t size) {
+    // Nothing to hold, and a zero-length slot would share its offset with the next real one.
+    if (size == 0)
+        return std::make_shared<ReceiveBufferRegion>();
+
+    std::unique_lock ul(state->m);
+
+    if (size > state->capacity) {
+        // Heap fallback: counted, not bounded.
+        state->utilization += size;
+        state->max_utilization = std::max(state->max_utilization, state->utilization);
+        auto *region = new ReceiveBufferRegion{new uint8_t[size], size};
+        return std::shared_ptr<ReceiveBufferRegion>(region, [s = state](ReceiveBufferRegion *r) {
+            delete[] r->data;
+            s->ReleaseHeap(r->size);
+            delete r;
+        });
+    }
+
+    std::optional<size_t> offset;
+    state->cv.wait(ul, [&] {
+        if (state->stopped)
+            return true;
+        offset = state->FindSpace(size);
+        return offset.has_value();
+    });
+    if (!offset)
+        return nullptr;
+
+    state->slots.push_back({*offset, size});
+    state->tail = *offset + size;
+    state->utilization += size;
+    state->max_utilization = std::max(state->max_utilization, state->utilization);
+
+    auto *region = new ReceiveBufferRegion{state->memory.get() + *offset, size};
+    return std::shared_ptr<ReceiveBufferRegion>(region, [s = state, off = *offset](ReceiveBufferRegion *r) {
+        s->Release(off);
+        delete r;
+    });
+}
+
+void ReceiveRingBuffer::Stop() {
+    std::unique_lock ul(state->m);
+    state->stopped = true;
+    state->cv.notify_all();
+}
+
+size_t ReceiveRingBuffer::GetCapacity() const {
+    return state->capacity;
+}
+
+size_t ReceiveRingBuffer::GetUtilization() const {
+    std::unique_lock ul(state->m);
+    return state->utilization;
+}
+
+size_t ReceiveRingBuffer::GetMaxUtilization() const {
+    std::unique_lock ul(state->m);
+    return state->max_utilization;
+}

image_puller/ReceiveRingBuffer.h

@@ -0,0 +1,52 @@
+// SPDX-FileCopyrightText: 2026 Filip Leonarski, Paul Scherrer Institute <filip.leonarski@psi.ch>
+// SPDX-License-Identifier: GPL-3.0-only
+
+#pragma once
+
+#include <condition_variable>
+#include <cstdint>
+#include <deque>
+#include <memory>
+#include <mutex>
+
+// A received message lives here from the moment it comes off the socket until the last consumer is
+// done with it - the writer has stored the image and the republish thread has sent it. Both consumers
+// hold a shared_ptr to the region, and the region's destructor is what gives the bytes back.
+struct ReceiveBufferRegion {
+    uint8_t *data = nullptr;
+    size_t size = 0;
+};
+
+// Bounds the memory a puller holds in received, not yet consumed, messages - by bytes, which is the
+// quantity that matters, where the queues in front of and behind it count messages. One image of a
+// large detector is tens of megabytes, so a queue of a thousand of them is a quantity of memory nobody
+// chose.
+//
+// Regions are handed out in arrival order and the space is taken back in arrival order too, which is
+// what makes it a ring: the writer and the republish thread may finish with messages out of order, so
+// a region released early is only remembered, and the space is reclaimed once everything received
+// before it is released as well. Republish holds at most its queue's worth of messages, so that is as
+// far as the reclaim can lag.
+//
+// A message larger than the whole ring goes on the heap instead, so an oversized START or END message
+// cannot wedge the receiver; everything else waits for space, which is the back-pressure that keeps the
+// sender from running ahead.
+class ReceiveRingBuffer {
+    struct State;
+    std::shared_ptr<State> state;
+public:
+    explicit ReceiveRingBuffer(size_t capacity);
+
+    // Blocks until the bytes are available; returns nullptr once Stop() has been called. A request
+    // larger than the capacity (so every request of a ring of capacity 0) is served from the heap,
+    // at once and whether stopped or not.
+    std::shared_ptr<ReceiveBufferRegion> Allocate(size_t size);
+
+    // Releases every waiter in Allocate; the regions already handed out stay valid.
+    void Stop();
+
+    size_t GetCapacity() const;
+    // Bytes held in live regions right now / at most since construction (heap fallbacks included).
+    size_t GetUtilization() const;
+    size_t GetMaxUtilization() const;
+};

image_puller/TCPImagePuller.cpp

@@ -46,8 +46,10 @@ static std::pair<std::string, uint16_t> ParseTcpAddressPull(const std::string &a
 TCPImagePuller::TCPImagePuller(const std::string &tcp_addr,
                                std::optional<int32_t> rcv_buffer_size,
                                const std::string &repub_address,
-                               const std::optional<int32_t> &repub_watermark)
-    : addr(tcp_addr),
+                               const std::optional<int32_t> &repub_watermark,
+                               const std::optional<size_t> &ring_size)
+    : ImagePuller(DefaultQueueSize, ring_size.value_or(0)),
+      addr(tcp_addr),
       receive_buffer_size(rcv_buffer_size) {
     auto parsed = ParseTcpAddressPull(tcp_addr);
     host = parsed.first;
@@ -130,7 +132,7 @@ void TCPImagePuller::CBORThread() {
             if (type == TCPFrameType::CANCEL) {
                 outside_fifo.PutBlocking(ret);
             } else {
-                ret.cbor = CBORStream2Deserialize(ret.tcp_msg->payload.data(), ret.tcp_msg->payload.size());
+                ret.cbor = CBORStream2Deserialize(ret.raw ? ret.raw->data : nullptr, ret.raw ? ret.raw->size : 0);
                 outside_fifo.PutBlocking(ret);
                 // A PREFLIGHT carries a StartMessage payload but starts nothing, so republishing it
                 // would announce a run to downstream consumers that is not going to happen.
@@ -138,11 +140,13 @@ void TCPImagePuller::CBORThread() {
                     if ((ret.cbor->msg_type == CBORImageType::START)
                         || (ret.cbor->msg_type == CBORImageType::END))
                         repub_fifo.PutBlocking(ret);
-                    else
-                        repub_fifo.Put(ret);
+                    else if (RepubCanHold(ret.raw->size) && repub_fifo.Put(ret))
+                        repub_bytes += ret.raw->size;
                 }
             }
-        } catch (const JFJochException &e) {
+        } catch (const std::exception &e) {
+            // std::exception: the deserialiser allocates a vector per per-image array of an END message,
+            // and a bad_alloc leaving this thread would end the process.
             logger.ErrorException(e);
         }
         ret = cbor_fifo.GetBlocking();
@@ -170,11 +174,13 @@ void TCPImagePuller::RepubThread() {
                     logger.Info("Republish active");
             } else {
                 if (repub_active)
-                    repub_socket->Send(ret.tcp_msg->payload.data(), ret.tcp_msg->payload.size(), true);
+                    repub_socket->Send(ret.raw->data, ret.raw->size, true);
             }
-        } catch (const JFJochException &e) {
+        } catch (const std::exception &e) {
             logger.ErrorException(e);
         }
+        if (ret.cbor && (ret.cbor->msg_type != CBORImageType::START) && (ret.cbor->msg_type != CBORImageType::END))
+            repub_bytes -= ret.raw->size;
         ret = repub_fifo.GetBlocking();
     }
     if (repub_active)
@@ -372,14 +378,21 @@ void TCPImagePuller::ReceiverThread() {
             ImagePullerOutput out;
             out.tcp_msg = std::make_shared<RawFrame>();
             out.tcp_msg->header = frame.header;
-            out.tcp_msg->payload.resize(frame.header.payload_size);
 
-            if (frame.header.payload_size > 0
-                && !ReadExact(out.tcp_msg->payload.data(), out.tcp_msg->payload.size())) {
-                logger.Info("TCP payload read failed, reconnecting to " + addr);
-                CloseSocket();
-                std::this_thread::sleep_for(std::chrono::milliseconds(20));
-                continue;
+            if (frame.header.payload_size > 0) {
+                // Read straight into the ring (or, with no ring, onto the heap). A full ring holds the
+                // read here, the socket buffer fills, and the pusher's send blocks - its BUSY heartbeats
+                // tell it apart from a dead writer, and it gives up only after a minute in which not one
+                // image was written.
+                out.raw = ring.Allocate(frame.header.payload_size);
+                if (!out.raw)
+                    break; // Stop() - the puller is being disconnected.
+                if (!ReadExact(out.raw->data, out.raw->size)) {
+                    logger.Info("TCP payload read failed, reconnecting to " + addr);
+                    CloseSocket();
+                    std::this_thread::sleep_for(std::chrono::milliseconds(20));
+                    continue;
+                }
             }
 
             cbor_fifo.PutBlocking(out);
@@ -438,6 +451,8 @@ void TCPImagePuller::Disconnect() {
     cbor_fifo.Stop();
     repub_fifo.Stop();
     outside_fifo.Stop();
+    // And ReceiverThread may be waiting for ring space that the stopped consumers will never free.
+    ring.Stop();
 
     if (receiver_thread.joinable())
         receiver_thread.join();

image_puller/TCPImagePuller.h

@@ -53,7 +53,8 @@ public:
     explicit TCPImagePuller(const std::string &tcp_addr,
                             std::optional<int32_t> rcv_buffer_size = {},
                             const std::string &repub_address = "",
-                            const std::optional<int32_t> &repub_watermark = {});
+                            const std::optional<int32_t> &repub_watermark = {},
+                            const std::optional<size_t> &ring_size = {});
 
     ~TCPImagePuller() override;
     bool SupportsAck() const override { return true; }

image_puller/ZMQImagePuller.cpp

@@ -4,10 +4,14 @@
 #include "ZMQImagePuller.h"
 #include "../frame_serialize/CBORStream2Serializer.h"
 
+#include <cstring>
+
 ZMQImagePuller::ZMQImagePuller(const std::string &in_addr,
                                const std::string &repub_address,
     const std::optional<int32_t> &rcv_watermark,
-    const std::optional<int32_t> &repub_watermark) :
+    const std::optional<int32_t> &repub_watermark,
+    const std::optional<size_t> &ring_size) :
+        ImagePuller(DefaultQueueSize, ring_size.value_or(0)),
         socket (ZMQSocketType::Pull), addr(in_addr) {
     auto start_time = std::chrono::steady_clock::now();
     puller_thread = std::thread(&ZMQImagePuller::PullerThread, this);
@@ -43,6 +47,8 @@ void ZMQImagePuller::Disconnect() {
     cbor_fifo.Stop();
     repub_fifo.Stop();
     outside_fifo.Stop();
+    // And PullerThread may be waiting for ring space that the stopped consumers will never free.
+    ring.Stop();
 
     if (puller_thread.joinable())
         puller_thread.join();
@@ -59,8 +65,7 @@ void ZMQImagePuller::Disconnect() {
 
 void ZMQImagePuller::PullerThread() {
     while (true) {
-        ImagePullerOutput ret;
-        ret.zmq_msg = std::make_shared<ZMQMessage>();
+        auto msg = std::make_shared<ZMQMessage>();
         bool received = false;
         while (!received) {
             if (disconnect) {
@@ -68,13 +73,33 @@ void ZMQImagePuller::PullerThread() {
                 return;
             }
             try {
-                received = socket.Receive(*ret.zmq_msg, false);
+                received = socket.Receive(*msg, false);
                 if (!received)
                     std::this_thread::sleep_for(std::chrono::milliseconds(1));
             } catch (const JFJochException &e) {
                 logger.ErrorException(e);
             }
         }
+        ImagePullerOutput ret;
+        if (ring.GetCapacity() == 0) {
+            // No ring: the ZeroMQ message itself is the buffer, kept alive by the region. No copy.
+            ret.raw = std::shared_ptr<ReceiveBufferRegion>(new ReceiveBufferRegion{msg->data(), msg->size()},
+                                                           [msg](ReceiveBufferRegion *r) { delete r; });
+        } else {
+            // Copied into the ring, and the ZeroMQ message is released with the next receive. When the
+            // ring is full this waits and the socket is not read: its queue fills to the watermark and
+            // the sender's send blocks - for as long as the sender is prepared to wait. The broker's
+            // ZeroMQ pusher waits one second, then drops the image and sends without waiting for the
+            // rest of the run, so a ring that is too small for the writer's stalls costs images.
+            ret.raw = ring.Allocate(msg->size());
+            if (!ret.raw) {
+                // Stop() - the puller is being disconnected.
+                cbor_fifo.PutBlocking(ImagePullerOutput{});
+                return;
+            }
+            if (msg->size() > 0)
+                memcpy(ret.raw->data, msg->data(), msg->size());
+        }
         cbor_fifo.PutBlocking(ret);
     }
 }
@@ -82,9 +107,9 @@ void ZMQImagePuller::PullerThread() {
 void ZMQImagePuller::CBORThread() {
     auto ret = cbor_fifo.GetBlocking();
 
-    while (ret.zmq_msg) {
+    while (ret.raw) {
         try {
-            ret.cbor = CBORStream2Deserialize(ret.zmq_msg->data(), ret.zmq_msg->size());
+            ret.cbor = CBORStream2Deserialize(ret.raw->data, ret.raw->size);
             // Even if we suspend consuming of the messages by the receiver,
             // it is still reasonable to have them republished
             // so republish functionality is not affected by Suspend()
@@ -94,10 +119,12 @@ void ZMQImagePuller::CBORThread() {
                 if ((ret.cbor->msg_type == CBORImageType::START)
                     || (ret.cbor->msg_type == CBORImageType::END))
                     repub_fifo.PutBlocking(ret);
-                else
-                    repub_fifo.Put(ret);
+                else if (RepubCanHold(ret.raw->size) && repub_fifo.Put(ret))
+                    repub_bytes += ret.raw->size;
             }
-        } catch (const JFJochException &e) {
+        } catch (const std::exception &e) {
+            // std::exception: the deserialiser allocates a vector per per-image array of an END message,
+            // and a bad_alloc leaving this thread would end the process.
             logger.ErrorException(e);
         }
         ret = cbor_fifo.GetBlocking();
@@ -111,7 +138,7 @@ void ZMQImagePuller::RepubThread() {
     auto ret = repub_fifo.GetBlocking();
     bool repub_active = false;
 
-    while (ret.zmq_msg) {
+    while (ret.raw) {
         try {
             if (ret.cbor->msg_type == CBORImageType::START) {
                 // Start message needs to be cleaned when running republish
@@ -125,11 +152,13 @@ void ZMQImagePuller::RepubThread() {
                     logger.Info("Republish active");
             } else {
                 if (repub_active)
-                    repub_socket->Send(ret.zmq_msg->data(), ret.zmq_msg->size(), true);
+                    repub_socket->Send(ret.raw->data, ret.raw->size, true);
             }
-        } catch (const JFJochException &e) {
+        } catch (const std::exception &e) {
             logger.ErrorException(e);
         }
+        if (ret.cbor && (ret.cbor->msg_type != CBORImageType::START) && (ret.cbor->msg_type != CBORImageType::END))
+            repub_bytes -= ret.raw->size;
         ret = repub_fifo.GetBlocking();
     }
     if (repub_active)

image_puller/ZMQImagePuller.h

@@ -15,7 +15,9 @@
 
 
 class ZMQImagePuller : public ImagePuller {
-    const uint32_t default_receive_watermark = 100;
+    // Messages, not bytes: what the socket holds beyond the ring buffer. Kept small for that reason -
+    // the ring is where the memory is accounted for.
+    const uint32_t default_receive_watermark = 10;
     const uint32_t default_repub_watermark = 220;
 
     // ZeroMQ receive timeout allows to check for abort value from time to time
@@ -47,7 +49,8 @@ public:
             const std::string &zmq_addr,
             const std::string &repub_address = "",
             const std::optional<int32_t> &rcv_watermark = {},
-            const std::optional<int32_t> &repub_watermark = {});
+            const std::optional<int32_t> &repub_watermark = {},
+            const std::optional<size_t> &ring_size = {});
     ~ZMQImagePuller();
 
     void Disconnect() override;

image_pusher/ZMQStream2PusherSocket.cpp

@@ -3,6 +3,8 @@
 
 #include "ZMQStream2PusherSocket.h"
 
+#include <spdlog/spdlog.h>
+
 void zmq_socket_free(void *data, void *hint) {
     auto z = (ZeroCopyReturnValue *) hint;
     z->release();
@@ -56,6 +58,10 @@ void ZMQStream2PusherSocket::WriterThread() {
         // When hitting timeout on SendZeroCopy consider it a transmission error and switch to non-blocking communication
         bool blocking_send = !transmission_error;
         if (!s.SendZeroCopy(e.z->GetImage(), e.z->GetImageSize(), zmq_socket_free, e.z, blocking_send)) {
+            if (!transmission_error)
+                spdlog::warn("ZeroMQ image socket {}: send blocked for over a second - image {} dropped, and "
+                             "images that find the queue full are dropped for the rest of the run",
+                             s.GetEndpointName(), e.z->GetImageNumber());
             transmission_error = true;
             // The slot is already released: on a failed send SendZeroCopy closes the message, and
             // closing a zmq_msg_init_data message runs its free function (zmq_socket_free).

tests/CMakeLists.txt

@@ -57,6 +57,7 @@ ADD_EXECUTABLE(jfjoch_test
         AcquisitionCountersTest.cpp
         IndexingUnitTest.cpp
         ThreadSafeFIFOSetTest.cpp
+        ReceiveRingBufferTest.cpp
         FPGANetworkTest.cpp
         JFPedestalTest.cpp
         JFCalibrationTest.cpp

tests/ReceiveRingBufferTest.cpp

@@ -0,0 +1,129 @@
+// 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 <future>
+#include <thread>
+
+#include "../image_puller/ReceiveRingBuffer.h"
+
+using namespace std::chrono_literals;
+
+TEST_CASE("ReceiveRingBuffer_sequential_reuse", "[ImagePuller][Unit]") {
+    ReceiveRingBuffer ring(1024);
+    // Far more messages than fit at once: each one is released before the next is taken.
+    for (int i = 0; i < 100; i++) {
+        auto r = ring.Allocate(300);
+        REQUIRE(r);
+        REQUIRE(r->size == 300);
+        r->data[0] = 1;
+        r->data[299] = 2;
+    }
+    CHECK(ring.GetUtilization() == 0);
+    CHECK(ring.GetMaxUtilization() == 300);
+    CHECK(ring.GetCapacity() == 1024);
+}
+
+TEST_CASE("ReceiveRingBuffer_wraps_to_the_start", "[ImagePuller][Unit]") {
+    ReceiveRingBuffer ring(1000);
+    auto a = ring.Allocate(400);
+    auto b = ring.Allocate(400);
+    REQUIRE(a);
+    REQUIRE(b);
+    uint8_t *a_data = a->data;
+
+    // 200 bytes left at the end and nothing free at the start: a third 400 has to wait.
+    auto f = std::async(std::launch::async, [&] { return ring.Allocate(400); });
+    CHECK(f.wait_for(50ms) == std::future_status::timeout);
+
+    a.reset(); // frees [0, 400): the waiter wraps round to the start
+    auto c = f.get();
+    REQUIRE(c);
+    CHECK(c->data == a_data);
+    CHECK(ring.GetUtilization() == 800);
+    // Never two live regions on the same bytes
+    CHECK((b->data + 400 <= c->data || c->data + 400 <= b->data));
+}
+
+TEST_CASE("ReceiveRingBuffer_full_ring_blocks", "[ImagePuller][Unit]") {
+    ReceiveRingBuffer ring(1000);
+    auto a = ring.Allocate(500);
+    auto b = ring.Allocate(500);   // tail has come round to head
+    REQUIRE(a);
+    REQUIRE(b);
+    uint8_t *a_data = a->data;
+    auto f = std::async(std::launch::async, [&] { return ring.Allocate(1); });
+    CHECK(f.wait_for(50ms) == std::future_status::timeout);
+    a.reset();
+    auto c = f.get();
+    REQUIRE(c);
+    CHECK(c->data == a_data);
+}
+
+TEST_CASE("ReceiveRingBuffer_out_of_order_release_reclaims_in_order", "[ImagePuller][Unit]") {
+    ReceiveRingBuffer ring(1000);
+    auto a = ring.Allocate(500);
+    auto b = ring.Allocate(500);
+    REQUIRE(a);
+    REQUIRE(b);
+
+    b.reset(); // the newer region goes first - as republish may do
+    CHECK(ring.GetUtilization() == 1000); // its bytes are behind a's and cannot be reused yet
+
+    auto f = std::async(std::launch::async, [&] { return ring.Allocate(600); });
+    CHECK(f.wait_for(50ms) == std::future_status::timeout);
+
+    a.reset(); // now both are reclaimed
+    auto c = f.get();
+    REQUIRE(c);
+    CHECK(ring.GetUtilization() == 600);
+}
+
+TEST_CASE("ReceiveRingBuffer_oversized_message_goes_to_heap", "[ImagePuller][Unit]") {
+    ReceiveRingBuffer ring(100);
+    auto small = ring.Allocate(100);
+    REQUIRE(small);
+    // Does not wait for the full ring - it never fits in it
+    auto big = ring.Allocate(1000);
+    REQUIRE(big);
+    REQUIRE(big->size == 1000);
+    big->data[999] = 1;
+    CHECK(ring.GetUtilization() == 1100);
+    CHECK(ring.GetMaxUtilization() == 1100);
+    big.reset();
+    CHECK(ring.GetUtilization() == 100);
+}
+
+TEST_CASE("ReceiveRingBuffer_zero_size", "[ImagePuller][Unit]") {
+    ReceiveRingBuffer ring(100);
+    auto a = ring.Allocate(0);
+    auto b = ring.Allocate(100);
+    REQUIRE(a);
+    REQUIRE(b);
+    CHECK(a->size == 0);
+    a.reset();
+    CHECK(ring.GetUtilization() == 100);
+    b.reset();
+    CHECK(ring.GetUtilization() == 0);
+}
+
+TEST_CASE("ReceiveRingBuffer_stop_releases_waiter", "[ImagePuller][Unit]") {
+    ReceiveRingBuffer ring(100);
+    auto a = ring.Allocate(100);
+    REQUIRE(a);
+    auto f = std::async(std::launch::async, [&] { return ring.Allocate(50); });
+    CHECK(f.wait_for(50ms) == std::future_status::timeout);
+    ring.Stop();
+    CHECK(f.get() == nullptr);
+}
+
+TEST_CASE("ReceiveRingBuffer_region_outlives_ring", "[ImagePuller][Unit]") {
+    std::shared_ptr<ReceiveBufferRegion> r;
+    {
+        ReceiveRingBuffer ring(100);
+        r = ring.Allocate(50);
+    }
+    REQUIRE(r);
+    r->data[49] = 1;
+    r.reset();
+}

tests/StreamWriterTest.cpp

@@ -11,6 +11,7 @@
 #include "../image_pusher/ZMQStream2Pusher.h"
 #include "../receiver/JFJochReceiverService.h"
 #include "../image_pusher/ZMQWriterNotificationPuller.h"
+#include "../image_puller/TestImagePuller.h"
 
 TEST_CASE("StreamWriterTest_ZMQ", "[StreamWriter]") {
     RegisterHDF5Filter();
@@ -246,3 +247,52 @@ TEST_CASE("StreamWriterTest_ZMQ_MasterAfterAllDataFiles", "[StreamWriter]") {
 
     std::filesystem::remove_all("subdir4");
 }
+
+// A failure that is not a JFJochException must end the run the same way as one that is: error state,
+// END processed, state Finalized. Before this test the writer caught only its own exception type at the
+// per-image and END sites, and anything else - here the JSON library refusing a sample name that is not
+// UTF-8 while publishing a file-done notification - left Run() with the state stuck in Receiving, no END
+// acknowledgement and no notification, and on TCP a broker waiting for that acknowledgement for ever.
+TEST_CASE("StreamWriterTest_NonJFJochException_EndsRunInErrorState", "[StreamWriter]") {
+    RegisterHDF5Filter();
+    Logger logger("StreamWriterTest_NonJFJochException");
+
+    DiffractionExperiment x(DetJF(2));
+    x.FilePrefix("subdir5/StreamWriterTest5").NumTriggers(1).ImagesPerTrigger(2).ImagesPerFile(1)
+            .Compression(CompressionAlgorithm::NO_COMPRESSION).OverwriteExistingFiles(true)
+            .SampleName("\xff\xfe not utf-8");
+
+    StartMessage start_message;
+    x.FillMessage(start_message);
+    start_message.write_master_file = true;
+
+    TestImagePuller puller;
+    // The file-done PUB socket is what makes the writer serialise the sample name to JSON.
+    StreamWriter writer(logger, puller, "ipc://*");
+
+    puller.Put({.cbor = std::make_shared<CBORStream2DeserializerOutput>(start_message)});
+    std::vector<int16_t> image(x.GetPixelsNum());
+    for (int i = 0; i < 2; i++) {
+        DataMessage message{};
+        message.image = CompressedImage(image, x.GetXPixelsNum(), x.GetYPixelsNum());
+        message.number = i;
+        puller.Put({.cbor = std::make_shared<CBORStream2DeserializerOutput>(message)});
+    }
+    EndMessage end_message{};
+    end_message.max_image_number = 2;
+    puller.Put({.cbor = std::make_shared<CBORStream2DeserializerOutput>(end_message)});
+
+    StreamWriterOutput out;
+    REQUIRE_NOTHROW(out = writer.Run());
+
+    // Image 0 filled its one-image data file; closing it published the notification, which threw.
+    CHECK(out.stats.state == StreamWriterState::Finalized);
+    CHECK(out.stats.processed_images == 0);
+    CHECK(out.data_file_stats.empty());
+    // The master is never written for a run that ended in error; the data file was renamed into
+    // place before the notification was attempted.
+    CHECK(!std::filesystem::exists("subdir5/StreamWriterTest5_master.h5"));
+    CHECK(std::filesystem::exists("subdir5/StreamWriterTest5_data_000001.h5"));
+
+    std::filesystem::remove_all("subdir5");
+}

tests/TCPImagePusherTest.cpp

@@ -41,8 +41,13 @@ TEST_CASE("TCPImageCommTest_2Writers_WithAck", "[TCP]") {
 
     TCPStreamPusher pusher("tcp://127.0.0.1:*", npullers);
     std::vector<std::unique_ptr<TCPImagePuller> > puller;
+    // The second writer receives into a ring that holds three images (with their CBOR framing): a
+    // full ring stalls its read, the pusher's send blocks, and nothing may be lost or reordered.
+    const size_t small_ring = 3 * x.GetPixelsNum() * sizeof(uint16_t) + 3 * 65536;
     for (int i = 0; i < npullers; i++)
-        puller.push_back(std::make_unique<TCPImagePuller>(pusher.GetAddress()[0], 64 * 1024 * 1024));
+        puller.push_back(std::make_unique<TCPImagePuller>(pusher.GetAddress()[0], 64 * 1024 * 1024, "", std::nullopt,
+                                                          (i == 1) ? std::optional<size_t>(small_ring) : std::nullopt));
+    REQUIRE(puller[1]->GetRingBufferCapacity() == small_ring);
 
     // Wait for all pullers to connect before starting data collection
     for (int attempt = 0; attempt < 100 && pusher.GetConnectedWriters() < static_cast<size_t>(npullers); ++attempt)

tests/ZMQImagePusherTest.cpp

@@ -420,3 +420,75 @@ TEST_CASE("ZMQImagePuller_RecoversFromFullQueue","[ZeroMQ]") {
 
     puller.Disconnect();
 }
+
+// The whole run goes through a ring that holds three images: the puller has to wait for space, the
+// socket queue fills, the pusher blocks, and nothing is lost or reordered. The peak ring use proves
+// the bound held.
+TEST_CASE("ZMQImageCommTest_SmallRingBuffer","[ZeroMQ]") {
+    const size_t nframes = 64;
+
+    Logger logger(Catch::getResultCapture().getCurrentTestName());
+
+    DiffractionExperiment x(DetJF(1));
+    x.Raw();
+    x.PedestalG0Frames(0).NumTriggers(1).UseInternalPacketGenerator(false).IncidentEnergy_keV(12.4)
+            .ImagesPerTrigger(nframes).Compression(CompressionAlgorithm::NO_COMPRESSION);
+    REQUIRE(x.GetImageNum() == nframes);
+
+    std::mt19937 g1(1387);
+    std::uniform_int_distribution<uint16_t> dist;
+    std::vector<uint16_t> image1(x.GetPixelsNum()*nframes);
+    for (auto &i: image1) i = dist(g1);
+
+    const size_t image_bytes = x.GetPixelsNum() * sizeof(uint16_t);
+    const size_t ring_size = 3 * image_bytes + 3 * 65536; // three images with their CBOR framing
+
+    ZMQStream2Pusher pusher({"ipc://*"});
+    std::vector<size_t> diff_size(1), diff_content(1), diff_split(1), nimages(1);
+
+    auto pusher_addr = pusher.GetAddress();
+    ZMQImagePuller puller(pusher_addr[0], "", {}, {}, ring_size);
+    REQUIRE(puller.GetRingBufferCapacity() == ring_size);
+
+    std::thread sender_thread = std::thread([&] {
+        std::vector<uint8_t> serialization_buffer(16*1024*1024);
+        CBORStream2Serializer serializer(serialization_buffer.data(), serialization_buffer.size());
+
+        StartMessage message {
+                .images_per_file = 16,
+                .write_master_file = true
+        };
+        EndMessage end_message{};
+
+        pusher.StartDataCollection(message);
+        for (int i = 0; i < nframes; i++) {
+            DataMessage data_message;
+            data_message.number = i;
+            data_message.image = CompressedImage(image1.data() + i * x.GetPixelsNum(),
+                                                 image_bytes,
+                                                 x.GetXPixelsNum(),
+                                                 x.GetYPixelsNum(),
+                                                 x.GetImageMode(),
+                                                 x.GetCompressionAlgorithm());
+            serializer.SerializeImage(data_message);
+            pusher.SendImage(serialization_buffer.data(), serializer.GetBufferSize(), i);
+        }
+
+        pusher.EndDataCollection(end_message);
+    });
+
+    std::thread puller_thread(test_puller, &puller, std::cref(x), std::cref(image1), 1, 0,
+                              std::ref(diff_split), std::ref(diff_size), std::ref(diff_content),
+                              std::ref(nimages));
+
+    sender_thread.join();
+    puller_thread.join();
+
+    puller.Disconnect();
+
+    REQUIRE(nimages[0] == nframes);
+    REQUIRE(diff_size[0] == 0);
+    REQUIRE(diff_content[0] == 0);
+    CHECK(puller.GetMaxRingBufferUtilization() <= ring_size);
+    CHECK(puller.GetMaxRingBufferUtilization() >= image_bytes);
+}

writer/FileWriter.cpp

@@ -87,9 +87,11 @@ void FileWriter::WriteHDF5(const DataMessage& msg) {
         if (files.size() <= file_number)
             files.resize(file_number + 1);
 
-        if (!files[file_number])
+        if (!files[file_number]) {
             files[file_number] = std::make_unique<HDF5DataFile>(start_message, file_number,
                                                                 HDF5Metadata::DataFileName(start_message, file_number));
+            open_files.insert(file_number);
+        }
         files[file_number]->Write(msg, image_number);
 
         if (files[file_number]->GetNumImages() == start_message.images_per_file) {
@@ -111,6 +113,7 @@ void FileWriter::CloseFile(uint64_t file_number) {
     auto file_stats = files[file_number]->Close();
     files[file_number].reset();
     closed_files.insert(file_number);
+    open_files.erase(file_number);
 
     AddStats(file_stats);
 }
@@ -119,14 +122,15 @@ void FileWriter::CloseOldFiles(uint64_t current_image_number) {
     if (start_message.images_per_file == 0)
         return;
 
-    for (uint64_t f = 0; f < files.size(); ++f) {
-        if (!files[f] || closed_files.contains(f))
-            continue;
-
+    // Ascending, so the first file that is not old enough ends the walk; CloseFile erases from the set,
+    // hence the iterator is advanced first.
+    for (auto it = open_files.begin(); it != open_files.end();) {
+        const uint64_t f = *it++;
         const uint64_t file_end_image = (f + 1) * start_message.images_per_file - 1;
-        if (current_image_number > file_end_image + close_file_lag_images) {
+        if (current_image_number > file_end_image + close_file_lag_images)
             CloseFile(f);
-        }
+        else
+            break;
     }
 }
 

writer/FileWriter.h

@@ -10,6 +10,7 @@
 #include "../common/ZMQWrappers.h"
 #include "HDF5NXmx.h"
 #include <functional>
+#include <set>
 #include <unordered_set>
 
 class FileWriter {
@@ -23,6 +24,10 @@ class FileWriter {
     std::function<void(const HDF5DataFileStatistics &)> file_closed_callback;
 
     std::unordered_set<uint64_t> closed_files;
+    // File numbers with a live HDF5DataFile, in order. CloseOldFiles walks this and not `files`, whose
+    // slots stay (empty) for every file of the run - a thousand of them for a long run, visited on every
+    // image.
+    std::set<uint64_t> open_files;
     constexpr static uint64_t close_file_lag_images = 1000;
     constexpr static uint64_t default_images_per_file = 1000;
     void CreateHDF5MasterFile(const StartMessage& msg);

writer/StreamWriter.cpp

@@ -178,7 +178,12 @@ void StreamWriter::ProcessDataImage() {
                 if (verbose)
                     logger.Info("Written");
                 NotifyTcpAck(TCPFrameType::DATA, true, false, TCPAckCode::None);
-            } catch (const JFJochException &e) {
+            } catch (const std::exception &e) {
+                // std::exception, not JFJochException, for the reason given in ProcessStartMessage: an
+                // exception that leaves this function skips the error state and the fatal ACK, and the
+                // broker then waits for an END ACK that never comes. Not every thrower below is ours -
+                // std::bad_alloc from a per-image array, filesystem_error from a path check, the JSON
+                // library when a file-done notification carries a string that is not UTF-8.
                 logger.ErrorException(e);
                 logger.Warning("Error writing image - switching to error state");
                 state = StreamWriterState::Error;
@@ -206,7 +211,11 @@ void StreamWriter::ProcessEndMessage() {
             if ((image_puller_output.cbor->end_message->max_image_number == 0) && (max_image_number > 0))
                 image_puller_output.cbor->end_message->max_image_number = max_image_number;
             file_writer->WriteHDF5(*image_puller_output.cbor->end_message);
-        } catch (const JFJochException &e) {
+        } catch (const std::exception &e) {
+            // std::exception for the same reason as in ProcessDataImage. Here the cost of a leak is the
+            // highest: the END ACK and the notification below are never sent, the state never reaches
+            // Finalized, and on TCP the broker waits for the ACK for as long as the writer keeps
+            // heartbeating - which it does, whatever state it is in.
             logger.ErrorException(e);
             logger.Error("Error writing end message - switching to error state");
             state = StreamWriterState::Error;
@@ -333,9 +342,9 @@ StreamWriterOutput StreamWriter::Run() {
     StreamWriterOutput ret;
     ret.data_file_stats = hdf5_data_file_statistics;
     ret.stats = GetStatistics();
-    logger.Info("Write task done. Images = {} Throughput = {:.0f} MB/s Frame rate = {:.0f} Hz max occupation of FIFO {}",
+    logger.Info("Write task done. Images = {} Throughput = {:.0f} MB/s Frame rate = {:.0f} Hz max occupation of FIFO {} ring buffer peak {} MB",
                 ret.stats.processed_images, ret.stats.performance_MBs, ret.stats.performance_Hz,
-                ret.stats.max_puller_fifo_utilization);
+                ret.stats.max_puller_fifo_utilization, image_puller.GetMaxRingBufferUtilization() >> 20);
     return ret;
 }
 
@@ -411,7 +420,8 @@ void StreamWriter::NotifyReceiverOnFinalizedWrite(const std::string &detector_up
         s.SendTimeout(std::chrono::seconds(1));
         s.Connect(detector_update_zmq_addr);
         s.Send(j.dump());
-    } catch (const JFJochException &e) {
+    } catch (const std::exception &e) {
+        // std::exception: json::dump throws its own type when run_name is not valid UTF-8.
         logger.ErrorException(e);
         logger.Error("Error sending notification to detector update socket");
     }

writer/jfjoch_writer.cpp

@@ -25,7 +25,8 @@ void print_usage() {
     logger.Info("-d<path>| --root_dir=<path>          Root directory for file writing");
     logger.Info("-r<int> | --zmq_repub_port=<int>     ZeroMQ port for PUSH socket to republish images");
     logger.Info("-f<int> | --zmq_file_port=<int>      ZeroMQ port for PUB socket to inform about finalized files");
-    logger.Info("-w<int> | --rcv_watermark=<int>      Receiving ZeroMQ socket watermark (default = 100)");
+    logger.Info("-w<int> | --rcv_watermark=<int>      Receiving ZeroMQ socket watermark in messages (default = 10)");
+    logger.Info("-m<int> | --rcv_ring_mb=<int>        Receive ring buffer in MB; holds images until written and republished (default = 2048)");
     logger.Info("-W<int> | --repub_watermark=<int>    Republish ZeroMQ socket watermark (default = 1000)");
     logger.Info("-b<int> | --rcv_buffer=<int>         Receiving TCP socket buffer in bytes (only with -T; disables autotuning, capped by net.core.rmem_max)");
     logger.Info("-v      | --verbose                  Verbose output");
@@ -74,6 +75,8 @@ int main(int argc, char **argv) {
     std::optional<int32_t> rcv_watermark;
     std::optional<int32_t> repub_watermark;
     std::optional<int32_t> rcv_buffer;
+    // jfjoch_writer always runs with a ring; the puller's own default (no ring) is for the broker.
+    size_t ring_size = static_cast<size_t>(2048) << 20;
     bool verbose = false;
 
     static struct option long_options[] = {
@@ -84,6 +87,7 @@ int main(int argc, char **argv) {
             {"rcv_watermark", required_argument, 0, 'w'},
             {"repub_watermark", required_argument, 0, 'W'},
             {"rcv_buffer", required_argument, 0, 'b'},
+            {"rcv_ring_mb", required_argument, 0, 'm'},
             {"verbose", no_argument, 0, 'v'},
             {"tcp", no_argument, 0, 'T'},
             {"nproc", required_argument, 0, 'j'},
@@ -93,7 +97,7 @@ int main(int argc, char **argv) {
     int option_index = 0;
     int opt;
     bool raw_tcp = false;
-    while ((opt = getopt_long(argc, argv, "?hH:r:f:R:d:W:w:b:vTj:",long_options, &option_index)) != -1 ) {
+    while ((opt = getopt_long(argc, argv, "?hH:r:f:R:d:W:w:b:m:vTj:",long_options, &option_index)) != -1 ) {
         switch (opt) {
             case 'j':
                 nforks = atoi(optarg);
@@ -129,6 +133,13 @@ int main(int argc, char **argv) {
                     exit(EXIT_FAILURE);
                 }
                 break;
+            case 'm':
+                if (atoi(optarg) <= 0) {
+                    std::cerr << "Ring buffer size must be positive number of MB" << std::endl;
+                    exit(EXIT_FAILURE);
+                }
+                ring_size = static_cast<size_t>(atoi(optarg)) << 20;
+                break;
             case 'W':
                 repub_watermark = atoi(optarg);
                 if (repub_watermark.value() <= 0) {
@@ -240,9 +251,12 @@ int main(int argc, char **argv) {
 
     std::unique_ptr<ImagePuller> puller;
     if (raw_tcp)
-        puller = std::make_unique<TCPImagePuller>(argv[first_argc], rcv_buffer, repub_zmq_addr, repub_watermark);
+        puller = std::make_unique<TCPImagePuller>(argv[first_argc], rcv_buffer, repub_zmq_addr, repub_watermark,
+                                                  ring_size);
     else
-        puller = std::make_unique<ZMQImagePuller>(argv[first_argc], repub_zmq_addr, rcv_watermark, repub_watermark);
+        puller = std::make_unique<ZMQImagePuller>(argv[first_argc], repub_zmq_addr, rcv_watermark, repub_watermark,
+                                                  ring_size);
+    logger.Info("Receive ring buffer {} MB", puller->GetRingBufferCapacity() >> 20);
 
 
     writer = new StreamWriter(logger,*puller,file_done_zmq_addr, verbose);