folly/Subprocess.cpp

   1 /*
   2  * Copyright 2013 Facebook, Inc.
   3  *
   4  * Licensed under the Apache License, Version 2.0 (the "License");
   5  * you may not use this file except in compliance with the License.
   6  * You may obtain a copy of the License at
   7  *
   8  *   http://www.apache.org/licenses/LICENSE-2.0
   9  *
  10  * Unless required by applicable law or agreed to in writing, software
  11  * distributed under the License is distributed on an "AS IS" BASIS,
  12  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13  * See the License for the specific language governing permissions and
  14  * limitations under the License.
  15  */
  16
  17 #include "folly/Subprocess.h"
  18
  19 #include <sys/prctl.h>
  20 #include <fcntl.h>
  21 #include <poll.h>
  22 #include <unistd.h>
  23 #include <wait.h>
  24
  25 #include <array>
  26 #include <algorithm>
  27 #include <system_error>
  28
  29 #include <boost/container/flat_set.hpp>
  30 #include <boost/range/adaptors.hpp>
  31
  32 #include <glog/logging.h>
  33
  34 #include "folly/Conv.h"
  35 #include "folly/Exception.h"
  36 #include "folly/FileUtil.h"
  37 #include "folly/ScopeGuard.h"
  38 #include "folly/String.h"
  39 #include "folly/io/Cursor.h"
  40
  41 extern char** environ;
  42
  43 constexpr int kExecFailure = 127;
  44 constexpr int kChildFailure = 126;
  45
  46 namespace folly {
  47
  48 ProcessReturnCode::State ProcessReturnCode::state() const {
  49   if (rawStatus_ == RV_NOT_STARTED) return NOT_STARTED;
  50   if (rawStatus_ == RV_RUNNING) return RUNNING;
  51   if (WIFEXITED(rawStatus_)) return EXITED;
  52   if (WIFSIGNALED(rawStatus_)) return KILLED;
  53   throw std::runtime_error(to<std::string>(
  54       "Invalid ProcessReturnCode: ", rawStatus_));
  55 }
  56
  57 void ProcessReturnCode::enforce(State expected) const {
  58   State s = state();
  59   if (s != expected) {
  60     throw std::logic_error(to<std::string>("Invalid state ", s,
  61                                            " expected ", expected));
  62   }
  63 }
  64
  65 int ProcessReturnCode::exitStatus() const {
  66   enforce(EXITED);
  67   return WEXITSTATUS(rawStatus_);
  68 }
  69
  70 int ProcessReturnCode::killSignal() const {
  71   enforce(KILLED);
  72   return WTERMSIG(rawStatus_);
  73 }
  74
  75 bool ProcessReturnCode::coreDumped() const {
  76   enforce(KILLED);
  77   return WCOREDUMP(rawStatus_);
  78 }
  79
  80 std::string ProcessReturnCode::str() const {
  81   switch (state()) {
  82   case NOT_STARTED:
  83     return "not started";
  84   case RUNNING:
  85     return "running";
  86   case EXITED:
  87     return to<std::string>("exited with status ", exitStatus());
  88   case KILLED:
  89     return to<std::string>("killed by signal ", killSignal(),
  90                            (coreDumped() ? " (core dumped)" : ""));
  91   }
  92   CHECK(false);  // unreached
  93 }
  94
  95 CalledProcessError::CalledProcessError(ProcessReturnCode rc)
  96   : returnCode_(rc),
  97     what_(returnCode_.str()) {
  98 }
  99
 100 SubprocessSpawnError::SubprocessSpawnError(const char* executable,
 101                                            int errCode,
 102                                            int errnoValue)
 103   : errnoValue_(errnoValue),
 104     what_(to<std::string>(errCode == kExecFailure ?
 105                             "failed to execute " :
 106                             "error preparing to execute ",
 107                           executable, ": ", errnoStr(errnoValue))) {
 108 }
 109
 110 namespace {
 111
 112 // Copy pointers to the given strings in a format suitable for posix_spawn
 113 std::unique_ptr<const char*[]> cloneStrings(const std::vector<std::string>& s) {
 114   std::unique_ptr<const char*[]> d(new const char*[s.size() + 1]);
 115   for (int i = 0; i < s.size(); i++) {
 116     d[i] = s[i].c_str();
 117   }
 118   d[s.size()] = nullptr;
 119   return d;
 120 }
 121
 122 // Check a wait() status, throw on non-successful
 123 void checkStatus(ProcessReturnCode returnCode) {
 124   if (returnCode.state() != ProcessReturnCode::EXITED ||
 125       returnCode.exitStatus() != 0) {
 126     throw CalledProcessError(returnCode);
 127   }
 128 }
 129
 130 }  // namespace
 131
 132 Subprocess::Options& Subprocess::Options::fd(int fd, int action) {
 133   if (action == Subprocess::PIPE) {
 134     if (fd == 0) {
 135       action = Subprocess::PIPE_IN;
 136     } else if (fd == 1 || fd == 2) {
 137       action = Subprocess::PIPE_OUT;
 138     } else {
 139       throw std::invalid_argument(
 140           to<std::string>("Only fds 0, 1, 2 are valid for action=PIPE: ", fd));
 141     }
 142   }
 143   fdActions_[fd] = action;
 144   return *this;
 145 }
 146
 147 Subprocess::Subprocess(
 148     const std::vector<std::string>& argv,
 149     const Options& options,
 150     const char* executable,
 151     const std::vector<std::string>* env)
 152   : pid_(-1),
 153     returnCode_(RV_NOT_STARTED) {
 154   if (argv.empty()) {
 155     throw std::invalid_argument("argv must not be empty");
 156   }
 157   if (!executable) executable = argv[0].c_str();
 158   spawn(cloneStrings(argv), executable, options, env);
 159 }
 160
 161 Subprocess::Subprocess(
 162     const std::string& cmd,
 163     const Options& options,
 164     const std::vector<std::string>* env)
 165   : pid_(-1),
 166     returnCode_(RV_NOT_STARTED) {
 167   if (options.usePath_) {
 168     throw std::invalid_argument("usePath() not allowed when running in shell");
 169   }
 170   const char* shell = getenv("SHELL");
 171   if (!shell) {
 172     shell = "/bin/sh";
 173   }
 174
 175   std::unique_ptr<const char*[]> argv(new const char*[4]);
 176   argv[0] = shell;
 177   argv[1] = "-c";
 178   argv[2] = cmd.c_str();
 179   argv[3] = nullptr;
 180   spawn(std::move(argv), shell, options, env);
 181 }
 182
 183 Subprocess::~Subprocess() {
 184   CHECK_NE(returnCode_.state(), ProcessReturnCode::RUNNING)
 185     << "Subprocess destroyed without reaping child";
 186   closeAll();
 187 }
 188
 189 namespace {
 190 void closeChecked(int fd) {
 191   checkUnixError(::close(fd), "close");
 192 }
 193
 194 struct ChildErrorInfo {
 195   int errCode;
 196   int errnoValue;
 197 };
 198
 199 void childError(int errFd, int errCode, int errnoValue) FOLLY_NORETURN;
 200 void childError(int errFd, int errCode, int errnoValue) {
 201   ChildErrorInfo info = {errCode, errnoValue};
 202   // Write the error information over the pipe to our parent process.
 203   // We can't really do anything else if this write call fails.
 204   writeNoInt(errFd, &info, sizeof(info));
 205   // exit
 206   _exit(errCode);
 207 }
 208
 209 }  // namespace
 210
 211 void Subprocess::closeAll() {
 212   for (auto& p : pipes_) {
 213     closeChecked(p.parentFd);
 214   }
 215   pipes_.clear();
 216 }
 217
 218 void Subprocess::setAllNonBlocking() {
 219   for (auto& p : pipes_) {
 220     int fd = p.parentFd;
 221     int flags = ::fcntl(fd, F_GETFL);
 222     checkUnixError(flags, "fcntl");
 223     int r = ::fcntl(fd, F_SETFL, flags | O_NONBLOCK);
 224     checkUnixError(r, "fcntl");
 225   }
 226 }
 227
 228 void Subprocess::spawn(
 229     std::unique_ptr<const char*[]> argv,
 230     const char* executable,
 231     const Options& optionsIn,
 232     const std::vector<std::string>* env) {
 233   if (optionsIn.usePath_ && env) {
 234     throw std::invalid_argument(
 235         "usePath() not allowed when overriding environment");
 236   }
 237
 238   // Make a copy, we'll mutate options
 239   Options options(optionsIn);
 240
 241   // On error, close all of the pipes_
 242   auto pipesGuard = makeGuard([&] {
 243     for (auto& p : this->pipes_) {
 244       CHECK_ERR(::close(p.parentFd));
 245     }
 246   });
 247
 248   // Create a pipe to use to receive error information from the child,
 249   // in case it fails before calling exec()
 250   int errFds[2];
 251   int r = ::pipe(errFds);
 252   checkUnixError(r, "pipe");
 253   SCOPE_EXIT {
 254     CHECK_ERR(::close(errFds[0]));
 255     if (errFds[1] >= 0) {
 256       CHECK_ERR(::close(errFds[1]));
 257     }
 258   };
 259   // Ask the child to close the read end of the error pipe.
 260   options.fdActions_[errFds[0]] = CLOSE;
 261   // Set the close-on-exec flag on the write side of the pipe.
 262   // This way the pipe will be closed automatically in the child if execve()
 263   // succeeds.  If the exec fails the child can write error information to the
 264   // pipe.
 265   r = fcntl(errFds[1], F_SETFD, FD_CLOEXEC);
 266   checkUnixError(r, "set FD_CLOEXEC");
 267
 268   // Perform the actual work of setting up pipes then forking and
 269   // executing the child.
 270   spawnInternal(std::move(argv), executable, options, env, errFds[1]);
 271
 272   // After spawnInternal() returns the child is alive.  We have to be very
 273   // careful about throwing after this point.  We are inside the constructor,
 274   // so if we throw the Subprocess object will have never existed, and the
 275   // destructor will never be called.
 276   //
 277   // We should only throw if we got an error via the errFd, and we know the
 278   // child has exited and can be immediately waited for.  In all other cases,
 279   // we have no way of cleaning up the child.
 280
 281   // Close writable side of the errFd pipe in the parent process
 282   CHECK_ERR(::close(errFds[1]));
 283   errFds[1] = -1;
 284
 285   // Read from the errFd pipe, to tell if the child ran into any errors before
 286   // calling exec()
 287   readChildErrorPipe(errFds[0], executable);
 288
 289   // We have fully succeeded now, so release the guard on pipes_
 290   pipesGuard.dismiss();
 291 }
 292
 293 void Subprocess::spawnInternal(
 294     std::unique_ptr<const char*[]> argv,
 295     const char* executable,
 296     Options& options,
 297     const std::vector<std::string>* env,
 298     int errFd) {
 299   // Parent work, pre-fork: create pipes
 300   std::vector<int> childFds;
 301   // Close all of the childFds as we leave this scope
 302   SCOPE_EXIT {
 303     // These are only pipes, closing them shouldn't fail
 304     for (int cfd : childFds) {
 305       CHECK_ERR(::close(cfd));
 306     }
 307   };
 308
 309   int r;
 310   for (auto& p : options.fdActions_) {
 311     if (p.second == PIPE_IN || p.second == PIPE_OUT) {
 312       int fds[2];
 313       r = ::pipe(fds);
 314       checkUnixError(r, "pipe");
 315       PipeInfo pinfo;
 316       pinfo.direction = p.second;
 317       int cfd;
 318       if (p.second == PIPE_IN) {
 319         // Child gets reading end
 320         pinfo.parentFd = fds[1];
 321         cfd = fds[0];
 322       } else {
 323         pinfo.parentFd = fds[0];
 324         cfd = fds[1];
 325       }
 326       p.second = cfd;  // ensure it gets dup2()ed
 327       pinfo.childFd = p.first;
 328       childFds.push_back(cfd);
 329       pipes_.push_back(pinfo);
 330     }
 331   }
 332
 333   // This should already be sorted, as options.fdActions_ is
 334   DCHECK(std::is_sorted(pipes_.begin(), pipes_.end()));
 335
 336   // Note that the const casts below are legit, per
 337   // http://pubs.opengroup.org/onlinepubs/009695399/functions/exec.html
 338
 339   char** argVec = const_cast<char**>(argv.get());
 340
 341   // Set up environment
 342   std::unique_ptr<const char*[]> envHolder;
 343   char** envVec;
 344   if (env) {
 345     envHolder = cloneStrings(*env);
 346     envVec = const_cast<char**>(envHolder.get());
 347   } else {
 348     envVec = environ;
 349   }
 350
 351   // Block all signals around vfork; see http://ewontfix.com/7/.
 352   //
 353   // As the child may run in the same address space as the parent until
 354   // the actual execve() system call, any (custom) signal handlers that
 355   // the parent has might alter parent's memory if invoked in the child,
 356   // with undefined results.  So we block all signals in the parent before
 357   // vfork(), which will cause them to be blocked in the child as well (we
 358   // rely on the fact that Linux, just like all sane implementations, only
 359   // clones the calling thread).  Then, in the child, we reset all signals
 360   // to their default dispositions (while still blocked), and unblock them
 361   // (so the exec()ed process inherits the parent's signal mask)
 362   //
 363   // The parent also unblocks all signals as soon as vfork() returns.
 364   sigset_t allBlocked;
 365   r = ::sigfillset(&allBlocked);
 366   checkUnixError(r, "sigfillset");
 367   sigset_t oldSignals;
 368
 369   r = pthread_sigmask(SIG_SETMASK, &allBlocked, &oldSignals);
 370   checkPosixError(r, "pthread_sigmask");
 371   SCOPE_EXIT {
 372     // Restore signal mask
 373     r = pthread_sigmask(SIG_SETMASK, &oldSignals, nullptr);
 374     CHECK_EQ(r, 0) << "pthread_sigmask: " << errnoStr(r);  // shouldn't fail
 375   };
 376
 377   pid_t pid = vfork();
 378   if (pid == 0) {
 379     int errnoValue = prepareChild(options, &oldSignals);
 380     if (errnoValue != 0) {
 381       childError(errFd, kChildFailure, errnoValue);
 382     }
 383
 384     errnoValue = runChild(executable, argVec, envVec, options);
 385     // If we get here, exec() failed.
 386     childError(errFd, kExecFailure, errnoValue);
 387   }
 388   // In parent.  Make sure vfork() succeeded.
 389   checkUnixError(pid, errno, "vfork");
 390
 391   // Child is alive.  We have to be very careful about throwing after this
 392   // point.  We are inside the constructor, so if we throw the Subprocess
 393   // object will have never existed, and the destructor will never be called.
 394   //
 395   // We should only throw if we got an error via the errFd, and we know the
 396   // child has exited and can be immediately waited for.  In all other cases,
 397   // we have no way of cleaning up the child.
 398   pid_ = pid;
 399   returnCode_ = ProcessReturnCode(RV_RUNNING);
 400 }
 401
 402 int Subprocess::prepareChild(const Options& options,
 403                              const sigset_t* sigmask) const {
 404   // While all signals are blocked, we must reset their
 405   // dispositions to default.
 406   for (int sig = 1; sig < NSIG; ++sig) {
 407     ::signal(sig, SIG_DFL);
 408   }
 409   // Unblock signals; restore signal mask.
 410   int r = pthread_sigmask(SIG_SETMASK, sigmask, nullptr);
 411   if (r != 0) {
 412     return r;  // pthread_sigmask() returns an errno value
 413   }
 414
 415   // Close parent's ends of all pipes
 416   for (auto& p : pipes_) {
 417     r = ::close(p.parentFd);
 418     if (r == -1) {
 419       return errno;
 420     }
 421   }
 422
 423   // Close all fds that we're supposed to close.
 424   // Note that we're ignoring errors here, in case some of these
 425   // fds were set to close on exec.
 426   for (auto& p : options.fdActions_) {
 427     if (p.second == CLOSE) {
 428       ::close(p.first);
 429     } else {
 430       r = ::dup2(p.second, p.first);
 431       if (r == -1) {
 432         return errno;
 433       }
 434     }
 435   }
 436
 437   // If requested, close all other file descriptors.  Don't close
 438   // any fds in options.fdActions_, and don't touch stdin, stdout, stderr.
 439   // Ignore errors.
 440   if (options.closeOtherFds_) {
 441     for (int fd = getdtablesize() - 1; fd >= 3; --fd) {
 442       if (options.fdActions_.count(fd) == 0) {
 443         ::close(fd);
 444       }
 445     }
 446   }
 447
 448   // Opt to receive signal on parent death, if requested
 449   if (options.parentDeathSignal_ != 0) {
 450     r = prctl(PR_SET_PDEATHSIG, options.parentDeathSignal_, 0, 0, 0);
 451     if (r == -1) {
 452       return errno;
 453     }
 454   }
 455
 456   return 0;
 457 }
 458
 459 int Subprocess::runChild(const char* executable,
 460                          char** argv, char** env,
 461                          const Options& options) const {
 462   // Now, finally, exec.
 463   int r;
 464   if (options.usePath_) {
 465     ::execvp(executable, argv);
 466   } else {
 467     ::execve(executable, argv, env);
 468   }
 469   return errno;
 470 }
 471
 472 void Subprocess::readChildErrorPipe(int pfd, const char* executable) {
 473   ChildErrorInfo info;
 474   auto rc = readNoInt(pfd, &info, sizeof(info));
 475   if (rc == 0) {
 476     // No data means the child executed successfully, and the pipe
 477     // was closed due to the close-on-exec flag being set.
 478     return;
 479   } else if (rc != sizeof(ChildErrorInfo)) {
 480     // An error occurred trying to read from the pipe, or we got a partial read.
 481     // Neither of these cases should really occur in practice.
 482     //
 483     // We can't get any error data from the child in this case, and we don't
 484     // know if it is successfully running or not.  All we can do is to return
 485     // normally, as if the child executed successfully.  If something bad
 486     // happened the caller should at least get a non-normal exit status from
 487     // the child.
 488     LOG(ERROR) << "unexpected error trying to read from child error pipe " <<
 489       "rc=" << rc << ", errno=" << errno;
 490     return;
 491   }
 492
 493   // We got error data from the child.  The child should exit immediately in
 494   // this case, so wait on it to clean up.
 495   wait();
 496
 497   // Throw to signal the error
 498   throw SubprocessSpawnError(executable, info.errCode, info.errnoValue);
 499 }
 500
 501 ProcessReturnCode Subprocess::poll() {
 502   returnCode_.enforce(ProcessReturnCode::RUNNING);
 503   DCHECK_GT(pid_, 0);
 504   int status;
 505   pid_t found = ::waitpid(pid_, &status, WNOHANG);
 506   checkUnixError(found, "waitpid");
 507   if (found != 0) {
 508     returnCode_ = ProcessReturnCode(status);
 509     pid_ = -1;
 510   }
 511   return returnCode_;
 512 }
 513
 514 bool Subprocess::pollChecked() {
 515   if (poll().state() == ProcessReturnCode::RUNNING) {
 516     return false;
 517   }
 518   checkStatus(returnCode_);
 519   return true;
 520 }
 521
 522 ProcessReturnCode Subprocess::wait() {
 523   returnCode_.enforce(ProcessReturnCode::RUNNING);
 524   DCHECK_GT(pid_, 0);
 525   int status;
 526   pid_t found;
 527   do {
 528     found = ::waitpid(pid_, &status, 0);
 529   } while (found == -1 && errno == EINTR);
 530   checkUnixError(found, "waitpid");
 531   DCHECK_EQ(found, pid_);
 532   returnCode_ = ProcessReturnCode(status);
 533   pid_ = -1;
 534   return returnCode_;
 535 }
 536
 537 void Subprocess::waitChecked() {
 538   wait();
 539   checkStatus(returnCode_);
 540 }
 541
 542 void Subprocess::sendSignal(int signal) {
 543   returnCode_.enforce(ProcessReturnCode::RUNNING);
 544   int r = ::kill(pid_, signal);
 545   checkUnixError(r, "kill");
 546 }
 547
 548 pid_t Subprocess::pid() const {
 549   return pid_;
 550 }
 551
 552 namespace {
 553
 554 std::pair<const uint8_t*, size_t> queueFront(const IOBufQueue& queue) {
 555   auto* p = queue.front();
 556   if (!p) return std::make_pair(nullptr, 0);
 557   return io::Cursor(p).peek();
 558 }
 559
 560 // fd write
 561 bool handleWrite(int fd, IOBufQueue& queue) {
 562   for (;;) {
 563     auto p = queueFront(queue);
 564     if (p.second == 0) {
 565       return true;  // EOF
 566     }
 567
 568     ssize_t n;
 569     do {
 570       n = ::write(fd, p.first, p.second);
 571     } while (n == -1 && errno == EINTR);
 572     if (n == -1 && errno == EAGAIN) {
 573       return false;
 574     }
 575     checkUnixError(n, "write");
 576     queue.trimStart(n);
 577   }
 578 }
 579
 580 // fd read
 581 bool handleRead(int fd, IOBufQueue& queue) {
 582   for (;;) {
 583     auto p = queue.preallocate(100, 65000);
 584     ssize_t n;
 585     do {
 586       n = ::read(fd, p.first, p.second);
 587     } while (n == -1 && errno == EINTR);
 588     if (n == -1 && errno == EAGAIN) {
 589       return false;
 590     }
 591     checkUnixError(n, "read");
 592     if (n == 0) {
 593       return true;
 594     }
 595     queue.postallocate(n);
 596   }
 597 }
 598
 599 bool discardRead(int fd) {
 600   static const size_t bufSize = 65000;
 601   // Thread unsafe, but it doesn't matter.
 602   static std::unique_ptr<char[]> buf(new char[bufSize]);
 603
 604   for (;;) {
 605     ssize_t n;
 606     do {
 607       n = ::read(fd, buf.get(), bufSize);
 608     } while (n == -1 && errno == EINTR);
 609     if (n == -1 && errno == EAGAIN) {
 610       return false;
 611     }
 612     checkUnixError(n, "read");
 613     if (n == 0) {
 614       return true;
 615     }
 616   }
 617 }
 618
 619 }  // namespace
 620
 621 std::pair<std::string, std::string> Subprocess::communicate(
 622     StringPiece input) {
 623   IOBufQueue inputQueue;
 624   inputQueue.wrapBuffer(input.data(), input.size());
 625
 626   auto outQueues = communicateIOBuf(std::move(inputQueue));
 627   auto outBufs = std::make_pair(outQueues.first.move(),
 628                                 outQueues.second.move());
 629   std::pair<std::string, std::string> out;
 630   if (outBufs.first) {
 631     outBufs.first->coalesce();
 632     out.first.assign(reinterpret_cast<const char*>(outBufs.first->data()),
 633                      outBufs.first->length());
 634   }
 635   if (outBufs.second) {
 636     outBufs.second->coalesce();
 637     out.second.assign(reinterpret_cast<const char*>(outBufs.second->data()),
 638                      outBufs.second->length());
 639   }
 640   return out;
 641 }
 642
 643 std::pair<IOBufQueue, IOBufQueue> Subprocess::communicateIOBuf(
 644     IOBufQueue input) {
 645   // If the user supplied a non-empty input buffer, make sure
 646   // that stdin is a pipe so we can write the data.
 647   if (!input.empty()) {
 648     // findByChildFd() will throw std::invalid_argument if no pipe for
 649     // STDIN_FILENO exists
 650     findByChildFd(STDIN_FILENO);
 651   }
 652
 653   std::pair<IOBufQueue, IOBufQueue> out;
 654
 655   auto readCallback = [&] (int pfd, int cfd) -> bool {
 656     if (cfd == STDOUT_FILENO) {
 657       return handleRead(pfd, out.first);
 658     } else if (cfd == STDERR_FILENO) {
 659       return handleRead(pfd, out.second);
 660     } else {
 661       // Don't close the file descriptor, the child might not like SIGPIPE,
 662       // just read and throw the data away.
 663       return discardRead(pfd);
 664     }
 665   };
 666
 667   auto writeCallback = [&] (int pfd, int cfd) -> bool {
 668     if (cfd == STDIN_FILENO) {
 669       return handleWrite(pfd, input);
 670     } else {
 671       // If we don't want to write to this fd, just close it.
 672       return true;
 673     }
 674   };
 675
 676   communicate(std::move(readCallback), std::move(writeCallback));
 677
 678   return out;
 679 }
 680
 681 void Subprocess::communicate(FdCallback readCallback,
 682                              FdCallback writeCallback) {
 683   returnCode_.enforce(ProcessReturnCode::RUNNING);
 684   setAllNonBlocking();
 685
 686   std::vector<pollfd> fds;
 687   fds.reserve(pipes_.size());
 688   std::vector<int> toClose;
 689   toClose.reserve(pipes_.size());
 690
 691   while (!pipes_.empty()) {
 692     fds.clear();
 693     toClose.clear();
 694
 695     for (auto& p : pipes_) {
 696       pollfd pfd;
 697       pfd.fd = p.parentFd;
 698       // Yes, backwards, PIPE_IN / PIPE_OUT are defined from the
 699       // child's point of view.
 700       pfd.events = (p.direction == PIPE_IN ?  POLLOUT : POLLIN);
 701       fds.push_back(pfd);
 702     }
 703
 704     int r;
 705     do {
 706       r = ::poll(fds.data(), fds.size(), -1);
 707     } while (r == -1 && errno == EINTR);
 708     checkUnixError(r, "poll");
 709
 710     for (int i = 0; i < pipes_.size(); ++i) {
 711       auto& p = pipes_[i];
 712       DCHECK_EQ(fds[i].fd, p.parentFd);
 713       short events = fds[i].revents;
 714
 715       bool closed = false;
 716       if (events & POLLOUT) {
 717         DCHECK(!(events & POLLIN));
 718         if (writeCallback(p.parentFd, p.childFd)) {
 719           toClose.push_back(i);
 720           closed = true;
 721         }
 722       }
 723
 724       if (events & POLLIN) {
 725         DCHECK(!(events & POLLOUT));
 726         if (readCallback(p.parentFd, p.childFd)) {
 727           toClose.push_back(i);
 728           closed = true;
 729         }
 730       }
 731
 732       if ((events & (POLLHUP | POLLERR)) && !closed) {
 733         toClose.push_back(i);
 734         closed = true;
 735       }
 736     }
 737
 738     // Close the fds in reverse order so the indexes hold after erase()
 739     for (int idx : boost::adaptors::reverse(toClose)) {
 740       auto pos = pipes_.begin() + idx;
 741       closeChecked(pos->parentFd);
 742       pipes_.erase(pos);
 743     }
 744   }
 745 }
 746
 747 int Subprocess::findByChildFd(int childFd) const {
 748   auto pos = std::lower_bound(
 749       pipes_.begin(), pipes_.end(), childFd,
 750       [] (const PipeInfo& info, int fd) { return info.childFd < fd; });
 751   if (pos == pipes_.end() || pos->childFd != childFd) {
 752     throw std::invalid_argument(folly::to<std::string>(
 753         "child fd not found ", childFd));
 754   }
 755   return pos - pipes_.begin();
 756 }
 757
 758 void Subprocess::closeParentFd(int childFd) {
 759   int idx = findByChildFd(childFd);
 760   closeChecked(pipes_[idx].parentFd);
 761   pipes_.erase(pipes_.begin() + idx);
 762 }
 763
 764 namespace {
 765
 766 class Initializer {
 767  public:
 768   Initializer() {
 769     // We like EPIPE, thanks.
 770     ::signal(SIGPIPE, SIG_IGN);
 771   }
 772 };
 773
 774 Initializer initializer;
 775
 776 }  // namespace
 777
 778 }  // namespace folly
 779