Centralize I/O error handling and make read/write functions portable

src/libmain/unix/stack.cc

@@ -73,7 +73,7 @@ std::function<void(siginfo_t * info, void * ctx)> stackOverflowHandler(defaultSt
 void defaultStackOverflowHandler(siginfo_t * info, void * ctx)
 {
     char msg[] = "error: stack overflow (possible infinite recursion)\n";
-    [[gnu::unused]] auto res = write(2, msg, strlen(msg));
+    [[gnu::unused]] auto res = ::write(2, msg, strlen(msg));
     _exit(1); // maybe abort instead?
 }
 

src/libutil/error.cc

@@ -422,13 +422,20 @@ std::ostream & showErrorInfo(std::ostream & out, const ErrorInfo & einfo, bool s
  */
 static void writeErr(std::string_view buf)
 {
+    Descriptor fd = getStandardError();
     while (!buf.empty()) {
-        auto n = write(STDERR_FILENO, buf.data(), buf.size());
+#ifdef _WIN32
+        DWORD n;
+        if (!WriteFile(fd, buf.data(), buf.size(), &n, NULL))
+            abort();
+#else
+        auto n = ::write(fd, buf.data(), buf.size());
         if (n < 0) {
             if (errno == EINTR)
                 continue;
             abort();
         }
+#endif
         buf = buf.substr(n);
     }
 }

src/libutil/file-descriptor.cc

@@ -8,10 +8,102 @@
 #ifdef _WIN32
 #  include <winnt.h>
 #  include <fileapi.h>
+#else
+#  include <poll.h>
 #endif
 
 namespace nix {
 
+namespace {
+
+enum class PollDirection { In, Out };
+
+/**
+ * Retry an I/O operation if it fails with EAGAIN/EWOULDBLOCK.
+ *
+ * On Unix, polls the fd and retries. On Windows, just calls `f` once.
+ *
+ * This retry logic is needed to handle non-blocking reads/writes. This
+ * is needed in the buildhook, because somehow the json logger file
+ * descriptor ends up being non-blocking and breaks remote-building.
+ *
+ * @todo Get rid of buildhook and remove this logic again
+ * (https://github.com/NixOS/nix/issues/12688)
+ */
+template<typename F>
+auto retryOnBlock([[maybe_unused]] Descriptor fd, [[maybe_unused]] PollDirection dir, F && f) -> decltype(f())
+{
+#ifndef _WIN32
+    while (true) {
+        try {
+            return std::forward<F>(f)();
+        } catch (SystemError & e) {
+            if (e.is(std::errc::resource_unavailable_try_again) || e.is(std::errc::operation_would_block)) {
+                struct pollfd pfd;
+                pfd.fd = fd;
+                pfd.events = dir == PollDirection::In ? POLLIN : POLLOUT;
+                if (poll(&pfd, 1, -1) == -1)
+                    throw SysError("poll on file descriptor failed");
+                continue;
+            }
+            throw;
+        }
+    }
+#else
+    return std::forward<F>(f)();
+#endif
+}
+
+} // namespace
+
+void readFull(Descriptor fd, char * buf, size_t count)
+{
+    while (count) {
+        checkInterrupt();
+        auto res = retryOnBlock(
+            fd, PollDirection::In, [&]() { return read(fd, {reinterpret_cast<std::byte *>(buf), count}); });
+        if (res == 0)
+            throw EndOfFile("unexpected end-of-file");
+        count -= res;
+        buf += res;
+    }
+}
+
+std::string readLine(Descriptor fd, bool eofOk, char terminator)
+{
+    std::string s;
+    while (1) {
+        checkInterrupt();
+        char ch;
+        // FIXME: inefficient
+        auto rd =
+            retryOnBlock(fd, PollDirection::In, [&]() { return read(fd, {reinterpret_cast<std::byte *>(&ch), 1}); });
+        if (rd == 0) {
+            if (eofOk)
+                return s;
+            else
+                throw EndOfFile("unexpected EOF reading a line");
+        } else {
+            if (ch == terminator)
+                return s;
+            s += ch;
+        }
+    }
+}
+
+void writeFull(Descriptor fd, std::string_view s, bool allowInterrupts)
+{
+    while (!s.empty()) {
+        if (allowInterrupts)
+            checkInterrupt();
+        auto res = retryOnBlock(fd, PollDirection::Out, [&]() {
+            return write(fd, {reinterpret_cast<const std::byte *>(s.data()), s.size()});
+        });
+        if (res > 0)
+            s.remove_prefix(res);
+    }
+}
+
 void writeLine(Descriptor fd, std::string s)
 {
     s += '\n';
@@ -67,8 +159,6 @@ void drainFD(Descriptor fd, Sink & sink, DrainFdSinkOpts opts)
                 && (e.is(std::errc::resource_unavailable_try_again) || e.is(std::errc::operation_would_block)))
                 break;
 #endif
-            if (e.is(std::errc::interrupted))
-                continue;
             throw;
         }
 
@@ -105,18 +195,8 @@ void copyFdRange(Descriptor fd, off_t offset, size_t nbytes, Sink & sink)
     std::array<std::byte, 64 * 1024> buf;
 
     while (left) {
-        checkInterrupt();
         auto limit = std::min<size_t>(left, buf.size());
-        /* Should be initialized before we read, because the `catch`
-           block either throws or continues. */
-        size_t n;
-        try {
-            n = readOffset(fd, offset, std::span(buf.data(), limit));
-        } catch (SystemError & e) {
-            if (e.is(std::errc::interrupted))
-                continue;
-            throw;
-        }
+        auto n = readOffset(fd, offset, std::span(buf.data(), limit));
         if (n == 0)
             throw EndOfFile("unexpected end-of-file");
         assert(n <= left);

src/libutil/include/nix/util/file-descriptor.hh

@@ -65,7 +65,7 @@ std::string readFile(Descriptor fd);
  * Platform-specific read into a buffer.
  *
  * Thin wrapper around ::read (Unix) or ReadFile (Windows).
- * Does NOT handle EINTR on Unix - caller must catch and retry if needed.
+ * Handles EINTR on Unix. Treats ERROR_BROKEN_PIPE as EOF on Windows.
  *
  * @param fd The file descriptor to read from
  * @param buffer The buffer to read into
@@ -74,6 +74,19 @@ std::string readFile(Descriptor fd);
  */
 size_t read(Descriptor fd, std::span<std::byte> buffer);
 
+/**
+ * Platform-specific write from a buffer.
+ *
+ * Thin wrapper around ::write (Unix) or WriteFile (Windows).
+ * Handles EINTR on Unix.
+ *
+ * @param fd The file descriptor to write to
+ * @param buffer The buffer to write from
+ * @return The number of bytes actually written
+ * @throws SystemError on failure
+ */
+size_t write(Descriptor fd, std::span<const std::byte> buffer);
+
 /**
  * Get the size of a file.
  *

src/libutil/serialise.cc

@@ -190,24 +190,7 @@ bool BufferedSource::hasData()
 
 size_t FdSource::readUnbuffered(char * data, size_t len)
 {
-#ifdef _WIN32
-    DWORD n;
-    checkInterrupt();
-    if (!::ReadFile(fd, data, len, &n, NULL)) {
-        _good = false;
-        throw windows::WinError("ReadFile when FdSource::readUnbuffered");
-    }
-#else
-    ssize_t n;
-    do {
-        checkInterrupt();
-        n = ::read(fd, data, len);
-    } while (n == -1 && errno == EINTR);
-    if (n == -1) {
-        _good = false;
-        throw SysError("reading from file");
-    }
-#endif
+    auto n = nix::read(fd, {reinterpret_cast<std::byte *>(data), len});
     if (n == 0) {
         _good = false;
         throw EndOfFile(std::string(*endOfFileError));

src/libutil/unix/file-descriptor.cc

@@ -5,129 +5,52 @@
 
 #include <fcntl.h>
 #include <unistd.h>
-#include <poll.h>
 #include <span>
 
 #include "util-config-private.hh"
 #include "util-unix-config-private.hh"
 
 namespace nix {
 
-namespace {
-
-// This function is needed to handle non-blocking reads/writes. This is needed in the buildhook, because
-// somehow the json logger file descriptor ends up being non-blocking and breaks remote-building.
-// TODO: get rid of buildhook and remove this function again (https://github.com/NixOS/nix/issues/12688)
-void pollFD(int fd, int events)
-{
-    struct pollfd pfd;
-    pfd.fd = fd;
-    pfd.events = events;
-    int ret = poll(&pfd, 1, -1);
-    if (ret == -1) {
-        throw SysError("poll on file descriptor failed");
-    }
-}
-} // namespace
-
 std::make_unsigned_t<off_t> getFileSize(Descriptor fd)
 {
     auto st = nix::fstat(fd);
     return st.st_size;
 }
 
-void readFull(int fd, char * buf, size_t count)
+size_t read(Descriptor fd, std::span<std::byte> buffer)
 {
-    while (count) {
+    ssize_t n;
+    do {
         checkInterrupt();
-        ssize_t res = ::read(fd, buf, count);
-        if (res == -1) {
-            switch (errno) {
-            case EINTR:
-                continue;
-            case EAGAIN:
-                pollFD(fd, POLLIN);
-                continue;
-            }
-            auto savedErrno = errno;
-            throw SysError(savedErrno, "reading from file %s", PathFmt(descriptorToPath(fd)));
-        }
-        if (res == 0)
-            throw EndOfFile("unexpected end-of-file");
-        count -= res;
-        buf += res;
-    }
-}
-
-void writeFull(int fd, std::string_view s, bool allowInterrupts)
-{
-    while (!s.empty()) {
-        if (allowInterrupts)
-            checkInterrupt();
-        ssize_t res = write(fd, s.data(), s.size());
-        if (res == -1) {
-            switch (errno) {
-            case EINTR:
-                continue;
-            case EAGAIN:
-                pollFD(fd, POLLOUT);
-                continue;
-            }
-            auto savedErrno = errno;
-            throw SysError(savedErrno, "writing to file %s", PathFmt(descriptorToPath(fd)));
-        }
-        if (res > 0)
-            s.remove_prefix(res);
-    }
+        n = ::read(fd, buffer.data(), buffer.size());
+    } while (n == -1 && errno == EINTR);
+    if (n == -1)
+        throw SysError("read of %1% bytes", buffer.size());
+    return static_cast<size_t>(n);
 }
 
-std::string readLine(int fd, bool eofOk, char terminator)
+size_t readOffset(Descriptor fd, off_t offset, std::span<std::byte> buffer)
 {
-    std::string s;
-    while (1) {
+    ssize_t n;
+    do {
         checkInterrupt();
-        char ch;
-        // FIXME: inefficient
-        ssize_t rd = ::read(fd, &ch, 1);
-        if (rd == -1) {
-            switch (errno) {
-            case EINTR:
-                continue;
-            case EAGAIN: {
-                pollFD(fd, POLLIN);
-                continue;
-            }
-            default: {
-                auto savedErrno = errno;
-                throw SysError(savedErrno, "reading a line from %s", PathFmt(descriptorToPath(fd)));
-            }
-            }
-        } else if (rd == 0) {
-            if (eofOk)
-                return s;
-            else
-                throw EndOfFile("unexpected EOF reading a line");
-        } else {
-            if (ch == terminator)
-                return s;
-            s += ch;
-        }
-    }
-}
-
-size_t read(Descriptor fd, std::span<std::byte> buffer)
-{
-    ssize_t n = ::read(fd, buffer.data(), buffer.size());
+        n = pread(fd, buffer.data(), buffer.size(), offset);
+    } while (n == -1 && errno == EINTR);
     if (n == -1)
-        throw SysError("read of %1% bytes", buffer.size());
+        throw SysError("pread of %1% bytes at offset %2%", buffer.size(), offset);
     return static_cast<size_t>(n);
 }
 
-size_t readOffset(Descriptor fd, off_t offset, std::span<std::byte> buffer)
+size_t write(Descriptor fd, std::span<const std::byte> buffer)
 {
-    ssize_t n = pread(fd, buffer.data(), buffer.size(), offset);
+    ssize_t n;
+    do {
+        checkInterrupt();
+        n = ::write(fd, buffer.data(), buffer.size());
+    } while (n == -1 && errno == EINTR);
     if (n == -1)
-        throw SysError("pread of %1% bytes at offset %2%", buffer.size(), offset);
+        throw SysError("write of %1% bytes", buffer.size());
     return static_cast<size_t>(n);
 }