diff --git a/doc/images/tcp_socket_states.drawio b/doc/images/tcp_socket_states.drawio new file mode 100644 index 000000000..7a1667a66 --- /dev/null +++ b/doc/images/tcp_socket_states.drawio @@ -0,0 +1,71 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/doc/images/tcp_socket_states.svg b/doc/images/tcp_socket_states.svg new file mode 100644 index 000000000..bcfffa6b1 --- /dev/null +++ b/doc/images/tcp_socket_states.svg @@ -0,0 +1,4 @@ + + + +
Invalid
Closed
Connecting
Connected
LostConnection
Disconnecting
TCPSocket()
connect()
connect()
on_success
on_error, disconnect()
read(): EOF or perm. error
disconnect()
reconnect()
disconnect()
on_success
on_error
 \ No newline at end of file diff --git a/include/ur_client_library/comm/producer.h b/include/ur_client_library/comm/producer.h index e509d9017..5843f57f7 100644 --- a/include/ur_client_library/comm/producer.h +++ b/include/ur_client_library/comm/producer.h @@ -69,12 +69,33 @@ class URProducer : public IProducer 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 } 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(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; diff --git a/include/ur_client_library/comm/stream.h b/include/ur_client_library/comm/stream.h index 8a573492e..b8da6bfee 100644 --- a/include/ur_client_library/comm/stream.h +++ b/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 diff --git a/include/ur_client_library/comm/tcp_server.h b/include/ur_client_library/comm/tcp_server.h index 465e79073..d95cbe0e2 100644 --- a/include/ur_client_library/comm/tcp_server.h +++ b/include/ur_client_library/comm/tcp_server.h @@ -40,6 +40,10 @@ #include "ur_client_library/comm/socket_t.h" +#ifndef _WIN32 +# include +#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 keep_running_{ false }; std::thread worker_thread_; std::atomic listen_fd_; int port_; - socket_t maxfd_; - fd_set masterfds_; - fd_set tempfds_; - uint32_t max_clients_allowed_; std::vector 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 pollfds_; + std::mutex clients_mutex_; std::mutex message_mutex_; std::mutex listen_fd_mutex_; diff --git a/include/ur_client_library/comm/tcp_socket.h b/include/ur_client_library/comm/tcp_socket.h index 5a1d468c5..2d442d44c 100644 --- a/include/ur_client_library/comm/tcp_socket.h +++ b/include/ur_client_library/comm/tcp_socket.h @@ -36,12 +36,16 @@ namespace comm */ enum class SocketState { - Invalid, ///< Socket is initialized or setup failed - Connected, ///< Socket is connected and ready to use - Disconnected, ///< Socket is disconnected and cannot be used - Closed ///< Connection to socket got closed + Invalid, ///< Socket is initialized but was never connected + Connecting, ///< A connection is in progress + Connected, ///< Socket is connected and ready to use + LostConnection, ///< Connection dropped unexpectedly; auto-reconnect is expected to pick it up + Disconnecting, ///< A deliberate disconnect() is in progress + Closed, ///< Connection to socket got closed }; +const std::string& socketStateToString(SocketState state); + /*! * \brief Class for TCP socket abstraction */ @@ -50,19 +54,64 @@ class TCPSocket private: std::atomic socket_fd_; std::atomic state_; + std::atomic target_state_; std::chrono::milliseconds reconnection_time_; bool reconnection_time_modified_deprecated_ = false; void setupOptions(); + // Performs an interruptible, non-blocking connect on an already-created socket. + // Polls in short slices so that a concurrent disconnect() 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); + + bool setupInternal(const std::string& host, const int port, const size_t max_num_tries, + const std::chrono::milliseconds reconnection_time); + + void freeFileDescriptor(); + protected: + /*! + * \brief Atomically moves state_ to `desired`, unless a deliberate disconnect() is in effect. + * + * Returns true if the state was set, false if a deliberate stop is + * active (in which case state_ is left untouched). This is how the connect/retry machinery + * updates its in-progress state without ever clobbering a teardown signal. + */ + bool setTargetStateUnlessStopRequested(SocketState desired); + + /*! + * \brief Query whether there has been a deliberate disconnect() request. + * + * True while a deliberate disconnect() is in progress or has completed (the "deliberate-stop + * set"). + */ + bool isStopRequested() const + { + return target_state_ == SocketState::Closed; + } + + /*! + * \brief Performs a blocking connect on an already-created socket. + * + * This is the platform-native, uninterruptible connect. It is used when the caller has + * already established that no deliberate disconnect() is in effect, and is willing to block + * until the connect succeeds or fails. The caller must ensure that the socket is in blocking + * mode before calling this. + */ + [[deprecated("Use connect() instead, which is interruptible by a concurrent disconnect()")]] static bool open(socket_t socket_fd, struct sockaddr* address, size_t address_len) { return ::connect(socket_fd, address, static_cast(address_len)) == 0; } + [[deprecated("Use the public method connect() instead")]] bool setup(const std::string& host, const int port, const size_t max_num_tries = 0, - const std::chrono::milliseconds reconnection_time = DEFAULT_RECONNECTION_TIME); + const std::chrono::milliseconds reconnection_time = DEFAULT_RECONNECTION_TIME) + { + return connect(host, port, max_num_tries, reconnection_time); + } std::unique_ptr recv_timeout_; @@ -133,10 +182,60 @@ class TCPSocket bool write(const uint8_t* buf, const size_t buf_len, size_t& written); /*! - * \brief Closes the connection to the socket. + * \brief Establishes a connection to the configured host/port. + * + * This is the explicit connection setup method. It clears any prior deliberate + * disconnect() (moving the socket to Connecting) and then attempts to connect, retrying up to + * max_num_tries times (unlimited when 0). + * + * \param host Host to connect to + * \param port Port to connect to + * \param max_num_tries Maximum number of connection attempts before failing. Unlimited when 0. + * \param reconnection_time Time between connection attempts + * + * \returns True on success, false if the connection could not be established or was aborted by + * a concurrent disconnect() + */ + bool connect(const std::string& host, const int port, const size_t max_num_tries = 0, + const std::chrono::milliseconds reconnection_time = DEFAULT_RECONNECTION_TIME); + + /*! + * \brief Reconnects to the configured host/port. + * + * This is the explicit reconnection method. It can only be called when the socket is in + * LostConnection state, and will attempt to reconnect, retrying up to max_num_tries times + * (unlimited when 0). Thus, when an explicit disconnect() is in effect, this will fail + * immediately. When a concurrent disconnect() is called while this is in progress, it will abort + * and return false. + * + * \param host Host to connect to + * \param port Port to connect to + * \param max_num_tries Maximum number of connection attempts before failing. Unlimited when 0. + * \param reconnection_time Time between connection attempts + * + * \returns True on success, false if the connection could not be established or was aborted by + * a concurrent disconnect() + */ + bool reconnect(const std::string& host, const int port, const size_t max_num_tries = 0, + const std::chrono::milliseconds reconnection_time = DEFAULT_RECONNECTION_TIME); + + /*! + * \brief Disconnects the client + * + * This overwrites connection attempts. When a deliberate disconnect() has been called, the + * socket will not execute any connection attempts until it has reached the CLOSED state and + * connect() is called again. */ void close(); + /*! + * \brief Alias function for close() to disconnect the client + */ + void disconnect() + { + close(); + } + /*! * \brief Setup Receive timeout used for this socket. * diff --git a/include/ur_client_library/comm/unique_fd.h b/include/ur_client_library/comm/unique_fd.h new file mode 100644 index 000000000..7864ce518 --- /dev/null +++ b/include/ur_client_library/comm/unique_fd.h @@ -0,0 +1,83 @@ +/* + * Copyright 2026, Universal Robots A/S + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#pragma once + +#include + +#include "ur_client_library/comm/socket_t.h" + +namespace urcl +{ +namespace comm +{ +/*! + * \brief RAII owner for a single socket descriptor. + * + * Owns the lifecycle of exactly one socket descriptor, closing it on destruction. To replace the + * held descriptor, call reset(new_fd), which closes the currently held descriptor (if any) before + * adopting the new one. This makes it impossible to accidentally leak a descriptor by overwriting + * the handle, e.g. when retrying a connection attempt. + * + * The descriptor is stored atomically, so a concurrent close/reset (as issued by a deliberate + * disconnect() from another thread) is safe against the connect path reading or replacing it. This + * matches the concurrency model TCPSocket relied on when the descriptor was a plain + * std::atomic. + * + * The class is non-copyable, mirroring std::unique_ptr's unique-ownership semantics. + */ +class UniqueFd +{ +public: + UniqueFd() = default; + ~UniqueFd() + { + reset(); + } + + UniqueFd(const UniqueFd&) = delete; + UniqueFd& operator=(const UniqueFd&) = delete; + + /*! + * \brief Closes the currently held descriptor (if valid) and adopts new_fd. + * + * \param new_fd The descriptor to adopt. Defaults to INVALID_SOCKET, which simply closes the + * currently held descriptor. + */ + void reset(socket_t new_fd = INVALID_SOCKET) + { + socket_t old_fd = fd_.exchange(new_fd); + if (old_fd >= 0 && old_fd != new_fd) + { + ::ur_close(old_fd); + } + } + + /*! + * \brief Returns the currently held descriptor without transferring ownership. + * + * \returns The held descriptor, or INVALID_SOCKET if none is held. + */ + socket_t get() const + { + return fd_.load(); + } + +private: + std::atomic fd_{ INVALID_SOCKET }; +}; +} // namespace comm +} // namespace urcl diff --git a/src/comm/tcp_server.cpp b/src/comm/tcp_server.cpp index 3bb5559d5..6f040e7e1 100644 --- a/src/comm/tcp_server.cpp +++ b/src/comm/tcp_server.cpp @@ -34,15 +34,19 @@ #include #include +#include #include "ur_client_library/comm/socket_t.h" #include +#ifndef _WIN32 +# include +#endif namespace urcl { namespace comm { TCPServer::TCPServer(const int port, const size_t max_num_tries, const std::chrono::milliseconds reconnection_time) - : port_(port), maxfd_(0), max_clients_allowed_(0) + : port_(port), max_clients_allowed_(0) { #ifdef _WIN32 WSAData data; @@ -74,9 +78,6 @@ void TCPServer::init() ur_setsockopt(listen_fd_, SOL_SOCKET, SO_KEEPALIVE, &flag, sizeof(int)); URCL_LOG_DEBUG("Created socket with FD %d", (int)listen_fd_); - - FD_ZERO(&masterfds_); - FD_ZERO(&tempfds_); } void TCPServer::shutdown() @@ -139,6 +140,8 @@ void TCPServer::shutdown() } // This will effectively deactivate the disconnection handler. client_fds_.clear(); + // Worker thread has been joined above, so the poll set is no longer in use; keep it consistent. + pollfds_.clear(); ur_close(shutdown_socket); ur_close(listen_fd_); listen_fd_ = INVALID_SOCKET; @@ -179,9 +182,6 @@ void TCPServer::bind(const size_t max_num_tries, const std::chrono::milliseconds } while (err == -1 && (connection_counter <= max_num_tries || max_num_tries == 0)); URCL_LOG_DEBUG("Bound %d:%d to FD %d", server_addr.sin_addr.s_addr, port_, (int)listen_fd_); - - FD_SET(listen_fd_, &masterfds_); - maxfd_ = listen_fd_; } void TCPServer::startListen() @@ -220,21 +220,6 @@ void TCPServer::handleConnect() return; } -#ifdef _WIN32 - bool set_size_exceeded = client_fds_.size() >= FD_SETSIZE - 1; // -1 because listen_fd_ also occupies one - // slot in masterfds_ -#else - bool set_size_exceeded = client_fd >= FD_SETSIZE; // On Unix-like systems, the client FD itself must be less than - // FD_SETSIZE, otherwise it cannot be added to the fd_set. -#endif - - if (set_size_exceeded) - { - URCL_LOG_ERROR("Accepted client FD %d exceeds FD_SETSIZE (%d). Closing connection.", (int)client_fd, FD_SETSIZE); - ur_close(client_fd); - return; - } - bool accepted = false; { @@ -242,11 +227,7 @@ void TCPServer::handleConnect() if (client_fds_.size() < max_clients_allowed_ || max_clients_allowed_ == 0) { client_fds_.push_back(client_fd); - FD_SET(client_fd, &masterfds_); - if (client_fd > maxfd_) - { - maxfd_ = client_fd; - } + rebuildPollfds(); accepted = true; } else @@ -266,48 +247,68 @@ void TCPServer::handleConnect() } } -void TCPServer::spin() +void TCPServer::rebuildPollfds() { - tempfds_ = masterfds_; - - timeval timeout; - timeout.tv_sec = 1; - timeout.tv_usec = 0; + // clear() + push_back() reuses the vector's existing capacity, so this does not allocate once + // the set has reached a given size. + pollfds_.clear(); + pollfds_.push_back({ static_cast(listen_fd_), POLLIN, 0 }); + for (const auto& client_fd : client_fds_) + { + pollfds_.push_back({ client_fd, POLLIN, 0 }); + } +} - // blocks until activity on any socket from tempfds - int sel = select(static_cast(maxfd_ + 1), &tempfds_, NULL, NULL, &timeout); - if (sel < 0) +void TCPServer::spin() +{ + // Poll the cached set (pollfds_) directly. It is kept in sync with client_fds_ via + // rebuildPollfds() on every connect/disconnect, so no allocation happens here in steady state. + // poll() is used on both platforms (WSAPoll() on Windows) because it has no FD_SETSIZE limit on + // file descriptor numbers, unlike select(). This matters when the hosting process holds many + // file descriptors (e.g. a JVM), pushing socket FDs past FD_SETSIZE (1024). + // + // Block for up to 1 s waiting for activity on any socket. A shutdown wakes this immediately by + // connecting to the listen socket (see shutdown()). +#ifdef _WIN32 + int ready = ::WSAPoll(pollfds_.data(), static_cast(pollfds_.size()), 1000); +#else + int ready = ::poll(pollfds_.data(), pollfds_.size(), 1000); +#endif + if (ready < 0) { - URCL_LOG_ERROR("select() failed. Shutting down socket event handler."); + URCL_LOG_ERROR("poll() failed. Shutting down socket event handler."); keep_running_ = false; return; } - if (!keep_running_ || sel == 0) + if (!keep_running_ || ready == 0) { return; } - if (FD_ISSET(listen_fd_, &tempfds_)) - { - URCL_LOG_DEBUG("Activity on listen FD %d", (int)listen_fd_); - handleConnect(); - } + // Snapshot the poll results into locals BEFORE handleConnect()/handleDisconnect() rebuild + // pollfds_, otherwise activity on an existing client that arrives in the same poll cycle as a + // new connection would be lost. + const bool listen_activity = (pollfds_[0].revents & POLLIN) != 0; std::vector disconnected_clients; std::vector client_fds_with_activity; + // pollfds_[0] is the listen socket; client entries start at index 1. + for (size_t i = 1; i < pollfds_.size(); ++i) { - std::lock_guard lk(clients_mutex_); - for (const auto& client_fd : client_fds_) + if (pollfds_[i].revents & (POLLIN | POLLHUP | POLLERR)) { - if (FD_ISSET(client_fd, &tempfds_)) - { - URCL_LOG_DEBUG("Activity on client FD %d", (int)client_fd); - client_fds_with_activity.push_back(client_fd); - } + URCL_LOG_DEBUG("Activity on client FD %d", (int)pollfds_[i].fd); + client_fds_with_activity.push_back(static_cast(pollfds_[i].fd)); } } + + if (listen_activity) + { + URCL_LOG_DEBUG("Activity on listen FD %d", (int)listen_fd_); + handleConnect(); + } // We handle client activity outside the clients_mutex_ lock to avoid holding it during potentially slow I/O and // message callbacks. // The clients_mutex_ lock is only needed to protect the client_fds_ vector, but once we have copied the FDs with @@ -331,7 +332,6 @@ void TCPServer::handleDisconnect(const socket_t fd) { std::lock_guard lk(clients_mutex_); ur_close(fd); - FD_CLR(fd, &masterfds_); for (size_t i = 0; i < client_fds_.size(); ++i) { @@ -341,15 +341,7 @@ void TCPServer::handleDisconnect(const socket_t fd) break; } } - - maxfd_ = listen_fd_; - for (const auto& client_fd : client_fds_) - { - if (client_fd > maxfd_) - { - maxfd_ = client_fd; - } - } + rebuildPollfds(); } { @@ -415,6 +407,16 @@ void TCPServer::start() { URCL_LOG_DEBUG("Starting worker thread"); keep_running_ = true; + // Seed the poll set with the listen socket before the worker thread starts polling it. Reserve + // room for the listen socket plus the expected number of clients up front so the first + // connections don't reallocate. A bounded server reserves its exact client limit; an unbounded + // one (max_clients_allowed_ == 0) reserves DEFAULT_RESERVED_CLIENTS as headroom. + { + std::lock_guard lk(clients_mutex_); + const uint32_t expected_clients = max_clients_allowed_ > 0 ? max_clients_allowed_ : DEFAULT_RESERVED_CLIENTS; + pollfds_.reserve(static_cast(expected_clients) + 1); + rebuildPollfds(); + } worker_thread_ = std::thread(&TCPServer::worker, this); } diff --git a/src/comm/tcp_socket.cpp b/src/comm/tcp_socket.cpp index 6af840e0d..d6384e94f 100644 --- a/src/comm/tcp_socket.cpp +++ b/src/comm/tcp_socket.cpp @@ -23,11 +23,14 @@ #include #include #include +#include #include #ifndef _WIN32 # include +# include # include +# include #endif #include "ur_client_library/log.h" @@ -37,8 +40,111 @@ namespace urcl { namespace comm { + +const std::string& socketStateToString(SocketState state) +{ + switch (state) + { + case SocketState::Invalid: + { + static const std::string invalid_state = "Invalid"; + return invalid_state; + } + case SocketState::Connecting: + { + static const std::string connecting_state = "Connecting"; + return connecting_state; + } + case SocketState::Connected: + { + static const std::string connected_state = "Connected"; + return connected_state; + } + case SocketState::LostConnection: + { + static const std::string lost_connection_state = "LostConnection"; + return lost_connection_state; + } + case SocketState::Disconnecting: + { + static const std::string disconnecting_state = "Disconnecting"; + return disconnecting_state; + } + case SocketState::Closed: + { + static const std::string closed_state = "Closed"; + return closed_state; + } + } + throw std::invalid_argument("Unknown socket state"); +} + +namespace +{ +// Time slice used while waiting for a non-blocking connect to resolve. Kept short so a +// concurrent disconnect() 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)) + : socket_fd_(INVALID_SOCKET) + , state_(SocketState::Invalid) + , target_state_(SocketState::Invalid) + , reconnection_time_(std::chrono::seconds(10)) { #ifdef _WIN32 WSAData data; @@ -72,15 +178,75 @@ void TCPSocket::setupOptions() } } -bool TCPSocket::setup(const std::string& host, const int port, const size_t max_num_tries, - const std::chrono::milliseconds reconnection_time) +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(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 disconnect() asks us to abort. + while (true) + { + if (isStopRequested()) + { + return false; + } + int ready = waitForSocketWritable(socket_fd, CONNECT_POLL_SLICE_MS); + if (ready < 0) + { + // poll() error (e.g. the fd was closed by disconnect()). + 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(&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::setupInternal(const std::string& host, const int port, const size_t max_num_tries, + const std::chrono::milliseconds reconnection_time) { // This can be removed once we remove the setReconnectionTime() method auto reconnection_time_resolved = reconnection_time; if (reconnection_time_modified_deprecated_) { - URCL_LOG_WARN("TCPSocket::setup(): Reconnection time was modified using `setReconnectionTime()` which is " - "deprecated. Please change your code to set reconnection_time through the `setup()` method " + URCL_LOG_WARN("TCPSocket::setupInternal(): Reconnection time was modified using `setReconnectionTime()` which is " + "deprecated. Please change your code to set reconnection_time through the `(re)connect()` method " "directly. The value passed to this function will be ignored."); reconnection_time_resolved = reconnection_time_; } @@ -88,10 +254,9 @@ bool TCPSocket::setup(const std::string& host, const int port, const size_t max_ if (state_ == SocketState::Connected) return false; - URCL_LOG_DEBUG("Setting up connection: %s:%d", host.c_str(), port); + state_ = SocketState::Connecting; - // gethostbyname() is deprecated so use getadderinfo() as described in: - // https://beej.us/guide/bgnet/html/#getaddrinfoprepare-to-launch + URCL_LOG_DEBUG("Setting up connection: %s:%d", host.c_str(), port); const char* host_name = host.empty() ? nullptr : host.c_str(); std::string service = std::to_string(port); @@ -106,31 +271,41 @@ bool TCPSocket::setup(const std::string& host, const int port, const size_t max_ bool connected = false; while (!connected) { + if (isStopRequested()) + 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); return false; } - // loop through the list of addresses untill we find one that's connectable + // loop through the list of addresses until we find one that's connectable for (struct addrinfo* p = result; p != nullptr; p = p->ai_next) { 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 (isStopRequested()) + { + freeaddrinfo(result); + freeFileDescriptor(); + return false; + } } freeaddrinfo(result); if (!connected) { - state_ = SocketState::Invalid; 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); + freeFileDescriptor(); return false; } else @@ -141,26 +316,116 @@ bool TCPSocket::setup(const std::string& host, const int port, const size_t max_ << std::chrono::duration_cast>(reconnection_time_resolved).count() << " seconds."; URCL_LOG_ERROR("%s", ss.str().c_str()); - std::this_thread::sleep_for(reconnection_time_resolved); + // Sleep in short slices so that a concurrent disconnect() (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 && !isStopRequested(); + slept += sleep_slice) + { + std::this_thread::sleep_for(sleep_slice); + } + if (isStopRequested()) + { + freeFileDescriptor(); + return false; + } } } } setupOptions(); - state_ = SocketState::Connected; + // Mark Connected only if no deliberate disconnect() slipped in while we were finishing up; a late + // disconnect() must win so we do not advertise a usable socket that was just torn down. + SocketState expected = SocketState::Connecting; + if (!state_.compare_exchange_strong(expected, SocketState::Connected)) + { + close(); + return false; + } URCL_LOG_DEBUG("Connection established for %s:%d", host.c_str(), port); return connected; } -void TCPSocket::close() +bool TCPSocket::connect(const std::string& host, const int port, const size_t max_num_tries, + const std::chrono::milliseconds reconnection_time) +{ + if (state_ == SocketState::Connected) + { + URCL_LOG_ERROR("Connect called on a socket that is already connected"); + return false; + } + target_state_ = SocketState::Connected; + if (!setupInternal(host, port, max_num_tries, reconnection_time)) + { + disconnect(); + return false; + } + return true; +} + +bool TCPSocket::reconnect(const std::string& host, const int port, const size_t max_num_tries, + const std::chrono::milliseconds reconnection_time) +{ + if (state_ != SocketState::LostConnection) + { + URCL_LOG_ERROR("Reconnect called on a socket that is not in LostConnection state"); + return false; + } + setTargetStateUnlessStopRequested(SocketState::Connected); + if (!setupInternal(host, port, max_num_tries, reconnection_time)) + { + // If we failed to reconnect, we need to set the target state back to LostConnection so that + auto expected_state = SocketState::Connecting; + state_.compare_exchange_strong(expected_state, SocketState::LostConnection); + return false; + } + return true; +} + +bool TCPSocket::setTargetStateUnlessStopRequested(SocketState desired) +{ + SocketState current_target = target_state_.load(); + if (current_target == desired) + { + return true; // already tracking target + } + + // Lock-free compare-and-swap: move state_ to `desired` unless a deliberate disconnect() + // (Disconnecting/Disconnected) is in effect. This is not an unbounded spin: each iteration + // either succeeds, or observes that another thread changed state_ and re-evaluates. We use + // compare_exchange_strong so there are no spurious retries; the loop can only re-iterate when a + // concurrent writer genuinely changed state_, and the only such writers (a racing disconnect(), + // or a paired close()) make a bounded number of writes, so it terminates promptly. + while (target_state_ != SocketState::Closed) + { + if (target_state_.compare_exchange_strong(current_target, desired)) + { + return true; // successfully moved to `desired` + } + } + return false; // a deliberate disconnect() is in effect; state_ left untouched +} + +void TCPSocket::freeFileDescriptor() { if (socket_fd_ >= 0) { - state_ = SocketState::Closed; ::ur_close(socket_fd_); socket_fd_ = INVALID_SOCKET; } } +void TCPSocket::close() +{ + if (state_ != SocketState::Closed) + { + // closing overwrites everything, so we do not need to check for stopRequested() here + target_state_ = SocketState::Closed; + state_ = SocketState::Disconnecting; + } + freeFileDescriptor(); + state_ = SocketState::Closed; +} + std::string TCPSocket::getIP() const { sockaddr_in name; @@ -202,7 +467,7 @@ bool TCPSocket::read(uint8_t* buf, const size_t buf_len, size_t& read) if (res == 0) { - state_ = SocketState::Disconnected; + state_ = SocketState::LostConnection; return false; } else if (res < 0) @@ -212,13 +477,13 @@ bool TCPSocket::read(uint8_t* buf, const size_t buf_len, size_t& read) int code = ::WSAGetLastError(); if (code != WSAETIMEDOUT && code != WSAEWOULDBLOCK) { - state_ = SocketState::Disconnected; + state_ = SocketState::LostConnection; } #else if (!(errno == EAGAIN || errno == EWOULDBLOCK)) { // any permanent error should be detected early - state_ = SocketState::Disconnected; + state_ = SocketState::LostConnection; } #endif return false; diff --git a/src/primary/primary_client.cpp b/src/primary/primary_client.cpp index 1ae6d4822..68959ecd9 100644 --- a/src/primary/primary_client.cpp +++ b/src/primary/primary_client.cpp @@ -67,7 +67,7 @@ PrimaryClient::PrimaryClient(const std::string& robot_ip, [[maybe_unused]] comm: PrimaryClient::~PrimaryClient() { URCL_LOG_INFO("Stopping primary client pipeline"); - pipeline_->stop(); + stop(); } void PrimaryClient::start(const size_t max_num_tries, const std::chrono::milliseconds reconnection_time) @@ -79,8 +79,8 @@ void PrimaryClient::start(const size_t max_num_tries, const std::chrono::millise void PrimaryClient::stop() { + stream_.disconnect(); pipeline_->stop(); - stream_.close(); } void PrimaryClient::addPrimaryConsumer(std::shared_ptr> primary_consumer) diff --git a/src/rtde/rtde_client.cpp b/src/rtde/rtde_client.cpp index bdac1a733..d47322a2d 100644 --- a/src/rtde/rtde_client.cpp +++ b/src/rtde/rtde_client.cpp @@ -87,11 +87,11 @@ RTDEClient::~RTDEClient() { prod_->setReconnectionCallback(nullptr); stop_reconnection_ = true; + disconnect(); if (reconnecting_thread_.joinable()) { reconnecting_thread_.join(); } - disconnect(); } bool RTDEClient::init(const size_t max_connection_attempts, const std::chrono::milliseconds reconnection_timeout, @@ -508,11 +508,8 @@ bool RTDEClient::setupInputs() void RTDEClient::disconnect() { - if (client_state_ > ClientState::UNINITIALIZED) - { - stream_.disconnect(); - writer_.stop(); - } + stream_.disconnect(); + writer_.stop(); client_state_ = ClientState::UNINITIALIZED; prod_->stopProducer(); stopBackgroundRead(); diff --git a/src/ur/dashboard_client_implementation_g5.cpp b/src/ur/dashboard_client_implementation_g5.cpp index 12a9cfec9..c923d396c 100644 --- a/src/ur/dashboard_client_implementation_g5.cpp +++ b/src/ur/dashboard_client_implementation_g5.cpp @@ -220,7 +220,7 @@ bool DashboardClientImplG5::connect(const size_t max_num_tries, const std::chron TCPSocket::setReceiveTimeout(tv); try { - if (TCPSocket::setup(host_, port_, max_num_tries, reconnection_time)) + if (TCPSocket::connect(host_, port_, max_num_tries, reconnection_time)) { URCL_LOG_INFO("%s", read().c_str()); ret_val = true; diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt index 5582edf8b..99f7e15c7 100644 --- a/tests/CMakeLists.txt +++ b/tests/CMakeLists.txt @@ -46,6 +46,9 @@ if (INTEGRATION_TESTS) WORKING_DIRECTORY ${CMAKE_CURRENT_SOURCE_DIR} EXTRA_ARGS ${INTEGRATION_TESTS_ROBOT_IP_ARG} ) + # 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) @@ -163,6 +166,17 @@ target_link_libraries(fake_primary_server_tests PRIVATE ur_client_library::urcl gtest_add_tests(TARGET fake_primary_server_tests ) +# Robot-free regression test for ~PrimaryClient() not blocking on a stuck +# reconnect thread. Uses the in-process FakePrimaryServer so it runs without a +# robot (unlike the INTEGRATION_TESTS-gated primary_client_test_headless). +add_executable(primary_client_reconnect_tests test_primary_client_reconnect.cpp fake_primary_server.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) + add_executable(rtde_data_package_tests test_rtde_data_package.cpp) @@ -259,6 +273,15 @@ 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(unique_fd_tests test_unique_fd.cpp) +target_link_libraries(unique_fd_tests PRIVATE ur_client_library::urcl GTest::gtest_main) +gtest_add_tests(TARGET unique_fd_tests +) add_executable(stream_tests test_stream.cpp) target_link_libraries(stream_tests PRIVATE ur_client_library::urcl GTest::gtest_main) diff --git a/tests/test_primary_client_reconnect.cpp b/tests/test_primary_client_reconnect.cpp new file mode 100644 index 000000000..1032bc19b --- /dev/null +++ b/tests/test_primary_client_reconnect.cpp @@ -0,0 +1,188 @@ +// -- BEGIN LICENSE BLOCK ---------------------------------------------- +// Copyright 2026 Universal Robots A/S +// +// Redistribution and use in source and binary forms, with or without +// modification, are permitted provided that the following conditions are met: +// +// * Redistributions of source code must retain the above copyright +// notice, this list of conditions and the following disclaimer. +// +// * Redistributions in binary form must reproduce the above copyright +// notice, this list of conditions and the following disclaimer in the +// documentation and/or other materials provided with the distribution. +// +// * Neither the name of the {copyright_holder} nor the names of its +// contributors may be used to endorse or promote products derived from +// this software without specific prior written permission. +// +// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" +// AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE +// IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE +// ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE +// LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR +// CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF +// SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS +// INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN +// CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) +// ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE +// POSSIBILITY OF SUCH DAMAGE. +// -- END LICENSE BLOCK ------------------------------------------------ + +#include + +#include +#include +#include +#include + +#include + +#include "fake_primary_server.h" + +using namespace urcl; + +// Regression test for ~PrimaryClient() blocking indefinitely when the pipeline's +// producer thread is stuck in its reconnect loop at teardown time. +// +// This is the PrimaryClient counterpart of +// RTDEClientTest.destructor_not_blocked_by_stuck_reconnect_thread (test_rtde_client.cpp). +// +// Root cause: when the robot drops the primary connection, TCPSocket::read() +// returns false and leaves the socket in SocketState::LostConnection. URProducer's +// tryGetImpl() then enters its reconnect path: it sleeps an (exponentially +// growing) backoff and calls stream_.reconnect(), which retries with no upper +// bound (max_num_tries == 0), sleeping reconnection_time between attempts. If +// ~PrimaryClient() simply called pipeline_->stop() (which joins the producer +// thread) without first closing the stream, the join would block for the full +// reconnect duration — effectively forever for an unreachable robot. +// +// Fix (two parts): +// 1. ~PrimaryClient()/PrimaryClient::stop() call stream_.disconnect() BEFORE +// joining the pipeline. disconnect() moves the socket into the deliberate-stop +// state (Disconnecting/Disconnected) and closes it. +// 2. TCPSocket::setup() honors that state both during its (non-blocking) connect +// attempt and during the between-attempt wait, so it aborts within ~100 ms +// regardless of platform. setup() never overwrites the deliberate-stop state, +// so it cannot be raced away (the bug that hung Windows CI). +// Together these abort the producer within ~100 ms of the destructor, so the join — +// and therefore the destructor — returns promptly. +// +// Unlike test_primary_client.cpp's robot-dependent fixtures, this test uses the +// in-process FakePrimaryServer, so it runs in the normal (non-INTEGRATION_TESTS) +// build and needs no robot. +TEST(PrimaryClientReconnectTest, destructor_not_blocked_by_stuck_reconnect_thread) +{ + comm::INotifier notifier; + + auto server = std::make_unique(primary_interface::UR_PRIMARY_PORT); + auto client = std::make_unique("127.0.0.1", notifier); + + // Unlimited reconnect attempts with a large reconnection time: if the fix is + // absent, the producer's reconnect path keeps the destructor blocked. + const std::chrono::milliseconds large_reconnect_timeout(5000); + ASSERT_NO_THROW(client->start(/*max_num_tries=*/0, large_reconnect_timeout)); + ASSERT_TRUE(server->waitForClient()) << "PrimaryClient never connected to the fake server"; + + // Drop the server. The producer's read() fails, the socket transitions to + // SocketState::LostConnection, and the producer enters its reconnect loop. + server.reset(); + + // Give the producer time to detect the drop and reach its reconnect sleep + // (initial backoff is 1 s, after which it sleeps inside TCPSocket::setup()). + std::this_thread::sleep_for(std::chrono::milliseconds(1500)); + + // The destructor must return quickly: disconnect() aborts the producer's connect + // attempt/back-off, so the pipeline join completes well under 2 s. Without the fix + // this blocks for at least the reconnect timeout (and indefinitely with unlimited + // retries against a dead port). Run the destructor on a worker with a watchdog so a + // regression fails fast with a clear message instead of hanging the test binary (the + // CTest TIMEOUT then reaps it). + std::packaged_task teardown([&client]() { client.reset(); }); + auto teardown_future = teardown.get_future(); + std::thread teardown_thread(std::move(teardown)); + + const auto t0 = std::chrono::steady_clock::now(); + if (teardown_future.wait_for(std::chrono::seconds(5)) == std::future_status::timeout) + { + teardown_thread.detach(); + FAIL() << "~PrimaryClient() did not return within 5 s — the producer reconnect thread was not aborted by " + "disconnect()"; + } + teardown_thread.join(); + const auto elapsed = std::chrono::steady_clock::now() - t0; + + EXPECT_LT(elapsed, std::chrono::seconds(2)) + << "~PrimaryClient() blocked for " << std::chrono::duration_cast(elapsed).count() + << " ms — the producer reconnect thread was not aborted by disconnect()"; +} + +// Regression test for the SECOND symptom reported in issue #368: calling +// PrimaryClient::stop() (the implementation of UrDriver::stopPrimaryClientCommunication()) +// hangs when the producer thread is stuck in its reconnect loop against an unreachable +// robot. +// +// This is distinct from the destructor test above: stop() is a restartable operation, so +// besides asserting that it returns promptly it must also leave the client in a state where +// a subsequent start() can reconnect. That exercises the implicit clear of the deliberate-stop +// state by connect() in URProducer::setupProducer()/PrimaryClient::reconnectStream() — a +// regression there would not hang teardown but would silently prevent the client from ever +// reconnecting after a stop(). +TEST(PrimaryClientReconnectTest, stop_not_blocked_by_stuck_reconnect_thread) +{ + comm::INotifier notifier; + + auto server = std::make_unique(primary_interface::UR_PRIMARY_PORT); + auto client = std::make_unique("127.0.0.1", notifier); + + // Unlimited reconnect attempts with a large reconnection time: if the fix is + // absent, the producer's reconnect path keeps stop()'s pipeline join blocked. + const std::chrono::milliseconds large_reconnect_timeout(5000); + ASSERT_NO_THROW(client->start(/*max_num_tries=*/0, large_reconnect_timeout)); + ASSERT_TRUE(server->waitForClient()) << "PrimaryClient never connected to the fake server"; + + // Drop the server. The producer's read() fails, the socket transitions to + // SocketState::LostConnection, and the producer enters its reconnect loop. + server.reset(); + + // Give the producer time to detect the drop and reach its reconnect sleep + // (initial backoff is 1 s, after which it sleeps inside TCPSocket::setup()). + std::this_thread::sleep_for(std::chrono::milliseconds(1500)); + + // stop() must return quickly: disconnect() aborts the producer's connect attempt/back-off, + // so the pipeline join completes well under 2 s. Without the fix this blocks for at least the + // reconnect timeout (and indefinitely with unlimited retries against a dead port). Run it on a + // worker with a watchdog so a regression fails fast with a clear message instead of hanging the + // test binary (the CTest TIMEOUT then reaps it). + std::packaged_task stop_task([&client]() { client->stop(); }); + auto stop_future = stop_task.get_future(); + std::thread stop_thread(std::move(stop_task)); + + const auto t0 = std::chrono::steady_clock::now(); + if (stop_future.wait_for(std::chrono::seconds(5)) == std::future_status::timeout) + { + stop_thread.detach(); + FAIL() << "PrimaryClient::stop() did not return within 5 s — the producer reconnect thread was not aborted by " + "disconnect()"; + } + stop_thread.join(); + const auto elapsed = std::chrono::steady_clock::now() - t0; + + EXPECT_LT(elapsed, std::chrono::seconds(2)) + << "PrimaryClient::stop() blocked for " << std::chrono::duration_cast(elapsed).count() + << " ms — the producer reconnect thread was not aborted by disconnect()"; + + // Restart-reuse check: bring up a fresh server and start() again. This must reconnect, + // proving that the deliberate-stop state set by stop() was cleared by connect() on restart. + auto server2 = std::make_unique(primary_interface::UR_PRIMARY_PORT); + ASSERT_NO_THROW(client->start(/*max_num_tries=*/0, large_reconnect_timeout)); + EXPECT_TRUE(server2->waitForClient(std::chrono::seconds(3))) << "PrimaryClient did not reconnect after " + "stop()/start() — the deliberate-stop state set by " + "stop() was not cleared " + "by connect() on restart"; +} + +int main(int argc, char* argv[]) +{ + ::testing::InitGoogleTest(&argc, argv); + return RUN_ALL_TESTS(); +} diff --git a/tests/test_reverse_interface.cpp b/tests/test_reverse_interface.cpp index 15f7ca246..7fa0a6c44 100644 --- a/tests/test_reverse_interface.cpp +++ b/tests/test_reverse_interface.cpp @@ -76,7 +76,7 @@ class ReverseInterfaceTest : public ::testing::Test Client(const int& port) { std::string host = "127.0.0.1"; - TCPSocket::setup(host, port); + TCPSocket::connect(host, port); timeval tv; tv.tv_sec = 1; tv.tv_usec = 0; diff --git a/tests/test_rtde_client.cpp b/tests/test_rtde_client.cpp index 98f8c712f..d8fbd95ee 100644 --- a/tests/test_rtde_client.cpp +++ b/tests/test_rtde_client.cpp @@ -31,6 +31,7 @@ #include #include #include +#include #include #include #include @@ -812,6 +813,91 @@ TEST_F(RTDEClientTest, test_initialization) EXPECT_GE(std::chrono::duration_cast(elapsed).count(), 20); } +// Regression test for the bug where ~RTDEClient() could block indefinitely when +// the reconnect thread was stuck inside TCPSocket::setup(). Fixed by: (1) calling +// stream_.disconnect() (followed by RTDEClient::disconnect()) before joining reconnecting_thread_ +// in ~RTDEClient(), and (2) making TCPSocket::setup() abort on the deliberate-stop state, +// both during the (non-blocking) connect attempt and during the between-attempt wait. +// +// See also TCPSocketTest.setup_interruptible_by_close and +// TCPSocketTest.setup_interruptible_during_blocking_connect in test_tcp_socket.cpp +// for lower-level unit tests of the same fix that run without INTEGRATION_TESTS. +TEST_F(RTDEClientTest, destructor_not_blocked_by_stuck_reconnect_thread) +{ + // Use a large reconnection timeout so that the blocking window is clearly + // observable if the fix is absent (5 s sleep > 2 s assertion threshold). + const std::chrono::milliseconds large_reconnect_timeout(5000); + + auto fake_rtde_server = std::make_unique(g_FAKE_RTDE_PORT); + // Skip the bootup-timestamp check inside isRobotBooted(). + fake_rtde_server->setStartTime(std::chrono::steady_clock::now() - std::chrono::seconds(52)); + + client_.reset(new rtde_interface::RTDEClient("localhost", notifier_, resources_output_recipe_, + resources_input_recipe_, 100, false, g_FAKE_RTDE_PORT)); + // Attempt init up to 10 times with a short between-attempt sleep to ensure + // the RTDE handshake succeeds even in environments where the fake server's + // response arrives slightly after the 1-second socket read timeout. + bool initialized = false; + for (int attempt = 0; attempt < 10 && !initialized; ++attempt) + { + try + { + // max_connection_attempts=0 (unlimited): TCPSocket::setup() sleeps + // large_reconnect_timeout between every failed connect attempt once the + // server is gone. Use a short initialization_timeout for fast retries. + client_->init(0, large_reconnect_timeout, 1, std::chrono::milliseconds(50)); + initialized = true; + } + catch (const UrException&) + { + // Recreate the client on each retry to start from a clean state. + client_.reset(new rtde_interface::RTDEClient("localhost", notifier_, resources_output_recipe_, + resources_input_recipe_, 100, false, g_FAKE_RTDE_PORT)); + } + } + if (!initialized) + { + GTEST_SKIP() << "Could not initialize RTDEClient with the fake server after 10 attempts; " + "this test requires a reliably responding RTDE server. " + "The TCPSocket-level regression test (TCPSocketTest.setup_interruptible_by_close) " + "verifies the underlying fix without a robot."; + } + + // start(true) arms the reconnect callback via the background read thread. + client_->start(true); + + // Drop the server — the background read thread detects the connection loss, + // calls reconnectCallback(), which launches reconnecting_thread_. That thread + // enters setupCommunication() -> TCPSocket::setup() and begins sleeping + // large_reconnect_timeout between retry attempts. + fake_rtde_server.reset(); + + // Give the reconnect thread time to reach the wait inside TCPSocket::setup(). + std::this_thread::sleep_for(std::chrono::milliseconds(500)); + + // The destructor must return quickly: disconnect() aborts setup()'s connect/wait, + // so the join completes in well under 2 s. Without the fix this would block for + // >= large_reconnect_timeout (5 s), or forever with unlimited attempts. + // Run the destructor on a worker with a watchdog so a regression fails fast with a + // clear message instead of hanging the test binary (the CTest TIMEOUT then reaps it). + std::packaged_task teardown([this]() { client_.reset(); }); + auto teardown_future = teardown.get_future(); + std::thread teardown_thread(std::move(teardown)); + + const auto t0 = std::chrono::steady_clock::now(); + if (teardown_future.wait_for(std::chrono::seconds(5)) == std::future_status::timeout) + { + teardown_thread.detach(); + FAIL() << "~RTDEClient() did not return within 5 s — reconnect thread was not aborted by disconnect()"; + } + teardown_thread.join(); + const auto elapsed = std::chrono::steady_clock::now() - t0; + + EXPECT_LT(elapsed, std::chrono::seconds(2)) + << "RTDEClient destructor blocked for " << std::chrono::duration_cast(elapsed).count() + << " ms — reconnect thread was not aborted by disconnect()"; +} + int main(int argc, char* argv[]) { ::testing::InitGoogleTest(&argc, argv); diff --git a/tests/test_script_command_interface.cpp b/tests/test_script_command_interface.cpp index cf1cd0add..63a4f07fd 100644 --- a/tests/test_script_command_interface.cpp +++ b/tests/test_script_command_interface.cpp @@ -49,7 +49,7 @@ class ScriptCommandInterfaceTest : public ::testing::Test Client(const int& port) { std::string host = "127.0.0.1"; - TCPSocket::setup(host, port); + TCPSocket::connect(host, port); timeval tv; tv.tv_sec = 1; tv.tv_usec = 0; diff --git a/tests/test_script_sender.cpp b/tests/test_script_sender.cpp index d800a8bac..bdd97991d 100644 --- a/tests/test_script_sender.cpp +++ b/tests/test_script_sender.cpp @@ -43,7 +43,7 @@ class ScriptSenderTest : public ::testing::Test Client(const int& port) { std::string host = "127.0.0.1"; - TCPSocket::setup(host, port); + TCPSocket::connect(host, port); timeval tv; tv.tv_sec = 1; tv.tv_usec = 0; diff --git a/tests/test_tcp_server.cpp b/tests/test_tcp_server.cpp index 08c8c4fb8..88640d0e5 100644 --- a/tests/test_tcp_server.cpp +++ b/tests/test_tcp_server.cpp @@ -34,8 +34,14 @@ #include #include #include +#include #include #include +#ifndef _WIN32 +# include +# include +# include +#endif #include "test_utils.h" #include @@ -52,7 +58,7 @@ class TCPServerTest : public ::testing::Test Client(const int& port) { std::string host = "127.0.0.1"; - TCPSocket::setup(host, port); + TCPSocket::connect(host, port); } void send(const std::string& text) @@ -269,7 +275,7 @@ TEST_F(TCPServerTest, check_shutting_down_server_while_listening) } EXPECT_FALSE(read_success); // If the read just would have timeouted, the client state would still be connected. - EXPECT_EQ(client.getState(), comm::SocketState::Disconnected); + EXPECT_EQ(client.getState(), comm::SocketState::LostConnection); } TEST_F(TCPServerTest, double_shutdown) @@ -497,6 +503,186 @@ TEST_F(TCPServerTest, shutdown_during_active_writes) writer.join(); } +// Verifies that the server receives data from many clients that all send simultaneously. This +// exercises the poll() revents loop across many client file descriptors and guards against +// missed read events when several sockets are readable at once. +TEST_F(TCPServerTest, receives_from_many_concurrent_clients) +{ + comm::TCPServer server(0); + + std::mutex mtx; + std::condition_variable cv; + std::atomic message_count{ 0 }; + + server.setMessageCallback([&](const socket_t, char*, int) { + message_count.fetch_add(1); + std::lock_guard lk(mtx); + cv.notify_all(); + }); + server.start(); + +#ifdef _WIN32 + // Windows allows a maximum of 64 sockets per process by default. + constexpr int num_clients = 50; +#else + constexpr int num_clients = 100; +#endif + + std::vector> clients; + for (int i = 0; i < num_clients; ++i) + { + clients.push_back(std::make_unique(server.getPort())); + } + + // Every client sends a single message concurrently. + std::vector senders; + for (auto& client : clients) + { + senders.emplace_back([&client]() { client->send("ping\n"); }); + } + for (auto& t : senders) + { + t.join(); + } + + // The server's poll() loop must observe activity on every client FD and deliver all messages. + std::unique_lock lk(mtx); + EXPECT_TRUE(cv.wait_for(lk, std::chrono::seconds(5), [&]() { return message_count.load() >= num_clients; })); + EXPECT_EQ(message_count.load(), num_clients); +} + +#ifndef _WIN32 +// Regression test for the FD_SETSIZE limitation of select(): a client whose accepted socket file +// descriptor number is >= FD_SETSIZE (1024) must still be serviced normally. This is the exact +// scenario that occurs when the hosting process (e.g. a JVM) holds many file descriptors. The old +// select()-based implementation rejected/crashed on such descriptors; poll() handles them. +TEST_F(TCPServerTest, services_client_with_high_fd_number) +{ + // Make sure we are allowed to open more than FD_SETSIZE descriptors; raise the soft limit if + // needed and skip the test if the hard limit does not allow it. + struct rlimit rl; + ASSERT_EQ(getrlimit(RLIMIT_NOFILE, &rl), 0); + const rlim_t needed = static_cast(FD_SETSIZE) + 64; + if (rl.rlim_cur < needed) + { + rl.rlim_cur = std::min(needed, rl.rlim_max); + if (setrlimit(RLIMIT_NOFILE, &rl) != 0 || rl.rlim_cur < needed) + { + GTEST_SKIP() << "Cannot raise RLIMIT_NOFILE above FD_SETSIZE; skipping high-fd test."; + } + } + + // Consume the low-numbered descriptors so that subsequently created sockets are assigned fd + // numbers beyond FD_SETSIZE. + std::vector fd_hogs; + while (true) + { + int fd = ::open("/dev/null", O_RDONLY); + if (fd < 0) + { + break; + } + fd_hogs.push_back(fd); + if (fd > static_cast(FD_SETSIZE) + 8) + { + break; + } + } + const bool pushed_past_limit = !fd_hogs.empty() && fd_hogs.back() > static_cast(FD_SETSIZE); + if (!pushed_past_limit) + { + for (int fd : fd_hogs) + { + ::close(fd); + } + GTEST_SKIP() << "Could not allocate descriptors beyond FD_SETSIZE; skipping."; + } + + TestableTcpServer server(port_); + server.start(); + + Client client(port_); + EXPECT_TRUE(server.waitForConnectionCallback(2000)); + + // The server-side accepted client FD should exceed FD_SETSIZE -- the case that breaks select(). + auto client_fds = server.getClientFDs(); + ASSERT_FALSE(client_fds.empty()); + EXPECT_GT(client_fds.back(), static_cast(FD_SETSIZE)); + + // Data must flow both ways on the high-numbered descriptor. + const std::string message = "high fd message\n"; + client.send(message); + EXPECT_TRUE(server.waitForMessageCallback(2000)); + EXPECT_EQ(server.getReceivedMessage(), message); + + size_t written; + const auto* data = reinterpret_cast(message.c_str()); + ASSERT_TRUE(server.write(data, message.size(), written)); + EXPECT_EQ(client.recv(), message); + + // Disconnect must also be detected on the high-numbered descriptor. + client.close(); + EXPECT_TRUE(server.waitForDisconnectionCallback(2000)); + + for (int fd : fd_hogs) + { + ::close(fd); + } +} +#endif + +// White-box regression test: the cached poll set must stay in sync with the connected client set +// as clients connect and disconnect. rebuildPollfds() runs before the connect/disconnect callbacks +// fire, so once waitForConnectionCallback()/waitForDisconnectionCallback() returns the poll set is +// already rebuilt and the invariant can be checked deterministically. +TEST_F(TCPServerTest, poll_set_tracks_client_set) +{ + TestableTcpServer server(port_); + server.start(); + EXPECT_EQ(server.getPollSetSize(), 1u); // listen socket only + + std::vector> clients; + for (int i = 0; i < 5; ++i) + { + clients.push_back(std::make_unique(port_)); + ASSERT_TRUE(server.waitForConnectionCallback()); + EXPECT_EQ(server.getPollSetSize(), server.getClientFDs().size() + 1); + } + + while (!clients.empty()) + { + clients.back()->close(); + clients.pop_back(); + ASSERT_TRUE(server.waitForDisconnectionCallback()); + EXPECT_EQ(server.getPollSetSize(), server.getClientFDs().size() + 1); + } + EXPECT_EQ(server.getPollSetSize(), 1u); // back to listen socket only +} + +// Regression test for the snapshot-before-mutate ordering in spin(): a new client connecting must +// not cause an already-connected client's message that arrives in the same poll cycle to be +// dropped. If the revents were read after handleConnect() rebuilt the poll set, the existing +// client's activity would be lost and waitForMessageCallback() would time out. +TEST_F(TCPServerTest, message_not_lost_when_client_connects_concurrently) +{ + TestableTcpServer server(0); + server.start(); + Client existing(server.getPort()); + ASSERT_TRUE(server.waitForConnectionCallback()); + + for (int i = 0; i < 50; ++i) + { + std::unique_ptr fresh; + std::thread connector([&]() { fresh = std::make_unique(server.getPort()); }); + existing.send("ping\n"); + ASSERT_TRUE(server.waitForMessageCallback(2000)); + connector.join(); + ASSERT_TRUE(server.waitForConnectionCallback(2000)); + fresh->close(); + ASSERT_TRUE(server.waitForDisconnectionCallback(2000)); + } +} + int main(int argc, char* argv[]) { ::testing::InitGoogleTest(&argc, argv); diff --git a/tests/test_tcp_socket.cpp b/tests/test_tcp_socket.cpp index bb6bd317a..da27ebe22 100644 --- a/tests/test_tcp_socket.cpp +++ b/tests/test_tcp_socket.cpp @@ -30,16 +30,49 @@ #include #include -#include #include +#include #include "test_utils.h" #include #include #include "ur_client_library/types.h" +#ifdef __linux__ +# include +#endif + using namespace urcl; +#ifdef __linux__ +namespace +{ +// Counts the process' currently open file descriptors by listing /proc/self/fd. The transient +// descriptor opened for the directory itself is present in every measurement, so it cancels out +// when comparing a before/after count. +size_t countOpenFds() +{ + DIR* dir = ::opendir("/proc/self/fd"); + EXPECT_NE(dir, nullptr) << "Could not open /proc/self/fd"; + if (dir == nullptr) + { + return 0; + } + size_t count = 0; + while (struct dirent* entry = ::readdir(dir)) + { + if (std::string(entry->d_name) == "." || std::string(entry->d_name) == "..") + { + continue; + } + ++count; + } + ::closedir(dir); + return count; +} +} // namespace +#endif // __linux__ + class TCPSocketTest : public ::testing::Test { protected: @@ -66,10 +99,40 @@ class TCPSocketTest : public ::testing::Test ip_ = ip; } - bool setup(const size_t max_num_tries = 0, - const std::chrono::milliseconds reconnection_time = std::chrono::seconds(10)) + bool connect(const size_t max_num_tries = 0, + const std::chrono::milliseconds reconnection_time = std::chrono::seconds(10)) + { + return TCPSocket::connect(ip_, port_, max_num_tries, reconnection_time); + } + + bool reconnect(const size_t max_num_tries = 0, + const std::chrono::milliseconds reconnection_time = std::chrono::seconds(10)) + { + return TCPSocket::reconnect(ip_, port_, max_num_tries, reconnection_time); + } + + bool setTargetStateUnlessStopRequested(comm::SocketState desired) + { + URCL_LOG_INFO("Setting target state to %s", socketStateToString(desired).c_str()); + return TCPSocket::setTargetStateUnlessStopRequested(desired); + } + + bool isStopRequested() const + { + return TCPSocket::isStopRequested(); + } + + void printState() + { + comm::SocketState state = getState(); + std::string state_str = socketStateToString(state); + std::cout << "Client state: " << state_str << ", stop_requested: " << std::boolalpha << isStopRequested() + << std::endl; + } + + void disconnect() { - return TCPSocket::setup(ip_, port_, max_num_tries, reconnection_time); + TCPSocket::close(); } void setupClientBeforeServer(const size_t max_num_tries = 0, @@ -107,7 +170,7 @@ class TCPSocketTest : public ::testing::Test std::chrono::milliseconds reconnection_time = std::chrono::seconds(10)) { std::string ip = "127.0.0.1"; - TCPSocket::setup(ip, port, max_num_tries, reconnection_time); + TCPSocket::connect(ip, port, max_num_tries, reconnection_time); done_setting_up_client_ = true; } }; @@ -125,7 +188,7 @@ TEST_F(TCPSocketTest, socket_state) EXPECT_EQ(toUnderlying(expected_state), toUnderlying(actual_state)); // Client state should be connected after setup - client_->setup(); + client_->connect(); expected_state = comm::SocketState::Connected; actual_state = client_->getState(); @@ -149,8 +212,8 @@ TEST_F(TCPSocketTest, setup_client_before_server) // Make sure that the client has tried to connect to the server, before creating the server std::this_thread::sleep_for(std::chrono::seconds(1)); - // Client state should be invalid as long as the server is not available - comm::SocketState expected_state = comm::SocketState::Invalid; + // Client state should be Connecting while it keeps retrying and the server is not available + comm::SocketState expected_state = comm::SocketState::Connecting; comm::SocketState actual_state = client_->getState(); EXPECT_EQ(toUnderlying(expected_state), toUnderlying(actual_state)); @@ -174,7 +237,7 @@ TEST_F(TCPSocketTest, get_ip) EXPECT_EQ(expected_ip, actual_ip); - client_->setup(); + client_->connect(); expected_ip = "127.0.0.1"; actual_ip = client_->getIP(); @@ -201,7 +264,7 @@ TEST_F(TCPSocketTest, read_on_non_connected_socket) TEST_F(TCPSocketTest, write_on_connected_socket) { - client_->setup(); + client_->connect(); std::string message = "test message"; const uint8_t* data = reinterpret_cast(message.c_str()); @@ -215,7 +278,7 @@ TEST_F(TCPSocketTest, write_on_connected_socket) TEST_F(TCPSocketTest, read_on_connected_socket) { - client_->setup(); + client_->connect(); // Make sure the client has connected to the server, before writing to the client EXPECT_TRUE(server_->waitForConnectionCallback()); @@ -248,7 +311,7 @@ TEST_F(TCPSocketTest, get_socket_fd) EXPECT_EQ(expected_fd, actual_fd); - client_->setup(); + client_->connect(); actual_fd = client_->getSocketFD(); EXPECT_NE(expected_fd, actual_fd); @@ -261,7 +324,7 @@ TEST_F(TCPSocketTest, get_socket_fd) TEST_F(TCPSocketTest, receive_timeout) { - client_->setup(); + client_->connect(); timeval tv; tv.tv_sec = 1; @@ -277,33 +340,103 @@ TEST_F(TCPSocketTest, receive_timeout) TEST_F(TCPSocketTest, setup_while_client_is_connected) { - client_->setup(); + client_->connect(); - EXPECT_FALSE(client_->setup()); + EXPECT_FALSE(client_->connect()); } TEST_F(TCPSocketTest, connect_non_running_robot) { Client client(12321, "127.0.0.1"); auto start = std::chrono::system_clock::now(); - EXPECT_FALSE(client.setup(2, std::chrono::milliseconds(500))); + EXPECT_FALSE(client.connect(2, std::chrono::milliseconds(500))); auto end = std::chrono::system_clock::now(); auto elapsed = end - start; // This is only a rough estimate, obviously EXPECT_LT(elapsed, std::chrono::milliseconds(7500)); } +// Regression test for the bug where TCPSocket::setup() could block the caller +// indefinitely when the wait between retry attempts was not interruptible. +// +// setup() is interrupted by disconnect(), which moves the socket into the deliberate-stop +// state (Disconnecting/Disconnected) and closes it. setup() never overwrites that state and +// aborts as soon as it observes it, so a teardown signal cannot be lost by setup()'s internal +// state updates. This is the path used by ~RTDEClient()/~PrimaryClient() before joining their +// reconnect threads. +TEST_F(TCPSocketTest, setup_interruptible_by_close) +{ + // Use a port with no listener so every connect attempt fails immediately, + // sending setup() into the between-attempt wait. + const int unused_port = 12322; + const std::chrono::milliseconds large_reconnect_timeout(5000); + + Client client(unused_port, "127.0.0.1"); + + // Run setup() with unlimited retries in a background thread. + std::thread setup_thread([&client, &large_reconnect_timeout]() { + // max_num_tries=0 (unlimited) → setup() waits large_reconnect_timeout after + // every failed connect attempt and never exits on its own. + client.connect(0, large_reconnect_timeout); + }); + + // Give the thread time to reach the between-attempt wait inside setup(). + std::this_thread::sleep_for(std::chrono::milliseconds(300)); + + // disconnect() moves the socket to the deliberate-stop state; the sliced wait in setup() + // detects it within 100 ms and setup() returns, allowing the thread to finish. + const auto t0 = std::chrono::steady_clock::now(); + client.disconnect(); + + setup_thread.join(); + const auto elapsed = std::chrono::steady_clock::now() - t0; + + // Without the fix, elapsed would be >= large_reconnect_timeout (5 s). + EXPECT_LT(elapsed, std::chrono::seconds(2)) << "TCPSocket::setup() was not interrupted by disconnect() within 2 s; " + "the between-attempt wait is not interruptible"; +} + +// Regression test for issue #368: a reconnect thread blocked inside connect() to a +// genuinely unreachable host (no SYN-ACK, no RST) must still be abortable. The +// previous fix only made the between-attempt *sleep* interruptible, not the connect +// itself, so this case could block for the full OS connect timeout. setup() now uses +// a non-blocking connect polled in short slices, so disconnect() aborts it promptly. +TEST_F(TCPSocketTest, setup_interruptible_during_blocking_connect) +{ + // 10.255.255.1 is in a private range and is (almost) never routable, so connect() + // hangs in SYN retransmit rather than failing fast like a refused localhost port. + const int unused_port = 12323; + const std::chrono::milliseconds large_reconnect_timeout(5000); + + Client client(unused_port, "10.255.255.1"); + + std::thread setup_thread([&client, &large_reconnect_timeout]() { client.connect(0, large_reconnect_timeout); }); + + // Give the thread time to enter the (blocking) connect attempt. + std::this_thread::sleep_for(std::chrono::milliseconds(500)); + + const auto t0 = std::chrono::steady_clock::now(); + client.disconnect(); + + setup_thread.join(); + const auto elapsed = std::chrono::steady_clock::now() - t0; + + EXPECT_LT(elapsed, std::chrono::seconds(2)) << "TCPSocket::setup() was not interrupted while blocked in connect() " + "within 2 s; " + "the connect attempt is not interruptible"; +} + TEST_F(TCPSocketTest, test_deprecated_reconnection_time_interface) { URCL_SILENCE_DEPRECATED_BEGIN client_->setReconnectionTime(std::chrono::milliseconds(100)); URCL_SILENCE_DEPRECATED_END - EXPECT_TRUE(client_->setup(2)); + EXPECT_TRUE(client_->connect(2)); } TEST_F(TCPSocketTest, test_read_on_socket_abruptly_closed) { - client_->setup(); + client_->connect(); // Make sure the client has connected to the server, before writing to the client EXPECT_TRUE(server_->waitForConnectionCallback()); @@ -320,7 +453,289 @@ TEST_F(TCPSocketTest, test_read_on_socket_abruptly_closed) char characters; size_t read_chars = 0; EXPECT_FALSE(client_->read((uint8_t*)&characters, 1, read_chars)); - EXPECT_EQ(client_->getState(), comm::SocketState::Disconnected); + EXPECT_EQ(client_->getState(), comm::SocketState::LostConnection); +} + +TEST_F(TCPSocketTest, test_socket_client_lifecycle) +{ + // After startup a client should be in state Invalid + EXPECT_EQ(client_->getState(), comm::SocketState::Invalid); + client_->printState(); + + client_->connect(); + EXPECT_EQ(client_->getState(), comm::SocketState::Connected); + client_->printState(); + + // Make sure the client has connected to the server, before writing to the client + EXPECT_TRUE(server_->waitForConnectionCallback()); + + std::atomic read_count(0); + auto read_fun = [this, &read_count]() { + bool read_success = true; + uint8_t buffer[64]; + size_t read_chars = 0; + while (read_success) + { + std::cout << "Reading from socket..." << std::endl; + read_success = client_->read(buffer, 64, read_chars); + if (read_success) + { + std::cout << "Read " << read_chars << " characters." << std::endl; + read_count.store(read_count.load() + 1); + } + } + }; + + // Scenario 1: Client is connected, server is shutdown, client should go into LostConnection + // state. Afterwards we should be able to restart the server and reconnect the client. + { + std::thread read_thread(read_fun); + + std::string send_message = "test message"; + size_t len = send_message.size(); + const uint8_t* data = reinterpret_cast(send_message.c_str()); + size_t written; + server_->write(data, len, written); + + ASSERT_NO_THROW(waitFor([&read_count]() { return read_count.load() > 0; }, std::chrono::seconds(1))); + std::cout << "Read count: " << read_count.load() << std::endl; + + EXPECT_EQ(client_->getState(), comm::SocketState::Connected); + client_->printState(); + + std::cout << "Shutting down server..." << std::endl; + server_->shutdown(); + + read_thread.join(); + EXPECT_EQ(client_->getState(), comm::SocketState::LostConnection); + client_->printState(); + + std::cout << "Restarting server..." << std::endl; + server_.reset(new TestableTcpServer(60001)); + server_->start(); + + std::cout << "Reconnecting client..." << std::endl; + ASSERT_TRUE(client_->connect()); + client_->printState(); + } + + // Scenario 2: Deliberate disconnect() while a read is in progress should abort the read and put + // the client into Disconnected state. We should be able to reconnect the client afterwards. + { + std::thread read_thread(read_fun); + client_->disconnect(); + read_thread.join(); + EXPECT_EQ(client_->getState(), comm::SocketState::Closed); + client_->printState(); + + client_->connect(); + EXPECT_EQ(client_->getState(), comm::SocketState::Connected) + << "Expected state" << comm::socketStateToString(comm::SocketState::Connected) + << ". Got state : " << socketStateToString(client_->getState()); + client_->printState(); + } + + // Scenario 3: Server shuts down, client disconnects. Connecting the client will go into a + // connection attempt loop. At some point we start the server again and the client should reconnect successfully. + // This simulates a client application with a auto-reconnect on read-failure. + { + std::thread read_thread(read_fun); + + std::cout << "Shutting down server..." << std::endl; + server_->shutdown(); + + read_thread.join(); + EXPECT_EQ(client_->getState(), comm::SocketState::LostConnection); + client_->printState(); + + std::cout << "Starting connection attempt loop..." << std::endl; + std::thread connect_thread([this]() { + bool result = client_->reconnect(0, std::chrono::milliseconds(100)); + EXPECT_TRUE(result) << "Client failed to reconnect after server restart"; + }); + + std::this_thread::sleep_for(std::chrono::milliseconds(200)); + server_.reset(new TestableTcpServer(60001)); + server_->start(); + + connect_thread.join(); + EXPECT_EQ(client_->getState(), comm::SocketState::Connected) + << "Expected state" << comm::socketStateToString(comm::SocketState::Connected) + << ". Got state: " << socketStateToString(client_->getState()); + client_->printState(); + } + + // Scenario 4: Same as scenario 3, but while reconnecting we do not start the server but + // disconnect the client. The client should abort the connection attempt loop and go into Disconnected state. + { + std::thread read_thread(read_fun); + + std::cout << "Shutting down server..." << std::endl; + server_->shutdown(); + + read_thread.join(); + EXPECT_EQ(client_->getState(), comm::SocketState::LostConnection) + << "Client state after reconnect: " << socketStateToString(client_->getState()); + client_->printState(); + + std::cout << "Starting connection attempt loop..." << std::endl; + std::thread connect_thread([this]() { + bool result = client_->reconnect(0, std::chrono::milliseconds(100)); + EXPECT_FALSE(result) << "Client should have aborted the connection attempt loop"; + }); + + std::this_thread::sleep_for(std::chrono::milliseconds(200)); + client_->disconnect(); + + connect_thread.join(); + EXPECT_EQ(client_->getState(), comm::SocketState::Closed) + << "Client state after reconnect: " << socketStateToString(client_->getState()); + client_->printState(); + } + + // Scenario 5: We attempt to connect to a non-reachable server. Disconnecting the client should abort the connection + // attempt loop and put the client into Disconnected state. + { + server_->shutdown(); + std::thread connect_thread([this]() { + bool result = client_->connect(0, std::chrono::milliseconds(100)); + EXPECT_FALSE(result) << "Client should have aborted the connection attempt loop"; + }); + + std::this_thread::sleep_for(std::chrono::milliseconds(200)); + client_->disconnect(); + + connect_thread.join(); + EXPECT_EQ(client_->getState(), comm::SocketState::Closed) + << "Client state after reconnect: " << socketStateToString(client_->getState()); + client_->printState(); + } + + // Scenario 6: Server shuts down, connection lost. Then, client gets disconnected. Server starts + // again, reconnect should fail, since it was deliberately disconnected. Connect should work. + { + server_.reset(new TestableTcpServer(60001)); + server_->start(); + client_->connect(); + std::thread read_thread(read_fun); + + std::cout << "Shutting down server..." << std::endl; + server_->shutdown(); + + read_thread.join(); + EXPECT_EQ(client_->getState(), comm::SocketState::LostConnection); + client_->printState(); + + std::cout << "Disconnecting client..." << std::endl; + client_->disconnect(); + EXPECT_EQ(client_->getState(), comm::SocketState::Closed); + client_->printState(); + + std::cout << "Starting server..." << std::endl; + server_.reset(new TestableTcpServer(60001)); + server_->start(); + + std::cout << "Attempting to reconnect client..." << std::endl; + bool result = client_->reconnect(0, std::chrono::milliseconds(100)); + EXPECT_FALSE(result) << "Client should not have reconnected after deliberate disconnect()"; + EXPECT_EQ(client_->getState(), comm::SocketState::Closed); + client_->printState(); + + std::cout << "Attempting to connect client..." << std::endl; + result = client_->connect(0, std::chrono::milliseconds(100)); + EXPECT_TRUE(result) << "Client should have connected after deliberate disconnect()"; + EXPECT_EQ(client_->getState(), comm::SocketState::Connected); + client_->printState(); + } +} + +TEST_F(TCPSocketTest, test_set_target_state_unless_stop_requested) +{ + ASSERT_EQ(client_->getState(), comm::SocketState::Invalid); + // Requesting to go to current state should return true + ASSERT_TRUE(client_->setTargetStateUnlessStopRequested(comm::SocketState::Invalid)); + // We should be able to set the target state to Connected, since we are not in a deliberate + // disconnect() state + ASSERT_TRUE(client_->setTargetStateUnlessStopRequested(comm::SocketState::Connected)); + // This should not have modified the state of the client, since we are not connected yet + ASSERT_EQ(client_->getState(), comm::SocketState::Invalid); + client_->connect(); + + // Make sure the client has connected to the server, before writing to the client + EXPECT_TRUE(server_->waitForConnectionCallback()); + + ASSERT_EQ(client_->getState(), comm::SocketState::Connected); + // Requesting to go to current state should return true + ASSERT_TRUE(client_->setTargetStateUnlessStopRequested(comm::SocketState::Connected)); + + client_->disconnect(); + ASSERT_EQ(client_->getState(), comm::SocketState::Closed); + ASSERT_TRUE(client_->isStopRequested()); + ASSERT_FALSE(client_->setTargetStateUnlessStopRequested(comm::SocketState::Connected)); + client_->printState(); +} + +// Regression test for a file-descriptor leak: each connection attempt creates a new socket +// descriptor, and a failed attempt must free it. Repeatedly failing to connect must therefore not +// grow the number of open descriptors. +TEST_F(TCPSocketTest, no_fd_leak_on_repeated_failed_connects) +{ +#ifndef __linux__ + GTEST_SKIP() << "Counting open file descriptors via /proc/self/fd is Linux-only"; +#else + // Port with no listener so every attempt fails quickly. + const int dead_port = 12321; + // A little slack absorbs unrelated descriptors (e.g. logging, DNS caches) that may be opened + // lazily on the first attempts; a real leak grows linearly with the iteration count and blows + // well past this. + const size_t slack = 4; + + // Warm up once so any one-off descriptor allocation happens before we take the baseline. + { + Client warmup(dead_port, "127.0.0.1"); + EXPECT_FALSE(warmup.connect(2, std::chrono::milliseconds(20))); + } + + const size_t baseline = countOpenFds(); + for (int i = 0; i < 50; ++i) + { + Client client(dead_port, "127.0.0.1"); + EXPECT_FALSE(client.connect(2, std::chrono::milliseconds(20))); + } + const size_t after = countOpenFds(); + + EXPECT_LE(after, baseline + slack) << "Open file descriptors grew from " << baseline << " to " << after + << " over 50 failed connection attempts; a descriptor is leaking"; +#endif // __linux__ +} + +// Regression test targeting the retry loop specifically: with unlimited retries the socket keeps +// creating new descriptors between attempts. Letting it retry several times and then disconnecting +// must not leave leaked descriptors behind. +TEST_F(TCPSocketTest, no_fd_leak_in_retry_loop) +{ +#ifndef __linux__ + GTEST_SKIP() << "Counting open file descriptors via /proc/self/fd is Linux-only"; +#else + const int dead_port = 12321; + const size_t slack = 4; + + const size_t baseline = countOpenFds(); + + Client client(dead_port, "127.0.0.1"); + // Unlimited retries with a short back-off, so the loop runs many iterations while we wait. + std::thread connect_thread([&client]() { client.connect(0, std::chrono::milliseconds(50)); }); + + // Give the retry loop time to run through a good number of failed attempts. + std::this_thread::sleep_for(std::chrono::milliseconds(500)); + + client.disconnect(); + connect_thread.join(); + + const size_t after = countOpenFds(); + EXPECT_LE(after, baseline + slack) << "Open file descriptors grew from " << baseline << " to " << after + << " while retrying to connect; the retry loop is leaking descriptors"; +#endif // __linux__ } int main(int argc, char* argv[]) diff --git a/tests/test_trajectory_point_interface.cpp b/tests/test_trajectory_point_interface.cpp index 5e0356c96..f648eac78 100644 --- a/tests/test_trajectory_point_interface.cpp +++ b/tests/test_trajectory_point_interface.cpp @@ -79,7 +79,7 @@ class TrajectoryPointInterfaceTest : public ::testing::Test Client(const int& port) { std::string host = "127.0.0.1"; - TCPSocket::setup(host, port); + TCPSocket::connect(host, port); timeval tv; tv.tv_sec = 1; tv.tv_usec = 0; diff --git a/tests/test_unique_fd.cpp b/tests/test_unique_fd.cpp new file mode 100644 index 000000000..028c3cd22 --- /dev/null +++ b/tests/test_unique_fd.cpp @@ -0,0 +1,148 @@ +// -- BEGIN LICENSE BLOCK ---------------------------------------------- +// Copyright 2026 Universal Robots A/S +// +// Redistribution and use in source and binary forms, with or without +// modification, are permitted provided that the following conditions are met: +// +// * Redistributions of source code must retain the above copyright +// notice, this list of conditions and the following disclaimer. +// +// * Redistributions in binary form must reproduce the above copyright +// notice, this list of conditions and the following disclaimer in the +// documentation and/or other materials provided with the distribution. +// +// * Neither the name of the {copyright_holder} nor the names of its +// contributors may be used to endorse or promote products derived from +// this software without specific prior written permission. +// +// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" +// AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE +// IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE +// ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE +// LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR +// CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF +// SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS +// INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN +// CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) +// ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE +// POSSIBILITY OF SUCH DAMAGE. +// -- END LICENSE BLOCK ------------------------------------------------ + +#include +#include + +#include + +using namespace urcl::comm; + +// UniqueFd is a thin RAII wrapper around a socket descriptor. On POSIX platforms a socket +// descriptor is an ordinary file descriptor, so we can allocate real descriptors with dup() and +// verify that they get closed by querying fcntl(F_GETFD), which fails with EBADF once the +// descriptor is closed. On Windows sockets are not fcntl-able, so the closing-behaviour tests are +// POSIX-only; the ownership semantics are identical across platforms. +#ifndef _WIN32 +# include +# include + +namespace +{ +// Allocates a fresh, valid descriptor for testing by duplicating stdin. +socket_t makeFd() +{ + socket_t fd = ::dup(0); + EXPECT_GE(fd, 0) << "Could not allocate a test file descriptor"; + return fd; +} + +bool isFdOpen(socket_t fd) +{ + return ::fcntl(fd, F_GETFD) != -1; +} +} // namespace + +TEST(UniqueFdTest, default_constructed_is_invalid) +{ + UniqueFd fd; + EXPECT_EQ(fd.get(), INVALID_SOCKET); +} + +TEST(UniqueFdTest, reset_adopts_descriptor) +{ + socket_t raw = makeFd(); + UniqueFd fd; + fd.reset(raw); + EXPECT_EQ(fd.get(), raw); + EXPECT_TRUE(isFdOpen(raw)); +} + +TEST(UniqueFdTest, reset_with_new_fd_closes_previous) +{ + socket_t first = makeFd(); + socket_t second = makeFd(); + + UniqueFd fd; + fd.reset(first); + fd.reset(second); + + // The previously held descriptor must have been closed, and the new one adopted. + EXPECT_FALSE(isFdOpen(first)); + EXPECT_TRUE(isFdOpen(second)); + EXPECT_EQ(fd.get(), second); +} + +TEST(UniqueFdTest, reset_without_argument_closes_and_invalidates) +{ + socket_t raw = makeFd(); + + UniqueFd fd; + fd.reset(raw); + fd.reset(); + + EXPECT_FALSE(isFdOpen(raw)); + EXPECT_EQ(fd.get(), INVALID_SOCKET); +} + +TEST(UniqueFdTest, reset_with_same_fd_is_a_noop) +{ + socket_t raw = makeFd(); + + UniqueFd fd; + fd.reset(raw); + // Resetting with the descriptor we already hold must not close it. + fd.reset(raw); + + EXPECT_TRUE(isFdOpen(raw)); + EXPECT_EQ(fd.get(), raw); +} + +TEST(UniqueFdTest, destructor_closes_descriptor) +{ + socket_t raw = makeFd(); + { + UniqueFd fd; + fd.reset(raw); + EXPECT_TRUE(isFdOpen(raw)); + } + EXPECT_FALSE(isFdOpen(raw)); +} + +TEST(UniqueFdTest, destructor_on_invalid_is_safe) +{ + // Destroying a UniqueFd that never held a descriptor must not attempt to close anything. + UniqueFd fd; + SUCCEED(); +} +#endif // _WIN32 + +TEST(UniqueFdTest, is_non_copyable) +{ + static_assert(!std::is_copy_constructible::value, "UniqueFd must not be copy constructible"); + static_assert(!std::is_copy_assignable::value, "UniqueFd must not be copy assignable"); + SUCCEED(); +} + +int main(int argc, char* argv[]) +{ + ::testing::InitGoogleTest(&argc, argv); + return RUN_ALL_TESTS(); +} diff --git a/tests/test_utils.h b/tests/test_utils.h index 31fc11022..08273b51d 100644 --- a/tests/test_utils.h +++ b/tests/test_utils.h @@ -165,6 +165,11 @@ class TestableTcpServer : public urcl::comm::TCPServer return client_fds_; } + size_t getPollSetSize() + { + return TCPServer::getPollSetSize(); + } + private: std::vector client_fds_; std::condition_variable connect_cv_;