transport(tcp): support multi-part sends (#421)

It is not an error for ::send or ::writev to write less bytes than the
buffer size, as long as the return value is a positive value.

In addition, EAGAIN and EWOULDBLOCK have different meanings for blocking
and non-blocking sockets.

include/faabric/transport/tcp/SendSocket.h

@@ -30,15 +30,21 @@ class SendSocket
             totalSize += sizeArray.at(i);
         }
 
-        auto nSent = ::writev(sock.get(), msgs.data(), N);
-        if (nSent != totalSize) {
-            SPDLOG_ERROR("Error sending many to {}:{} ({}/{}): {}",
-                         host,
-                         port,
-                         nSent,
-                         totalSize,
-                         std::strerror(errno));
-            throw std::runtime_error("TCP SendSocket sendMany error!");
+        size_t totalNumSent = 0;
+        while (totalNumSent < totalSize) {
+            auto nSent = ::writev(sock.get(), msgs.data(), N);
+            if (nSent == -1) {
+                SPDLOG_ERROR("Error sending many to {}:{} ({}/{}): {} (no: {})",
+                             host,
+                             port,
+                             nSent,
+                             totalSize,
+                             std::strerror(errno),
+                             errno);
+                throw std::runtime_error("TCP SendSocket sendMany error!");
+            }
+
+            totalNumSent += nSent;
         }
     }
 

src/transport/tcp/RecvSocket.cpp

@@ -133,19 +133,24 @@ void RecvSocket::recvOne(int conn, uint8_t* buffer, size_t bufferSize)
                 continue;
             }
 #endif
+            if (errno == EAGAIN || errno == EWOULDBLOCK) {
+                SPDLOG_ERROR("TCP recv socket timed-out receiving in {}", conn);
+            } else {
+                SPDLOG_ERROR("TCP recv socket error in {}: {} (no: {})",
+                             conn,
+                             std::strerror(errno),
+                             errno);
+            }
 
-            SPDLOG_ERROR("TCP Server error receiving in {}: {}",
-                         conn,
-                         std::strerror(errno));
             throw std::runtime_error("TCP error receiving!");
         }
-#ifndef FAABRIC_USE_SPINLOCK
+
+        // Handle peer disconnection separately
         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;

src/transport/tcp/SendSocket.cpp

@@ -68,16 +68,22 @@ void SendSocket::dial()
 
 void SendSocket::sendOne(const uint8_t* buffer, size_t bufferSize)
 {
-    size_t sent = ::send(sock.get(), buffer, bufferSize, 0);
-    if (sent != bufferSize) {
-        SPDLOG_ERROR(
-          "TCP client error sending TCP message to {}:{} ({}/{}): {}",
-          host,
-          port,
-          sent,
-          bufferSize,
-          std::strerror(errno));
-        throw std::runtime_error("TCP client error sending message!");
+    size_t totalNumSent = 0;
+
+    while (totalNumSent < bufferSize) {
+        size_t nSent = ::send(sock.get(), buffer, bufferSize, 0);
+        if (nSent == -1) {
+            SPDLOG_ERROR(
+              "TCP client error sending TCP message to {}:{} ({}/{}): {}",
+              host,
+              port,
+              nSent,
+              bufferSize,
+              std::strerror(errno));
+            throw std::runtime_error("TCP client error sending message!");
+        }
+
+        totalNumSent += nSent;
     }
 }
 }