fix(TCPSocket): make connection attempts cancellable to unblock teardown

Replace the state-based, sleep-polling reconnect interruption with a dedicated,
sticky cancellation flag (stop_requested_) plus requestStop()/clearStop(). The
flag is orthogonal to SocketState, so setup()'s internal state resets can no
longer race away a teardown signal -- the lost-wakeup that hung ~PrimaryClient()
(and the Windows CI test) indefinitely.

setup() now connects with a non-blocking socket polled in short slices
(poll()/WSAPoll()), so a connect attempt against a genuinely unreachable host is
interruptible too -- not just the between-attempt back-off. This makes the
destructors return promptly instead of blocking for the reconnection timeout
(or forever with unlimited attempts).

- TCPSocket: add stop_requested_ + requestStop()/clearStop(); interruptible,
  non-blocking openInterruptible(); honor the flag in setup()'s connect loop and
  back-off.
- PrimaryClient/RTDEClient: call requestStop() before joining the reconnect
  thread; clear the flag on (re)start (URProducer::setupProducer, RTDEClient::init).
- tests: drive setup() interruption via requestStop(); add a non-routable-address
  blocking-connect regression test; guard the teardown tests with a watchdog;
  add CTest TIMEOUT properties so a hang fails fast instead of timing out the job.

Author: urrsk

include/ur_client_library/comm/producer.h

