comms/uniflow/tcp: refuse a bind() on a transport that is already bound

Summary:
bind() guarded only shutdown_, then set state_ = Initialized unconditionally.
Initialized is the one state connect() admits, so a second bind() re-armed that
gate on a live transport and nothing rejected the second connect() that follows.

Two things go wrong on that path. bind() itself calls servers_.clear(), dropping
the listener out from under an established connection. Then the second connect()
reaches establishLanes(), whose first act is lanes_.clear() -- and a TcpLane owns
its reader and sender std::threads, so that destroys two joinable threads.
~std::thread on a joinable thread calls std::terminate unconditionally. The lane's
Conn is destroyed in the same sweep while both threads are still using it, so
there is a use-after-free behind the terminate as well.

This needs caller misuse: the Transport contract is bind-then-connect once, and
nothing in tree re-binds -- MultiTransport::bind() fails the whole bind when a
transport returns empty info, and Uniflow::establishConnection() binds each
freshly created transport once. But the failure mode is a process-wide abort with
no diagnostic, and on AMD this transport shares a MultiTransport with RDMA, so it
would take down transports that did nothing wrong. The refusal costs three lines
next to the shutdown_ check that is already there.

Error is treated as terminal here too. A bind() that failed leaves state_ = Error,
and this guard therefore refuses a retry. Nothing retries today, and a transport
whose listener never came up should not be revived by a second attempt against
the same host_.

Differential Revision: D117884613

Author: Peng Chen

comms/uniflow/transport/tcp/TcpTransport.cpp

@@ -328,6 +328,18 @@ TransportInfo TcpTransport::bind() {
     UNIFLOW_LOG_ERROR("TcpTransport::bind: transport is already shut down");
     return TransportInfo{};
   }
+  // Refuse a re-bind rather than re-arming Initialized, which is the only state
+  // connect() admits. Re-arming lets a second connect() reach
+  // establishLanes(), whose first act is to clear lanes_ -- destroying the live
+  // lanes' joinable reader/sender threads, and ~std::thread on a joinable
+  // thread calls std::terminate. The servers_.clear() below would already have
+  // dropped the listener out from under the current connection. Error is
+  // terminal here too: nothing retries a failed bind, and a transport that
+  // never came up should not be revived.
+  if (state_ != TransportState::Disconnected) {
+    UNIFLOW_LOG_ERROR("TcpTransport::bind: transport is already bound");
+    return TransportInfo{};
+  }
   // One listener per device when striping, otherwise a single listener on host_
   // with egress left to the routing table. Each device's listener binds that
   // device's own address *and* sets SO_BINDTODEVICE, because accepted sockets

comms/uniflow/transport/tcp/tests/unit/TcpTransportConnectTest.cpp

@@ -8,6 +8,7 @@
 #include <cstdint>
 #include <cstring>
 #include <memory>
+#include <thread>
 #include <vector>
 
 #include "comms/uniflow/executor/ScopedEventBaseThread.h"
@@ -192,6 +193,42 @@ TEST_F(TcpTransportConnectTest, BindAfterShutdownIsRefused) {
       << "bind() must not re-open a shut-down transport";
 }
 
+// bind() re-arms the Initialized state that connect() gates on, so without an
+// already-bound guard a second bind() lets a second connect() through on a live
+// transport. establishLanes() clears lanes_ as its first act, which destroys
+// the current lanes' joinable reader/sender threads -- and ~std::thread on a
+// joinable thread calls std::terminate, taking down every other transport in
+// the process with it.
+TEST_F(TcpTransportConnectTest, BindAfterConnectIsRefused) {
+  auto peerEvb =
+      std::make_unique<ScopedEventBaseThread>("tcp-connect-test-peer");
+  auto peer = std::make_unique<TcpTransport>(
+      /*deviceId=*/-1,
+      peerEvb->getEventBase(),
+      registry_,
+      controller::TcpSocketConfig{},
+      /*host=*/"127.0.0.1");
+
+  const TransportInfo selfInfo = transport_->bind();
+  const TransportInfo peerInfo = peer->bind();
+  ASSERT_FALSE(selfInfo.empty());
+  ASSERT_FALSE(peerInfo.empty());
+  // Each side dials or accepts by endpoint order, so both connects have to be
+  // in flight at once for the handshake to complete.
+  std::thread dialer([&]() { (void)peer->connect(selfInfo); });
+  const Status status = transport_->connect(peerInfo);
+  dialer.join();
+  ASSERT_FALSE(status.hasError()) << status.error().message();
+  ASSERT_EQ(transport_->state(), TransportState::Connected);
+
+  EXPECT_TRUE(transport_->bind().empty())
+      << "bind() must not re-arm a transport that is already connected";
+  EXPECT_EQ(transport_->state(), TransportState::Connected)
+      << "a refused bind() must leave the live connection intact";
+
+  peer->shutdown();
+}
+
 // The factory's topology blob used to be a default-constructed *addressing*
 // struct, so it advertised 127.0.0.1:0 rather than a real endpoint and carried
 // no version. canConnect() therefore validated nothing, and a wire-format