Further improve thread safety

Author: urfeex

include/ur_client_library/comm/socket_t.h

@@ -71,20 +71,24 @@ typedef int socket_t;
 #  define MSG_NOSIGNAL 0
 #endif
 
-inline std::system_error makeSocketError(const std::string& message)
+/*!
+ * \brief Get the last socket error as an std::error_code
+ *
+ * On Windows, this will use WSAGetLastError and the system category, while on other platforms it
+ * will use errno and the generic category.
+ *
+ * \return The last socket error
+ */
+inline std::error_code getLastSocketErrorCode()
 {
 #ifdef _WIN32
-  return std::system_error(std::error_code(WSAGetLastError(), std::system_category()), message);
+  return std::error_code(WSAGetLastError(), std::system_category());
 #else
-  return std::system_error(std::error_code(errno, std::generic_category()), message);
+  return std::error_code(errno, std::generic_category());
 #endif
 }
 
-inline int getLastSocketError()
+inline std::system_error makeSocketError(const std::string& message)
 {
-#ifdef _WIN32
-  return WSAGetLastError();
-#else
-  return errno;
-#endif
+  return std::system_error(getLastSocketErrorCode(), message);
 }

include/ur_client_library/comm/tcp_server.h

@@ -82,6 +82,9 @@ class TCPServer
    * \note: For thread safety, this will block when there is an active callback around connection
    * (connection, disconnection, shutdown). Thus, trying to call setConnectCallback e.g. by a
    * handler function registered as disconnectCallback will result in a deadlock.
