diff --git a/image_pusher/TCPStreamPusher.cpp b/image_pusher/TCPStreamPusher.cpp index 5a0fb681..f7fdd041 100644 --- a/image_pusher/TCPStreamPusher.cpp +++ b/image_pusher/TCPStreamPusher.cpp @@ -554,11 +554,25 @@ void TCPStreamPusher::AcceptorThread() { if (new_fd < 0) continue; - std::lock_guard lg(connections_mutex); + std::vector> dead; + bool rejected = false; + size_t total = 0; + { + std::lock_guard lg(connections_mutex); + dead = DetachDeadConnections(); - RemoveDeadConnections(); + if (connections.size() >= max_connections) { + rejected = true; + } else { + SetupNewConnection(new_fd, next_socket_number++); + total = connections.size(); + } + } - if (connections.size() >= max_connections) { + // Outside the lock: this joins writer threads that may be mid-send. + CloseDeadConnections(dead); + + if (rejected) { logger.Warning("Max connections (" + std::to_string(max_connections) + ") reached, rejecting new connection"); shutdown(new_fd, SHUT_RDWR); @@ -566,9 +580,8 @@ void TCPStreamPusher::AcceptorThread() { continue; } - SetupNewConnection(new_fd, next_socket_number++); logger.Info("Accepted writer connection (socket_number=" + std::to_string(next_socket_number - 1) + - ", total=" + std::to_string(connections.size()) + ")"); + ", total=" + std::to_string(total) + ")"); } } @@ -605,29 +618,40 @@ void TCPStreamPusher::SetupNewConnection(int new_fd, uint32_t socket_number) { connections.emplace_back(std::move(c)); } -void TCPStreamPusher::RemoveDeadConnections() { - // Must be called with connections_mutex held. - // We move dead connections out, release the mutex implicitly (caller still holds it), - // then join their futures. Actually — we can join right here since PersistentAckThread - // doesn't take connections_mutex, so no deadlock. +std::vector> TCPStreamPusher::DetachDeadConnections() { + // Must be called with connections_mutex held. Only unlinks them from the pool - which is quick - + // and hands them back for the caller to close once the mutex is released. Tearing them down here + // would join a writer thread that can be sitting in a send for up to max_backpressure_timeout, + // with connections_mutex held the whole time. + std::vector> dead; auto it = connections.begin(); while (it != connections.end()) { auto c = *it; if (c->broken || !c->connected || !IsConnectionAlive(*c)) { c->connected = false; c->broken = true; - StopDataCollectionThreads(*c); - CloseFd(c->fd); - - if (c->persistent_ack_future.valid()) - c->persistent_ack_future.get(); - - logger.Info("Removed dead connection (socket_number=" + std::to_string(c->socket_number) + ")"); + dead.push_back(c); it = connections.erase(it); } else { ++it; } } + return dead; +} + +void TCPStreamPusher::CloseDeadConnections(const std::vector> &dead) { + // Blocking: joins the writer thread, which may be inside a send. Never call with + // connections_mutex held. Safe to run unlocked - these are already out of the pool, so nobody + // else can reach them. + for (const auto &c : dead) { + StopDataCollectionThreads(*c); + CloseFd(c->fd); + + if (c->persistent_ack_future.valid()) + c->persistent_ack_future.get(); + + logger.Info("Removed dead connection (socket_number=" + std::to_string(c->socket_number) + ")"); + } } void TCPStreamPusher::KeepaliveThread() { @@ -642,8 +666,16 @@ void TCPStreamPusher::KeepaliveThread() { if (data_collection_active) continue; - std::lock_guard lg(connections_mutex); - for (auto& cptr : connections) { + // Copy the pool out and send with the mutex released. SendFrame blocks until the peer + // liveness or backpressure timeout, and holding connections_mutex across that stalls + // SendImage and every /statistics poll behind an idle-time keepalive. + std::vector> local_connections; + { + std::lock_guard lg(connections_mutex); + local_connections = connections; + } + + for (auto& cptr : local_connections) { auto& c = *cptr; if (c.broken || !c.connected) continue; @@ -657,7 +689,12 @@ void TCPStreamPusher::KeepaliveThread() { } } - RemoveDeadConnections(); + std::vector> dead; + { + std::lock_guard lg(connections_mutex); + dead = DetachDeadConnections(); + } + CloseDeadConnections(dead); } } @@ -799,12 +836,26 @@ void TCPStreamPusher::StartDataCollection(StartMessage& message) { total_data_acked_bad.store(0, std::memory_order_relaxed); total_data_acked_total.store(0, std::memory_order_relaxed); + // Stopping a writer thread joins it, and closing a detached connection joins it too, so both + // happen with connections_mutex released; the mutex is only taken to read and write the pool. + std::vector> current; + { + std::lock_guard lg(connections_mutex); + current = connections; + } + for (auto& c : current) + StopDataCollectionThreads(*c); + + std::vector> dead; + { + std::lock_guard lg(connections_mutex); + dead = DetachDeadConnections(); + } + CloseDeadConnections(dead); + std::vector> local_connections; { std::lock_guard lg(connections_mutex); - for (auto& c : connections) - StopDataCollectionThreads(*c); - RemoveDeadConnections(); if (connections.empty()) throw JFJochException(JFJochExceptionCategory::InputParameterInvalid, "No writers connected to " + endpoint); diff --git a/image_pusher/TCPStreamPusher.h b/image_pusher/TCPStreamPusher.h index c2ec0edf..2f5c795a 100644 --- a/image_pusher/TCPStreamPusher.h +++ b/image_pusher/TCPStreamPusher.h @@ -145,7 +145,10 @@ class TCPStreamPusher : public ImagePusher { void KeepaliveThread(); void SetupNewConnection(int new_fd, uint32_t socket_number); - void RemoveDeadConnections(); + // Unlink dead connections from the pool (connections_mutex held) and close them (mutex released). + // Split because closing joins a writer thread that can be blocked in a send. + std::vector> DetachDeadConnections(); + void CloseDeadConnections(const std::vector> &dead); void TearDownConnection(Connection& c); void StartDataCollectionThreads(Connection& c);