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)); +}