From 6df0bd3a6a2cfbf08c1723da912cc60938ce5665 Mon Sep 17 00:00:00 2001 From: Filip Leonarski Date: Wed, 11 Nov 2026 13:07:20 +0100 Subject: [PATCH] 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 Claude-Session: https://claude.ai/code/session_01HVeDQsU5a8AJUdJPgWevAK --- common/ImageBuffer.cpp | 8 ++--- common/ImageBuffer.h | 4 +-- image_analysis/MXAnalysisWithoutFPGA.cpp | 13 +++++++++ image_analysis/MXAnalysisWithoutFPGA.h | 8 +++-- .../dark_mask_analysis/DarkMaskAnalysis.cpp | 2 +- .../dark_mask_analysis/DarkMaskAnalysis.h | 2 +- image_analysis/indexing/FFTIndexerGPU.cu | 10 +++++++ image_analysis/indexing/FFTIndexerGPU.h | 1 + image_analysis/indexing/Indexer.h | 2 ++ image_analysis/indexing/IndexerThreadPool.cpp | 14 +++++++++ image_analysis/indexing/IndexerThreadPool.h | 9 ++++++ image_puller/ImagePuller.h | 23 +++++++-------- image_puller/ReceiveRingBuffer.cpp | 29 ++++++++----------- image_puller/ReceiveRingBuffer.h | 23 ++------------- image_puller/TCPImagePuller.cpp | 27 +++++------------ image_puller/TCPImagePuller.h | 3 +- image_puller/ZMQImagePuller.cpp | 19 ++++-------- image_puller/ZMQImagePuller.h | 3 +- 18 files changed, 99 insertions(+), 101 deletions(-) diff --git a/common/ImageBuffer.cpp b/common/ImageBuffer.cpp index ee496f880..faa52b817 100644 --- a/common/ImageBuffer.cpp +++ b/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(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); diff --git a/common/ImageBuffer.h b/common/ImageBuffer.h index 0e856a446..1f87eca9d 100644 --- a/common/ImageBuffer.h +++ b/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); diff --git a/image_analysis/MXAnalysisWithoutFPGA.cpp b/image_analysis/MXAnalysisWithoutFPGA.cpp index 010d895c3..1a68b6481 100644 --- a/image_analysis/MXAnalysisWithoutFPGA.cpp +++ b/image_analysis/MXAnalysisWithoutFPGA.cpp @@ -446,6 +446,19 @@ void MXAnalysisWithoutFPGA::RebuildROI() { roi = std::make_unique(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)) diff --git a/image_analysis/MXAnalysisWithoutFPGA.h b/image_analysis/MXAnalysisWithoutFPGA.h index 8d3e0708a..63e7d2eb4 100644 --- a/image_analysis/MXAnalysisWithoutFPGA.h +++ b/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 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 diff --git a/image_analysis/dark_mask_analysis/DarkMaskAnalysis.cpp b/image_analysis/dark_mask_analysis/DarkMaskAnalysis.cpp index 3a53bc3d3..f12cc17b1 100644 --- a/image_analysis/dark_mask_analysis/DarkMaskAnalysis.cpp +++ b/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 buffer) { +void DarkMaskAnalysis::AnalyzeImage(const DataMessage &data, std::vector &buffer) { if (data.image.GetWidth() * data.image.GetHeight() != mask.size()) throw JFJochException(JFJochExceptionCategory::InputParameterInvalid, "Invalid size of input image"); diff --git a/image_analysis/dark_mask_analysis/DarkMaskAnalysis.h b/image_analysis/dark_mask_analysis/DarkMaskAnalysis.h index c9e8bb4e3..90e1a9eaf 100644 --- a/image_analysis/dark_mask_analysis/DarkMaskAnalysis.h +++ b/image_analysis/dark_mask_analysis/DarkMaskAnalysis.h @@ -21,7 +21,7 @@ class DarkMaskAnalysis { template void check(const T *ptr); public: DarkMaskAnalysis(const DarkMaskSettings& settings, size_t det_size); - void AnalyzeImage(const DataMessage &msg, std::vector buffer); + void AnalyzeImage(const DataMessage &msg, std::vector &buffer); std::vector GetMask() const; }; diff --git a/image_analysis/indexing/FFTIndexerGPU.cu b/image_analysis/indexing/FFTIndexerGPU.cu index 234820928..d048e093f 100644 --- a/image_analysis/indexing/FFTIndexerGPU.cu +++ b/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 dir_x(nDirections), dir_y(nDirections), dir_z(nDirections); for (int i = 0; i < nDirections; i++) { diff --git a/image_analysis/indexing/FFTIndexerGPU.h b/image_analysis/indexing/FFTIndexerGPU.h index 99cf0c3e5..03658bacf 100644 --- a/image_analysis/indexing/FFTIndexerGPU.h +++ b/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; }; diff --git a/image_analysis/indexing/Indexer.h b/image_analysis/indexing/Indexer.h index 903a38737..0ecb7ec57 100644 --- a/image_analysis/indexing/Indexer.h +++ b/image_analysis/indexing/Indexer.h @@ -86,6 +86,8 @@ protected: virtual std::optional RunSeverityOnly(const std::vector &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 diff --git a/image_analysis/indexing/IndexerThreadPool.cpp b/image_analysis/indexing/IndexerThreadPool.cpp index d849acb01..013ea4e51 100644 --- a/image_analysis/indexing/IndexerThreadPool.cpp +++ b/image_analysis/indexing/IndexerThreadPool.cpp @@ -259,6 +259,10 @@ void IndexerThread::Worker(int threadid) { { std::unique_lock 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(std::cref(settings), i, construction, beside)); } +size_t IndexerThreadPool::DeviceBytes() const { + std::map 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(); diff --git a/image_analysis/indexing/IndexerThreadPool.h b/image_analysis/indexing/IndexerThreadPool.h index 1a61be816..3f2f62b48 100644 --- a/image_analysis/indexing/IndexerThreadPool.h +++ b/image_analysis/indexing/IndexerThreadPool.h @@ -101,6 +101,9 @@ class IndexerThread { std::unique_ptr result = nullptr; std::unique_ptr 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; }; diff --git a/image_puller/ImagePuller.h b/image_puller/ImagePuller.h index b399fb63e..fec35d311 100644 --- a/image_puller/ImagePuller.h +++ b/image_puller/ImagePuller.h @@ -7,6 +7,7 @@ #include #include #include +#include #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 raw; std::shared_ptr tcp_msg; std::shared_ptr 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 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 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; } }; diff --git a/image_puller/ReceiveRingBuffer.cpp b/image_puller/ReceiveRingBuffer.cpp index c01b86c1e..f8533e608 100644 --- a/image_puller/ReceiveRingBuffer.cpp +++ b/image_puller/ReceiveRingBuffer.cpp @@ -7,24 +7,21 @@ #include 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 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 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(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 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 ReceiveRingBuffer::Allocate(size_t size) { if (size == 0) return std::make_shared(); - 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 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; } diff --git a/image_puller/ReceiveRingBuffer.h b/image_puller/ReceiveRingBuffer.h index c1831117a..55eaad63d 100644 --- a/image_puller/ReceiveRingBuffer.h +++ b/image_puller/ReceiveRingBuffer.h @@ -9,37 +9,20 @@ #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. +// 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; 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 Allocate(size_t size); // Releases every waiter in Allocate; the regions already handed out stay valid. diff --git a/image_puller/TCPImagePuller.cpp b/image_puller/TCPImagePuller.cpp index 2f27b2ec9..d85fa29c7 100644 --- a/image_puller/TCPImagePuller.cpp +++ b/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(); diff --git a/image_puller/TCPImagePuller.h b/image_puller/TCPImagePuller.h index 151e931e1..ff229c11c 100644 --- a/image_puller/TCPImagePuller.h +++ b/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"}; diff --git a/image_puller/ZMQImagePuller.cpp b/image_puller/ZMQImagePuller.cpp index e606d4d4d..377e88906 100644 --- a/image_puller/ZMQImagePuller.cpp +++ b/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(new ReceiveBufferRegion{const_cast(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(); diff --git a/image_puller/ZMQImagePuller.h b/image_puller/ZMQImagePuller.h index 84770ab38..ed8b83c98 100644 --- a/image_puller/ZMQImagePuller.h +++ b/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;