diff --git a/image_pusher/TCPStreamPusher.cpp b/image_pusher/TCPStreamPusher.cpp index f7fdd041..9a082d3a 100644 --- a/image_pusher/TCPStreamPusher.cpp +++ b/image_pusher/TCPStreamPusher.cpp @@ -166,6 +166,13 @@ void TCPStreamPusher::TearDownConnection(Connection& c) { c.connected = false; c.broken = true; CloseFd(c.fd); + JoinPersistentAck(c); +} + +// The persistent ack reader is joined both here and by the acceptor when it reaps a dead +// connection, so the join takes the same per-connection lock the writer join does. +void TCPStreamPusher::JoinPersistentAck(Connection& c) { + std::lock_guard teardown(c.teardown_mutex); if (c.persistent_ack_future.valid()) c.persistent_ack_future.get(); } @@ -641,14 +648,14 @@ std::vector> TCPStreamPusher::Detac 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. + // connections_mutex held. These are out of the pool, but a control-plane call that copied the + // pool before the erase still holds a shared_ptr to them, so the joins below take the + // connection's own teardown lock rather than assuming exclusive access. for (const auto &c : dead) { StopDataCollectionThreads(*c); CloseFd(c->fd); - if (c->persistent_ack_future.valid()) - c->persistent_ack_future.get(); + JoinPersistentAck(*c); logger.Info("Removed dead connection (socket_number=" + std::to_string(c->socket_number) + ")"); } @@ -732,6 +739,10 @@ void TCPStreamPusher::StartDataCollectionThreads(Connection& c) { } void TCPStreamPusher::StopDataCollectionThreads(Connection& c) { + // One thread at a time per connection: the acceptor can be reaping this connection while the + // control plane is starting or ending a run on it, and both end in writer_future.get(). + std::lock_guard teardown(c.teardown_mutex); + if (!c.active) return; diff --git a/image_pusher/TCPStreamPusher.h b/image_pusher/TCPStreamPusher.h index 2f5c795a..db6dcc1c 100644 --- a/image_pusher/TCPStreamPusher.h +++ b/image_pusher/TCPStreamPusher.h @@ -53,6 +53,12 @@ class TCPStreamPusher : public ImagePusher { // Persistent ack/keepalive reader (runs as long as the connection is alive) std::future persistent_ack_future; + // Serialises tearing this connection down. Both futures below are joined from more than one + // path - the acceptor reaping a dead connection, and the control plane starting or ending a + // run - and calling get() on one future from two threads is undefined and invalidates it. + // Held only around the joins, never around a send, and no thread it joins takes it. + std::mutex teardown_mutex; + std::mutex send_mutex; std::mutex ack_mutex; std::condition_variable ack_cv; @@ -153,6 +159,7 @@ class TCPStreamPusher : public ImagePusher { void StartDataCollectionThreads(Connection& c); void StopDataCollectionThreads(Connection& c); + void JoinPersistentAck(Connection& c); bool WaitForAck(Connection& c, TCPFrameType ack_for, std::chrono::milliseconds timeout, std::string* error_text); bool WaitForEndAck(Connection& c, std::chrono::milliseconds liveness_timeout, std::string* error_text);