+   *
+   * \note: Calling write() from within the callback will cause a deadlock. If you want to write from the connection
+   * callback, you can use writeUnchecked(), as the callback will be triggered from a thread-protected context.
    */
   void setConnectCallback(std::function<void(const socket_t)> func)
   {
@@ -98,6 +101,9 @@ class TCPServer
    * \note: For thread safety, this will block when there is an active callback around connection
    * (connection, disconnection, shutdown). Thus, trying to call setDisconnectCallback e.g. by a
    * handler function registered as connectCallback will result in a deadlock.
+   *
+   * \note: The socket will already be closed when the disconnect callback is triggered, thus
+   * trying to write to the socket from the disconnect callback will fail.
    */
   void setDisconnectCallback(std::function<void(const socket_t)> func)
   {
@@ -113,6 +119,9 @@ class TCPServer
    *
    * \note: For thread safety, this will block when there is an active message callback. Thus, trying to call
    * setMessageCallback e.g. from a handler function registered as messageCallback will result in a deadlock.
+   *
+   * \note: Calling write() from within the callback will cause a deadlock. If you want to write from the message
+   * callback, you can use writeUnchecked(), as the callback will be triggered from a thread-protected context.
    */
   void setMessageCallback(std::function<void(const socket_t, char*, int)> func)
   {
@@ -129,9 +138,7 @@ class TCPServer
   void start();
 
   /*!
-   * \brief Shut down the event listener thread. After calling this, no events will be handled
-   * anymore, but the socket will remain open and bound to the port. Call start() in order to
-   * restart event handling.
+   * \brief Shutdown the server and close all client connections.
    */
   void shutdown();
 
@@ -148,6 +155,21 @@ class TCPServer
    */
   bool write(const socket_t fd, const uint8_t* buf, const size_t buf_len, size_t& written);
 
+  /*!
+   * \brief Writes to a filedescriptor without verifying that it is a client or even a valid
+   * filedescriptor. It is the caller's responsibility to ensure that the filedescriptor is valid
+   * and belongs to a client.
+   *
+   * \param[in] fd File descriptor belonging to the client the data should be sent to. The file
+   * descriptor will be given from the connection callback.
+   * \param[in] buf Buffer of bytes to write
+   * \param[in] buf_len Number of bytes in the buffer
+   * \param[out] written Number of bytes actually written
+   *
+   * \returns True on success, false otherwise
+   */
+  bool writeUnchecked(const socket_t fd, const uint8_t* buf, const size_t buf_len, size_t& written);
+
   /*!
    * \brief Get the maximum number of clients allowed to connect to this server
    *
@@ -217,6 +239,7 @@ class TCPServer
   std::vector<socket_t> client_fds_;
   std::mutex clients_mutex_;
   std::mutex message_mutex_;
+  std::mutex listen_fd_mutex_;
 
   static const int INPUT_BUFFER_SIZE = 4096;
   char input_buffer_[INPUT_BUFFER_SIZE];

include/ur_client_library/log.h

@@ -89,19 +89,4 @@ void setLogLevel(LogLevel level);
  * \param fmt Format string
  */
 void log(const char* file, int line, LogLevel level, const char* fmt, ...);
-
-/*!
- * \brief Cross-platform replacement for strerror.
- *
- * On MSVC, strerror triggers C4996 (deprecated). This function uses std::error_code instead,
- * which is portable and thread-safe.
- *
- * \param errnum Error number (on POSIX typically errno)
- * \returns Human-readable error message
- */
-inline std::string strerrorPortable(int errnum)
-{
-  return std::error_code(errnum, std::system_category()).message();
-}
-
 }  // namespace urcl

src/comm/tcp_server.cpp

@@ -34,6 +34,7 @@
 
 #include <sstream>
 #include <cstring>
+#include "ur_client_library/comm/socket_t.h"
 #include <fcntl.h>
 
 namespace urcl
@@ -80,11 +81,23 @@ void TCPServer::init()
 
 void TCPServer::shutdown()
 {
+  std::unique_lock<std::mutex> listen_lk(listen_fd_mutex_, std::try_to_lock);
   if (listen_fd_ == INVALID_SOCKET)
   {
-    URCL_LOG_DEBUG("TCPServer is already shut down.");
+    URCL_LOG_INFO("Listen FD already closed by another thread. Nothing to do here.");
     return;
   }
+  if (!listen_lk.owns_lock())
+  {
+    URCL_LOG_WARN("Could not acquire lock for listen FD when shutting down. Is there another thread shutting the "
+                  "server down already? Waiting for lock to be released.");
+    listen_lk.lock();
+    if (listen_fd_ == INVALID_SOCKET)
+    {
+      URCL_LOG_INFO("Listen FD already closed by another thread. Nothing to do here.");
+      return;
+    }
+  }
   keep_running_ = false;
 
   socket_t shutdown_socket = ::socket(AF_INET, SOCK_STREAM, 0);
@@ -147,9 +160,9 @@ void TCPServer::bind(const size_t max_num_tries, const std::chrono::milliseconds
     err = ::bind(listen_fd_, (struct sockaddr*)&server_addr, sizeof(server_addr));
     if (err == -1)
     {
+      auto error_code = getLastSocketErrorCode();
       std::ostringstream ss;
-      ss << "Failed to bind socket for port " << port_
-         << " to address. Reason: " << strerrorPortable(getLastSocketError());
+      ss << "Failed to bind socket for port " << port_ << " to address. Reason: " << error_code.message();
 
       if (connection_counter++ < max_num_tries || max_num_tries == 0)
       {
@@ -160,7 +173,7 @@ void TCPServer::bind(const size_t max_num_tries, const std::chrono::milliseconds
       }
       else
       {
-        throw makeSocketError(ss.str());
+        throw std::system_error(error_code, ss.str());
       }
     }
   } while (err == -1 && (connection_counter <= max_num_tries || max_num_tries == 0));
@@ -185,7 +198,7 @@ void TCPServer::startListen()
   if (getsockname(listen_fd_, (struct sockaddr*)&sin, &len) == -1)
   {
     URCL_LOG_ERROR("getsockname() failed to get port number for listening socket: %s",
-                   strerrorPortable(getLastSocketError()).c_str());
+                   getLastSocketErrorCode().message().c_str());
   }
 
   else
@@ -202,12 +215,26 @@ void TCPServer::handleConnect()
   socket_t client_fd = accept(listen_fd_, (struct sockaddr*)&client_addr, &addrlen);
   if (client_fd == INVALID_SOCKET)
   {
-    std::ostringstream ss;
-    ss << "Failed to accept connection request on port  " << port_;
-    throw makeSocketError(ss.str());
+    URCL_LOG_ERROR("Failed to accept connection request on port %d. Reason: %s", port_,
+                   getLastSocketErrorCode().message().c_str());
+    return;
   }
 
-  if (client_fd >= FD_SETSIZE)
+  auto set_size_exceeded = [this, &client_fd]() {
+#ifdef _WIN32
+    (void)client_fd;  // Avoid unused variable warning, since on Windows we only check the number of
+                      // clients, not the client FD itself.
+    return client_fds_.size() >= FD_SETSIZE - 1;  // -1 because listen_fd_ also occupies one
+                                                  // slot in masterfds_
+#else
+    (void)this;                      // Avoid unused variable warning, since on Unix-like systems we only check the
+                                     // client FD itself, not the number of clients.
+    return 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);
@@ -267,14 +294,17 @@ void TCPServer::spin()
 
   std::vector<socket_t> disconnected_clients;
 
-  for (const auto& client_fd : client_fds_)
   {
-    if (FD_ISSET(client_fd, &tempfds_))
+    std::lock_guard<std::mutex> lk(clients_mutex_);
+    for (const auto& client_fd : client_fds_)
     {
-      URCL_LOG_DEBUG("Activity on client FD %d", (int)client_fd);
-      if (!readData(client_fd))
+      if (FD_ISSET(client_fd, &tempfds_))
       {
-        disconnected_clients.push_back(client_fd);
+        URCL_LOG_DEBUG("Activity on client FD %d", (int)client_fd);
+        if (!readData(client_fd))
+        {
+          disconnected_clients.push_back(client_fd);
+        }
       }
     }
   }
@@ -388,13 +418,20 @@ bool TCPServer::write(const socket_t fd, const uint8_t* buf, const size_t buf_le
     }
   }
 
+  // We don't use a lock around the send call here, since writing on a closed socket would raise
+  // an error anyway, and the client FD is only removed from client_fds_ after the socket is
+  // closed. Thus, even if the client gets disconnected right after the check, the send call will
+  // just fail and return false, which is the expected behavior.
+  return writeUnchecked(fd, buf, buf_len, written);
+}
+
+bool TCPServer::writeUnchecked(const socket_t fd, const uint8_t* buf, const size_t buf_len, size_t& written)
+{
   size_t remaining = buf_len;
 
   // handle partial sends
   while (written < buf_len)
   {
-    // We don't use a lock around the send call here, since writing on a closed socket would raise
-    // an error anyway.
     ssize_t sent =
         ::send(fd, reinterpret_cast<const char*>(buf + written), static_cast<socklen_t>(remaining), MSG_NOSIGNAL);