Merge pull request #235 from fischman/ppoll

pselect->ppoll on linux to raise the process limit roof

3 files changed, 86 insertions, 7 deletions

diff --git a/src/subprocess.cc b/src/subprocess.cc
index d4a7d03..99de93f 100644
--- a/src/subprocess.cc
+++ b/src/subprocess.cc
@@ -1,4 +1,4 @@
-// Copyright 2011 Google Inc. All Rights Reserved.
+// Copyright 2012 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.
@@ -42,10 +42,12 @@ bool Subprocess::Start(SubprocessSet* set, const string& command) {
   if (pipe(output_pipe) < 0)
     Fatal("pipe: %s", strerror(errno));
   fd_ = output_pipe[0];
-  // fd_ may be a member of the pselect set in SubprocessSet::DoWork.  Check
-  // that it falls below the system limit.
+#if !defined(linux)
+  // On linux we use ppoll in DoWork(); elsewhere we use pselect and so must
+  // avoid overly-large FDs.
   if (fd_ >= FD_SETSIZE)
     Fatal("pipe: %s", strerror(EMFILE));
+#endif  // !linux
   SetCloseOnExec(fd_);
 
   pid_ = fork();
@@ -180,6 +182,54 @@ Subprocess *SubprocessSet::Add(const string &command) {
   return subprocess;
 }
 
+#ifdef linux
+bool SubprocessSet::DoWork() {
+  vector<pollfd> fds;
+  nfds_t nfds = 0;
+
+  for (vector<Subprocess*>::iterator i = running_.begin();
+       i != running_.end(); ++i) {
+    int fd = (*i)->fd_;
+    if (fd < 0)
+      continue;
+    pollfd pfd = { fd, POLLIN | POLLPRI | POLLRDHUP, 0 };
+    fds.push_back(pfd);
+    ++nfds;
+  }
+
+  int ret = ppoll(&fds.front(), nfds, NULL, &old_mask_);
+  if (ret == -1) {
+    if (errno != EINTR) {
+      perror("ninja: ppoll");
+      return false;
+    }
+    bool interrupted = interrupted_;
+    interrupted_ = false;
+    return interrupted;
+  }
+
+  nfds_t cur_nfd = 0;
+  for (vector<Subprocess*>::iterator i = running_.begin();
+       i != running_.end(); ) {
+    int fd = (*i)->fd_;
+    if (fd < 0)
+      continue;
+    assert(fd == fds[cur_nfd].fd);
+    if (fds[cur_nfd++].revents) {
+      (*i)->OnPipeReady();
+      if ((*i)->Done()) {
+        finished_.push(*i);
+        i = running_.erase(i);
+        continue;
+      }
+    }
+    ++i;
+  }
+
+  return false;
+}
+
+#else  // linux
 bool SubprocessSet::DoWork() {
   fd_set set;
   int nfds = 0;
@@ -213,7 +263,7 @@ bool SubprocessSet::DoWork() {
       (*i)->OnPipeReady();
       if ((*i)->Done()) {
         finished_.push(*i);
-        running_.erase(i);
+        i = running_.erase(i);
         continue;
       }
     }
@@ -222,6 +272,7 @@ bool SubprocessSet::DoWork() {
 
   return false;
 }
+#endif  // linux
 
 Subprocess* SubprocessSet::NextFinished() {
   if (finished_.empty())
diff --git a/src/subprocess.h b/src/subprocess.h
index 8e0d7f1..3294416 100644
--- a/src/subprocess.h
+++ b/src/subprocess.h
@@ -1,4 +1,4 @@
-// Copyright 2011 Google Inc. All Rights Reserved.
+// Copyright 2012 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.
@@ -68,7 +68,7 @@ struct Subprocess {
   friend struct SubprocessSet;
 };
 
-/// SubprocessSet runs a pselect() loop around a set of Subprocesses.
+/// SubprocessSet runs a ppoll/pselect() loop around a set of Subprocesses.
 /// DoWork() waits for any state change in subprocesses; finished_
 /// is a queue of subprocesses as they finish.
 struct SubprocessSet {
diff --git a/src/subprocess_test.cc b/src/subprocess_test.cc
index 5b3e8a3..c155012 100644
--- a/src/subprocess_test.cc
+++ b/src/subprocess_test.cc
@@ -1,4 +1,4 @@
-// Copyright 2011 Google Inc. All Rights Reserved.
+// Copyright 2012 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.
@@ -142,3 +142,31 @@ TEST_F(SubprocessTest, SetWithMulti) {
   }
 }
 
+#ifdef linux
+TEST_F(SubprocessTest, SetWithLots) {
+  // Arbitrary big number; needs to be over 1024 to confirm we're no longer
+  // hostage to pselect.
+  const size_t kNumProcs = 1025;
+
+  // Make sure [ulimit -n] isn't going to stop us from working.
+  rlimit rlim;
+  ASSERT_EQ(0, getrlimit(RLIMIT_NOFILE, &rlim));
+  ASSERT_GT(rlim.rlim_cur, kNumProcs)
+      << "Raise [ulimit -n] well above " << kNumProcs
+      << " to make this test go";
+
+  vector<Subprocess*> procs;
+  for (size_t i = 0; i < kNumProcs; ++i) {
+    Subprocess* subproc = subprocs_.Add("/bin/echo");
+    ASSERT_NE((Subprocess *) 0, subproc);
+    procs.push_back(subproc);
+  }
+  while (!subprocs_.running_.empty())
+    subprocs_.DoWork();
+  for (size_t i = 0; i < procs.size(); ++i) {
+    ASSERT_EQ(ExitSuccess, procs[i]->Finish());
+    ASSERT_NE("", procs[i]->GetOutput());
+  }
+  ASSERT_EQ(kNumProcs, subprocs_.finished_.size());
+}
+#endif  // linux