windows: subprocess implementation for Windows

Heavily based on a patch from Sergey Nenakhov <nenakhov.sergey@gmail.com>.

4 files changed, 243 insertions, 6 deletions

diff --git a/src/build.cc b/src/build.cc
index 03c692c..913bc31 100644
--- a/src/build.cc
+++ b/src/build.cc
@@ -291,7 +291,7 @@ bool RealCommandRunner::StartCommand(Edge* edge) {
   string command = edge->EvaluateCommand();
   Subprocess* subproc = new Subprocess;
   subproc_to_edge_.insert(make_pair(subproc, edge));
-  if (!subproc->Start(command))
+  if (!subproc->Start(&subprocs_, command))
     return false;
 
   subprocs_.Add(subproc);
diff --git a/src/subprocess-win32.cc b/src/subprocess-win32.cc
new file mode 100644
index 0000000..2efec00
--- /dev/null
+++ b/src/subprocess-win32.cc
@@ -0,0 +1,211 @@
+// Copyright 2011 Google Inc. All Rights Reserved.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#include "subprocess.h"
+
+#include <stdio.h>
+#include <windows.h>
+
+#include <algorithm>
+
+#include "util.h"
+
+namespace {
+
+void Win32Fatal(const char* function) {
+  DWORD err = GetLastError();
+
+  char* msg_buf;
+  FormatMessageA(
+        FORMAT_MESSAGE_ALLOCATE_BUFFER |
+        FORMAT_MESSAGE_FROM_SYSTEM |
+        FORMAT_MESSAGE_IGNORE_INSERTS,
+        NULL,
+        err,
+        MAKELANGID(LANG_NEUTRAL, SUBLANG_DEFAULT),
+        (char*)&msg_buf,
+        0,
+        NULL);
+  Fatal("%s: %s", function, msg_buf);
+  LocalFree(msg_buf);
+}
+
+}  // anonymous namespace
+
+Subprocess::Subprocess() : child_(NULL) , overlapped_() {
+}
+
+Subprocess::~Subprocess() {
+  // Reap child if forgotten.
+  if (child_)
+    Finish();
+}
+
+HANDLE Subprocess::SetupPipe(HANDLE ioport) {
+  char pipe_name[32];
+  snprintf(pipe_name, sizeof(pipe_name),
+           "\\\\.\\pipe\\ninja_%p_out", ::GetModuleHandle(NULL));
+
+  pipe_ = ::CreateNamedPipeA(pipe_name,
+                             PIPE_ACCESS_INBOUND | FILE_FLAG_OVERLAPPED,
+                             PIPE_TYPE_BYTE,
+                             PIPE_UNLIMITED_INSTANCES,
+                             0, 0, INFINITE, NULL);
+  if (pipe_ == INVALID_HANDLE_VALUE)
+    Win32Fatal("CreateNamedPipe");
+
+  if (!CreateIoCompletionPort(pipe_, ioport, (ULONG_PTR)this, 0))
+    Win32Fatal("CreateIoCompletionPort");
+
+  memset(&overlapped_, 0, sizeof(overlapped_));
+  if (!ConnectNamedPipe(pipe_, &overlapped_) &&
+      GetLastError() != ERROR_IO_PENDING) {
+    Win32Fatal("ConnectNamedPipe");
+  }
+
+  // Get the write end of the pipe as a handle inheritable across processes.
+  HANDLE output_write_handle = CreateFile(pipe_name, GENERIC_WRITE, 0,
+                                          NULL, OPEN_EXISTING, 0, NULL);
+  HANDLE output_write_child;
+  if (!DuplicateHandle(GetCurrentProcess(), output_write_handle,
+                       GetCurrentProcess(), &output_write_child,
+                       0, TRUE, DUPLICATE_SAME_ACCESS)) {
+    Win32Fatal("DuplicateHandle");
+  }
+  CloseHandle(output_write_handle);
+
+  return output_write_child;
+}
+
+bool Subprocess::Start(struct SubprocessSet* set, const string& command) {
+  HANDLE child_pipe = SetupPipe(set->ioport_);
+
+  STARTUPINFOA startup_info = {};
+  startup_info.cb = sizeof(STARTUPINFO);
+  startup_info.dwFlags = STARTF_USESTDHANDLES;
+  startup_info.hStdOutput = child_pipe;
+  startup_info.hStdInput  = NULL;
+  startup_info.hStdError  = NULL;  // TODO: handle child stderr as well.
+
+  PROCESS_INFORMATION process_info;
+
+  string full_command = "cmd /c " + command;
+  printf("running %s\n", full_command.c_str());
+  if (!CreateProcessA(NULL, (char*)full_command.c_str(), NULL, NULL,
+                      /* inherit handles */ TRUE, 0,
+                      NULL, NULL,
+                      &startup_info, &process_info)) {
+    Win32Fatal("CreateProcess");
+  }
+
+  // Close pipe channel only used by the child.
+  if (child_pipe)
+    CloseHandle(child_pipe);
+
+  CloseHandle(process_info.hThread);
+  child_ = process_info.hProcess;
+
+  return true;
+}
+
+void Subprocess::OnPipeReady() {
+  DWORD bytes;
+  if (!GetOverlappedResult(pipe_, &overlapped_, &bytes, FALSE)) {
+    if (GetLastError() == ERROR_BROKEN_PIPE) {
+      CloseHandle(pipe_);
+      pipe_ = NULL;
+      return;
+    }
+    Win32Fatal("GetOverlappedResult");
+  }
+
+  if (bytes)
+    buf_.append(overlapped_buf_, bytes);
+
+  memset(&overlapped_, 0, sizeof(overlapped_));
+  if (!::ReadFile(pipe_, overlapped_buf_, sizeof(overlapped_buf_),
+                  &bytes, &overlapped_)) {
+    if (GetLastError() != ERROR_IO_PENDING)
+      Win32Fatal("ReadFile");
+  }
+
+  // Even if we read any bytes in the readfile call, we'll enter this
+  // function again later and get them at that point.
+}
+
+bool Subprocess::Finish() {
+  // TODO: add error handling for all of these.
+  WaitForSingleObject(child_, INFINITE);
+
+  DWORD exit_code = 0;
+  GetExitCodeProcess(child_, &exit_code);
+
+  CloseHandle(child_);
+  child_ = NULL;
+
+  return exit_code == 0;
+}
+
+bool Subprocess::Done() const {
+  return pipe_ == NULL;
+}
+
+const string& Subprocess::GetOutput() const {
+  return buf_;
+}
+
+SubprocessSet::SubprocessSet() {
+  ioport_ = ::CreateIoCompletionPort(INVALID_HANDLE_VALUE, NULL, 0, 1);
+  if (!ioport_)
+    Win32Fatal("CreateIoCompletionPort");
+}
+
+SubprocessSet::~SubprocessSet() {
+  CloseHandle(ioport_);
+}
+
+void SubprocessSet::Add(Subprocess* subprocess) {
+  running_.push_back(subprocess);
+}
+
+void SubprocessSet::DoWork() {
+  DWORD bytes_read;
+  Subprocess* subproc;
+  OVERLAPPED* overlapped;
+
+  if (!GetQueuedCompletionStatus(ioport_, &bytes_read, (PULONG_PTR)&subproc,
+                                 &overlapped, INFINITE)) {
+    if (GetLastError() != ERROR_BROKEN_PIPE)
+      Win32Fatal("GetQueuedCompletionStatus");
+  }
+
+  subproc->OnPipeReady();
+
+  if (subproc->Done()) {
+    vector<Subprocess*>::iterator end =
+        std::remove(running_.begin(), running_.end(), subproc);
+    if (running_.end() != end) {
+      finished_.push(subproc);
+      running_.resize(end - running_.begin());
+    }
+  }
+}
+
+Subprocess* SubprocessSet::NextFinished() {
+  if (finished_.empty())
+    return NULL;
+  Subprocess* subproc = finished_.front();
+  finished_.pop();
+  return subproc;
+}
diff --git a/src/subprocess.cc b/src/subprocess.cc
index 907c248..c2e306c 100644
--- a/src/subprocess.cc
+++ b/src/subprocess.cc
@@ -37,7 +37,7 @@ Subprocess::~Subprocess() {
     Finish();
 }
 
-bool Subprocess::Start(const string& command) {
+bool Subprocess::Start(SubprocessSet* set, const string& command) {
   int output_pipe[2];
   if (pipe(output_pipe) < 0)
     Fatal("pipe: %s", strerror(errno));
@@ -84,7 +84,7 @@ bool Subprocess::Start(const string& command) {
   return true;
 }
 
-void Subprocess::OnFDReady() {
+void Subprocess::OnPipeReady() {
   char buf[4 << 10];
   ssize_t len = read(fd_, buf, sizeof(buf));
   if (len > 0) {
@@ -120,6 +120,9 @@ const string& Subprocess::GetOutput() const {
   return buf_;
 }
 
+SubprocessSet::SubprocessSet() {}
+SubprocessSet::~SubprocessSet() {}
+
 void SubprocessSet::Add(Subprocess* subprocess) {
   running_.push_back(subprocess);
 }
@@ -151,7 +154,7 @@ void SubprocessSet::DoWork() {
   for (size_t i = 0; i < fds.size(); ++i) {
     if (fds[i].revents) {
       Subprocess* subproc = fd_to_subprocess[fds[i].fd];
-      subproc->OnFDReady();
+      subproc->OnPipeReady();
       if (subproc->Done()) {
         finished_.push(subproc);
         std::remove(running_.begin(), running_.end(), subproc);
diff --git a/src/subprocess.h b/src/subprocess.h
index 934a220..9828bf4 100644
--- a/src/subprocess.h
+++ b/src/subprocess.h
@@ -20,6 +20,10 @@
 #include <queue>
 using namespace std;
 
+#ifdef _WIN32
+#include <windows.h>
+#endif
+
 /// Subprocess wraps a single async subprocess.  It is entirely
 /// passive: it expects the caller to notify it when its fds are ready
 /// for reading, as well as call Finish() to reap the child once done()
@@ -27,8 +31,8 @@ using namespace std;
 struct Subprocess {
   Subprocess();
   ~Subprocess();
-  bool Start(const string& command);
-  void OnFDReady();
+  bool Start(struct SubprocessSet* set, const string& command);
+  void OnPipeReady();
   /// Returns true on successful process exit.
   bool Finish();
 
@@ -38,8 +42,20 @@ struct Subprocess {
 
  private:
   string buf_;
+
+#ifdef _WIN32
+  /// Set up pipe_ as the parent-side pipe of the subprocess; return the
+  /// other end of the pipe, usable in the child process.
+  HANDLE SetupPipe(HANDLE ioport);
+
+  HANDLE child_;
+  HANDLE pipe_;
+  OVERLAPPED overlapped_;
+  char overlapped_buf_[4 << 10];
+#else
   int fd_;
   pid_t pid_;
+#endif
 
   friend struct SubprocessSet;
 };
@@ -48,12 +64,19 @@ struct Subprocess {
 /// DoWork() waits for any state change in subprocesses; finished_
 /// is a queue of subprocesses as they finish.
 struct SubprocessSet {
+  SubprocessSet();
+  ~SubprocessSet();
+
   void Add(Subprocess* subprocess);
   void DoWork();
   Subprocess* NextFinished();
 
   vector<Subprocess*> running_;
   queue<Subprocess*> finished_;
+
+#ifdef _WIN32
+  HANDLE ioport_;
+#endif
 };
 
 #endif // NINJA_SUBPROCESS_H_