transport(tcp): set timeout on send sockets (#419)

include/faabric/transport/tcp/SocketOptions.h

@@ -14,5 +14,6 @@ bool isNonBlocking(int connFd);
 void setBusyPolling(int connFd);
 
 // Set timeout for blocking sockets
-void setTimeoutMs(int connFd, int timeoutMs);
+void setRecvTimeoutMs(int connFd, int timeoutMs);
+void setSendTimeoutMs(int connFd, int timeoutMs);
 }

src/transport/tcp/RecvSocket.cpp

@@ -111,7 +111,7 @@ void RecvSocket::setSocketOptions(int connFd)
     }
 
     // Set the timeout
-    setTimeoutMs(connFd, SocketTimeoutMs);
+    setRecvTimeoutMs(connFd, SocketTimeoutMs);
 #endif
 }
 

src/transport/tcp/SendSocket.cpp

@@ -20,6 +20,7 @@ void SendSocket::setSocketOptions(int connFd)
 {
     setNoDelay(connFd);
     setQuickAck(connFd);
+    setSendTimeoutMs(connFd, SocketTimeoutMs);
 }
 
 void SendSocket::dial()
@@ -67,7 +68,7 @@ void SendSocket::dial()
 
 void SendSocket::sendOne(const uint8_t* buffer, size_t bufferSize)
 {
-    size_t sent = send(sock.get(), buffer, bufferSize, 0);
+    size_t sent = ::send(sock.get(), buffer, bufferSize, 0);
     if (sent != bufferSize) {
         SPDLOG_ERROR(
           "TCP client error sending TCP message to {}:{} ({}/{}): {}",

src/transport/tcp/SocketOptions.cpp

@@ -104,7 +104,7 @@ void setBusyPolling(int connFd)
     }
 }
 
-void setTimeoutMs(int connFd, int timeoutMs)
+void setRecvTimeoutMs(int connFd, int timeoutMs)
 {
     struct timeval timeVal;
     timeVal.tv_sec = timeoutMs / 1000;
@@ -119,4 +119,20 @@ void setTimeoutMs(int connFd, int timeoutMs)
         throw std::runtime_error("Error setting recv timeout");
     }
 }
+
+void setSendTimeoutMs(int connFd, int timeoutMs)
+{
+    struct timeval timeVal;
+    timeVal.tv_sec = timeoutMs / 1000;
+    timeVal.tv_usec = 0;
+
+    int ret = ::setsockopt(
+      connFd, SOL_SOCKET, SO_SNDTIMEO, (const char*)&timeVal, sizeof(timeVal));
+    if (ret == -1) {
+        SPDLOG_ERROR("Error setting send timeout for socket {}: {}",
+                     connFd,
+                     std::strerror(errno));
+        throw std::runtime_error("Error setting send timeout");
+    }
+}
 }

tests/test/transport/test_tcp_sockets.cpp

@@ -60,15 +60,16 @@ TEST_CASE("Test setting socket options", "[transport]")
         dst.listen();
         conn = dst.accept();
 
-        // Set options on the socket given by accept (i.e. the send socket)
+        // Set options on the socket given by accept (i.e. the recv socket)
         setReuseAddr(conn);
         setNoDelay(conn);
         setQuickAck(conn);
         setQuickAck(conn);
         setBusyPolling(conn);
         setNonBlocking(conn);
         setBlocking(conn);
-        setTimeoutMs(conn, SocketTimeoutMs);
+        setRecvTimeoutMs(conn, SocketTimeoutMs);
+        setSendTimeoutMs(conn, SocketTimeoutMs);
 
         REQUIRE(!isNonBlocking(conn));
 
@@ -88,7 +89,8 @@ TEST_CASE("Test setting socket options", "[transport]")
     REQUIRE_THROWS(setBusyPolling(conn));
     REQUIRE_THROWS(setNonBlocking(conn));
     REQUIRE_THROWS(setBlocking(conn));
-    REQUIRE_THROWS(setTimeoutMs(conn, SocketTimeoutMs));
+    REQUIRE_THROWS(setRecvTimeoutMs(conn, SocketTimeoutMs));
+    REQUIRE_THROWS(setSendTimeoutMs(conn, SocketTimeoutMs));
 }
 
 TEST_CASE("Test send/recv one message using raw TCP sockets", "[transport]")