libutil: asyncify runProgram2 stdout stream

fun side effect: post-build hooks are no longer single-threaded!

Change-Id: Ie5c75e588e4b2b8d515d4f1ae36de45b0a0807f7

lix/libfetchers/git.cc

@@ -956,8 +956,7 @@ struct GitInputScheme : InputScheme
             });
             Finally const _wait([&] { proc.waitAndCheck(); });
 
-            AsyncSourceInputStream stdout{*proc.getStdout()};
-            TRY_AWAIT(unpackTarfile(stdout, tmpDir));
+            TRY_AWAIT(unpackTarfile(*proc.getStdout(), tmpDir));
         }
 
         auto storePath = TRY_AWAIT(

lix/libstore/build/derivation-goal.cc

@@ -1,4 +1,5 @@
 #include "lix/libstore/build/derivation-goal.hh"
+#include "lix/libutil/async-io.hh"
 #include "lix/libutil/async.hh"
 #include "lix/libutil/file-descriptor.hh"
 #include "lix/libutil/file-system.hh"
@@ -11,6 +12,7 @@
 #include "lix/libstore/common-protocol-impl.hh" // IWYU pragma: keep
 #include "lix/libstore/local-store.hh" // TODO remove, along with remaining downcasts
 #include "lix/libstore/build/substitution-goal.hh"
+#include "lix/libutil/logging.hh"
 #include "lix/libutil/result.hh"
 #include "lix/libutil/rpc.hh"
 #include "lix/libutil/strings.hh"
@@ -830,15 +832,14 @@ void DerivationGoal::cleanupPostOutputsRegisteredModeNonCheck()
 {
 }
 
-void runPostBuildHook(
-    Store & store,
-    Logger & logger,
-    const StorePath & drvPath,
-    const StorePathSet & outputPaths)
-{
+static kj::Promise<Result<void>> runPostBuildHook(
+    Store & store, Logger & logger, const StorePath & drvPath, const StorePathSet & outputPaths
+)
+try {
     auto hook = settings.postBuildHook;
-    if (hook == "")
-        return;
+    if (hook == "") {
+        co_return result::success();
+    }
 
     Activity act(
         logger,
@@ -860,7 +861,7 @@ void runPostBuildHook(
         .captureStdout = true,
         .redirections = {{.dup = STDERR_FILENO, .from = STDOUT_FILENO}},
     });
-    Finally const _wait([&] {
+    auto wait = kj::defer([&] {
         try {
             proc.waitAndCheck();
         } catch (nix::Error & e) {
@@ -876,26 +877,27 @@ void runPostBuildHook(
     auto & hookStdout = *proc.getStdout();
     std::string currentLine;
     std::vector<char> buffer(8192);
-    try {
-        while (true) {
-            const auto got = hookStdout.read(buffer.data(), buffer.size());
-            const std::string_view data{buffer.data(), got};
-            for (auto c : data) {
-                if (c == '\n') {
-                    act.result(resPostBuildLogLine, currentLine);
-                    currentLine.clear();
-                } else {
-                    currentLine += c;
-                }
+    while (const auto got = TRY_AWAIT(hookStdout.readRange(buffer.data(), 1, buffer.size()))) {
+        const std::string_view data{buffer.data(), *got};
+        for (auto c : data) {
+            if (c == '\n') {
+                act.result(resPostBuildLogLine, currentLine);
+                currentLine.clear();
+            } else {
+                currentLine += c;
             }
         }
-    } catch (EndOfFile &) {
     }
 
     if (currentLine != "") {
         currentLine += '\n';
         act.result(resPostBuildLogLine, currentLine);
     }
+
+    wait.run();
+    co_return result::success();
+} catch (...) {
+    co_return result::current_exception();
 }
 
 std::string DerivationGoal::buildErrorContents(const std::string & exitMsg, bool diskFull)
@@ -993,12 +995,7 @@ try {
         StorePathSet outputPaths;
         for (auto & [_, output] : builtOutputs)
             outputPaths.insert(output.outPath);
-        runPostBuildHook(
-            worker.store,
-            *logger,
-            drvPath,
-            outputPaths
-        );
+        TRY_AWAIT(runPostBuildHook(worker.store, *logger, drvPath, outputPaths));
 
         cleanupPostOutputsRegisteredModeNonCheck();
 

lix/libutil/processes.cc

@@ -256,7 +256,7 @@ try {
     try {
         auto proc = runProgram2(options);
         Finally const _wait([&] { proc.waitAndCheck(); });
-        stdout = proc.getStdout()->drain();
+        stdout = TRY_AWAIT(proc.getStdout()->drain());
     } catch (ExecError & e) {
         status = e.status;
     }
@@ -269,8 +269,7 @@ try {
 RunningProgram::RunningProgram(PathView program, Pid pid, AutoCloseFD stdout)
     : program(program)
     , pid(std::move(pid))
-    , stdoutSource(stdout ? std::make_unique<FdSource>(stdout.get()) : nullptr)
-    , stdout_(std::move(stdout))
+    , stdout(stdout ? std::make_unique<AsyncFdIoStream>(std::move(stdout)) : nullptr)
 {
 }
 
@@ -285,9 +284,9 @@ RunningProgram::~RunningProgram()
     }
 }
 
-std::tuple<pid_t, std::unique_ptr<Source>, int> RunningProgram::release()
+std::tuple<pid_t, std::unique_ptr<AsyncFdIoStream>> RunningProgram::release()
 {
-    return {pid.release(), std::move(stdoutSource), stdout_.release()};
+    return {pid.release(), std::move(stdout)};
 }
 
 int RunningProgram::kill()

lix/libutil/processes.hh

@@ -1,6 +1,7 @@
 #pragma once
 ///@file
 
+#include "lix/libutil/async-io.hh"
 #include "lix/libutil/result.hh"
 #include "lix/libutil/types.hh"
 #include "lix/libutil/error.hh"
@@ -108,8 +109,7 @@ struct [[nodiscard("you must call RunningProgram::wait()")]] RunningProgram
 private:
     Path program;
     Pid pid;
-    std::unique_ptr<Source> stdoutSource;
-    AutoCloseFD stdout_;
+    std::unique_ptr<AsyncFdIoStream> stdout;
 
     RunningProgram(PathView program, Pid pid, AutoCloseFD stdout);
 
@@ -121,7 +121,7 @@ public:
 
     explicit operator bool() const { return bool(pid); }
 
-    std::tuple<pid_t, std::unique_ptr<Source>, int> release();
+    std::tuple<pid_t, std::unique_ptr<AsyncFdIoStream>> release();
 
     int kill();
     [[nodiscard]]
@@ -130,10 +130,13 @@ public:
 
     std::optional<int> getStdoutFD() const
     {
-        return stdout_ ? std::optional(stdout_.get()) : std::nullopt;
+        return stdout ? std::optional(stdout->getFD()) : std::nullopt;
     }
 
-    Source * getStdout() const { return stdoutSource.get(); };
+    AsyncFdIoStream * getStdout() const
+    {
+        return stdout.get();
+    };
 };
 
 kj::Promise<Result<std::pair<int, std::string>>> runProgram(RunOptions options);