libutil: return sources from runProgram2

this much more closely mimics what is actually happening: we're reading
data from somewhere else, actively, rather than passively waiting. with
the data flow matching the underlying system interactions better we can
remove a few sinkToSource calls that merely exists to undo the mismatch
caused by not treating subprocess output as a data source to begin with

Change-Id: If4abfc2f8398fb5e88c9b91a8bdefd5504bb2d11

src/libfetchers/git.cc

@@ -688,17 +688,14 @@ struct GitInputScheme : InputScheme
 
             filter = isNotDotGitDirectory;
         } else {
-            // FIXME: should pipe this, or find some better way to extract a
+            auto proc = runProgram2({
-            // revision.
-            auto source = sinkToSource([&](Sink & sink) {
-                runProgram2({
                 .program = "git",
                 .args = { "-C", repoDir, "--git-dir", gitDir, "archive", input.getRev()->gitRev() },
-                    .standardOut = &sink
+                .captureStdout = true,
-                }).wait();
             });
+            Finally const _wait([&] { proc.wait(); });
 
-            unpackTarfile(*source, tmpDir);
+            unpackTarfile(*proc.stdout(), tmpDir);
         }
 
         auto storePath = store->addToStore(name, tmpDir, FileIngestionMethod::Recursive, htSHA256, filter);

src/libstore/build/derivation-goal.cc

@@ -923,12 +923,16 @@ void runPostBuildHook(
     };
     LogSink sink(act);
 
-    runProgram2({
+    auto proc = runProgram2({
         .program = settings.postBuildHook,
         .environment = hookEnvironment,
-        .standardOut = &sink,
+        .captureStdout = true,
         .mergeStderrToStdout = true,
-    }).wait();
+    });
+    Finally const _wait([&] { proc.wait(); });
+
+    // FIXME just process the data, without a wrapper sink class
+    proc.stdout()->drainInto(sink);
 }
 
 void DerivationGoal::buildDone()

src/libutil/processes.cc

@@ -243,18 +243,28 @@ std::string runProgram(Path program, bool searchPath, const Strings & args, bool
 // Output = error code + "standard out" output stream
 std::pair<int, std::string> runProgram(RunOptions && options)
 {
-    StringSink sink;
+    options.captureStdout = true;
-    options.standardOut = &sink;
 
     int status = 0;
+    std::string stdout;
 
     try {
-        runProgram2(options).wait();
+        auto proc = runProgram2(options);
+        Finally const _wait([&] { proc.wait(); });
+        stdout = proc.stdout()->drain();
     } catch (ExecError & e) {
         status = e.status;
     }
 
-    return {status, std::move(sink.s)};
+    return {status, std::move(stdout)};
+}
+
+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))
+{
 }
 
 RunningProgram::~RunningProgram()
@@ -281,7 +291,7 @@ RunningProgram runProgram2(const RunOptions & options)
 
     /* Create a pipe. */
     Pipe out;
-    if (options.standardOut) out.create();
+    if (options.captureStdout) out.create();
 
     ProcessOptions processOptions;
 
@@ -299,7 +309,7 @@ RunningProgram runProgram2(const RunOptions & options)
     Pid pid{startProcess([&]() {
         if (options.environment)
             replaceEnv(*options.environment);
-        if (options.standardOut && dup2(out.writeSide.get(), STDOUT_FILENO) == -1)
+        if (options.captureStdout && dup2(out.writeSide.get(), STDOUT_FILENO) == -1)
             throw SysError("dupping stdout");
         if (options.mergeStderrToStdout)
             if (dup2(STDOUT_FILENO, STDERR_FILENO) == -1)
@@ -332,10 +342,11 @@ RunningProgram runProgram2(const RunOptions & options)
 
     out.writeSide.close();
 
-    if (options.standardOut)
+    return RunningProgram{
-        *options.standardOut << drainFDSource(out.readSide.get());
+        options.program,
-
+        std::move(pid),
-    return RunningProgram{options.program, std::move(pid)};
+        options.captureStdout ? std::move(out.readSide) : AutoCloseFD{}
+    };
 }
 
 std::string statusToString(int status)

src/libutil/processes.hh

@@ -3,6 +3,7 @@
 
 #include "types.hh"
 #include "error.hh"
+#include "file-descriptor.hh"
 
 #include <sys/types.h>
 #include <sys/stat.h>
@@ -82,7 +83,7 @@ struct RunOptions
     std::optional<uid_t> gid;
     std::optional<Path> chdir;
     std::optional<std::map<std::string, std::string>> environment;
-    Sink * standardOut = nullptr;
+    bool captureStdout = false;
     bool mergeStderrToStdout = false;
     bool isInteractive = false;
 };
@@ -94,14 +95,18 @@ struct [[nodiscard("you must call RunningProgram::wait()")]] RunningProgram
 private:
     Path program;
     Pid pid;
+    std::unique_ptr<Source> stdoutSource;
+    AutoCloseFD stdout_;
 
-    RunningProgram(Path program, Pid pid) : program(std::move(program)), pid(std::move(pid)) {}
+    RunningProgram(PathView program, Pid pid, AutoCloseFD stdout);
 
 public:
     RunningProgram() = default;
     ~RunningProgram();
 
     void wait();
+
+    Source * stdout() const { return stdoutSource.get(); }
 };
 
 std::pair<int, std::string> runProgram(RunOptions && options);