@@ -133,6 +133,9 @@ class URProducer : public IProducer<T>
   void setupProducer(const size_t max_num_tries = 0,
                      const std::chrono::milliseconds reconnection_time = std::chrono::seconds(10)) override
   {
+    // Clear any cancellation request left over from a previous teardown so the stream can be
+    // (re)used. Safe here because this runs on the controlling thread before the producer loop.
+    stream_.clearStop();
     timeval tv;
     tv.tv_sec = 1;
     tv.tv_usec = 0;

include/ur_client_library/comm/tcp_socket.h

@@ -53,8 +53,22 @@ class TCPSocket
   std::chrono::milliseconds reconnection_time_;
   bool reconnection_time_modified_deprecated_ = false;
 
+  // Cancellation token used to abort a connection attempt that is currently in
+  // progress (either waiting inside connect() or sleeping between attempts).
+  // This is intentionally orthogonal to state_: a SocketState::Closed left over
+  // from a deliberate close()+connect() reconnect must NOT abort a fresh
+  // attempt, whereas requestStop() (used at teardown) must abort reliably and
+  // cannot be clobbered by setup()'s internal state_ resets.
+  std::atomic<bool> stop_requested_{ false };
+
   void setupOptions();
 
+  // Performs an interruptible, non-blocking connect on an already-created socket.
+  // Polls in short slices so that a concurrent requestStop() aborts the attempt
+  // promptly on all platforms (POSIX close() of a blocked connect() is reliable,
+  // Winsock's is not). Restores blocking mode on success.
+  bool openInterruptible(socket_t socket_fd, struct sockaddr* address, size_t address_len);
+
 protected:
   static bool open(socket_t socket_fd, struct sockaddr* address, size_t address_len)
   {
@@ -137,6 +151,26 @@ class TCPSocket
    */
   void close();
 
+  /*!
+   * \brief Requests that any in-progress (or future) connection attempt be aborted.
+   *
+   * Sets a sticky cancellation flag and closes the socket. A reconnect thread that is
+   * waiting inside setup() (either in connect() or in the between-attempt back-off) returns
+   * promptly instead of blocking until the reconnection timeout expires. Use this at teardown
+   * (e.g. from a destructor) before joining a reconnect thread. The flag stays set until
+   * clearStop() is called, so it cannot be lost by a racing internal state reset.
+   */
+  void requestStop();
+
+  /*!
+   * \brief Clears the cancellation flag set by requestStop().
+   *
+   * Must be called on a controlled (re)start before attempting to connect again, so a socket
+   * that was previously stopped can be reused. Never call this concurrently with a reconnect
+   * thread.
+   */
+  void clearStop();
+
   /*!
    * \brief Setup Receive timeout used for this socket.
    *

src/comm/tcp_socket.cpp

@@ -27,7 +27,9 @@
 
 #ifndef _WIN32
 #  include <arpa/inet.h>
+#  include <fcntl.h>
 #  include <netinet/tcp.h>
+#  include <poll.h>
 #endif
 
 #include "ur_client_library/log.h"
@@ -37,6 +39,67 @@ namespace urcl
 {
 namespace comm
 {
+namespace
+{
+// Time slice used while waiting for a non-blocking connect to resolve. Kept short so a
+// concurrent requestStop() aborts the wait promptly even if closing the socket does not
+// itself wake the wait (as is the case on Windows).
+constexpr int CONNECT_POLL_SLICE_MS = 100;
+
+// Toggle the blocking mode of a socket. Returns true on success.
+bool setSocketBlocking(socket_t socket_fd, bool blocking)
+{
+#ifdef _WIN32
+  u_long mode = blocking ? 0 : 1;
+  return ::ioctlsocket(socket_fd, FIONBIO, &mode) == 0;
+#else
+  int flags = ::fcntl(socket_fd, F_GETFL, 0);
+  if (flags < 0)
+  {
+    return false;
+  }
+  flags = blocking ? (flags & ~O_NONBLOCK) : (flags | O_NONBLOCK);
+  return ::fcntl(socket_fd, F_SETFL, flags) == 0;
+#endif
+}
+
+// True if the last connect() call indicated that the connection is being established
+// asynchronously (the expected result for a non-blocking socket).
+bool connectInProgress()
+{
+#ifdef _WIN32
+  return ::WSAGetLastError() == WSAEWOULDBLOCK;
+#else
+  return errno == EINPROGRESS;
+#endif
+}
+
+// Waits up to timeout_ms for the socket to become writable (connect resolved).
+// Returns >0 if the socket is ready/has an event, 0 on timeout, <0 on error.
+//
+// poll() is used on both platforms (WSAPoll() on Windows). It avoids select()'s FD_SETSIZE
+// limitation entirely and keeps a single mental model. On Windows this relies on the WSAPoll
+// connect-failure fix introduced in Windows 10 version 2004 / Windows Server 2019: a failed
+// non-blocking connect is reported as (POLLHUP | POLLERR | POLLWRNORM). The caller treats any
+// returned event as "connect resolved" and consults SO_ERROR for the actual outcome, so it does
+// not depend on which particular revents flag is set.
+int waitForSocketWritable(socket_t socket_fd, int timeout_ms)
+{
+#ifdef _WIN32
+  WSAPOLLFD pfd;
+  pfd.fd = socket_fd;
+  pfd.events = POLLWRNORM;  // == POLLOUT
+  pfd.revents = 0;
+  return ::WSAPoll(&pfd, 1, timeout_ms);
+#else
+  struct pollfd pfd;
+  pfd.fd = socket_fd;
+  pfd.events = POLLOUT;
+  pfd.revents = 0;
+  return ::poll(&pfd, 1, timeout_ms);
+#endif
+}
+}  // namespace
 TCPSocket::TCPSocket()
   : socket_fd_(INVALID_SOCKET), state_(SocketState::Invalid), reconnection_time_(std::chrono::seconds(10))
 {
@@ -72,6 +135,66 @@ void TCPSocket::setupOptions()
   }
 }
 
+bool TCPSocket::openInterruptible(socket_t socket_fd, struct sockaddr* address, size_t address_len)
+{
+  if (!setSocketBlocking(socket_fd, false))
+  {
+    return false;
+  }
+
+  int connect_res = ::connect(socket_fd, address, static_cast<socklen_t>(address_len));
+  bool connected = false;
+  if (connect_res == 0)
+  {
+    // Connected immediately (common for loopback).
+    connected = true;
+  }
+  else if (connectInProgress())
+  {
+    // Poll in short slices until the connect resolves, the OS connect timeout expires,
+    // or a concurrent requestStop() asks us to abort.
+    while (true)
+    {
+      if (stop_requested_)
+      {
+        return false;
+      }
+      int ready = waitForSocketWritable(socket_fd, CONNECT_POLL_SLICE_MS);
+      if (ready < 0)
+      {
+        // poll() error (e.g. the fd was closed by requestStop()).
+        return false;
+      }
+      if (ready == 0)
+      {
+        // Timeout slice elapsed without the connect resolving: re-check stop and keep waiting.
+        continue;
+      }
+      // The socket reported an event: query SO_ERROR to find out whether the connect succeeded.
+      int so_error = 0;
+      socklen_t len = sizeof(so_error);
+      if (::getsockopt(socket_fd, SOL_SOCKET, SO_ERROR, reinterpret_cast<char*>(&so_error), &len) < 0)
+      {
+        return false;
+      }
+      connected = (so_error == 0);
+      break;
+    }
+  }
+  else
+  {
+    // Immediate, permanent failure (e.g. connection refused).
+    connected = false;
+  }
+
+  if (connected && !setSocketBlocking(socket_fd, true))
+  {
+    // Could not restore blocking mode; treat the connection as failed.
+    return false;
+  }
+  return connected;
+}
+
 bool TCPSocket::setup(const std::string& host, const int port, const size_t max_num_tries,
                       const std::chrono::milliseconds reconnection_time)
 {
@@ -85,13 +208,17 @@ bool TCPSocket::setup(const std::string& host, const int port, const size_t max_
     reconnection_time_resolved = reconnection_time_;
   }
 
+  // Honor a pending stop before doing anything else (e.g. a connect attempt issued just after
+  // requestStop()). Checked before touching state_ so the cancellation cannot be lost.
+  if (stop_requested_)
+    return false;
+
   if (state_ == SocketState::Connected)
     return false;
 
-  // Clear any pre-existing Closed/Disconnected/Invalid state so that the
-  // between-attempt sleep below (which exits early when state becomes Closed)
-  // can only be short-circuited by a concurrent external close() call, not by
-  // a Closed state left over from a previous disconnect().
+  // Clear any leftover Closed/Disconnected state from a previous (deliberate) disconnect() so
+  // it does not interfere with this fresh attempt. Cancellation is tracked via stop_requested_,
+  // not via state_, so this reset can no longer erase a teardown signal.
   state_ = SocketState::Invalid;
 
   URCL_LOG_DEBUG("Setting up connection: %s:%d", host.c_str(), port);
@@ -112,6 +239,9 @@ bool TCPSocket::setup(const std::string& host, const int port, const size_t max_
   bool connected = false;
   while (!connected)
   {
+    if (stop_requested_)
+      return false;
+
     if (getaddrinfo(host_name, service.c_str(), &hints, &result) != 0)
     {
       URCL_LOG_ERROR("Failed to get address for %s:%d", host.c_str(), port);
@@ -122,18 +252,28 @@ bool TCPSocket::setup(const std::string& host, const int port, const size_t max_
     {
       socket_fd_ = ::socket(p->ai_family, p->ai_socktype, p->ai_protocol);
 
-      if (socket_fd_ != -1 && open(socket_fd_, p->ai_addr, p->ai_addrlen))
+      if (socket_fd_ != -1 && openInterruptible(socket_fd_, p->ai_addr, p->ai_addrlen))
       {
         connected = true;
         break;
       }
+
+      if (stop_requested_)
+      {
+        freeaddrinfo(result);
+        return false;
+      }
     }
 
     freeaddrinfo(result);
 
     if (!connected)
     {
       state_ = SocketState::Invalid;
+      if (stop_requested_)
+      {
+        return false;
+      }
       if (++connect_counter >= max_num_tries && max_num_tries > 0)
       {
         URCL_LOG_ERROR("Failed to establish connection for %s:%d after %d tries", host.c_str(), port, max_num_tries);
@@ -147,16 +287,15 @@ bool TCPSocket::setup(const std::string& host, const int port, const size_t max_
            << std::chrono::duration_cast<std::chrono::duration<float>>(reconnection_time_resolved).count()
            << " seconds.";
         URCL_LOG_ERROR("%s", ss.str().c_str());
-        // Sleep in short slices so that an external close() (e.g. from ~RTDEClient calling
-        // disconnect() before joining the reconnect thread) can interrupt the wait promptly.
+        // Sleep in short slices so that a concurrent requestStop() (e.g. from ~RTDEClient or
+        // ~PrimaryClient before joining the reconnect thread) can interrupt the back-off promptly.
         const auto sleep_slice = std::chrono::milliseconds(100);
-        for (auto slept = std::chrono::milliseconds(0);
-             slept < reconnection_time_resolved && state_ != SocketState::Closed;
+        for (auto slept = std::chrono::milliseconds(0); slept < reconnection_time_resolved && !stop_requested_;
              slept += sleep_slice)
         {
           std::this_thread::sleep_for(sleep_slice);
         }
-        if (state_ == SocketState::Closed)
+        if (stop_requested_)
         {
           return false;
         }
@@ -179,6 +318,19 @@ void TCPSocket::close()
   }
 }
 
+void TCPSocket::requestStop()
+{
+  // Set the flag before closing so that a reconnect thread observing the closed socket is
+  // guaranteed to also see the cancellation request (and therefore stop instead of retrying).
+  stop_requested_ = true;
+  close();
+}
+
+void TCPSocket::clearStop()
+{
+  stop_requested_ = false;
+}
+
 std::string TCPSocket::getIP() const
 {
   sockaddr_in name;

src/primary/primary_client.cpp

@@ -67,12 +67,12 @@ PrimaryClient::PrimaryClient(const std::string& robot_ip, [[maybe_unused]] comm:
 PrimaryClient::~PrimaryClient()
 {
   URCL_LOG_INFO("Stopping primary client pipeline");
-  // Close the stream BEFORE stopping (joining) the pipeline. The pipeline's
+  // Request a stop on the stream BEFORE stopping (joining) the pipeline. The pipeline's
   // producer thread may be sleeping inside its reconnect backoff or blocked in
-  // TCPSocket::setup(); closing the stream sets SocketState::Closed, which wakes
-  // both paths so pipeline_->stop()'s join returns promptly instead of blocking
-  // until the (potentially unbounded) reconnect timeout expires.
-  stream_.close();
+  // TCPSocket::setup(); requestStop() sets the cancellation flag and closes the socket so
+  // both paths abort within one poll slice and pipeline_->stop()'s join returns promptly
+  // instead of blocking until the (potentially unbounded) reconnect timeout expires.
+  stream_.requestStop();
   pipeline_->stop();
 }
 
@@ -85,9 +85,10 @@ void PrimaryClient::start(const size_t max_num_tries, const std::chrono::millise
 
 void PrimaryClient::stop()
 {
-  // Close the stream before joining the pipeline so a producer thread stuck in
-  // its reconnect path is woken and the join returns promptly (see ~PrimaryClient).
-  stream_.close();
+  // Request a stop on the stream before joining the pipeline so a producer thread stuck in
+  // its reconnect path is aborted and the join returns promptly (see ~PrimaryClient). A
+  // subsequent start() clears the flag again via URProducer::setupProducer().
+  stream_.requestStop();
   pipeline_->stop();
 }
 

src/rtde/rtde_client.cpp

@@ -87,9 +87,11 @@ RTDEClient::~RTDEClient()
 {
   prod_->setReconnectionCallback(nullptr);
   stop_reconnection_ = true;
-  // Disconnect before joining the reconnect thread so that any sleep inside
-  // TCPSocket::setup() (which checks for SocketState::Closed) wakes promptly
-  // instead of blocking the destructor for the full reconnection_timeout.
+  // Request a stop on the stream before joining the reconnect thread. requestStop() sets the
+  // cancellation flag and closes the socket so that any connect attempt or back-off sleep inside
+  // TCPSocket::setup() aborts within one poll slice, instead of blocking the destructor for the
+  // full reconnection_timeout (or indefinitely with max_connection_attempts == 0).
+  stream_.requestStop();
   disconnect();
   if (reconnecting_thread_.joinable())
   {
@@ -115,6 +117,10 @@ bool RTDEClient::init(const size_t max_connection_attempts, const std::chrono::m
   max_initialization_attempts_ = max_initialization_attempts;
   initialization_timeout_ = initialization_timeout;
 
+  // Clear any cancellation request from a previous teardown so this (re)initialization can
+  // connect. Runs on the controlling thread before the reconnect thread can be started.
+  stream_.clearStop();
+
   prod_->setReconnectionCallback(nullptr);
 
   unsigned int attempts = 0;

tests/CMakeLists.txt

@@ -39,6 +39,9 @@ if (INTEGRATION_TESTS)
   gtest_add_tests(TARGET      rtde_tests
                   WORKING_DIRECTORY ${CMAKE_CURRENT_SOURCE_DIR}
   )
+  # Bound this teardown regression test so a hang fails CI fast instead of timing out the job.
+  set_tests_properties(RTDEClientTest.destructor_not_blocked_by_stuck_reconnect_thread
+                       PROPERTIES TIMEOUT 60)
   if (CHECK_RTDE_DOCS_RECIPE)
     find_package(Python3 COMPONENTS Interpreter REQUIRED)
     add_custom_target(generate_outputs ALL COMMAND ${Python3_EXECUTABLE} ${CMAKE_CURRENT_SOURCE_DIR}/resources/generate_rtde_outputs.py)
@@ -159,6 +162,9 @@ add_executable(primary_client_reconnect_tests test_primary_client_reconnect.cpp
 target_link_libraries(primary_client_reconnect_tests PRIVATE ur_client_library::urcl GTest::gtest_main)
 gtest_add_tests(TARGET      primary_client_reconnect_tests
 )
+# Bound this teardown regression test so a hang fails CI in ~1 min instead of timing out the job.
+set_tests_properties(PrimaryClientReconnectTest.destructor_not_blocked_by_stuck_reconnect_thread
+                     PROPERTIES TIMEOUT 60)
 
 
 
@@ -256,6 +262,10 @@ add_executable(tcp_socket_tests test_tcp_socket.cpp)
 target_link_libraries(tcp_socket_tests PRIVATE ur_client_library::urcl GTest::gtest_main)
 gtest_add_tests(TARGET tcp_socket_tests
 )
+# Bound the interruptible-setup regression tests so a hang fails CI fast instead of timing out the job.
+set_tests_properties(TCPSocketTest.setup_interruptible_by_close
+                     TCPSocketTest.setup_interruptible_during_blocking_connect
+                     PROPERTIES TIMEOUT 60)
 
 add_executable(stream_tests test_stream.cpp)
 target_link_libraries(stream_tests PRIVATE ur_client_library::urcl GTest::gtest_main)