folly/Subprocess.cpp

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