WIP: Add notifyfile support

src/build.cc

@@ -470,7 +470,8 @@ bool RealCommandRunner::CanRunMore() const {
 
 bool RealCommandRunner::StartCommand(Edge* edge) {
   string command = edge->EvaluateCommand();
-  Subprocess* subproc = subprocs_.Add(command, edge->use_console());
+  Subprocess* subproc = subprocs_.Add(command, edge->use_console(),
+                                      edge->GetUnescapedNotifyfile());
   if (!subproc)
     return false;
   subproc_to_edge_.insert(make_pair(subproc, edge));
@@ -486,11 +487,17 @@ bool RealCommandRunner::WaitForCommand(Result* result) {
       return false;
   }
 
+  map<const Subprocess*, Edge*>::iterator e = subproc_to_edge_.find(subproc);
+  result->edge = e->second;
+
+  if (!subproc->Done()) {
+    result->notify = subproc->GetNotifyPaths();
+    return true;
+  }
+
   result->status = subproc->Finish();
   result->output = subproc->GetOutput();
 
-  map<const Subprocess*, Edge*>::iterator e = subproc_to_edge_.find(subproc);
-  result->edge = e->second;
   subproc_to_edge_.erase(e);
 
   delete subproc;
@@ -634,6 +641,24 @@ bool Builder::Build(string* err) {
         return false;
       }
 
+      if (!result.notify.empty()) {
+        // Command is not finished, but some outputs are already finished
+        for (const StringPiece& path : result.notify) {
+          for (vector<Node*>::iterator o = result.edge->outputs_.begin();
+               o != result.edge->outputs_.end(); ++o) {
+            StringPiece opath((*o)->path());
+            if (opath == path) {
+              result.edge->outputs_paths_ready_.push_back(opath);
+              if (!plan_.NodeFinished(*o, err)) {
+                return false;
+              }
+              break;
+            }
+          }
+        }
+        continue;
+      }
+
       --pending_commands;
       if (!FinishCommand(&result, err)) {
         Cleanup();

src/build.h

@@ -80,18 +80,19 @@ struct Plan {
   /// by information loaded from a dyndep file.
   bool DyndepsLoaded(DependencyScan* scan, const Node* node,
                      const DyndepFile& ddf, std::string* err);
-private:
-  bool RefreshDyndepDependents(DependencyScan* scan, const Node* node, std::string* err);
-  void UnmarkDependents(const Node* node, std::set<Node*>* dependents);
-  bool AddSubTarget(const Node* node, const Node* dependent, std::string* err,
-                    std::set<Edge*>* dyndep_walk);
 
   /// Update plan with knowledge that the given node is up to date.
   /// If the node is a dyndep binding on any of its dependents, this
   /// loads dynamic dependencies from the node's path.
   /// Returns 'false' if loading dyndep info fails and 'true' otherwise.
   bool NodeFinished(Node* node, std::string* err);
 
+private:
+  bool RefreshDyndepDependents(DependencyScan* scan, const Node* node, std::string* err);
+  void UnmarkDependents(const Node* node, std::set<Node*>* dependents);
+  bool AddSubTarget(const Node* node, const Node* dependent, std::string* err,
+                    std::set<Edge*>* dyndep_walk);
+
   /// Enumerate possible steps we want for an edge.
   enum Want
   {
@@ -144,6 +145,7 @@ struct CommandRunner {
     Edge* edge;
     ExitStatus status;
     std::string output;
+    std::vector<StringPiece> notify;
     bool success() const { return status == ExitSuccess; }
   };
   /// Wait for a command to complete, or return false if interrupted.

src/graph.cc

@@ -316,7 +316,7 @@ bool DependencyScan::LoadDyndeps(Node* node, DyndepFile* ddf,
 bool Edge::AllInputsReady() const {
   for (vector<Node*>::const_iterator i = inputs_.begin();
        i != inputs_.end(); ++i) {
-    if ((*i)->in_edge() && !(*i)->in_edge()->outputs_ready())
+    if ((*i)->in_edge() && !(*i)->in_edge()->outputs_ready((*i)->path()))
       return false;
   }
   return true;
@@ -432,6 +432,11 @@ std::string Edge::GetUnescapedRspfile() const {
   return env.LookupVariable("rspfile");
 }
 
+std::string Edge::GetUnescapedNotifyfile() const {
+  EdgeEnv env(this, EdgeEnv::kDoNotEscape);
+  return env.LookupVariable("notifyfile");
+}
+
 void Edge::Dump(const char* prefix) const {
   printf("%s[ ", prefix);
   for (vector<Node*>::const_iterator i = inputs_.begin();

src/graph.h

@@ -18,6 +18,7 @@
 #include <set>
 #include <string>
 #include <vector>
+#include <algorithm>
 
 #include "dyndep.h"
 #include "eval_env.h"
@@ -168,6 +169,8 @@ struct Edge {
   std::string GetUnescapedDyndep() const;
   /// Like GetBinding("rspfile"), but without shell escaping.
   std::string GetUnescapedRspfile() const;
+  /// Like GetBinding("notifyfile"), but without shell escaping.
+  std::string GetUnescapedNotifyfile() const;
 
   void Dump(const char* prefix="") const;
 
@@ -188,6 +191,13 @@ struct Edge {
   int weight() const { return 1; }
   bool outputs_ready() const { return outputs_ready_; }
 
+  // Some output paths might be ready before the whole edge is ready.
+  std::vector<StringPiece> outputs_paths_ready_;
+  bool outputs_ready(const StringPiece& path) const {
+    return outputs_ready_ ||
+           std::find(outputs_paths_ready_.begin(), outputs_paths_ready_.end(), path) != outputs_paths_ready_.end();
+  }
+
   // There are three types of inputs.
   // 1) explicit deps, which show up as $in on the command line;
   // 2) implicit deps, which the target depends on implicitly (e.g. C headers),

src/subprocess-posix.cc

@@ -22,6 +22,7 @@
 #include <stdio.h>
 #include <string.h>
 #include <sys/wait.h>
+#include <sys/stat.h>
 #include <spawn.h>
 
 #if defined(USE_PPOLL)
@@ -36,13 +37,17 @@ extern char** environ;
 
 using namespace std;
 
-Subprocess::Subprocess(bool use_console) : fd_(-1), pid_(-1),
-                                           use_console_(use_console) {
-}
+Subprocess::Subprocess(bool use_console, const std::string& notify_file)
+    : notify_file_(notify_file), notify_pos_(0),
+      fd_(-1), notify_fd_(-1), pid_(-1), use_console_(use_console) {}
 
 Subprocess::~Subprocess() {
   if (fd_ >= 0)
     close(fd_);
+  if (notify_fd_ >= 0) {
+    close(notify_fd_);
+    unlink(notify_file_.c_str());
+  }
   // Reap child if forgotten.
   if (pid_ != -1)
     Finish();
@@ -61,6 +66,15 @@ bool Subprocess::Start(SubprocessSet* set, const string& command) {
 #endif  // !USE_PPOLL
   SetCloseOnExec(fd_);
 
+  if (!notify_file_.empty()) {
+    unlink(notify_file_.c_str());
+    if (mkfifo(notify_file_.c_str(), 0666) < 0)
+      Fatal("mkfifo: %s", strerror(errno));
+    notify_fd_ = open(notify_file_.c_str(), O_RDONLY | O_NONBLOCK);
+    if (notify_fd_ < 0)
+      Fatal("open notify file: %s", strerror(errno));
+  }
+
   posix_spawn_file_actions_t action;
   int err = posix_spawn_file_actions_init(&action);
   if (err != 0)
@@ -147,6 +161,19 @@ void Subprocess::OnPipeReady() {
   }
 }
 
+bool Subprocess::OnNotifyReady() {
+  char buf[4 << 10];
+  ssize_t len = read(notify_fd_, buf, sizeof(buf));
+  if (len > 0) {
+    notify_buf_.append(buf, len);
+  } else {
+    close(notify_fd_);
+    unlink(notify_file_.c_str());
+    notify_fd_ = -1;
+  }
+  return notify_buf_.find('\n', notify_pos_) != std::string::npos;
+}
+
 ExitStatus Subprocess::Finish() {
   assert(pid_ != -1);
   int status;
@@ -184,6 +211,17 @@ const string& Subprocess::GetOutput() const {
   return buf_;
 }
 
+const std::vector<StringPiece> Subprocess::GetNotifyPaths() {
+  std::vector<StringPiece> paths;
+  size_t found;
+  while ((found = notify_buf_.find('\n', notify_pos_)) != std::string::npos) {
+    paths.push_back(
+        StringPiece(notify_buf_.data() + notify_pos_, found - notify_pos_));
+    notify_pos_ = found + 1;
+  }
+  return paths;
+}
+
 int SubprocessSet::interrupted_;
 
 void SubprocessSet::SetInterruptedFlag(int signum) {
@@ -238,8 +276,9 @@ SubprocessSet::~SubprocessSet() {
     Fatal("sigprocmask: %s", strerror(errno));
 }
 
-Subprocess *SubprocessSet::Add(const string& command, bool use_console) {
-  Subprocess *subprocess = new Subprocess(use_console);
+Subprocess* SubprocessSet::Add(const string& command, bool use_console,
+                               const std::string& notify_file) {
+  Subprocess* subprocess = new Subprocess(use_console, notify_file);
   if (!subprocess->Start(this, command)) {
     delete subprocess;
     return 0;
@@ -256,11 +295,17 @@ bool SubprocessSet::DoWork() {
   for (vector<Subprocess*>::iterator i = running_.begin();
        i != running_.end(); ++i) {
     int fd = (*i)->fd_;
-    if (fd < 0)
-      continue;
-    pollfd pfd = { fd, POLLIN | POLLPRI, 0 };
-    fds.push_back(pfd);
-    ++nfds;
+    if (fd >= 0) {
+      pollfd pfd = { fd, POLLIN | POLLPRI, 0 };
+      fds.push_back(pfd);
+      ++nfds;
+    }
+    fd = (*i)->notify_fd_;
+    if (fd >= 0) {
+      pollfd pfd = { fd, POLLIN | POLLPRI, 0 };
+      fds.push_back(pfd);
+      ++nfds;
+    }
   }
 
   interrupted_ = 0;
@@ -280,18 +325,31 @@ bool SubprocessSet::DoWork() {
   nfds_t cur_nfd = 0;
   for (vector<Subprocess*>::iterator i = running_.begin();
        i != running_.end(); ) {
+    bool pipe_ready = false;
+    bool notify_ready = false;
     int fd = (*i)->fd_;
-    if (fd < 0)
-      continue;
-    assert(fd == fds[cur_nfd].fd);
-    if (fds[cur_nfd++].revents) {
+    if (fd >= 0) {
+      assert(fd == fds[cur_nfd].fd);
+      pipe_ready = fds[cur_nfd++].revents != 0;
+    }
+    fd = (*i)->notify_fd_;
+    if (fd >= 0) {
+      assert(fd == fds[cur_nfd].fd);
+      notify_ready = fds[cur_nfd++].revents != 0;
+    }
+    if (pipe_ready) {
       (*i)->OnPipeReady();
       if ((*i)->Done()) {
         finished_.push(*i);
         i = running_.erase(i);
         continue;
       }
     }
+    if (notify_ready) {
+      if ((*i)->OnNotifyReady()) {
+        finished_.push(*i);
+      }
+    }
     ++i;
   }
 
@@ -312,6 +370,12 @@ bool SubprocessSet::DoWork() {
       if (nfds < fd+1)
         nfds = fd+1;
     }
+    fd = (*i)->notify_fd_;
+    if (fd > 0) {
+      FD_SET(fd, &set);
+      if (nfds < fd + 1)
+        nfds = fd + 1;
+    }
   }
 
   interrupted_ = 0;
@@ -339,6 +403,12 @@ bool SubprocessSet::DoWork() {
         continue;
       }
     }
+    fd = (*i)->notify_fd_;
+    if (fd >= 0 && FD_ISSET(fd, &set)) {
+      if ((*i)->OnNotifyReady()) {
+        finished_.push(*i);
+      }
+    }
     ++i;
   }
 

src/subprocess.h

@@ -34,6 +34,7 @@
 #endif
 
 #include "exit_status.h"
+#include "string_piece.h"
 
 /// Subprocess wraps a single async subprocess.  It is entirely
 /// passive: it expects the caller to notify it when its fds are ready
@@ -49,13 +50,18 @@ struct Subprocess {
   bool Done() const;
 
   const std::string& GetOutput() const;
+  const std::vector<StringPiece> GetNotifyPaths();
 
  private:
-  Subprocess(bool use_console);
+  Subprocess(bool use_console, const std::string& notify_file);
   bool Start(struct SubprocessSet* set, const std::string& command);
   void OnPipeReady();
+  bool OnNotifyReady();
 
   std::string buf_;
+  std::string notify_file_;
+  std::string notify_buf_;
+  size_t notify_pos_;
 
 #ifdef _WIN32
   /// Set up pipe_ as the parent-side pipe of the subprocess; return the
@@ -69,6 +75,7 @@ struct Subprocess {
   bool is_reading_;
 #else
   int fd_;
+  int notify_fd_;
   pid_t pid_;
 #endif
   bool use_console_;
@@ -83,7 +90,8 @@ struct SubprocessSet {
   SubprocessSet();
   ~SubprocessSet();
 
-  Subprocess* Add(const std::string& command, bool use_console = false);
+  Subprocess* Add(const std::string& command, bool use_console = false,
+                  const std::string& notify_file = "");
   bool DoWork();
   Subprocess* NextFinished();
   void Clear();