macOS connection quality fixes

include/TCPClient.h

@@ -31,7 +31,7 @@ class TCPClient final : public ICommunicationClient {
   public:
     TCPClient(std::string ip,
               const std::shared_ptr<BlockingQueue<std::unique_ptr<std::vector<uint8_t>>>> &rx_queue)
-        : port{3000}, m_ip{std::move(ip)}, m_stop_flag(false),
+        : port{3001}, m_ip{std::move(ip)}, m_stop_flag(false),
           m_thread(std::thread(&TCPClient::rx_thread, this)), m_rx_queue(rx_queue) {
     }
     ~TCPClient() override;

src/TCPClient.cpp

@@ -4,6 +4,7 @@
 
 #include <chrono>
 #include <iostream>
+#include <thread>
 #include <vector>
 
 #include "TCPClient.h"
@@ -16,6 +17,8 @@ constexpr int PORT = 3001;
 constexpr auto QUEUE_ADD_TIMEOUT = std::chrono::milliseconds(100);
 constexpr auto RX_SLEEP_ON_ERROR = std::chrono::milliseconds(100);
 constexpr auto SOCKET_TIMEOUT_MS = 2500;
+constexpr int CONNECT_MAX_RETRIES = 5;
+constexpr auto CONNECT_RETRY_DELAY = std::chrono::milliseconds(500);
 
 // todo: - add authentication
 //       - encryption
@@ -55,9 +58,36 @@ int TCPClient::init() {
     setsockopt(this->m_socket, SOL_SOCKET, SO_SNDTIMEO, &timeout, sizeof(timeout));
 #endif
 
-    if (connect(this->m_socket, reinterpret_cast<sockaddr *>(&serv_addr), sizeof(serv_addr)) < 0) {
+    bool connected = false;
-        spdlog::error("[TCP] Connection failed to connect");
+    for (int attempt = 0; attempt < CONNECT_MAX_RETRIES; ++attempt) {
+        if (attempt > 0) {
+            CLOSE_SOCKET(this->m_socket);
+            if ((this->m_socket = socket(AF_INET, SOCK_STREAM, 0)) < 0) {
+                spdlog::error("[TCP] Failed to recreate socket on retry");
+                return -2;
+            }
+#ifdef _WIN32
+            setsockopt(this->m_socket, SOL_SOCKET, SO_RCVTIMEO, (char *)&timeout, sizeof(timeout));
+            setsockopt(this->m_socket, SOL_SOCKET, SO_SNDTIMEO, (char *)&timeout, sizeof(timeout));
+#else
+            setsockopt(this->m_socket, SOL_SOCKET, SO_RCVTIMEO, &timeout, sizeof(timeout));
+            setsockopt(this->m_socket, SOL_SOCKET, SO_SNDTIMEO, &timeout, sizeof(timeout));
+#endif
+        }
+
+        if (connect(this->m_socket, reinterpret_cast<sockaddr *>(&serv_addr), sizeof(serv_addr)) ==
+            0) {
+            connected = true;
+            break;
+        }
+        spdlog::warn("[TCP] Connection attempt {}/{} failed, retrying...", attempt + 1,
+                     CONNECT_MAX_RETRIES);
         print_errno();
+        std::this_thread::sleep_for(CONNECT_RETRY_DELAY);
+    }
+
+    if (!connected) {
+        spdlog::error("[TCP] Connection failed to connect after {} attempts", CONNECT_MAX_RETRIES);
         deinit();
         return -1;
     }

src/UDPClient.cpp

@@ -50,7 +50,9 @@ int UDPClient::init() {
     setsockopt(m_tx_socket, SOL_SOCKET, SO_REUSEADDR, (char *)&opt, sizeof(opt));
 #else
     setsockopt(m_rx_socket, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt));
+    setsockopt(m_rx_socket, SOL_SOCKET, SO_REUSEPORT, &opt, sizeof(opt));
     setsockopt(m_tx_socket, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt));
+    setsockopt(m_tx_socket, SOL_SOCKET, SO_REUSEPORT, &opt, sizeof(opt));
 #endif
 
     timeval timeout{};
@@ -80,7 +82,7 @@ int UDPClient::init() {
         return -1;
     }
 
-    ip_mreq mreq;
+    ip_mreq mreq{};
     mreq.imr_multiaddr.s_addr = inet_addr(RECV_MCAST.c_str());
     mreq.imr_interface.s_addr = INADDR_ANY;
 
@@ -106,7 +108,8 @@ int UDPClient::init() {
 
     in_addr tx_iface{};
     tx_iface.s_addr = mreq.imr_interface.s_addr;
-    if (setsockopt(m_tx_socket, IPPROTO_IP, IP_MULTICAST_IF, (char *)&tx_iface, sizeof(tx_iface)) < 0) {
+    if (setsockopt(m_tx_socket, IPPROTO_IP, IP_MULTICAST_IF, (char *)&tx_iface, sizeof(tx_iface)) <
+        0) {
         spdlog::error("[UDP] Failed to set multicast TX interface");
         print_errno();
         deinit();
@@ -115,7 +118,8 @@ int UDPClient::init() {
 
     // Set multicast TTL > 1 so packets leave the local subnet if needed (default is 1).
     constexpr int mcast_ttl = 32;
-    if (setsockopt(m_tx_socket, IPPROTO_IP, IP_MULTICAST_TTL, (char *)&mcast_ttl, sizeof(mcast_ttl)) < 0) {
+    if (setsockopt(m_tx_socket, IPPROTO_IP, IP_MULTICAST_TTL, (char *)&mcast_ttl,
+                   sizeof(mcast_ttl)) < 0) {
         spdlog::warn("[UDP] Failed to set multicast TTL");
         print_errno();
     }
@@ -126,6 +130,27 @@ int UDPClient::init() {
         deinit();
         return -1;
     }
+
+    in_addr tx_iface{};
+    tx_iface.s_addr = mreq.imr_interface.s_addr; // INADDR_ANY lets the OS pick
+    if (setsockopt(m_tx_socket, IPPROTO_IP, IP_MULTICAST_IF, &tx_iface, sizeof(tx_iface)) < 0) {
+        spdlog::error("[UDP] Failed to set multicast TX interface");
+        print_errno();
+        deinit();
+        return -1;
+    }
+
+    constexpr int mcast_ttl = 32;
+    if (setsockopt(m_tx_socket, IPPROTO_IP, IP_MULTICAST_TTL, &mcast_ttl, sizeof(mcast_ttl)) < 0) {
+        spdlog::warn("[UDP] Failed to set multicast TTL");
+        print_errno();
+    }
+
+    constexpr int loop = 0;
+    if (setsockopt(m_tx_socket, IPPROTO_IP, IP_MULTICAST_LOOP, &loop, sizeof(loop)) < 0) {
+        spdlog::warn("[UDP] Failed to disable multicast loopback");
+        print_errno();
+    }
 #endif
 
     this->m_initialized = true;
@@ -158,18 +183,17 @@ int UDPClient::send_msg(void *sendbuff, const uint32_t len) {
     *reinterpret_cast<uint32_t *>(buffer.data()) = static_cast<uint32_t>(len);
     std::memcpy(buffer.data() + 4, sendbuff, len);
 
-    sockaddr_in mcast_dest = {
+    sockaddr_in mcast_dest{};
-        .sin_family = AF_INET,
+    mcast_dest.sin_family = AF_INET;
-        .sin_port = htons(TX_PORT),
+    mcast_dest.sin_port = htons(TX_PORT);
-    };
     inet_pton(AF_INET, SEND_MCAST.c_str(), &mcast_dest.sin_addr);
 
 #ifdef _WIN32
     return sendto(m_tx_socket, reinterpret_cast<const char *>(buffer.data()), buffer.size(), 0,
                   reinterpret_cast<sockaddr *>(&mcast_dest), sizeof(mcast_dest));
 #else
-    return sendto(m_tx_socket, buffer.data(), buffer.size(), 0,
+    return static_cast<int>(sendto(m_tx_socket, buffer.data(), buffer.size(), 0,
-                  reinterpret_cast<sockaddr *>(&mcast_dest), sizeof(mcast_dest));
+                                   reinterpret_cast<sockaddr *>(&mcast_dest), sizeof(mcast_dest)));
 #endif
 }
 

src/librpc.cpp

@@ -40,9 +40,9 @@ int MessagingInterface::send(uint8_t *buffer, const size_t size, const uint8_t d
     }
 
     Flatbuffers::MPIMessageBuilder builder;
-    const auto [mpi_buffer, mpi_size] = builder.build_mpi_message(
+    const auto [mpi_buffer, mpi_size] =
-        Messaging::MessageType_PTP, PC_MODULE_ID, destination, 0, durable, tag,
+        builder.build_mpi_message(Messaging::MessageType_PTP, PC_MODULE_ID, destination, 0, durable,
-        std::vector<uint8_t>(buffer, buffer + size));
+                                  tag, std::vector<uint8_t>(buffer, buffer + size));
 
     std::shared_lock lock(m_client_mutex);
     if (durable) {