Robustify TCPServer implementation for thread safety and windows support (#458)

This PR hardens the TCPServer implementation in the following points:

* Add thread safety between connect / disconnect, write / read and
shutdown events.
* Improved windows-specific error case handling

Author: urfeex

include/ur_client_library/comm/socket_t.h

@@ -16,6 +16,7 @@
 
 #pragma once
 
+#include <system_error>
 #ifdef _WIN32
 
 #  define NOMINMAX
@@ -33,6 +34,7 @@
 
 typedef SOCKET socket_t;
 typedef SSIZE_T ssize_t;
+typedef int socklen_t;
 
 static inline int ur_setsockopt(socket_t s, int level, int optname, const void* optval, unsigned int optlen)
 {
@@ -64,3 +66,29 @@ typedef int socket_t;
 #  define ur_close close
 
 #endif  //  _WIN32
+
+#ifndef MSG_NOSIGNAL
+#  define MSG_NOSIGNAL 0
+#endif
+
+/*!
+ * \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::error_code(WSAGetLastError(), std::system_category());
+#else
+  return std::error_code(errno, std::generic_category());
+#endif
+}
+
+inline std::system_error makeSocketError(const std::string& message)
+{
+  return std::system_error(getLastSocketErrorCode(), message);
+}

include/ur_client_library/comm/tcp_server.h

@@ -34,6 +34,7 @@
 #include <cstddef>
 #include <cstdint>
 #include <functional>
+#include <mutex>
 #include <thread>
 #include <vector>
 
@@ -79,9 +80,13 @@ class TCPServer
    *
    * \param func Function handling the event information. The file descriptor created by the
    * connection event will be passed to the function.
+   *
+   * \note: The connection callback will be triggered with the socket being accepted. Hence, it
+   * is possible to send data from the connection callback directly.
    */
   void setConnectCallback(std::function<void(const socket_t)> func)
   {
+    std::lock_guard<std::mutex> lk(callback_mutex_);
     new_connection_callback_ = func;
   }
 
@@ -90,9 +95,13 @@ class TCPServer
    *
    * \param func Function handling the event information. The file descriptor created by the
    * connection event will be passed to the function.
+   *
+   * \note: The socket will already be closed when the disconnect callback is triggered, thus
+   * trying to interact with the socket from the disconnect callback will fail.
    */
   void setDisconnectCallback(std::function<void(const socket_t)> func)
   {
+    std::lock_guard<std::mutex> lk(callback_mutex_);
     disconnect_callback_ = func;
   }
 
@@ -104,6 +113,7 @@ class TCPServer
    */
   void setMessageCallback(std::function<void(const socket_t, char*, int)> func)
   {
+    std::lock_guard<std::mutex> lk(message_mutex_);
     message_callback_ = func;
   }
 
@@ -116,9 +126,11 @@ 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.
+   *
+   * \note: This should not be called from within any of the registered callback functions, as
+   * it will cause a deadlock. If you want to shutdown the server from a callback, you can e.g.
+   * start a new thread that calls shutdown() from there.
    */
   void shutdown();
 
@@ -135,6 +147,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
    *
@@ -182,15 +209,15 @@ class TCPServer
   void handleDisconnect(const socket_t fd);
 
   //! read data from socket
-  void readData(const socket_t fd);
+  bool readData(const socket_t fd);
 
   //! Event handler. Blocks until activity on any client or connection attempt
   void spin();
 
   //! Runs spin() as long as keep_running_ is set to true.
   void worker();
 
-  std::atomic<bool> keep_running_;
+  std::atomic<bool> keep_running_{ false };
   std::thread worker_thread_;
 
   std::atomic<socket_t> listen_fd_;
@@ -202,6 +229,10 @@ class TCPServer
 
   uint32_t max_clients_allowed_;
   std::vector<socket_t> client_fds_;
+  std::mutex clients_mutex_;
+  std::mutex message_mutex_;
+  std::mutex listen_fd_mutex_;
+  std::mutex callback_mutex_;
 
   static const int INPUT_BUFFER_SIZE = 4096;
   char input_buffer_[INPUT_BUFFER_SIZE];

include/ur_client_library/log.h

@@ -88,5 +88,4 @@ void setLogLevel(LogLevel level);
  * \param fmt Format string
  */
 void log(const char* file, int line, LogLevel level, const char* fmt, ...);
-
 }  // namespace urcl