diff --git a/common/JfjochTCP.h b/common/JfjochTCP.h index e945613b..f43709d7 100644 --- a/common/JfjochTCP.h +++ b/common/JfjochTCP.h @@ -6,7 +6,7 @@ #include constexpr uint32_t JFJOCH_TCP_MAGIC = 0x4A464A54; // JFJT -constexpr uint32_t JFJOCH_TCP_VERSION = 2; +constexpr uint32_t JFJOCH_TCP_VERSION = 3; enum class TCPFrameType : uint16_t { START = 1, @@ -16,6 +16,11 @@ enum class TCPFrameType : uint16_t { ACK = 5, CANCEL = 6, KEEPALIVE = 7, + // Writer -> pusher liveness/backpressure heartbeat. Tells the pusher the writer + // is alive but busy (e.g. stalled on a slow filesystem) and that it should keep + // waiting for a proper ACK rather than declaring the connection dead. Carries the + // writer's current FIFO occupancy in the ack_fifo_* fields. + BUSY = 8, }; enum class TCPAckCode : uint16_t { diff --git a/image_puller/TCPImagePuller.cpp b/image_puller/TCPImagePuller.cpp index f1e83ee2..7f93b561 100644 --- a/image_puller/TCPImagePuller.cpp +++ b/image_puller/TCPImagePuller.cpp @@ -54,6 +54,7 @@ TCPImagePuller::TCPImagePuller(const std::string &tcp_addr, receiver_thread = std::thread(&TCPImagePuller::ReceiverThread, this); cbor_thread = std::thread(&TCPImagePuller::CBORThread, this); + heartbeat_thread = std::thread(&TCPImagePuller::HeartbeatThread, this); if (!repub_address.empty()) { repub_socket = std::make_unique(ZMQSocketType::Push); @@ -91,6 +92,8 @@ bool TCPImagePuller::SendAll(const void *buf, size_t len) { } bool TCPImagePuller::SendAck(const PullerAckMessage &ack) { + std::lock_guard lg(send_mutex); + TcpFrameHeader h{}; h.type = static_cast(TCPFrameType::ACK); h.run_number = ack.run_number; @@ -319,7 +322,12 @@ void TCPImagePuller::ReceiverThread() { TcpFrameHeader pong{}; pong.type = static_cast(TCPFrameType::KEEPALIVE); pong.payload_size = 0; - if (!SendAll(&pong, sizeof(pong))) { + bool pong_ok; + { + std::lock_guard lg(send_mutex); + pong_ok = SendAll(&pong, sizeof(pong)); + } + if (!pong_ok) { logger.Info("Keepalive pong send failed, reconnecting to " + addr); CloseSocket(); std::this_thread::sleep_for(std::chrono::milliseconds(20)); @@ -365,6 +373,37 @@ void TCPImagePuller::ReceiverThread() { cbor_fifo.PutBlocking(ImagePullerOutput{}); } +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. + while (!disconnect) { + // Sleep in small slices so shutdown stays prompt. + for (int i = 0; i < 5 && !disconnect; i++) + std::this_thread::sleep_for(HeartbeatInterval / 5); + if (disconnect) + break; + + { + std::unique_lock ul(fd_mutex); + if (fd < 0) + continue; // Not connected; ReceiverThread is (re)establishing the link. + } + + TcpFrameHeader h{}; + h.type = static_cast(TCPFrameType::BUSY); + h.payload_size = 0; + 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. + std::lock_guard lg(send_mutex); + SendAll(&h, sizeof(h)); + } +} + void TCPImagePuller::Disconnect() { if (disconnect.exchange(true)) return; @@ -377,4 +416,6 @@ void TCPImagePuller::Disconnect() { cbor_thread.join(); if (repub_thread.joinable()) repub_thread.join(); + if (heartbeat_thread.joinable()) + heartbeat_thread.join(); } diff --git a/image_puller/TCPImagePuller.h b/image_puller/TCPImagePuller.h index 5b5f444f..3a0c795e 100644 --- a/image_puller/TCPImagePuller.h +++ b/image_puller/TCPImagePuller.h @@ -29,11 +29,17 @@ class TCPImagePuller : public ImagePuller { std::thread receiver_thread; std::thread cbor_thread; 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. + std::mutex send_mutex; Logger logger{"TCPImagePuller"}; static constexpr uint32_t default_repub_watermark = 220; static constexpr auto RepubTimeout = std::chrono::milliseconds(100); + static constexpr auto HeartbeatInterval = std::chrono::milliseconds(250); bool ReadExact(void *buf, size_t size); bool SendAll(const void *buf, size_t len); @@ -42,6 +48,7 @@ class TCPImagePuller : public ImagePuller { void ReceiverThread(); void CBORThread(); void RepubThread(); + void HeartbeatThread(); public: explicit TCPImagePuller(const std::string &tcp_addr, std::optional rcv_buffer_size = {}, diff --git a/image_pusher/TCPStreamPusher.cpp b/image_pusher/TCPStreamPusher.cpp index 892b1bbc..88678d3b 100644 --- a/image_pusher/TCPStreamPusher.cpp +++ b/image_pusher/TCPStreamPusher.cpp @@ -15,6 +15,13 @@ #include #endif +namespace { +int64_t SteadyNowNs() { + return std::chrono::duration_cast( + std::chrono::steady_clock::now().time_since_epoch()).count(); +} +} + std::pair> TCPStreamPusher::ParseTcpAddress(const std::string& addr) { const std::string prefix = "tcp://"; if (addr.rfind(prefix, 0) != 0) @@ -197,10 +204,6 @@ bool TCPStreamPusher::SendAll(Connection& c, const void* buf, size_t len, bool a bool zc_used = false; uint32_t zc_first = 0; uint32_t zc_last = 0; - // Watchdog on *lack of progress*, not total time: a slow but advancing transfer - // (a writer applying back-pressure) must not be torn down, only a socket that - // stops moving bytes altogether. Reset on every byte actually handed to the kernel. - auto last_progress = std::chrono::steady_clock::now(); bool try_zerocopy = false; #if defined(MSG_ZEROCOPY) @@ -216,7 +219,13 @@ bool TCPStreamPusher::SendAll(Connection& c, const void* buf, size_t len, bool a return false; } - if (std::chrono::steady_clock::now() - last_progress >= send_total_timeout) { + // 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); + 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"); c.broken = true; CloseFd(c.fd); if (zc_used_out) *zc_used_out = zc_used; @@ -291,9 +300,6 @@ bool TCPStreamPusher::SendAll(Connection& c, const void* buf, size_t len, bool a } #endif - if (rc > 0) - last_progress = std::chrono::steady_clock::now(); - sent += static_cast(rc); } @@ -584,6 +590,9 @@ void TCPStreamPusher::PersistentAckThread(Connection* c) { break; } + // Any well-formed frame from the peer is a sign of life. + c->last_peer_activity_ns.store(SteadyNowNs(), std::memory_order_relaxed); + const auto frame_type = static_cast(h.type); // Keepalive pong from the writer @@ -596,6 +605,17 @@ void TCPStreamPusher::PersistentAckThread(Connection* c) { continue; } + // Busy/backpressure heartbeat: the writer is alive but stalled. The liveness + // timestamp update above is all we need; just record reported FIFO occupancy. + if (frame_type == TCPFrameType::BUSY) { + c->last_ack_fifo_occupancy.store(h.ack_fifo_occupancy, std::memory_order_relaxed); + if (h.payload_size > 0) { + std::vector discard(h.payload_size); + ReadExactPersistent(*c, discard.data(), discard.size()); + } + continue; + } + if (frame_type != TCPFrameType::ACK) { c->broken = true; logger.Error("Unexpected frame type " + std::to_string(h.type) + " on socket " + std::to_string(c->socket_number)); @@ -737,6 +757,7 @@ void TCPStreamPusher::SetupNewConnection(int new_fd, uint32_t socket_number) { auto now = std::chrono::steady_clock::now(); c->last_keepalive_sent = now; c->last_keepalive_recv = now; + c->last_peer_activity_ns.store(SteadyNowNs(), std::memory_order_relaxed); auto* raw = c.get(); c->persistent_ack_future = std::async(std::launch::async, &TCPStreamPusher::PersistentAckThread, this, raw); @@ -910,26 +931,26 @@ bool TCPStreamPusher::WaitForAck(Connection& c, TCPFrameType ack_for, std::chron return ack_ok; } -bool TCPStreamPusher::WaitForEndAck(Connection& c, std::chrono::milliseconds no_progress_timeout, std::string* error_text) { +bool TCPStreamPusher::WaitForEndAck(Connection& c, std::chrono::milliseconds liveness_timeout, std::string* error_text) { std::unique_lock ul(c.ack_mutex); - // After END is sent the writer may still be draining a backlog of DATA frames. - // Each DATA ACK that arrives is forward progress, so the END timer is effectively - // armed only once the writer falls silent: the timeout fires when neither a DATA - // ACK nor the END ACK is seen for the whole window, not while images still drain. - uint64_t last_acked = c.data_acked_total.load(std::memory_order_relaxed); - + // After END is sent the writer may still be flushing a backlog to a slow filesystem. + // It keeps proving it is alive via BUSY heartbeats and DATA ACKs (both refresh + // last_peer_activity_ns), so we wait through arbitrarily long stalls and only give up + // if the peer falls completely silent for the whole liveness window. while (!c.end_ack_received && !c.broken) { - const bool progressed = c.ack_cv.wait_for(ul, no_progress_timeout, [&] { - return c.end_ack_received || c.broken.load() - || c.data_acked_total.load(std::memory_order_relaxed) != last_acked; + c.ack_cv.wait_for(ul, liveness_timeout, [&] { + return c.end_ack_received || c.broken.load(); }); - if (!progressed) { - if (error_text) *error_text = "END ACK timeout (writer stalled)"; + if (c.end_ack_received || c.broken) + break; + + const int64_t silent_ns = SteadyNowNs() - c.last_peer_activity_ns.load(std::memory_order_relaxed); + if (silent_ns > std::chrono::duration_cast(liveness_timeout).count()) { + if (error_text) *error_text = "END ACK timeout (writer silent)"; return false; } - last_acked = c.data_acked_total.load(std::memory_order_relaxed); } if (c.broken) { @@ -1121,7 +1142,7 @@ bool TCPStreamPusher::EndDataCollection(const EndMessage& message) { } std::string ack_err; - if (!WaitForEndAck(c, std::chrono::seconds(10), &ack_err)) + if (!WaitForEndAck(c, peer_liveness_timeout, &ack_err)) ret = false; } diff --git a/image_pusher/TCPStreamPusher.h b/image_pusher/TCPStreamPusher.h index dfa62c03..2ae61271 100644 --- a/image_pusher/TCPStreamPusher.h +++ b/image_pusher/TCPStreamPusher.h @@ -95,6 +95,11 @@ class TCPStreamPusher : public ImagePusher { std::chrono::steady_clock::time_point last_keepalive_sent{}; std::chrono::steady_clock::time_point last_keepalive_recv{}; + + // Last time ANY frame (ACK / keepalive pong / busy heartbeat) was received from + // the peer, as steady_clock nanoseconds. Used to keep a healthy-but-busy writer + // alive through long backpressure while still detecting a genuinely dead peer. + std::atomic last_peer_activity_ns{0}; }; std::string endpoint; @@ -116,7 +121,11 @@ class TCPStreamPusher : public ImagePusher { std::future keepalive_future; std::chrono::milliseconds send_poll_timeout{250}; - std::chrono::milliseconds send_total_timeout{3000}; + // Maximum time a send (or the post-END ACK wait) may block with NO sign of life from + // the peer before the connection is declared dead. A busy writer refreshes its liveness + // 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}; int64_t images_per_file = 1; uint64_t run_number = 0; @@ -161,7 +170,7 @@ class TCPStreamPusher : public ImagePusher { bool WaitForZeroCopyDrain(Connection& c, std::chrono::milliseconds timeout); bool WaitForAck(Connection& c, TCPFrameType ack_for, std::chrono::milliseconds timeout, std::string* error_text); - bool WaitForEndAck(Connection& c, std::chrono::milliseconds no_progress_timeout, std::string* error_text); + bool WaitForEndAck(Connection& c, std::chrono::milliseconds liveness_timeout, std::string* error_text); public: explicit TCPStreamPusher(const std::string& addr, size_t in_max_connections,