More work on the scheduler for windows

- Get a rump derivation goal: hook instance will come later, local
  derivation goal will come after that.

- Start cleaning up the channel / waiting code with an abstraction.

src/libutil/muxable-pipe.hh

@@ -0,0 +1,82 @@
+#pragma once
+///@file
+
+#include "file-descriptor.hh"
+#ifdef _WIN32
+#  include "windows-async-pipe.hh"
+#endif
+
+#ifndef _WIN32
+#  include <poll.h>
+#else
+#  include <ioapiset.h>
+#  include "windows-error.hh"
+#endif
+
+namespace nix {
+
+/**
+ * An "muxable pipe" is a type of pipe supporting endpoints that wait
+ * for events on multiple pipes at once.
+ *
+ * On Unix, this is just a regular anonymous pipe. On Windows, this has
+ * to be a named pipe because we need I/O Completion Ports to wait on
+ * multiple pipes.
+ */
+using MuxablePipe =
+#ifndef _WIN32
+    Pipe
+#else
+    windows::AsyncPipe
+#endif
+    ;
+
+/**
+ * Use pool() (Unix) / I/O Completion Ports (Windows) to wait for the
+ * input side of any logger pipe to become `available'.  Note that
+ * `available' (i.e., non-blocking) includes EOF.
+ */
+struct MuxablePipePollState
+{
+#ifndef _WIN32
+    std::vector<struct pollfd> pollStatus;
+    std::map<int, size_t> fdToPollStatus;
+#else
+    OVERLAPPED_ENTRY oentries[0x20] = {0};
+    ULONG removed;
+    bool gotEOF = false;
+
+#endif
+
+    /**
+     * Check for ready (Unix) / completed (Windows) operations
+     */
+    void poll(
+#ifdef _WIN32
+        HANDLE ioport,
+#endif
+        std::optional<unsigned int> timeout);
+
+    using CommChannel =
+#ifndef _WIN32
+        Descriptor
+#else
+        windows::AsyncPipe *
+#endif
+        ;
+
+    /**
+     * Process for ready (Unix) / completed (Windows) operations,
+     * calling the callbacks as needed.
+     *
+     * @param handleRead callback to be passed read data.
+     *
+     * @param handleEOF callback for when the `MuxablePipe` has closed.
+     */
+    void iterate(
+        std::set<CommChannel> & channels,
+        std::function<void(Descriptor fd, std::string_view data)> handleRead,
+        std::function<void(Descriptor fd)> handleEOF);
+};
+
+}

src/libutil/processes.hh

@@ -118,8 +118,6 @@ public:
     { }
 };
 
