From 3cd96b8607a38c6f9928f39f81a96197f50858ca Mon Sep 17 00:00:00 2001 From: Filip Leonarski Date: Thu, 25 Jun 2026 13:52:37 +0200 Subject: [PATCH] TCPStreamPusher: hard backpressure cap so a wedged writer can't hang the run The peer-liveness timeout only catches a *silent* writer. A misbehaving writer that keeps sending BUSY heartbeats while never draining (e.g. a permanently wedged filesystem) would otherwise block SendAll -- and, through it, the queued SendImage path and the end-of-run frame_transformation_futures.get() -- forever. Add a progress-based cap in SendAll: if no bytes leave the socket for max_backpressure_timeout (default 60s, tunable via SetMaxBackpressureTimeout) the connection is declared dead regardless of heartbeats. It is one global cap, enforced everywhere SendAll runs, so it bounds both mid-run stalls and finalization. Generous relative to the 15s liveness window, since a heartbeating peer is given more grace than a silent one -- but finite. Add TCPImageCommTest_WedgedWriter_DroppedByBackpressureCap: a writer that ACKs START then stalls forever while heartbeating (cap 1.5s, liveness 5s) must have its connection dropped, and neither the producers nor EndDataCollection may hang. Verified to hang (timeout) with the cap disabled. Co-Authored-By: Claude Opus 4.8 (1M context) --- image_pusher/TCPStreamPusher.cpp | 21 ++++++++++-- image_pusher/TCPStreamPusher.h | 12 +++++++ tests/TCPImagePusherTest.cpp | 59 ++++++++++++++++++++++++++++++++ 3 files changed, 90 insertions(+), 2 deletions(-) diff --git a/image_pusher/TCPStreamPusher.cpp b/image_pusher/TCPStreamPusher.cpp index d8291661..f4c9b5ac 100644 --- a/image_pusher/TCPStreamPusher.cpp +++ b/image_pusher/TCPStreamPusher.cpp @@ -197,16 +197,19 @@ bool TCPStreamPusher::IsConnectionAlive(const Connection& c) const { bool TCPStreamPusher::SendAll(Connection& c, const void* buf, size_t len) { const auto* p = static_cast(buf); size_t sent = 0; + int64_t last_progress_ns = SteadyNowNs(); while (sent < len) { const int local_fd = c.fd.load(); if (local_fd < 0 || c.broken) return false; + const int64_t now_ns = SteadyNowNs(); + // Treat backpressure as fatal only if the peer shows NO sign of life. A busy // writer keeps refreshing last_peer_activity_ns via BUSY heartbeats / ACKs, so - // we wait through arbitrarily long stalls; only a silent peer trips the timeout. - const int64_t silent_ns = SteadyNowNs() - c.last_peer_activity_ns.load(std::memory_order_relaxed); + // we wait through long stalls; only a silent peer trips this timeout. + const int64_t silent_ns = now_ns - c.last_peer_activity_ns.load(std::memory_order_relaxed); if (silent_ns > std::chrono::duration_cast(peer_liveness_timeout).count()) { logger.Warning("No liveness from writer on socket " + std::to_string(c.socket_number) + " for " + std::to_string(silent_ns / 1000000) + " ms while sending; marking broken"); @@ -215,6 +218,19 @@ bool TCPStreamPusher::SendAll(Connection& c, const void* buf, size_t len) { return false; } + // Hard backpressure cap: a writer may keep heartbeating yet never drain (e.g. a + // permanently wedged filesystem). If no bytes leave the socket for this long, give + // up regardless of heartbeats so the run -- and its finalization -- cannot hang. + if (now_ns - last_progress_ns > + std::chrono::duration_cast(max_backpressure_timeout).count()) { + logger.Warning("No send progress to writer on socket " + std::to_string(c.socket_number) + + " for " + std::to_string((now_ns - last_progress_ns) / 1000000) + + " ms despite heartbeats; marking broken"); + c.broken = true; + CloseFd(c.fd); + return false; + } + pollfd pfd{}; pfd.fd = local_fd; pfd.events = POLLOUT; @@ -247,6 +263,7 @@ bool TCPStreamPusher::SendAll(Connection& c, const void* buf, size_t len) { } sent += static_cast(rc); + last_progress_ns = SteadyNowNs(); } return true; diff --git a/image_pusher/TCPStreamPusher.h b/image_pusher/TCPStreamPusher.h index 7ef05382..c2ec0edf 100644 --- a/image_pusher/TCPStreamPusher.h +++ b/image_pusher/TCPStreamPusher.h @@ -109,6 +109,12 @@ class TCPStreamPusher : public ImagePusher { // every ~250 ms via BUSY heartbeats (and via DATA ACKs), so genuine backpressure of any // duration is tolerated; only a truly silent (frozen/dead) peer trips this. std::chrono::milliseconds peer_liveness_timeout{15000}; + // Hard upper bound on backpressure: if the socket accepts no bytes for this long the + // writer is wedged and is declared dead even if it keeps heartbeating, so a + // misbehaving writer cannot block the run (or its finalization) forever. Generous + // relative to peer_liveness_timeout, since a heartbeating peer is given more grace + // than a silent one — but still finite. + std::chrono::milliseconds max_backpressure_timeout{60000}; int64_t images_per_file = 1; uint64_t run_number = 0; @@ -160,6 +166,12 @@ public: /// before data collection starts. void SetPeerLivenessTimeout(std::chrono::milliseconds t) { peer_liveness_timeout = t; } + /// Hard upper bound on backpressure. Even while the peer keeps heartbeating, if no + /// bytes can be sent for this long the writer is declared dead so a wedged writer + /// cannot block the run or its finalization forever. Must be set before data + /// collection starts. + void SetMaxBackpressureTimeout(std::chrono::milliseconds t) { max_backpressure_timeout = t; } + std::vector GetAddress() const override { return {endpoint}; } /// Returns the number of currently connected writers (can be called at any time) diff --git a/tests/TCPImagePusherTest.cpp b/tests/TCPImagePusherTest.cpp index 20cf1440..9b04c63d 100644 --- a/tests/TCPImagePusherTest.cpp +++ b/tests/TCPImagePusherTest.cpp @@ -930,3 +930,62 @@ TEST_CASE("TCPImageCommTest_StalledWriter_SurvivesViaHeartbeat", "[TCP]") { image_buffer.Finalize(std::chrono::seconds(5)); } + +TEST_CASE("TCPImageCommTest_WedgedWriter_DroppedByBackpressureCap", "[TCP]") { + // A writer that keeps heartbeating but never drains (e.g. a permanently wedged + // filesystem) must not block the run or its finalization forever. The hard + // backpressure cap tears the connection down even though BUSY keeps arriving, and + // well before the (longer) peer-liveness timeout that those heartbeats keep at bay. + constexpr int64_t N = 1000; + constexpr auto liveness = std::chrono::milliseconds(5000); // kept fresh by heartbeats + constexpr auto max_backpressure = std::chrono::milliseconds(1500); + + TCPStreamPusher pusher("tcp://127.0.0.1:*", 1, 16 * 1024); + pusher.SetPeerLivenessTimeout(liveness); + pusher.SetMaxBackpressureTimeout(max_backpressure); + + StallableWriterDouble writer(pusher.GetAddress()[0], 16 * 1024); // never released + REQUIRE(writer.Connected()); + + for (int attempt = 0; attempt < 200 && pusher.GetConnectedWriters() < 1; ++attempt) + std::this_thread::sleep_for(std::chrono::milliseconds(25)); + REQUIRE(pusher.GetConnectedWriters() == 1); + + ImageBuffer image_buffer(16 * 1024 * 1024); + image_buffer.StartMeasurement(static_cast(4096)); + + StartMessage start{.images_per_file = 1000, .write_master_file = true}; + pusher.StartDataCollection(start); // writer ACKs START, then stalls forever + + auto sender = std::async(std::launch::async, [&] { + for (int64_t i = 0; i < N; i++) { + ZeroCopyReturnValue *slot = nullptr; + while ((slot = image_buffer.GetImageSlot()) == nullptr) + std::this_thread::sleep_for(std::chrono::milliseconds(1)); + std::memset(slot->GetImage(), 0, 256); + slot->SetImageNumber(i); + slot->SetImageSize(256); + slot->ReadyToSend(); + pusher.SendImage(*slot); + } + }); + + // The cap must fire and drop the connection despite continuous heartbeats. + bool dropped = false; + for (int attempt = 0; attempt < 400 && !dropped; ++attempt) { + if (pusher.GetConnectedWriters() == 0) + dropped = true; + else + std::this_thread::sleep_for(std::chrono::milliseconds(25)); + } + CHECK(dropped); + + // Neither the producers nor finalization may hang once the writer is wedged. + REQUIRE(sender.wait_for(std::chrono::seconds(10)) == std::future_status::ready); + sender.get(); + + EndMessage end{}; + CHECK(pusher.EndDataCollection(end) == false); // bounded, and reports failure + + image_buffer.Finalize(std::chrono::seconds(5)); +}