report write errors, handle SIGPIPE

slsSupportLib/src/DataSocket.cpp

@@ -23,6 +23,14 @@ namespace sls {
 DataSocket::DataSocket(int socketId) : sockfd_(socketId) {
     int value = 1;
     setsockopt(sockfd_, SOL_SOCKET, SO_REUSEADDR, &value, sizeof(value));
+#ifdef SO_NOSIGPIPE
+    // macOS/BSD: suppress SIGPIPE when sending to a peer that closed the
+    // connection, so a failed send returns an error instead of killing the
+    // process. On Linux we instead pass MSG_NOSIGNAL to send() (see Send()).
+    int nosigpipe = 1;
+    setsockopt(sockfd_, SOL_SOCKET, SO_NOSIGPIPE, &nosigpipe,
+               sizeof(nosigpipe));
+#endif
 }
 
 DataSocket::~DataSocket() {
@@ -88,28 +96,56 @@ std::string DataSocket::Receive(size_t length) {
     return buff;
 }
 int DataSocket::Send(const void *buffer, size_t size) {
+    int bytes_expected = static_cast<int>(size); // signed size
     int bytes_sent = 0;
-    int data_size = static_cast<int>(size); // signed size
-    while (bytes_sent < (data_size)) {
-        auto this_send = ::write(getSocketId(), buffer, size);
+    ssize_t this_send = 0; // last write result, kept for diagnostics
+    // Linux: avoid SIGPIPE on a broken connection by using send() with
+    // MSG_NOSIGNAL. macOS/BSD lack the flag and use SO_NOSIGPIPE instead
+    // (set in the constructor).
+#ifdef MSG_NOSIGNAL
+    const int send_flags = MSG_NOSIGNAL;
+#else
+    const int send_flags = 0;
+#endif
+    Timer timer;
+    while (bytes_sent < bytes_expected) {
+        this_send = ::send(
+            getSocketId(),
+            reinterpret_cast<const char *>(buffer) + bytes_sent,
+            bytes_expected - bytes_sent, send_flags);
         if (this_send <= 0)
             break;
         bytes_sent += this_send;
     }
-    if (bytes_sent != data_size) {
+    if (bytes_sent == bytes_expected) {
+        return bytes_sent;
+    } else {
+        int err = errno; // capture before any other call can clobber it
         std::ostringstream ss;
         ss << "TCP socket sent " << bytes_sent << " bytes instead of "
-           << data_size << " bytes ("
+           << bytes_expected << " bytes ("
            << getFunctionNameFromEnum(static_cast<detFuncs>(fnum_)) << ')';
+        if (this_send == 0)
+            ss << ": 0 bytes sent";
+        else if (this_send < 0)
+            ss << ": write error:  " << std::strerror(err) << " ("
+               << errno_name(err) << ")";
+        ss << " after " << timer.elapsed_ms() << " ms";
         throw SocketError(ss.str());
     }
-    return bytes_sent;
 }
 
 int DataSocket::Send(const std::string &s) { return Send(&s[0], s.size()); }
 
 int DataSocket::write(void *buffer, size_t size) {
-    return ::write(getSocketId(), buffer, size);
+    // Use send() with MSG_NOSIGNAL (Linux) to avoid SIGPIPE on a broken
+    // connection; macOS/BSD rely on SO_NOSIGPIPE set in the constructor.
+#ifdef MSG_NOSIGNAL
+    const int send_flags = MSG_NOSIGNAL;
+#else
+    const int send_flags = 0;
+#endif
+    return ::send(getSocketId(), buffer, size, send_flags);
 }
 
 int DataSocket::read(void *buffer, size_t size) {

slsSupportLib/tests/test-Sockets.cpp

@@ -3,13 +3,17 @@
 #include "catch.hpp"
 #include "sls/ClientSocket.h"
 #include "sls/ServerSocket.h"
+#include "sls/Timer.h"
 #include "sls/sls_detector_defs.h"
 #include "sls/sls_detector_exceptions.h"
 #include "sls/sls_detector_funcs.h"
+#include <atomic>
 #include <chrono>
 #include <future>
 #include <iostream>
 #include <string>
+#include <sys/socket.h>
+#include <sys/time.h>
 #include <thread>
 #include <unistd.h>
 
@@ -85,6 +89,23 @@ void short_reply_server(uint16_t port, size_t retval_bytes_to_send) {
     s.close();
 }
 
+// Server that accepts a connection but never reads from it, so a client
+// trying to send more than fits in the kernel buffers will stall. A small
+// receive buffer keeps the amount the test must send modest. Stays open until
+// the client signals it is done (or a safety timeout) so it never closes
+// mid-transfer and races the client's Send.
+void non_reading_server(uint16_t port, std::atomic<bool> *client_done) {
+    auto server = ServerSocket(port);
+    auto s = server.accept();
+    int rcvbuf = 1024;
+    setsockopt(s.getSocketId(), SOL_SOCKET, SO_RCVBUF, &rcvbuf, sizeof(rcvbuf));
+    // Intentionally never Receive(): let the client's send path back up.
+    Timer t;
+    while (!client_done->load() && t.elapsed_ms() < 10000)
+        std::this_thread::sleep_for(std::chrono::milliseconds(10));
+    s.close();
+}
+
 TEST_CASE("The server recive the same message as we send", "[support]") {
     std::vector<char> received_message(100, '\0');
     std::vector<char> sent_message(100, '\0');
@@ -267,4 +288,47 @@ TEST_CASE("ServerSocket replies with a too short message", "[support]") {
                Catch::Matchers::Contains("connection closed by peer (EOF)"));
 }
 
+TEST_CASE("Client cannot send the expected number of bytes", "[support]") {
+    constexpr uint16_t port = 1963;
+
+    // Server accepts but never reads; it stays open until we tell it the
+    // client is done, so it cannot close mid-transfer.
+    std::atomic<bool> client_done{false};
+    auto s = std::async(std::launch::async, non_reading_server, port,
+                        &client_done);
+    std::this_thread::sleep_for(std::chrono::milliseconds(100));
+
+    auto client = DetectorSocket("localhost", port);
+
+    // Shrink the send buffer and add a short send timeout so the write stalls
+    // and returns before all the data is sent.
+    int sndbuf = 4096;
+    setsockopt(client.getSocketId(), SOL_SOCKET, SO_SNDBUF, &sndbuf,
+               sizeof(sndbuf));
+    struct timeval tv {};
+    tv.tv_sec = 0;
+    tv.tv_usec = 300000; // 300 ms
+    setsockopt(client.getSocketId(), SOL_SOCKET, SO_SNDTIMEO, &tv, sizeof(tv));
+
+    // Much more than fits in the (shrunken) send + receive buffers, so the
+    // send cannot complete while the server refuses to read.
+    std::vector<char> big_message(8 * 1024 * 1024, '\0');
+
+    std::string error_message;
+    try {
+        client.Send(big_message.data(), big_message.size());
+        FAIL("Send should have thrown when it could not send all bytes");
+    } catch (const SocketError &e) {
+        error_message = e.what();
+    }
+    client_done = true;
+    client.close();
+    s.get();
+
+    // Fewer bytes were sent than expected, reported as a write error (the
+    // send timed out with EAGAIN/EWOULDBLOCK).
+    CHECK_THAT(error_message, Catch::Matchers::Contains("bytes instead of"));
+    CHECK_THAT(error_message, Catch::Matchers::Contains("write error:"));
+}
+
 } // namespace sls