cmUVProcessChain: Manually create pipes between processes

Previously, we were letting libuv create the output pipe for each
process. However, in order to facilitate use cases where process
spawn is expected to fail (in the case of a missing executable), we
want pipe creation to be separate from process creation. Manually
create all of the pipes.

1 files changed, 68 insertions, 23 deletions

diff --git a/Source/cmUVProcessChain.cxx b/Source/cmUVProcessChain.cxx
index 436ada2..6517e60 100644
--- a/Source/cmUVProcessChain.cxx
+++ b/Source/cmUVProcessChain.cxx
@@ -5,11 +5,9 @@
 #include "cmUVProcessChain.h"
 
 #include <array>
-#include <cassert>
 #include <csignal>
 #include <cstdio>
 #include <istream> // IWYU pragma: keep
-#include <iterator>
 #include <type_traits>
 #include <utility>
 
@@ -53,6 +51,7 @@ struct cmUVProcessChain::InternalData
   {
     cmUVProcessChain::InternalData* Data;
     cm::uv_process_ptr Process;
+    cm::uv_pipe_ptr InputPipe;
     cm::uv_pipe_ptr OutputPipe;
     bool Finished = false;
     Status ProcessStatus;
@@ -66,14 +65,17 @@ struct cmUVProcessChain::InternalData
 
   StreamData<std::istream> OutputStreamData;
   StreamData<std::istream> ErrorStreamData;
+  cm::uv_pipe_ptr TempOutputPipe;
   cm::uv_pipe_ptr TempErrorPipe;
 
   unsigned int ProcessesCompleted = 0;
   std::vector<std::unique_ptr<ProcessData>> Processes;
 
   bool Prepare(const cmUVProcessChainBuilder* builder);
-  bool AddCommand(const cmUVProcessChainBuilder::ProcessConfiguration& config,
-                  bool first, bool last);
+  bool SpawnProcess(
+    std::size_t index,
+    const cmUVProcessChainBuilder::ProcessConfiguration& config, bool first,
+    bool last);
   void Finish();
 
   static const Status* GetStatus(const ProcessData& data);
@@ -168,9 +170,9 @@ cmUVProcessChain cmUVProcessChainBuilder::Start() const
     return chain;
   }
 
-  for (auto it = this->Processes.begin(); it != this->Processes.end(); ++it) {
-    if (!chain.Data->AddCommand(*it, it == this->Processes.begin(),
-                                it == std::prev(this->Processes.end()))) {
+  for (std::size_t i = 0; i < this->Processes.size(); i++) {
+    if (!chain.Data->SpawnProcess(i, this->Processes[i], i == 0,
+                                  i == this->Processes.size() - 1)) {
       return chain;
     }
   }
@@ -247,12 +249,27 @@ bool cmUVProcessChain::InternalData::Prepare(
         outputData.Stdio.flags = UV_INHERIT_FD;
         outputData.Stdio.data.fd = errorData.Stdio.data.fd;
       } else {
+        int pipeFd[2];
+        if (cmGetPipes(pipeFd) < 0) {
+          return false;
+        }
+
         if (outputData.BuiltinStream.init(*this->Loop, 0) < 0) {
           return false;
         }
-        outputData.Stdio.flags =
-          static_cast<uv_stdio_flags>(UV_CREATE_PIPE | UV_WRITABLE_PIPE);
-        outputData.Stdio.data.stream = outputData.BuiltinStream;
+        if (uv_pipe_open(outputData.BuiltinStream, pipeFd[0]) < 0) {
+          return false;
+        }
+        outputData.Stdio.flags = UV_INHERIT_FD;
+        outputData.Stdio.data.fd = pipeFd[1];
+
+        if (this->TempOutputPipe.init(*this->Loop, 0) < 0) {
+          return false;
+        }
+        if (uv_pipe_open(this->TempOutputPipe, outputData.Stdio.data.fd) < 0) {
+          return false;
+        }
+
         outputData.Streambuf.open(outputData.BuiltinStream);
       }
       break;
@@ -263,16 +280,46 @@ bool cmUVProcessChain::InternalData::Prepare(
       break;
   }
 
+  bool first = true;
+  for (std::size_t i = 0; i < this->Builder->Processes.size(); i++) {
+    this->Processes.emplace_back(cm::make_unique<ProcessData>());
+    auto& process = *this->Processes.back();
+    process.Data = this;
+
+    if (!first) {
+      auto& prevProcess = *this->Processes[i - 1];
+
+      int pipeFd[2];
+      if (cmGetPipes(pipeFd) < 0) {
+        return false;
+      }
+
+      if (prevProcess.OutputPipe.init(*this->Loop, 0) < 0) {
+        return false;
+      }
+      if (uv_pipe_open(prevProcess.OutputPipe, pipeFd[1]) < 0) {
+        return false;
+      }
+      if (process.InputPipe.init(*this->Loop, 0) < 0) {
+        return false;
+      }
+      if (uv_pipe_open(process.InputPipe, pipeFd[0]) < 0) {
+        return false;
+      }
+    }
+
+    first = false;
+  }
+
   return true;
 }
 
-bool cmUVProcessChain::InternalData::AddCommand(
+bool cmUVProcessChain::InternalData::SpawnProcess(
+  std::size_t index,
   const cmUVProcessChainBuilder::ProcessConfiguration& config, bool first,
   bool last)
 {
-  this->Processes.emplace_back(cm::make_unique<ProcessData>());
-  auto& process = *this->Processes.back();
-  process.Data = this;
+  auto& process = *this->Processes[index];
 
   auto options = uv_process_options_t();
 
@@ -296,20 +343,14 @@ bool cmUVProcessChain::InternalData::AddCommand(
   if (first) {
     stdio[0].flags = UV_IGNORE;
   } else {
-    assert(this->Processes.size() >= 2);
-    auto& prev = *this->Processes[this->Processes.size() - 2];
     stdio[0].flags = UV_INHERIT_STREAM;
-    stdio[0].data.stream = prev.OutputPipe;
+    stdio[0].data.stream = process.InputPipe;
   }
   if (last) {
     stdio[1] = this->OutputStreamData.Stdio;
   } else {
-    if (process.OutputPipe.init(*this->Loop, 0) < 0) {
-      return false;
-    }
     stdio[1] = uv_stdio_container_t();
-    stdio[1].flags =
-      static_cast<uv_stdio_flags>(UV_CREATE_PIPE | UV_WRITABLE_PIPE);
+    stdio[1].flags = UV_INHERIT_STREAM;
     stdio[1].data.stream = process.OutputPipe;
   }
   stdio[2] = this->ErrorStreamData.Stdio;
@@ -325,11 +366,15 @@ bool cmUVProcessChain::InternalData::AddCommand(
     processData->Data->ProcessesCompleted++;
   };
 
-  return process.Process.spawn(*this->Loop, options, &process) >= 0;
+  bool result = process.Process.spawn(*this->Loop, options, &process) >= 0;
+  process.InputPipe.reset();
+  process.OutputPipe.reset();
+  return result;
 }
 
 void cmUVProcessChain::InternalData::Finish()
 {
+  this->TempOutputPipe.reset();
   this->TempErrorPipe.reset();
   this->Valid = true;
 }