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