From e5ffe833a9b4f45005c4c1484d8568b845211fce Mon Sep 17 00:00:00 2001
From: Carlos <carlos@carlossegarra.com>
Date: Tue, 16 Apr 2024 10:12:31 +0100
Subject: [PATCH] tests(tcp): add test for blocking recv sockets (#420)

Add test for blocking RECV TCP sockets and add a catch for when we are
trying to receive from a client that has already disconnected.
---
 src/transport/tcp/RecvSocket.cpp          |  7 ++++++
 src/transport/tcp/SocketOptions.cpp       |  2 +-
 tasks/dev.py                              |  4 ++++
 tests/test/transport/test_tcp_sockets.cpp | 28 ++++++++++++++++++++---
 4 files changed, 37 insertions(+), 4 deletions(-)

diff --git a/src/transport/tcp/RecvSocket.cpp b/src/transport/tcp/RecvSocket.cpp
index 49d72fb33..d979b4d67 100644
--- a/src/transport/tcp/RecvSocket.cpp
+++ b/src/transport/tcp/RecvSocket.cpp
@@ -139,6 +139,13 @@ void RecvSocket::recvOne(int conn, uint8_t* buffer, size_t bufferSize)
                          std::strerror(errno));
             throw std::runtime_error("TCP error receiving!");
         }
+#ifndef FAABRIC_USE_SPINLOCK
+        if (got == 0 && bufferSize != 0) {
+            SPDLOG_ERROR(
+              "TCP socket trying to receive from disconnected client");
+            throw std::runtime_error("TCP socket client disconnected");
+        }
+#endif
 
         buffer += got;
         numRecvd += got;
diff --git a/src/transport/tcp/SocketOptions.cpp b/src/transport/tcp/SocketOptions.cpp
index 55c7f17fc..5dbb40764 100644
--- a/src/transport/tcp/SocketOptions.cpp
+++ b/src/transport/tcp/SocketOptions.cpp
@@ -108,7 +108,7 @@ void setRecvTimeoutMs(int connFd, int timeoutMs)
 {
     struct timeval timeVal;
     timeVal.tv_sec = timeoutMs / 1000;
-    timeVal.tv_usec = 0;
+    timeVal.tv_usec = (timeoutMs % 1000) * 1e3;
 
     int ret = ::setsockopt(
       connFd, SOL_SOCKET, SO_RCVTIMEO, (const char*)&timeVal, sizeof(timeVal));
diff --git a/tasks/dev.py b/tasks/dev.py
index e72823144..ad17b6897 100644
--- a/tasks/dev.py
+++ b/tasks/dev.py
@@ -23,6 +23,7 @@ def cmake(
     coverage=False,
     prof=False,
     cpu=None,
+    disable_spinlock=False,
 ):
     """
     Configures the build
@@ -52,6 +53,9 @@ def cmake(
         "-DFAABRIC_USE_SANITISER={}".format(sanitiser),
         "-DFAABRIC_SELF_TRACING=ON" if prof else "",
         "-DFAABRIC_CODE_COVERAGE=ON" if coverage else "",
+        "-DFAABRIC_USE_SPINLOCK={}".format(
+            "OFF" if disable_spinlock else "ON"
+        ),
         "-DFAABRIC_TARGET_CPU={}".format(cpu) if cpu else "",
         PROJ_ROOT,
     ]
diff --git a/tests/test/transport/test_tcp_sockets.cpp b/tests/test/transport/test_tcp_sockets.cpp
index 2b675392e..de271a5ff 100644
--- a/tests/test/transport/test_tcp_sockets.cpp
+++ b/tests/test/transport/test_tcp_sockets.cpp
@@ -96,8 +96,10 @@ TEST_CASE("Test setting socket options", "[transport]")
 TEST_CASE("Test send/recv one message using raw TCP sockets", "[transport]")
 {
     RecvSocket dst(TEST_PORT);
+    std::latch endLatch(2);
 
     std::vector<int> msg;
+    bool timeout = false;
 
     SECTION("Small message")
     {
@@ -109,13 +111,23 @@ TEST_CASE("Test send/recv one message using raw TCP sockets", "[transport]")
         msg = std::vector<int>(300, 200);
     }
 
+    SECTION("Do not send a message")
+    {
+        msg = std::vector<int>(3, 2);
+        timeout = true;
+    }
+
     // Start sending socket thread
     std::jthread sendThread([&] {
         SendSocket src(LOCALHOST, TEST_PORT);
         // Connect to receiving socket
         src.dial();
 
-        src.sendOne(BYTES(msg.data()), sizeof(int) * msg.size());
+        if (!timeout) {
+            src.sendOne(BYTES(msg.data()), sizeof(int) * msg.size());
+        }
+
+        endLatch.arrive_and_wait();
     });
 
     // Prepare receiving socket
@@ -123,9 +135,19 @@ TEST_CASE("Test send/recv one message using raw TCP sockets", "[transport]")
     int conn = dst.accept();
 
     std::vector<int> actual(msg.size());
-    dst.recvOne(conn, BYTES(actual.data()), sizeof(int) * actual.size());
 
-    REQUIRE(actual == msg);
+    // To test the timeout we must set the socket as blocking
+    if (timeout) {
+        setRecvTimeoutMs(conn, 200);
+        REQUIRE_THROWS(
+          dst.recvOne(conn, BYTES(actual.data()), sizeof(int) * actual.size()));
+    } else {
+        dst.recvOne(conn, BYTES(actual.data()), sizeof(int) * actual.size());
+
+        REQUIRE(actual == msg);
+    }
+
+    endLatch.arrive_and_wait();
 }
 
 TEST_CASE("Test using scatter send functionality (sendMany)", "[transport]")