From 5f6f257998eefa8f6a6b667aeac06f384d1ab68c Mon Sep 17 00:00:00 2001 From: Filip Leonarski Date: Tue, 13 Oct 2026 12:27:08 +0200 Subject: [PATCH 1/4] Writer: a failure that is not a JFJochException ends the run in error state too ProcessDataImage, ProcessEndMessage and NotifyReceiverOnFinalizedWrite caught only JFJochException, while ProcessStartMessage and FinalizeDataCollection already caught std::exception. Anything else - std::bad_alloc from a per-image array, filesystem_error from a path check, the JSON library refusing a string that is not UTF-8 while publishing a file-done notification - left CollectImages, and Run() logged "Exception not properly handled" and returned with the state never reaching Finalized: no END acknowledgement, no writer notification, the master's .tmp left behind until the next START. On the TCP image stream the broker then waited for the END acknowledgement indefinitely, because the writer's heartbeat thread keeps reporting it alive whatever state it is in. Now the same catch as at the other sites: error state, fatal ACK, END processed, Finalized. The new StreamWriter test provokes the JSON case through a sample name that is not UTF-8 with the file-done socket enabled, and checks the run ends Finalized with no master written. Co-Authored-By: Claude Fable 5.1 --- tests/StreamWriterTest.cpp | 50 ++++++++++++++++++++++++++++++++++++++ writer/StreamWriter.cpp | 16 +++++++++--- 2 files changed, 63 insertions(+), 3 deletions(-) diff --git a/tests/StreamWriterTest.cpp b/tests/StreamWriterTest.cpp index 662477ccf..1b767fb64 100644 --- a/tests/StreamWriterTest.cpp +++ b/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(start_message)}); + std::vector 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(message)}); + } + EndMessage end_message{}; + end_message.max_image_number = 2; + puller.Put({.cbor = std::make_shared(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"); +} diff --git a/writer/StreamWriter.cpp b/writer/StreamWriter.cpp index af1bde36f..12bc6b8fa 100644 --- a/writer/StreamWriter.cpp +++ b/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; @@ -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"); } From 1be6093429ecb3b1045a05c61af13d7a2bffc2ac Mon Sep 17 00:00:00 2001 From: Filip Leonarski Date: Tue, 13 Oct 2026 12:51:34 +0200 Subject: [PATCH 2/4] Writer: received images live in a byte-bounded ring buffer until written and republished Every stage between the socket and the HDF5 write was bounded by a message count and never by bytes: ZeroMQ receive watermark 100, cbor_fifo 200, outside_fifo 1000, repub_fifo 200 - up to some 1500 images in flight per writer process, each a whole compressed image, so 6-12 GB for a JUNGFRAU 4M and 30-60 GB for a 9M, and the writer's memory was a quantity nobody had chosen. ReceiveRingBuffer is a fixed ring of bytes owned by the ImagePuller base class, so the ZeroMQ and TCP pullers share it. A message is received straight into it (TCP reads the payload into the ring; ZeroMQ copies the message in) and travels as a shared_ptr region in ImagePullerOutput; the deserialised image points into that region, republish sends it as is, and the bytes come back when the last holder lets go. Space is reclaimed in arrival order: a region released out of turn - republish finishes with messages in its own order - is marked and waits for the ones in front of it. Because of that, what republish may hold is capped at a quarter of the ring, and an image over the cap is not republished, exactly as one that finds the republish queue full; otherwise a stalled republish consumer would pin the ring and slow the writer to its rate. A message larger than the whole ring (a START or END of a very long run) goes on the heap, so it cannot wedge the receiver. A full ring stalls the receive thread, and what that does depends on the stream. On TCP the pusher waits behind the writer's BUSY heartbeats and gives up only after a minute in which no image at all was written. On ZeroMQ the pusher's send blocks for one second, then drops the image and sends without waiting for the rest of the run - so on ZeroMQ a ring that is smaller than the writer's stalls costs images, where the old 1500-message queues absorbed them. That is the trade: the budget is now a number, and it has to be chosen. jfjoch_writer sets it with -m/--rcv_ring_mb, default 2048 MB per writer process; the ZeroMQ receive watermark default goes from 100 to 10 messages, since it now only counts what sits beyond the ring. The ring's peak use is in the end-of-run log line, and docs/JFJOCH_WRITER.md has a Memory section with the sizing rule. The puller's own default is no ring at all, which keeps every other user as it was: the broker's DECTRIS stream2 puller (JFJochServices, JFJochReceiverService) stays zero-copy with the ZeroMQ message as the buffer, and the broker's direct HDF5 writer (HDF5FilePusher -> FileWriter) never goes through a puller. The puller threads now also catch std::exception, as the writer does since the previous commit: deserialising the END message of a million-image run allocates a vector per per-image array, and a bad_alloc leaving the thread ended the process. Tests: ReceiveRingBufferTest (reuse, wrap-around, full ring, out-of-order release, heap fallback, zero size, stop, a region outliving its ring); ZMQImageCommTest_SmallRingBuffer (64 one-megabyte images through a three-image ring, nothing lost, peak within the bound); one of the two writers in TCPImageCommTest_2Writers_WithAck runs on a three-image ring. Co-Authored-By: Claude Fable 5.1 --- docs/JFJOCH_WRITER.md | 30 +++++- image_puller/CMakeLists.txt | 4 +- image_puller/ImagePuller.cpp | 10 +- image_puller/ImagePuller.h | 26 ++++- image_puller/ReceiveRingBuffer.cpp | 146 +++++++++++++++++++++++++++++ image_puller/ReceiveRingBuffer.h | 52 ++++++++++ image_puller/TCPImagePuller.cpp | 45 ++++++--- image_puller/TCPImagePuller.h | 3 +- image_puller/ZMQImagePuller.cpp | 53 ++++++++--- image_puller/ZMQImagePuller.h | 7 +- tests/CMakeLists.txt | 1 + tests/ReceiveRingBufferTest.cpp | 129 +++++++++++++++++++++++++ tests/TCPImagePusherTest.cpp | 7 +- tests/ZMQImagePusherTest.cpp | 72 ++++++++++++++ writer/StreamWriter.cpp | 4 +- writer/jfjoch_writer.cpp | 22 ++++- 16 files changed, 568 insertions(+), 43 deletions(-) create mode 100644 image_puller/ReceiveRingBuffer.cpp create mode 100644 image_puller/ReceiveRingBuffer.h create mode 100644 tests/ReceiveRingBufferTest.cpp diff --git a/docs/JFJOCH_WRITER.md b/docs/JFJOCH_WRITER.md index 3bd9f8823..1ebca2e6e 100644 --- a/docs/JFJOCH_WRITER.md +++ b/docs/JFJOCH_WRITER.md @@ -21,7 +21,8 @@ Options: -d | --root_dir= Root directory for file writing (-R is a deprecated alias) -r | --zmq_repub_port= ZeroMQ port for PUSH socket to republish images -f | --zmq_file_port= ZeroMQ port for PUB socket for notifications on finalized files --w | --rcv_watermark= Receiving ZeroMQ socket watermark (default = 100) +-w | --rcv_watermark= Receiving ZeroMQ socket watermark in messages (default = 10) +-m | --rcv_ring_mb= Receive ring buffer in MB; holds images until written and republished (default = 2048) -W | --repub_watermark= Republish ZeroMQ socket watermark (default = 1000) -b | --rcv_buffer= 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 diff --git a/image_puller/CMakeLists.txt b/image_puller/CMakeLists.txt index 2ac60e680..4fb07c230 100644 --- a/image_puller/CMakeLists.txt +++ b/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) diff --git a/image_puller/ImagePuller.cpp b/image_puller/ImagePuller.cpp index 98e10f485..4855a61b0 100644 --- a/image_puller/ImagePuller.cpp +++ b/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(); diff --git a/image_puller/ImagePuller.h b/image_puller/ImagePuller.h index 41c8533dd..2d1d433dc 100644 --- a/image_puller/ImagePuller.h +++ b/image_puller/ImagePuller.h @@ -3,6 +3,7 @@ #pragma once +#include #include #include @@ -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 payload; bool end = false; }; struct ImagePullerOutput { - std::shared_ptr 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 raw; std::shared_ptr tcp_msg; std::shared_ptr cbor; }; @@ -42,16 +46,32 @@ struct ImagePullerStatistics { class ImagePuller { protected: ThreadSafeFIFO 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 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 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; } diff --git a/image_puller/ReceiveRingBuffer.cpp b/image_puller/ReceiveRingBuffer.cpp new file mode 100644 index 000000000..c01b86c1e --- /dev/null +++ b/image_puller/ReceiveRingBuffer.cpp @@ -0,0 +1,146 @@ +// SPDX-FileCopyrightText: 2026 Filip Leonarski, Paul Scherrer Institute +// SPDX-License-Identifier: GPL-3.0-only + +#include "ReceiveRingBuffer.h" + +#include +#include + +struct ReceiveRingBuffer::State { + // Uninitialised on purpose: a ring that is never filled should not cost its capacity in + // resident memory. + std::unique_ptr 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 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(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 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(capacity)) {} + +std::shared_ptr 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(); + + 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(region, [s = state](ReceiveBufferRegion *r) { + delete[] r->data; + s->ReleaseHeap(r->size); + delete r; + }); + } + + std::optional 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(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; +} diff --git a/image_puller/ReceiveRingBuffer.h b/image_puller/ReceiveRingBuffer.h new file mode 100644 index 000000000..c1831117a --- /dev/null +++ b/image_puller/ReceiveRingBuffer.h @@ -0,0 +1,52 @@ +// SPDX-FileCopyrightText: 2026 Filip Leonarski, Paul Scherrer Institute +// SPDX-License-Identifier: GPL-3.0-only + +#pragma once + +#include +#include +#include +#include +#include + +// 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; +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 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; +}; diff --git a/image_puller/TCPImagePuller.cpp b/image_puller/TCPImagePuller.cpp index fdb5e792c..122729127 100644 --- a/image_puller/TCPImagePuller.cpp +++ b/image_puller/TCPImagePuller.cpp @@ -46,8 +46,10 @@ static std::pair ParseTcpAddressPull(const std::string &a TCPImagePuller::TCPImagePuller(const std::string &tcp_addr, std::optional rcv_buffer_size, const std::string &repub_address, - const std::optional &repub_watermark) - : addr(tcp_addr), + const std::optional &repub_watermark, + const std::optional &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(); 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(); diff --git a/image_puller/TCPImagePuller.h b/image_puller/TCPImagePuller.h index 3a0c795ed..151e931e1 100644 --- a/image_puller/TCPImagePuller.h +++ b/image_puller/TCPImagePuller.h @@ -53,7 +53,8 @@ public: explicit TCPImagePuller(const std::string &tcp_addr, std::optional rcv_buffer_size = {}, const std::string &repub_address = "", - const std::optional &repub_watermark = {}); + const std::optional &repub_watermark = {}, + const std::optional &ring_size = {}); ~TCPImagePuller() override; bool SupportsAck() const override { return true; } diff --git a/image_puller/ZMQImagePuller.cpp b/image_puller/ZMQImagePuller.cpp index 5078dd593..641ae41b7 100644 --- a/image_puller/ZMQImagePuller.cpp +++ b/image_puller/ZMQImagePuller.cpp @@ -4,10 +4,14 @@ #include "ZMQImagePuller.h" #include "../frame_serialize/CBORStream2Serializer.h" +#include + ZMQImagePuller::ZMQImagePuller(const std::string &in_addr, const std::string &repub_address, const std::optional &rcv_watermark, - const std::optional &repub_watermark) : + const std::optional &repub_watermark, + const std::optional &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(); + auto msg = std::make_shared(); 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(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) diff --git a/image_puller/ZMQImagePuller.h b/image_puller/ZMQImagePuller.h index 071b5fc18..84770ab38 100644 --- a/image_puller/ZMQImagePuller.h +++ b/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 &rcv_watermark = {}, - const std::optional &repub_watermark = {}); + const std::optional &repub_watermark = {}, + const std::optional &ring_size = {}); ~ZMQImagePuller(); void Disconnect() override; diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt index bd6eb3127..6f4632394 100644 --- a/tests/CMakeLists.txt +++ b/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 diff --git a/tests/ReceiveRingBufferTest.cpp b/tests/ReceiveRingBufferTest.cpp new file mode 100644 index 000000000..a020242e0 --- /dev/null +++ b/tests/ReceiveRingBufferTest.cpp @@ -0,0 +1,129 @@ +// SPDX-FileCopyrightText: 2026 Filip Leonarski, Paul Scherrer Institute +// SPDX-License-Identifier: GPL-3.0-only + +#include +#include +#include + +#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 r; + { + ReceiveRingBuffer ring(100); + r = ring.Allocate(50); + } + REQUIRE(r); + r->data[49] = 1; + r.reset(); +} diff --git a/tests/TCPImagePusherTest.cpp b/tests/TCPImagePusherTest.cpp index 19b5b0ae9..d19a67a80 100644 --- a/tests/TCPImagePusherTest.cpp +++ b/tests/TCPImagePusherTest.cpp @@ -41,8 +41,13 @@ TEST_CASE("TCPImageCommTest_2Writers_WithAck", "[TCP]") { TCPStreamPusher pusher("tcp://127.0.0.1:*", npullers); std::vector > 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(pusher.GetAddress()[0], 64 * 1024 * 1024)); + puller.push_back(std::make_unique(pusher.GetAddress()[0], 64 * 1024 * 1024, "", std::nullopt, + (i == 1) ? std::optional(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(npullers); ++attempt) diff --git a/tests/ZMQImagePusherTest.cpp b/tests/ZMQImagePusherTest.cpp index beaa18517..f6e541e52 100644 --- a/tests/ZMQImagePusherTest.cpp +++ b/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 dist; + std::vector 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 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 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); +} diff --git a/writer/StreamWriter.cpp b/writer/StreamWriter.cpp index 12bc6b8fa..efac1d632 100644 --- a/writer/StreamWriter.cpp +++ b/writer/StreamWriter.cpp @@ -342,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; } diff --git a/writer/jfjoch_writer.cpp b/writer/jfjoch_writer.cpp index f5c4001f0..c8ee45b92 100644 --- a/writer/jfjoch_writer.cpp +++ b/writer/jfjoch_writer.cpp @@ -25,7 +25,8 @@ void print_usage() { logger.Info("-d| --root_dir= Root directory for file writing"); logger.Info("-r | --zmq_repub_port= ZeroMQ port for PUSH socket to republish images"); logger.Info("-f | --zmq_file_port= ZeroMQ port for PUB socket to inform about finalized files"); - logger.Info("-w | --rcv_watermark= Receiving ZeroMQ socket watermark (default = 100)"); + logger.Info("-w | --rcv_watermark= Receiving ZeroMQ socket watermark in messages (default = 10)"); + logger.Info("-m | --rcv_ring_mb= Receive ring buffer in MB; holds images until written and republished (default = 2048)"); logger.Info("-W | --repub_watermark= Republish ZeroMQ socket watermark (default = 1000)"); logger.Info("-b | --rcv_buffer= 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 rcv_watermark; std::optional repub_watermark; std::optional 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(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(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 puller; if (raw_tcp) - puller = std::make_unique(argv[first_argc], rcv_buffer, repub_zmq_addr, repub_watermark); + puller = std::make_unique(argv[first_argc], rcv_buffer, repub_zmq_addr, repub_watermark, + ring_size); else - puller = std::make_unique(argv[first_argc], repub_zmq_addr, rcv_watermark, repub_watermark); + puller = std::make_unique(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); From 6202d36268b22e4a21c012d02912e1b3acfaace3 Mon Sep 17 00:00:00 2001 From: Filip Leonarski Date: Tue, 13 Oct 2026 12:51:34 +0200 Subject: [PATCH 3/4] Writer: CloseOldFiles walks the open data files, not every file slot of the run The slots of closed files stay in `files` (empty) for the whole run, and CloseOldFiles visited all of them on every image: a thousand slots per image for a million-image run. Keep the open file numbers in an ordered set and walk that; being ordered, the walk ends at the first file that is not old enough to close. Co-Authored-By: Claude Fable 5.1 --- writer/FileWriter.cpp | 18 +++++++++++------- writer/FileWriter.h | 5 +++++ 2 files changed, 16 insertions(+), 7 deletions(-) diff --git a/writer/FileWriter.cpp b/writer/FileWriter.cpp index 9ac107781..5d6b42432 100644 --- a/writer/FileWriter.cpp +++ b/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(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; } } diff --git a/writer/FileWriter.h b/writer/FileWriter.h index 1bd0965ae..c4ba5494e 100644 --- a/writer/FileWriter.h +++ b/writer/FileWriter.h @@ -10,6 +10,7 @@ #include "../common/ZMQWrappers.h" #include "HDF5NXmx.h" #include +#include #include class FileWriter { @@ -23,6 +24,10 @@ class FileWriter { std::function file_closed_callback; std::unordered_set 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 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); From a5b4529f9cf7fb229c72abbd4c2fdd1b8069a883 Mon Sep 17 00:00:00 2001 From: Filip Leonarski Date: Tue, 13 Oct 2026 12:51:34 +0200 Subject: [PATCH 4/4] ZeroMQ pusher: say so when a stalled writer switches the socket to dropping A send that blocked for over a second drops that image and turns the socket non-blocking for the rest of the run, so every later image that finds the queue full is dropped too - without a word in the log. Log the moment it happens, once per run, with the image number. Co-Authored-By: Claude Fable 5.1 --- image_pusher/ZMQStream2PusherSocket.cpp | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/image_pusher/ZMQStream2PusherSocket.cpp b/image_pusher/ZMQStream2PusherSocket.cpp index 2b9334a97..64475b649 100644 --- a/image_pusher/ZMQStream2PusherSocket.cpp +++ b/image_pusher/ZMQStream2PusherSocket.cpp @@ -3,6 +3,8 @@ #include "ZMQStream2PusherSocket.h" +#include + 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).