fix: teardown blocks indefinitely when a reconnect thread is active and the robot is unreachable (#368)

doc/images/tcp_socket_states.drawio

@@ -0,0 +1,71 @@
+<mxfile host="Electron" agent="ur_client_library">
+  <diagram id="tcp_socket_states" name="TCPSocket state machine">
+    <mxGraphModel dx="1417" dy="858" grid="1" gridSize="10" guides="1" tooltips="1" connect="1" arrows="1" fold="1" page="1" pageScale="1" pageWidth="1169" pageHeight="826" math="0" shadow="0">
+      <root>
+        <mxCell id="0" />
+        <mxCell id="1" parent="0" />
+        <mxCell id="start" parent="1" style="ellipse;fillColor=#000000;strokeColor=#000000;html=1;" value="" vertex="1">
+          <mxGeometry height="20" width="20" x="80" y="100" as="geometry" />
+        </mxCell>
+        <mxCell id="Invalid" parent="1" style="rounded=1;whiteSpace=wrap;html=1;fillColor=#a0522d;strokeColor=#6D1F00;fontSize=14;fontColor=#ffffff;" value="Invalid" vertex="1">
+          <mxGeometry height="50" width="160" x="10" y="170" as="geometry" />
+        </mxCell>
+        <mxCell id="Closed" parent="1" style="rounded=1;whiteSpace=wrap;html=1;fillColor=#f8cecc;strokeColor=#b85450;fontSize=14;" value="Closed" vertex="1">
+          <mxGeometry height="50" width="160" x="220" y="170" as="geometry" />
+        </mxCell>
+        <mxCell id="Connecting" parent="1" style="rounded=1;whiteSpace=wrap;html=1;fillColor=#eeeeee;strokeColor=#36393d;fontSize=14;" value="Connecting" vertex="1">
+          <mxGeometry height="50" width="160" x="220" y="300" as="geometry" />
+        </mxCell>
+        <mxCell id="Connected" parent="1" style="rounded=1;whiteSpace=wrap;html=1;fillColor=#d5e8d4;strokeColor=#000000;fontSize=14;" value="Connected" vertex="1">
+          <mxGeometry height="50" width="160" x="340" y="450" as="geometry" />
+        </mxCell>
+        <mxCell id="LostConnection" parent="1" style="rounded=1;whiteSpace=wrap;html=1;fillColor=#ffe6cc;strokeColor=#000000;fontSize=14;" value="LostConnection" vertex="1">
+          <mxGeometry height="50" width="160" x="220" y="590" as="geometry" />
+        </mxCell>
+        <mxCell id="Disconnecting" parent="1" style="rounded=1;whiteSpace=wrap;html=1;fillColor=#eeeeee;strokeColor=#36393d;fontSize=14;" value="Disconnecting" vertex="1">
+          <mxGeometry height="50" width="160" x="600" y="450" as="geometry" />
+        </mxCell>
+        <mxCell id="e_start_invalid" edge="1" parent="1" source="start" style="edgeStyle=orthogonalEdgeStyle;rounded=0;html=1;endArrow=classic;fontSize=11;strokeColor=#7f7f7f;fontColor=#7f7f7f;noEdgeStyle=1;" target="Invalid" value="TCPSocket()">
+          <mxGeometry relative="1" as="geometry" />
+        </mxCell>
+        <mxCell id="e_invalid_connecting" edge="1" parent="1" source="Invalid" style="edgeStyle=orthogonalEdgeStyle;rounded=0;html=1;endArrow=classic;fontSize=11;strokeColor=#1f77b4;fontColor=#1f77b4;noEdgeStyle=1;" target="Connecting" value="connect()">
+          <mxGeometry relative="1" as="geometry" />
+        </mxCell>
+        <mxCell id="e_closed_connecting" edge="1" parent="1" source="Closed" style="edgeStyle=orthogonalEdgeStyle;rounded=0;html=1;endArrow=classic;fontSize=11;strokeColor=#1f77b4;fontColor=#1f77b4;noEdgeStyle=1;" target="Connecting" value="connect()">
+          <mxGeometry relative="1" as="geometry" />
+        </mxCell>
+        <mxCell id="e_connecting_connected" edge="1" parent="1" source="Connecting" style="edgeStyle=orthogonalEdgeStyle;rounded=0;html=1;endArrow=classic;fontSize=11;strokeColor=#2ca02c;fontColor=#2ca02c;noEdgeStyle=1;" target="Connected" value="on_&lt;span style=&quot;background-color: light-dark(#ffffff, var(--ge-dark-color, #121212)); color: light-dark(rgb(44, 160, 44), rgb(57, 156, 57));&quot;&gt;success&lt;/span&gt;">
+          <mxGeometry relative="1" as="geometry" />
+        </mxCell>
+        <mxCell id="e_connecting_disconnecting" edge="1" parent="1" source="Connecting" style="edgeStyle=orthogonalEdgeStyle;rounded=0;html=1;endArrow=classic;fontSize=11;strokeColor=#d62728;fontColor=#d62728;noEdgeStyle=1;" target="Disconnecting" value="on_error, disconnect()">
+          <mxGeometry relative="1" x="-0.2568" y="8" as="geometry">
+            <mxPoint as="offset" />
+          </mxGeometry>
+        </mxCell>
+        <mxCell id="e_connected_lost" edge="1" parent="1" source="Connected" style="edgeStyle=orthogonalEdgeStyle;rounded=0;html=1;endArrow=classic;fontSize=11;strokeColor=#8c564b;fontColor=#8c564b;noEdgeStyle=1;" target="LostConnection" value="read(): EOF or perm. error">
+          <mxGeometry relative="1" as="geometry" />
+        </mxCell>
+        <mxCell id="e_connected_disconnecting" edge="1" parent="1" source="Connected" style="edgeStyle=orthogonalEdgeStyle;rounded=0;html=1;endArrow=classic;fontSize=11;strokeColor=#d62728;fontColor=#d62728;noEdgeStyle=1;" target="Disconnecting" value="disconnect()">
+          <mxGeometry relative="1" as="geometry" />
+        </mxCell>
+        <mxCell id="e_lost_reconnecting" edge="1" parent="1" source="LostConnection" style="edgeStyle=orthogonalEdgeStyle;rounded=0;html=1;endArrow=classic;fontSize=11;strokeColor=#17becf;fontColor=#17becf;noEdgeStyle=1;" target="Connecting" value="reconnect()">
+          <mxGeometry relative="1" as="geometry" />
+        </mxCell>
+        <mxCell id="e_lost_disconnecting" edge="1" parent="1" source="LostConnection" style="edgeStyle=orthogonalEdgeStyle;rounded=0;html=1;endArrow=classic;fontSize=11;strokeColor=#d62728;fontColor=#d62728;noEdgeStyle=1;" target="Disconnecting" value="disconnect()">
+          <mxGeometry relative="1" as="geometry" />
+        </mxCell>
+        <mxCell id="e_disconnecting_disconnected" edge="1" parent="1" source="Disconnecting" style="edgeStyle=orthogonalEdgeStyle;rounded=0;html=1;endArrow=classic;fontSize=11;strokeColor=#82b366;fontColor=#d62728;noEdgeStyle=1;fillColor=#d5e8d4;" target="Closed" value="&lt;div&gt;&lt;font style=&quot;color: rgb(151, 208, 119);&quot;&gt;on_success&lt;/font&gt;&lt;/div&gt;">
+          <mxGeometry relative="1" x="0.0013" as="geometry">
+            <mxPoint as="offset" />
+            <mxPoint x="249.56985294117658" y="177" as="targetPoint" />
+          </mxGeometry>
+        </mxCell>
+        <mxCell id="EPpyj3g_8WyPdl9NbQM6-1" edge="1" parent="1" source="Connecting" style="rounded=0;orthogonalLoop=1;jettySize=auto;html=1;entryX=0.25;entryY=0;entryDx=0;entryDy=0;exitX=0.25;exitY=1;exitDx=0;exitDy=0;fillColor=#f8cecc;strokeColor=#b85450;fontColor=#CC0000;" target="LostConnection" value="on_error">
+          <mxGeometry relative="1" y="-30" as="geometry">
+            <mxPoint as="offset" />
+          </mxGeometry>
+        </mxCell>
+      </root>
+    </mxGraphModel>
+  </diagram>
+</mxfile>

include/ur_client_library/comm/producer.h

@@ -69,12 +69,33 @@ class URProducer : public IProducer<T>
       if (!running_)
         return false;
 
-      if (stream_.getState() == SocketState::Connected)
+      const SocketState state = stream_.getState();
+
+      switch (state)
       {
-        continue;
+        case SocketState::Invalid:
+          URCL_LOG_WARN("Stream is invalid. Connect it before trying to read!");
+          break;
+        case SocketState::Connecting:
+          URCL_LOG_WARN("Stream is connecting but not ready, yet. Re-attempting to read!");
+          continue;
+          break;
+        case SocketState::Connected:
+          URCL_LOG_WARN("Stream is connected but failed to read from it. Re-attempting to read!");
+          continue;
+          break;
+        case SocketState::LostConnection:
+          URCL_LOG_WARN("Lost connection to stream, attempting to reconnect...");
+          break;
+        case SocketState::Disconnecting:
+          URCL_LOG_WARN("Stream is disconnecting. Connect it before trying to read!");
+          break;
+        case SocketState::Closed:
+          URCL_LOG_WARN("Stream is disconnected. Connect it before trying to read!");
+          break;
       }
 
-      if (stream_.closed())
+      if (stream_.closed() || stream_.stopRequested())
         return false;
 
       if (on_reconnect_cb_)
@@ -85,9 +106,23 @@ class URProducer : public IProducer<T>
       }
 
       URCL_LOG_WARN("Failed to read from stream, reconnecting in %ld seconds...", timeout_.count());