-#ifndef _WIN32
-
 /**
  * Convert the exit status of a child as returned by wait() into an
  * error string.
@@ -128,6 +126,4 @@ std::string statusToString(int status);
 
 bool statusOk(int status);
 
-#endif
-
 }

src/libutil/unix/muxable-pipe.cc

@@ -0,0 +1,47 @@
+#include <poll.h>
+
+#include "logging.hh"
+#include "util.hh"
+#include "muxable-pipe.hh"
+
+namespace nix {
+
+void MuxablePipePollState::poll(std::optional<unsigned int> timeout)
+{
+    if (::poll(pollStatus.data(), pollStatus.size(), timeout ? *timeout : -1) == -1) {
+        if (errno == EINTR)
+            return;
+        throw SysError("waiting for input");
+    }
+}
+
+void MuxablePipePollState::iterate(
+    std::set<MuxablePipePollState::CommChannel> & channels,
+    std::function<void(Descriptor fd, std::string_view data)> handleRead,
+    std::function<void(Descriptor fd)> handleEOF)
+{
+    std::set<Descriptor> fds2(channels);
+    std::vector<unsigned char> buffer(4096);
+    for (auto & k : fds2) {
+        const auto fdPollStatusId = get(fdToPollStatus, k);
+        assert(fdPollStatusId);
+        assert(*fdPollStatusId < pollStatus.size());
+        if (pollStatus.at(*fdPollStatusId).revents) {
+            ssize_t rd = ::read(fromDescriptorReadOnly(k), buffer.data(), buffer.size());
+            // FIXME: is there a cleaner way to handle pt close
+            // than EIO? Is this even standard?
+            if (rd == 0 || (rd == -1 && errno == EIO)) {
+                handleEOF(k);
+                channels.erase(k);
+            } else if (rd == -1) {
+                if (errno != EINTR)
+                    throw SysError("read failed");
+            } else {
+                std::string_view data((char *) buffer.data(), rd);
+                handleRead(k, data);
+            }
+        }
+    }
+}
+
+}

src/libutil/windows/muxable-pipe.cc

@@ -0,0 +1,70 @@
+#include <ioapiset.h>
+#include "windows-error.hh"
+
+#include "logging.hh"
+#include "util.hh"
+#include "muxable-pipe.hh"
+
+namespace nix {
+
+void MuxablePipePollState::poll(HANDLE ioport, std::optional<unsigned int> timeout)
+{
+    /* We are on at least Windows Vista / Server 2008 and can get many
+       (countof(oentries)) statuses in one API call. */
+    if (!GetQueuedCompletionStatusEx(
+            ioport, oentries, sizeof(oentries) / sizeof(*oentries), &removed, timeout ? *timeout : INFINITE, false)) {
+        windows::WinError winError("GetQueuedCompletionStatusEx");
+        if (winError.lastError != WAIT_TIMEOUT)
+            throw winError;
+        assert(removed == 0);
+    } else {
+        assert(0 < removed && removed <= sizeof(oentries) / sizeof(*oentries));
+    }
+}
+
+void MuxablePipePollState::iterate(
+    std::set<MuxablePipePollState::CommChannel> & channels,
+    std::function<void(Descriptor fd, std::string_view data)> handleRead,
+    std::function<void(Descriptor fd)> handleEOF)
+{
+    auto p = channels.begin();
+    while (p != channels.end()) {
+        decltype(p) nextp = p;
+        ++nextp;
+        for (ULONG i = 0; i < removed; i++) {
+            if (oentries[i].lpCompletionKey == ((ULONG_PTR) ((*p)->readSide.get()) ^ 0x5555)) {
+                printMsg(lvlVomit, "read %s bytes", oentries[i].dwNumberOfBytesTransferred);
+                if (oentries[i].dwNumberOfBytesTransferred > 0) {
+                    std::string data{
+                        (char *) (*p)->buffer.data(),
+                        oentries[i].dwNumberOfBytesTransferred,
+                    };
+                    handleRead((*p)->readSide.get(), data);
+                }
+
+                if (gotEOF) {
+                    handleEOF((*p)->readSide.get());
+                    nextp = channels.erase(p); // no need to maintain `channels`?
+                } else {
+                    BOOL rc = ReadFile(
+                        (*p)->readSide.get(), (*p)->buffer.data(), (*p)->buffer.size(), &(*p)->got, &(*p)->overlapped);
+                    if (rc) {
+                        // here is possible (but not obligatory) to call
+                        // `handleRead` and repeat ReadFile immediately
+                    } else {
+                        windows::WinError winError("ReadFile(%s, ..)", (*p)->readSide.get());
+                        if (winError.lastError == ERROR_BROKEN_PIPE) {
+                            handleEOF((*p)->readSide.get());
+                            nextp = channels.erase(p); // no need to maintain `channels` ?
+                        } else if (winError.lastError != ERROR_IO_PENDING)
+                            throw winError;
+                    }
+                }
+                break;
+            }
+        }
+        p = nextp;
+    }
+}
+
+}

src/libutil/windows/processes.cc

@@ -16,16 +16,6 @@
 #include <sys/types.h>
 #include <unistd.h>
 
-#ifdef __APPLE__
-# include <sys/syscall.h>
-#endif
-
-#ifdef __linux__
-# include <sys/prctl.h>
-# include <sys/mman.h>
-#endif
-
-
 namespace nix {
 
 std::string runProgram(Path program, bool lookupPath, const Strings & args,
@@ -34,15 +24,31 @@ std::string runProgram(Path program, bool lookupPath, const Strings & args,
     throw UnimplementedError("Cannot shell out to git on Windows yet");
 }
 
+
 // Output = error code + "standard out" output stream
 std::pair<int, std::string> runProgram(RunOptions && options)
 {
     throw UnimplementedError("Cannot shell out to git on Windows yet");
 }
 
+
 void runProgram2(const RunOptions & options)
 {
     throw UnimplementedError("Cannot shell out to git on Windows yet");
 }
 
+std::string statusToString(int status)
+{
+    if (status != 0)
+        return fmt("with exit code %d", status);
+    else
+        return "succeeded";
+}
+
+
+bool statusOk(int status)
+{
+    return status == 0;
+}
+
 }

src/libutil/windows/windows-async-pipe.hh

@@ -5,6 +5,13 @@
 
 namespace nix::windows {
 
+/***
+ * An "async pipe" is a pipe that supports I/O Completion Ports so
+ * multiple pipes can be listened too.
+ *
+ * Unfortunately, only named pipes support that on windows, so we use
+ * those with randomized temp file names.
+ */
 class AsyncPipe
 {
 public: