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 d3184d2c..cf09670c 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,7 +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; - const auto deadline = std::chrono::steady_clock::now() + send_total_timeout; bool try_zerocopy = false; #if defined(MSG_ZEROCOPY) @@ -213,7 +219,13 @@ bool TCPStreamPusher::SendAll(Connection& c, const void* buf, size_t len, bool a return false; } - if (std::chrono::steady_clock::now() >= deadline) { + // 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; @@ -573,6 +585,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 @@ -585,6 +600,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 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)); @@ -726,6 +752,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); diff --git a/image_pusher/TCPStreamPusher.h b/image_pusher/TCPStreamPusher.h index 1b6eb0ef..0c6bc7c4 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 may block on backpressure 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 peer trips this. + std::chrono::milliseconds peer_liveness_timeout{15000}; int64_t images_per_file = 1; uint64_t run_number = 0;