-      std::this_thread::sleep_for(timeout_);
+      // Sleep in small slices so the producer can be stopped (running_ == false)
+      // or woken by a stream close (e.g. from a destructor calling stop()) within
+      // ~100 ms, instead of blocking for the full (exponentially growing) timeout.
+      // Without this, a thread joining the pipeline at teardown could block for up
+      // to 120 s while this thread sleeps here.
+      const auto sleep_slice = std::chrono::milliseconds(100);
+      const auto sleep_total = std::chrono::duration_cast<std::chrono::milliseconds>(timeout_);
+      for (auto slept = std::chrono::milliseconds(0);
+           slept < sleep_total && running_ && !stream_.closed() && !stream_.stopRequested(); slept += sleep_slice)
+      {
+        std::this_thread::sleep_for(sleep_slice);
+      }
+
+      if (!running_ || stream_.closed() || stream_.stopRequested())
+        return false;
 
-      if (stream_.connect())
+      if (stream_.reconnect())
         continue;
 
       auto next = timeout_ * 2;

include/ur_client_library/comm/stream.h

@@ -63,16 +63,32 @@ class URStream : public TCPSocket
   bool connect(const size_t max_num_tries = 0,
                const std::chrono::milliseconds reconnection_time = std::chrono::seconds(10))
   {
-    return TCPSocket::setup(host_, port_, max_num_tries, reconnection_time);
+    return TCPSocket::connect(host_, port_, max_num_tries, reconnection_time);
   }
 
   /*!
-   * \brief Disconnects from the configured socket.
+   * \brief Re-establishes the connection after an unexpected drop, without clearing a deliberate
+   * disconnect(). Used by the automatic reconnect path.
+   *
+   * \param max_num_tries Maximum number of connection attempts before failing. Unlimited when 0.
+   * \param reconnection_time time in between connection attempts to the server
+   *
+   * \returns True on success, false if it could not reconnect or a deliberate disconnect() is in
+   * effect
+   */
+  bool reconnect(const size_t max_num_tries = 0,
+                 const std::chrono::milliseconds reconnection_time = std::chrono::seconds(10))
+  {
+    return TCPSocket::reconnect(host_, port_, max_num_tries, reconnection_time);
+  }
+
+  /*!
+   * \brief Deliberately disconnects from the configured socket, leaving it ready to connect again.
    */
   void disconnect()
   {
     URCL_LOG_DEBUG("Disconnecting from %s:%d", host_.c_str(), port_);
-    TCPSocket::close();
+    TCPSocket::disconnect();
   }
 
   /*!
@@ -83,6 +99,15 @@ class URStream : public TCPSocket
     return getState() == SocketState::Closed;
   }
 
+  /*!
+   * \brief Returns whether a deliberate disconnect() is in progress or in effect (the socket will
+   * not auto-reconnect until connect() is called again).
+   */
+  bool stopRequested()
+  {
+    return TCPSocket::isStopRequested();
+  }
+
   /*!
    * \brief Reads a full UR package out of a socket. For this, it looks into the package and reads
    * the byte length from the socket directly. It returns as soon as all bytes for the package are

include/ur_client_library/comm/tcp_server.h

@@ -40,6 +40,10 @@
 
 #include "ur_client_library/comm/socket_t.h"
 
+#ifndef _WIN32
+#  include <poll.h>
+#endif
+
 namespace urcl
 {
 namespace comm
@@ -198,6 +202,14 @@ class TCPServer
     return port_;
   }
 
+protected:
+  // Test hook: number of file descriptors currently in the cached poll set
+  // (listen socket + one per connected client).
+  size_t getPollSetSize() const
+  {
+    return pollfds_.size();
+  }
+
 private:
   void init();
   void bind(const size_t max_num_tries, const std::chrono::milliseconds reconnection_time);
@@ -217,18 +229,35 @@ class TCPServer
   //! Runs spin() as long as keep_running_ is set to true.
   void worker();
 
+  //! Rebuilds the cached poll set (pollfds_) from the listen socket and client_fds_.
+  void rebuildPollfds();
+
+  // Number of client slots the poll set reserves up front when the client count is unbounded
+  // (max_clients_allowed_ == 0). Bounded servers reserve their exact limit instead. This is just
+  // headroom so the first few connects (incl. transient overlap during a robot reconnect) don't
+  // reallocate; the vector still grows geometrically beyond it.
+  static constexpr uint32_t DEFAULT_RESERVED_CLIENTS = 8;
+
   std::atomic<bool> keep_running_{ false };
   std::thread worker_thread_;
 
   std::atomic<socket_t> listen_fd_;
   int port_;
 
-  socket_t maxfd_;
-  fd_set masterfds_;
-  fd_set tempfds_;
-
   uint32_t max_clients_allowed_;
   std::vector<socket_t> client_fds_;
+
+#ifdef _WIN32
+  using PollFd = WSAPOLLFD;
+#else
+  using PollFd = struct pollfd;
+#endif
+  // Cached poll set: index 0 is the listen socket, the rest mirror client_fds_. Only touched by
+  // the worker thread (seeded in start() before the thread launches, rebuilt in
+  // handleConnect()/handleDisconnect(), cleared in shutdown() after the worker is joined), so
+  // spin() can poll() on it directly without per-iteration allocation or extra locking.
+  std::vector<PollFd> pollfds_;
+
   std::mutex clients_mutex_;
   std::mutex message_mutex_;
   std::mutex listen_fd_mutex_;