Stream receivers, shared pieces: puller drop count, image slots kept free, analysis of a stream image

Taken from rc186 (merge 2250e4019, net of 7f2ebd820) as the stream receivers need them:
- ImagePuller::GetDroppedImages (15266b21e), and the puller comments as rc186 has them
  (0d26ecd13, 3047ccd13); ReceiveRingBuffer's mutex renamed (0d26ecd13).
- ImageBuffer::GetImageSlot(keep_free), for JFJochReceiverRugnuxStills, and the unused slots_idle
  counter dropped (8bad85f9f, b0db94658). The default keeps every existing call as it was.
- MXAnalysisWithoutFPGA::AnalyzeStreamImage, which keeps the saturated and error counts a stream
  states (4c3a7852e).
- DarkMaskAnalysis::AnalyzeImage takes its buffer by reference (5d1fb1a48).
- The device memory the indexers hold: Indexer/FFTIndexerGPU/IndexerThreadPool::DeviceBytes
  (0df766328), read by the live stills model step.

Left out: nothing of these files; their FPGA-side callers keep their calls and behaviour.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01HVeDQsU5a8AJUdJPgWevAK

common/ImageBuffer.cpp

@@ -69,7 +69,6 @@ void ImageBuffer::StartMeasurement(size_t in_location_size) {
     slots_sending = 0;
     slots_preparation = 0;
     slots_in_preview = 0;
-    slots_idle = 0;
 
     // Clear queue and vectors
     while (!free_slots.empty())
@@ -109,15 +108,13 @@ bool ImageBuffer::CheckIfBufferReturned(std::chrono::microseconds timeout) {
     return (slots_preparation == 0) && (slots_sending == 0);
 }
 
-ZeroCopyReturnValue *ImageBuffer::GetImageSlot() {
+ZeroCopyReturnValue *ImageBuffer::GetImageSlot(int64_t keep_free) {
     std::unique_lock ul(m);
 
-    if (!disable_requests && !free_slots.empty()) {
+    if (!disable_requests && static_cast<int64_t>(free_slots.size()) > keep_free) {
         auto i = free_slots.front();
         free_slots.pop();
 
-        if (v[i].status == ImageBufferEntryStatus::Idle)
-            --slots_idle;
         v[i].status = ImageBufferEntryStatus::InPreparation;
         v[i].image_number = INT64_MIN;
         v[i].image_size = 0;
@@ -181,7 +178,6 @@ void ImageBuffer::ReleaseSlot(uint32_t location) {
     if (v[location].status != ImageBufferEntryStatus::Sending)
         return;
 
-    ++slots_idle;
     --slots_sending;
     v[location].status = ImageBufferEntryStatus::Idle;
     free_slots.push(location);

common/ImageBuffer.h

@@ -47,7 +47,6 @@ class ImageBuffer {
     int64_t slots_preparation = 0;
     int64_t slots_sending = 0;
     int64_t slots_in_preview = 0;
-    int64_t slots_idle = 0;
 
     bool disable_requests = true; // All requests fail in this state
     int64_t counter = 0; // Counter is increased when something changes in the image buffer
@@ -76,7 +75,8 @@ public:
     void StartMeasurement(size_t buffer_location_size);
     void StartMeasurement(const DiffractionExperiment& experiment);
 
-    ZeroCopyReturnValue *GetImageSlot();
+    // Nullptr where no more than `keep_free` slots are free.
+    ZeroCopyReturnValue *GetImageSlot(int64_t keep_free = 0);
     void ReadyToSend(uint32_t location, int64_t image_number, size_t image_size, bool indexed);
     void ReleaseSlot(uint32_t location);
 

image_analysis/MXAnalysisWithoutFPGA.cpp

@@ -446,6 +446,19 @@ void MXAnalysisWithoutFPGA::RebuildROI() {
     roi = std::make_unique<ROIIntegrationCPU>(experiment);
 }
 
+void MXAnalysisWithoutFPGA::AnalyzeStreamImage(DataMessage &output,
+                                               AzimuthalIntegrationProfile &profile,
+                                               const SpotFindingSettings &spot_finding_settings) {
+    // In an unsigned image from the card saturated and error pixels have the same value: only the card knows which is which
+    const auto saturated_pixel_count = output.saturated_pixel_count;
+    const auto error_pixel_count = output.error_pixel_count;
+    Analyze(output, profile, spot_finding_settings);
+    if (saturated_pixel_count)
+        output.saturated_pixel_count = saturated_pixel_count;
+    if (error_pixel_count)
+        output.error_pixel_count = error_pixel_count;
+}
+
 void MXAnalysisWithoutFPGA::AnalyzeROIOnly(DataMessage &output) {
     if ((output.image.GetWidth() != xpixels)
         || (output.image.GetWidth() * output.image.GetHeight() != npixels))

image_analysis/MXAnalysisWithoutFPGA.h

@@ -60,9 +60,9 @@ class MXAnalysisWithoutFPGA {
     ImageSpotFinder &FixedThresholdFinder();
     // Self-calibrating finder, used when spot settings request adaptive detection. Kept alongside the
     // default finder because the choice arrives with the per-image settings, not at construction. It is
-    // an AdaptiveSpotFinderCPU by default; on the GPU path, when the fused engine is enabled (rugnux
-    // offline only), it is instead an AdaptiveSpotFinderGPU that also computes the azimuthal profile,
-    // aliased through fused_adaptive so Analyze() can take that profile and skip the separate azint pass.
+    // an AdaptiveSpotFinderCPU by default; on the GPU path, when the fused engine is enabled (rugnux,
+    // the viewer and the receivers), it is instead an AdaptiveSpotFinderGPU that also computes the
+    // azimuthal profile, aliased through fused_adaptive so Analyze() can take that profile and skip the separate azint pass.
     std::unique_ptr<ImageSpotFinder> adaptiveSpotFinder;
     AdaptiveSpotFinderGPU *fused_adaptive = nullptr;
     // The CPU finder, where it gives the profile too (the azimuthal integration being on the CPU).
@@ -117,6 +117,8 @@ public:
     MXAnalysisWithoutFPGA(const DiffractionExperiment &experiment, const AzimuthalIntegrationMapping &integration,
                      const PixelMask &mask, IndexAndRefine &indexer, bool enable_fused_adaptive_gpu = false);
     void Analyze(DataMessage &output, AzimuthalIntegrationProfile &profile, const SpotFindingSettings &spot_finding_settings);
+    // Analyze for an image from a stream: saturated and error counts the stream states stay, a DECTRIS message has none.
+    void AnalyzeStreamImage(DataMessage &output, AzimuthalIntegrationProfile &profile, const SpotFindingSettings &spot_finding_settings);
 
     // Surgical ROI-only paths used when a full re-analysis is not wanted: rebuild the
     // ROI engine after the ROI set changes, recompute ROIs after preprocessing a new

image_analysis/dark_mask_analysis/DarkMaskAnalysis.cpp

@@ -19,7 +19,7 @@ void DarkMaskAnalysis::check(const T *ptr) {
     }
 }
 
-void DarkMaskAnalysis::AnalyzeImage(const DataMessage &data, std::vector<uint8_t> buffer) {
+void DarkMaskAnalysis::AnalyzeImage(const DataMessage &data, std::vector<uint8_t> &buffer) {
     if (data.image.GetWidth() * data.image.GetHeight() != mask.size())
         throw JFJochException(JFJochExceptionCategory::InputParameterInvalid, "Invalid size of input image");
 

image_analysis/dark_mask_analysis/DarkMaskAnalysis.h

@@ -21,7 +21,7 @@ class DarkMaskAnalysis {
     template <class T> void check(const T *ptr);
 public:
     DarkMaskAnalysis(const DarkMaskSettings& settings, size_t det_size);
-    void AnalyzeImage(const DataMessage &msg, std::vector<uint8_t> buffer);
+    void AnalyzeImage(const DataMessage &msg, std::vector<uint8_t> &buffer);
     std::vector<uint32_t> GetMask() const;
 };
 

image_analysis/indexing/FFTIndexerGPU.cu

@@ -263,6 +263,16 @@ FFTIndexerGPU::FFTIndexerGPU(const IndexingSettings &settings)
 }
 
 
+size_t FFTIndexerGPU::DeviceBytes() const {
+    const size_t batch = std::min(nDirections, FFT_BATCH_DIRECTIONS);
+    const size_t out_len = histogram_size / 2 + 1;
+    const size_t directions = nDirections;
+    return batch * histogram_size * sizeof(float) + batch * out_len * sizeof(cufftComplex)
+           + directions * out_len * sizeof(float) + directions * sizeof(FFTResult)
+           + 3 * FFT_MAX_SPOTS * sizeof(float) + 3 * directions * sizeof(float)
+           + plan.WorkBytes() + (nDirections % batch != 0 ? plan_last.WorkBytes() : 0);
+}
+
 void FFTIndexerGPU::DirectionsChanged() {
     std::vector<float> dir_x(nDirections), dir_y(nDirections), dir_z(nDirections);
     for (int i = 0; i < nDirections; i++) {

image_analysis/indexing/FFTIndexerGPU.h

@@ -58,6 +58,7 @@ public:
     FFTIndexerGPU(const FFTIndexerGPU &i) = delete;
     const FFTIndexerGPU &operator=(const FFTIndexerGPU &i) = delete;
     ~FFTIndexerGPU() override = default;
+    size_t DeviceBytes() const override;
 };
 
 

image_analysis/indexing/Indexer.h

@@ -86,6 +86,8 @@ protected:
     virtual std::optional<SpindleSeverity> RunSeverityOnly(const std::vector<Coord> &coord) { return {}; }
 public:
     virtual ~Indexer() = default;
+    // The device memory the indexer holds for as long as it lives; 0 for one on the CPU.
+    virtual size_t DeviceBytes() const { return 0; }
     void Setup(const DiffractionExperiment& experiment);
     // severity_only = true runs RunSeverityOnly instead of indexing: no lattice comes back and
     // `executed` stays false, since nothing was an indexing attempt. Used when the seed escalation

image_analysis/indexing/IndexerThreadPool.cpp

@@ -259,6 +259,10 @@ void IndexerThread::Worker(int threadid) {
 
     {
         std::unique_lock<std::mutex> lock(m);
+        device_ = get_gpu();
+        for (const auto *i : {fft_indexer.get(), ffbidx_indexer.get(), fftw_indexer.get()})
+            if (i)
+                device_bytes_ += i->DeviceBytes();
         state = TaskState::IDLE;
     }
     c_running.notify_all();
@@ -405,6 +409,16 @@ IndexerThreadPool::IndexerThreadPool(const IndexingSettings &settings, IndexerCo
         tasks.emplace_back(std::make_unique<IndexerThread>(std::cref(settings), i, construction, beside));
 }
 
+size_t IndexerThreadPool::DeviceBytes() const {
+    std::map<int, size_t> by_card;
+    for (const auto &t : tasks)
+        by_card[t->Device()] += t->DeviceBytes();
+    size_t most = 0;
+    for (const auto &[card, bytes] : by_card)
+        most = std::max(most, bytes);
+    return most;
+}
+
 void IndexerThreadPool::ReleaseIndexers() {
     for (auto &t : tasks)
         t->Release();

image_analysis/indexing/IndexerThreadPool.h

@@ -101,6 +101,9 @@ class IndexerThread {
     std::unique_ptr<IndexerResult> result = nullptr;
     std::unique_ptr<TaskInput> task_input = nullptr;
     std::thread worker_thread;
+    // The card the worker is pinned to, and the device memory of the indexers it built before it was ready
+    int device_ = 0;
+    size_t device_bytes_ = 0;
 
     void Worker(int threadid);
 public:
@@ -112,6 +115,8 @@ public:
     // them again on the next frame that needs one. Returns when they are gone.
     void Release();
     void Finalize();
+    [[nodiscard]] int Device() const { return device_; }
+    [[nodiscard]] size_t DeviceBytes() const { return device_bytes_; }
 };
 
 class IndexerThreadPool {
@@ -140,6 +145,10 @@ public:
     // 0.6-1.2 GB of it - for a stretch with nothing to index. A frame that comes after all builds its
     // indexer again, as on first use, and the indexer it gets computes what the dropped one would.
     void ReleaseIndexers();
+    // The device memory the indexers built with the pool (IndexerConstruction::Preconstruct) hold, on the
+    // card they hold most on: what is resident beside anything else that plans its GPU memory on a card while
+    // the pool lives (model validation, molecular replacement, refinement).
+    [[nodiscard]] size_t DeviceBytes() const;
 };
 
 

image_puller/ImagePuller.h

@@ -7,6 +7,7 @@
 #include <chrono>
 #include <string>
 #include <memory>
+#include <vector>
 
 #include "../common/ZMQWrappers.h"
 #include "../frame_serialize/CBORStream2Deserializer.h"
@@ -32,14 +33,11 @@ struct RawFrame {
 };
 
 struct ImagePullerOutput {
-    // 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.
+    // The message as it came off the socket, in the ring buffer; held there as long as any copy of this struct lives
     std::shared_ptr<ReceiveBufferRegion> raw;
     std::shared_ptr<RawFrame> tcp_msg;
     std::shared_ptr<CBORStream2DeserializerOutput> cbor;
-    // When the message came off the socket - before it was queued or deserialised, so a consumer that
-    // falls behind does not shift it.
+    // When the message came off the socket, before it was queued or deserialised
     std::chrono::steady_clock::time_point received{};
 };
 
@@ -50,16 +48,12 @@ 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.
+    // Holds every received message until it is consumed. The queues count messages, the ring bounds bytes.
+    // With size 0 there is no ring: messages are held on the heap, bounded by the queues alone.
     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.
+    // Bytes republish holds now. Republish keeps the oldest unsent images and the ring is reclaimed in arrival order,
+    // so a slow republish consumer would pin everything received after them. That is why it is capped at a share of the ring.
     std::atomic<size_t> repub_bytes{0};
     bool RepubCanHold(size_t bytes) const {
         return ring.GetCapacity() == 0 || repub_bytes.load() + bytes <= ring.GetCapacity() / 4;
@@ -81,6 +75,9 @@ public:
     virtual bool SendAck(const PullerAckMessage &) { return false; }
 
     virtual void Disconnect() = 0;
+
+    // Images the puller had no room for, which the receiver therefore never gets.
+    virtual size_t GetDroppedImages() const { return 0; }
 };
 
 

image_puller/ReceiveRingBuffer.cpp

@@ -7,24 +7,21 @@
 #include <optional>
 
 struct ReceiveRingBuffer::State {
-    // Uninitialised on purpose: a ring that is never filled should not cost its capacity in
-    // resident memory.
+    // Uninitialised: 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::mutex state_mutex;
     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.
+    // Live regions in arrival order; only the oldest can give back space
     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.
+    // Next byte to hand out; the oldest live byte is slots.front().offset
     size_t tail = 0;
 
     size_t utilization = 0;
@@ -35,14 +32,12 @@ struct ReceiveRingBuffer::State {
             : 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.
+    // Where a region of `size` bytes would start, nothing if it does not fit now; 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.)
+        // tail has come round to head: treated as full
         if (tail == head)
             return std::nullopt;
         if (tail > head) {
@@ -60,7 +55,7 @@ struct ReceiveRingBuffer::State {
     }
 
     void Release(size_t offset) {
-        std::unique_lock ul(m);
+        std::unique_lock ul(state_mutex);
         for (auto &s: slots) {
             if (s.offset == offset) {
                 s.released = true;
@@ -77,7 +72,7 @@ struct ReceiveRingBuffer::State {
     }
 
     void ReleaseHeap(size_t size) {
-        std::unique_lock ul(m);
+        std::unique_lock ul(state_mutex);
         utilization -= size;
     }
 };
@@ -89,7 +84,7 @@ std::shared_ptr<ReceiveBufferRegion> ReceiveRingBuffer::Allocate(size_t size) {
     if (size == 0)
         return std::make_shared<ReceiveBufferRegion>();
 
-    std::unique_lock ul(state->m);
+    std::unique_lock ul(state->state_mutex);
 
     if (size > state->capacity) {
         // Heap fallback: counted, not bounded.
@@ -126,7 +121,7 @@ std::shared_ptr<ReceiveBufferRegion> ReceiveRingBuffer::Allocate(size_t size) {
 }
 
 void ReceiveRingBuffer::Stop() {
-    std::unique_lock ul(state->m);
+    std::unique_lock ul(state->state_mutex);
     state->stopped = true;
     state->cv.notify_all();
 }
@@ -136,11 +131,11 @@ size_t ReceiveRingBuffer::GetCapacity() const {
 }
 
 size_t ReceiveRingBuffer::GetUtilization() const {
-    std::unique_lock ul(state->m);
+    std::unique_lock ul(state->state_mutex);
     return state->utilization;
 }
 
 size_t ReceiveRingBuffer::GetMaxUtilization() const {
-    std::unique_lock ul(state->m);
+    std::unique_lock ul(state->state_mutex);
     return state->max_utilization;
 }

image_puller/ReceiveRingBuffer.h

@@ -9,37 +9,20 @@
 #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.
+// A received message, from the socket until the last consumer lets go of the shared_ptr to it
 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.
+// Bounds the bytes a puller holds in received messages, handed out and taken back in arrival order
 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.
+    // Blocks for the bytes; nullptr once stopped. A request larger than the capacity is served from the heap, at once
     std::shared_ptr<ReceiveBufferRegion> Allocate(size_t size);
 
     // Releases every waiter in Allocate; the regions already handed out stay valid.

image_puller/TCPImagePuller.cpp

@@ -134,8 +134,7 @@ void TCPImagePuller::CBORThread() {
             } else {
                 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.
+                // A PREFLIGHT starts nothing, so it is not republished
                 if (repub_socket && (type != TCPFrameType::PREFLIGHT)) {
                     if ((ret.cbor->msg_type == CBORImageType::START)
                         || (ret.cbor->msg_type == CBORImageType::END))
@@ -145,8 +144,7 @@ void TCPImagePuller::CBORThread() {
                 }
             }
         } 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.
+            // std::exception: a bad_alloc in the deserialiser leaving this thread would end the process
             logger.ErrorException(e);
         }
         ret = cbor_fifo.GetBlocking();
@@ -236,10 +234,7 @@ bool TCPImagePuller::EnsureConnected() {
         if (receive_buffer_size)
             setsockopt(new_fd, SOL_SOCKET, SO_RCVBUF, &receive_buffer_size.value(), sizeof(int32_t));
 
-        // OS-level TCP keep-alive: detect a silently-dead pusher (no RST, e.g. crash or
-        // network partition) even while recv() is blocked in a long inter-image gap. Mirror
-        // the pusher's settings so detection is symmetric (~60s) and stays well clear of the
-        // 250 ms app-level BUSY heartbeat.
+        // Detects a silently dead pusher while recv() blocks; the pusher's settings, well clear of the 250 ms heartbeat
         int one = 1;
         setsockopt(new_fd, SOL_SOCKET, SO_KEEPALIVE, &one, sizeof(one));
         int idle = 30;
@@ -380,10 +375,7 @@ void TCPImagePuller::ReceiverThread() {
             out.tcp_msg->header = frame.header;
 
             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.
+                // A full ring holds the read here: the socket buffer fills and the pusher's send blocks
                 out.raw = ring.Allocate(frame.header.payload_size);
                 if (!out.raw)
                     break; // Stop() - the puller is being disconnected.
@@ -411,10 +403,7 @@ void TCPImagePuller::ReceiverThread() {
 }
 
 void TCPImagePuller::HeartbeatThread() {
-    // While connected, periodically tell the pusher we are alive even if the
-    // consuming pipeline is stalled (e.g. blocked on a slow filesystem). This lets
-    // the pusher distinguish a busy-but-healthy writer from a dead one and keep
-    // waiting through arbitrarily long backpressure instead of dropping the run.
+    // Tells the pusher we are alive even if the consuming pipeline is stalled
     while (!disconnect) {
         // Sleep in small slices so shutdown stays prompt.
         for (int i = 0; i < 5 && !disconnect; i++)
@@ -434,8 +423,7 @@ void TCPImagePuller::HeartbeatThread() {
         h.ack_fifo_occupancy = cbor_fifo.GetCurrentUtilization();
         h.ack_fifo_max_occupancy = cbor_fifo.Size();
 
-        // Best effort: a failure here just means the socket is gone, which
-        // ReceiverThread will detect and reconnect on its own.
+        // Best effort: ReceiverThread notices a socket that is gone
         std::lock_guard lg(send_mutex);
         SendAll(&h, sizeof(h));
     }
@@ -447,8 +435,7 @@ void TCPImagePuller::Disconnect() {
 
     CloseSocket();
 
-    // Same reason as in ZMQImagePuller: once the puller is going away nobody drains outside_fifo,
-    // so a full queue would leave CBORThread blocked in PutBlocking and the joins below hanging.
+    // As in ZMQImagePuller: nobody drains outside_fifo any more, and the joins below would hang
     cbor_fifo.Stop();
     repub_fifo.Stop();
     outside_fifo.Stop();

image_puller/TCPImagePuller.h

@@ -31,8 +31,7 @@ class TCPImagePuller : public ImagePuller {
     std::thread repub_thread;
     std::thread heartbeat_thread;
 
-    // Serializes all writer->pusher sends (ACKs, keepalive pongs, busy heartbeats)
-    // so that concurrently-sent frames cannot interleave on the socket.
+    // Frames sent to the pusher (ACKs, pongs, busy heartbeats) must not interleave on the socket
     std::mutex send_mutex;
 
     Logger logger{"TCPImagePuller"};

image_puller/ZMQImagePuller.cpp

@@ -40,10 +40,7 @@ ZMQImagePuller::~ZMQImagePuller() {
 void ZMQImagePuller::Disconnect() {
     disconnect = 1;
 
-    // The threads below hand messages on through bounded blocking queues, and by the time a puller
-    // is disconnected nothing drains outside_fifo any more - the receiver that did is gone. A full
-    // queue would then park CBORThread in PutBlocking for ever, back up cbor_fifo and park
-    // PullerThread too, and the joins below would never return. Stop() releases them.
+    // Nothing drains outside_fifo any more: without Stop() a full queue would leave the joins below hanging
     cbor_fifo.Stop();
     repub_fifo.Stop();
     outside_fifo.Stop();
@@ -87,11 +84,8 @@ void ZMQImagePuller::PullerThread() {
             ret.raw = std::shared_ptr<ReceiveBufferRegion>(new ReceiveBufferRegion{const_cast<uint8_t *>(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.
+            // When the ring is full this waits and the socket is not read, so the sender's send blocks.
+            // The broker's ZeroMQ pusher waits one second, then drops the image, so a ring too small for the writer's stalls costs images.
             ret.raw = ring.Allocate(msg->size());
             if (!ret.raw) {
                 // Stop() - the puller is being disconnected.
@@ -111,9 +105,7 @@ void ZMQImagePuller::CBORThread() {
     while (ret.raw) {
         try {
             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()
+            // Suspend() of the receiver does not affect republish
             if (!suspend)
                 outside_fifo.PutBlocking(ret);
             if (repub_socket) {
@@ -124,8 +116,7 @@ void ZMQImagePuller::CBORThread() {
                     repub_bytes += ret.raw->size;
             }
         } 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.
+            // std::exception: a bad_alloc in the deserialiser leaving this thread would end the process
             logger.ErrorException(e);
         }
         ret = cbor_fifo.GetBlocking();

image_puller/ZMQImagePuller.h

@@ -15,8 +15,7 @@
 
 
 class ZMQImagePuller : public ImagePuller {
-    // 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.
+    // Messages, not bytes: small, as the ring is where the memory is accounted for
     const uint32_t default_receive_watermark = 10;
     const uint32_t default_repub_watermark = 220;