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   checkUnixError(::pipe2(errFds, O_CLOEXEC), "pipe2");
 260 #else
 261   checkUnixError(::pipe(errFds), "pipe");
 262 #endif
 263   SCOPE_EXIT {
 264     CHECK_ERR(::close(errFds[0]));
 265     if (errFds[1] >= 0) {
 266       CHECK_ERR(::close(errFds[1]));
 267     }
 268   };
 269
 270 #if !FOLLY_HAVE_PIPE2
 271   // Ask the child to close the read end of the error pipe.
 272   checkUnixError(fcntl(errFds[0], F_SETFD, FD_CLOEXEC), "set FD_CLOEXEC");
 273   // Set the close-on-exec flag on the write side of the pipe.
 274   // This way the pipe will be closed automatically in the child if execve()
 275   // succeeds.  If the exec fails the child can write error information to the
 276   // pipe.
 277   checkUnixError(fcntl(errFds[1], F_SETFD, FD_CLOEXEC), "set FD_CLOEXEC");
 278 #endif
 279
 280   // Perform the actual work of setting up pipes then forking and
 281   // executing the child.
 282   spawnInternal(std::move(argv), executable, options, env, errFds[1]);
 283
 284   // After spawnInternal() returns the child is alive.  We have to be very
 285   // careful about throwing after this point.  We are inside the constructor,
 286   // so if we throw the Subprocess object will have never existed, and the
 287   // destructor will never be called.
 288   //
 289   // We should only throw if we got an error via the errFd, and we know the
 290   // child has exited and can be immediately waited for.  In all other cases,
 291   // we have no way of cleaning up the child.
 292
 293   // Close writable side of the errFd pipe in the parent process
 294   CHECK_ERR(::close(errFds[1]));
 295   errFds[1] = -1;
 296
 297   // Read from the errFd pipe, to tell if the child ran into any errors before
 298   // calling exec()
 299   readChildErrorPipe(errFds[0], executable);
 300
 301   // We have fully succeeded now, so release the guard on pipes_
 302   pipesGuard.dismiss();
 303 }
 304
 305 void Subprocess::spawnInternal(
 306     std::unique_ptr<const char*[]> argv,
 307     const char* executable,
 308     Options& options,
 309     const std::vector<std::string>* env,
 310     int errFd) {
 311   // Parent work, pre-fork: create pipes
 312   std::vector<int> childFds;
 313   // Close all of the childFds as we leave this scope
 314   SCOPE_EXIT {
 315     // These are only pipes, closing them shouldn't fail
 316     for (int cfd : childFds) {
 317       CHECK_ERR(::close(cfd));
 318     }
 319   };
 320
 321   int r;
 322   for (auto& p : options.fdActions_) {
 323     if (p.second == PIPE_IN || p.second == PIPE_OUT) {
 324       int fds[2];
 325       // We're setting both ends of the pipe as close-on-exec. The child
 326       // doesn't need to reset the flag on its end, as we always dup2() the fd,
 327       // and dup2() fds don't share the close-on-exec flag.
 328 #if FOLLY_HAVE_PIPE2
 329       r = ::pipe2(fds, O_CLOEXEC);
 330       checkUnixError(r, "pipe2");
 331 #else
 332       r = ::pipe(fds);
 333       checkUnixError(r, "pipe");
 334       r = fcntl(fds[0], F_SETFD, FD_CLOEXEC);
 335       checkUnixError(r, "set FD_CLOEXEC");
 336       r = fcntl(fds[1], F_SETFD, FD_CLOEXEC);
 337       checkUnixError(r, "set FD_CLOEXEC");
 338 #endif
 339       PipeInfo pinfo;
 340       pinfo.direction = p.second;
 341       int cfd;
 342       if (p.second == PIPE_IN) {
 343         // Child gets reading end
 344         pinfo.parentFd = fds[1];
 345         cfd = fds[0];
 346       } else {
 347         pinfo.parentFd = fds[0];
 348         cfd = fds[1];
 349       }
 350       p.second = cfd;  // ensure it gets dup2()ed
 351       pinfo.childFd = p.first;
 352       childFds.push_back(cfd);
 353       pipes_.push_back(pinfo);
 354     }
 355   }
 356
 357   // This should already be sorted, as options.fdActions_ is
 358   DCHECK(std::is_sorted(pipes_.begin(), pipes_.end()));
 359
 360   // Note that the const casts below are legit, per
 361   // http://pubs.opengroup.org/onlinepubs/009695399/functions/exec.html
 362
 363   char** argVec = const_cast<char**>(argv.get());
 364
 365   // Set up environment
 366   std::unique_ptr<const char*[]> envHolder;
 367   char** envVec;
 368   if (env) {
 369     envHolder = cloneStrings(*env);
 370     envVec = const_cast<char**>(envHolder.get());
 371   } else {
 372     envVec = environ;
 373   }
 374
 375   // Block all signals around vfork; see http://ewontfix.com/7/.
 376   //
 377   // As the child may run in the same address space as the parent until
 378   // the actual execve() system call, any (custom) signal handlers that
 379   // the parent has might alter parent's memory if invoked in the child,
 380   // with undefined results.  So we block all signals in the parent before
 381   // vfork(), which will cause them to be blocked in the child as well (we
 382   // rely on the fact that Linux, just like all sane implementations, only
 383   // clones the calling thread).  Then, in the child, we reset all signals
 384   // to their default dispositions (while still blocked), and unblock them
 385   // (so the exec()ed process inherits the parent's signal mask)
 386   //
 387   // The parent also unblocks all signals as soon as vfork() returns.
 388   sigset_t allBlocked;
 389   r = sigfillset(&allBlocked);
 390   checkUnixError(r, "sigfillset");
 391   sigset_t oldSignals;
 392
 393   r = pthread_sigmask(SIG_SETMASK, &allBlocked, &oldSignals);
 394   checkPosixError(r, "pthread_sigmask");
 395   SCOPE_EXIT {
 396     // Restore signal mask
 397     r = pthread_sigmask(SIG_SETMASK, &oldSignals, nullptr);
 398     CHECK_EQ(r, 0) << "pthread_sigmask: " << errnoStr(r);  // shouldn't fail
 399   };
 400
 401   // Call c_str() here, as it's not necessarily safe after fork.
 402   const char* childDir =
 403     options.childDir_.empty() ? nullptr : options.childDir_.c_str();
 404   pid_t pid = vfork();
 405   if (pid == 0) {
 406     int errnoValue = prepareChild(options, &oldSignals, childDir);
 407     if (errnoValue != 0) {
 408       childError(errFd, kChildFailure, errnoValue);
 409     }
 410
 411     errnoValue = runChild(executable, argVec, envVec, options);
 412     // If we get here, exec() failed.
 413     childError(errFd, kExecFailure, errnoValue);
 414   }
 415   // In parent.  Make sure vfork() succeeded.
 416   checkUnixError(pid, errno, "vfork");
 417
 418   // Child is alive.  We have to be very careful about throwing after this
 419   // point.  We are inside the constructor, so if we throw the Subprocess
 420   // object will have never existed, and the destructor will never be called.
 421   //
 422   // We should only throw if we got an error via the errFd, and we know the
 423   // child has exited and can be immediately waited for.  In all other cases,
 424   // we have no way of cleaning up the child.
 425   pid_ = pid;
 426   returnCode_ = ProcessReturnCode(RV_RUNNING);
 427 }
 428
 429 int Subprocess::prepareChild(const Options& options,
 430                              const sigset_t* sigmask,
 431                              const char* childDir) const {
 432   // While all signals are blocked, we must reset their
 433   // dispositions to default.
 434   for (int sig = 1; sig < NSIG; ++sig) {
 435     ::signal(sig, SIG_DFL);
 436   }
 437
 438   {
 439     // Unblock signals; restore signal mask.
 440     int r = pthread_sigmask(SIG_SETMASK, sigmask, nullptr);
 441     if (r != 0) {
 442       return r;  // pthread_sigmask() returns an errno value
 443     }
 444   }
 445
 446   // Change the working directory, if one is given
 447   if (childDir) {
 448     if (::chdir(childDir) == -1) {
 449       return errno;
 450     }
 451   }
 452
 453   // We don't have to explicitly close the parent's end of all pipes,
 454   // as they all have the FD_CLOEXEC flag set and will be closed at
 455   // exec time.
 456
 457   // Close all fds that we're supposed to close.
 458   for (auto& p : options.fdActions_) {
 459     if (p.second == CLOSE) {
 460       if (::close(p.first) == -1) {
 461         return errno;
 462       }
 463     } else if (p.second != p.first) {
 464       if (::dup2(p.second, p.first) == -1) {
 465         return errno;
 466       }
 467     }
 468   }
 469
 470   // If requested, close all other file descriptors.  Don't close
 471   // any fds in options.fdActions_, and don't touch stdin, stdout, stderr.
 472   // Ignore errors.
 473   if (options.closeOtherFds_) {
 474     for (int fd = getdtablesize() - 1; fd >= 3; --fd) {
 475       if (options.fdActions_.count(fd) == 0) {
 476         ::close(fd);
 477       }
 478     }
 479   }
 480
 481 #if __linux__
 482   // Opt to receive signal on parent death, if requested
 483   if (options.parentDeathSignal_ != 0) {
 484     if (prctl(PR_SET_PDEATHSIG, options.parentDeathSignal_, 0, 0, 0) == -1) {
 485       return errno;
 486     }
 487   }
 488 #endif
 489
 490   return 0;
 491 }
 492
 493 int Subprocess::runChild(const char* executable,
 494                          char** argv, char** env,
 495                          const Options& options) const {
 496   // Now, finally, exec.
 497   int r;
 498   if (options.usePath_) {
 499     ::execvp(executable, argv);
 500   } else {
 501     ::execve(executable, argv, env);
 502   }
 503   return errno;
 504 }
 505
 506 void Subprocess::readChildErrorPipe(int pfd, const char* executable) {
 507   ChildErrorInfo info;
 508   auto rc = readNoInt(pfd, &info, sizeof(info));
 509   if (rc == 0) {
 510     // No data means the child executed successfully, and the pipe
 511     // was closed due to the close-on-exec flag being set.
 512     return;
 513   } else if (rc != sizeof(ChildErrorInfo)) {
 514     // An error occurred trying to read from the pipe, or we got a partial read.
 515     // Neither of these cases should really occur in practice.
 516     //
 517     // We can't get any error data from the child in this case, and we don't
 518     // know if it is successfully running or not.  All we can do is to return
 519     // normally, as if the child executed successfully.  If something bad
 520     // happened the caller should at least get a non-normal exit status from
 521     // the child.
 522     LOG(ERROR) << "unexpected error trying to read from child error pipe " <<
 523       "rc=" << rc << ", errno=" << errno;
 524     return;
 525   }
 526
 527   // We got error data from the child.  The child should exit immediately in
 528   // this case, so wait on it to clean up.
 529   wait();
 530
 531   // Throw to signal the error
 532   throw SubprocessSpawnError(executable, info.errCode, info.errnoValue);
 533 }
 534
 535 ProcessReturnCode Subprocess::poll() {
 536   returnCode_.enforce(ProcessReturnCode::RUNNING);
 537   DCHECK_GT(pid_, 0);
 538   int status;
 539   pid_t found = ::waitpid(pid_, &status, WNOHANG);
 540   checkUnixError(found, "waitpid");
 541   if (found != 0) {
 542     returnCode_ = ProcessReturnCode(status);
 543     pid_ = -1;
 544   }
 545   return returnCode_;
 546 }
 547
 548 bool Subprocess::pollChecked() {
 549   if (poll().state() == ProcessReturnCode::RUNNING) {
 550     return false;
 551   }
 552   checkStatus(returnCode_);
 553   return true;
 554 }
 555
 556 ProcessReturnCode Subprocess::wait() {
 557   returnCode_.enforce(ProcessReturnCode::RUNNING);
 558   DCHECK_GT(pid_, 0);
 559   int status;
 560   pid_t found;
 561   do {
 562     found = ::waitpid(pid_, &status, 0);
 563   } while (found == -1 && errno == EINTR);
 564   checkUnixError(found, "waitpid");
 565   DCHECK_EQ(found, pid_);
 566   returnCode_ = ProcessReturnCode(status);
 567   pid_ = -1;
 568   return returnCode_;
 569 }
 570
 571 void Subprocess::waitChecked() {
 572   wait();
 573   checkStatus(returnCode_);
 574 }
 575
 576 void Subprocess::sendSignal(int signal) {
 577   returnCode_.enforce(ProcessReturnCode::RUNNING);
 578   int r = ::kill(pid_, signal);
 579   checkUnixError(r, "kill");
 580 }
 581
 582 pid_t Subprocess::pid() const {
 583   return pid_;
 584 }
 585
 586 namespace {
 587
 588 std::pair<const uint8_t*, size_t> queueFront(const IOBufQueue& queue) {
 589   auto* p = queue.front();
 590   if (!p) return std::make_pair(nullptr, 0);
 591   return io::Cursor(p).peek();
 592 }
 593
 594 // fd write
 595 bool handleWrite(int fd, IOBufQueue& queue) {
 596   for (;;) {
 597     auto p = queueFront(queue);
 598     if (p.second == 0) {
 599       return true;  // EOF
 600     }
 601
 602     ssize_t n = writeNoInt(fd, p.first, p.second);
 603     if (n == -1 && errno == EAGAIN) {
 604       return false;
 605     }
 606     checkUnixError(n, "write");
 607     queue.trimStart(n);
 608   }
 609 }
 610
 611 // fd read
 612 bool handleRead(int fd, IOBufQueue& queue) {
 613   for (;;) {
 614     auto p = queue.preallocate(100, 65000);
 615     ssize_t n = readNoInt(fd, p.first, p.second);
 616     if (n == -1 && errno == EAGAIN) {
 617       return false;
 618     }
 619     checkUnixError(n, "read");
 620     if (n == 0) {
 621       return true;
 622     }
 623     queue.postallocate(n);
 624   }
 625 }
 626
 627 bool discardRead(int fd) {
 628   static const size_t bufSize = 65000;
 629   // Thread unsafe, but it doesn't matter.
 630   static std::unique_ptr<char[]> buf(new char[bufSize]);
 631
 632   for (;;) {
 633     ssize_t n = readNoInt(fd, buf.get(), bufSize);
 634     if (n == -1 && errno == EAGAIN) {
 635       return false;
 636     }
 637     checkUnixError(n, "read");
 638     if (n == 0) {
 639       return true;
 640     }
 641   }
 642 }
 643
 644 }  // namespace
 645
 646 std::pair<std::string, std::string> Subprocess::communicate(
 647     StringPiece input) {
 648   IOBufQueue inputQueue;
 649   inputQueue.wrapBuffer(input.data(), input.size());
 650
 651   auto outQueues = communicateIOBuf(std::move(inputQueue));
 652   auto outBufs = std::make_pair(outQueues.first.move(),
 653                                 outQueues.second.move());
 654   std::pair<std::string, std::string> out;
 655   if (outBufs.first) {
 656     outBufs.first->coalesce();
 657     out.first.assign(reinterpret_cast<const char*>(outBufs.first->data()),
 658                      outBufs.first->length());
 659   }
 660   if (outBufs.second) {
 661     outBufs.second->coalesce();
 662     out.second.assign(reinterpret_cast<const char*>(outBufs.second->data()),
 663                      outBufs.second->length());
 664   }
 665   return out;
 666 }
 667
 668 std::pair<IOBufQueue, IOBufQueue> Subprocess::communicateIOBuf(
 669     IOBufQueue input) {
 670   // If the user supplied a non-empty input buffer, make sure
 671   // that stdin is a pipe so we can write the data.
 672   if (!input.empty()) {
 673     // findByChildFd() will throw std::invalid_argument if no pipe for
 674     // STDIN_FILENO exists
 675     findByChildFd(STDIN_FILENO);
 676   }
 677
 678   std::pair<IOBufQueue, IOBufQueue> out;
 679
 680   auto readCallback = [&] (int pfd, int cfd) -> bool {
 681     if (cfd == STDOUT_FILENO) {
 682       return handleRead(pfd, out.first);
 683     } else if (cfd == STDERR_FILENO) {
 684       return handleRead(pfd, out.second);
 685     } else {
 686       // Don't close the file descriptor, the child might not like SIGPIPE,
 687       // just read and throw the data away.
 688       return discardRead(pfd);
 689     }
 690   };
 691
 692   auto writeCallback = [&] (int pfd, int cfd) -> bool {
 693     if (cfd == STDIN_FILENO) {
 694       return handleWrite(pfd, input);
 695     } else {
 696       // If we don't want to write to this fd, just close it.
 697       return true;
 698     }
 699   };
 700
 701   communicate(std::move(readCallback), std::move(writeCallback));
 702
 703   return out;
 704 }
 705
 706 void Subprocess::communicate(FdCallback readCallback,
 707                              FdCallback writeCallback) {
 708   returnCode_.enforce(ProcessReturnCode::RUNNING);
 709   setAllNonBlocking();
 710
 711   std::vector<pollfd> fds;
 712   fds.reserve(pipes_.size());
 713   std::vector<int> toClose;
 714   toClose.reserve(pipes_.size());
 715
 716   while (!pipes_.empty()) {
 717     fds.clear();
 718     toClose.clear();
 719
 720     for (auto& p : pipes_) {
 721       pollfd pfd;
 722       pfd.fd = p.parentFd;
 723       // Yes, backwards, PIPE_IN / PIPE_OUT are defined from the
 724       // child's point of view.
 725       if (!p.enabled) {
 726         // Still keeping fd in watched set so we get notified of POLLHUP /
 727         // POLLERR
 728         pfd.events = 0;
 729       } else if (p.direction == PIPE_IN) {
 730         pfd.events = POLLOUT;
 731       } else {
 732         pfd.events = POLLIN;
 733       }
 734       fds.push_back(pfd);
 735     }
 736
 737     int r;
 738     do {
 739       r = ::poll(fds.data(), fds.size(), -1);
 740     } while (r == -1 && errno == EINTR);
 741     checkUnixError(r, "poll");
 742
 743     for (int i = 0; i < pipes_.size(); ++i) {
 744       auto& p = pipes_[i];
 745       DCHECK_EQ(fds[i].fd, p.parentFd);
 746       short events = fds[i].revents;
 747
 748       bool closed = false;
 749       if (events & POLLOUT) {
 750         DCHECK(!(events & POLLIN));
 751         if (writeCallback(p.parentFd, p.childFd)) {
 752           toClose.push_back(i);
 753           closed = true;
 754         }
 755       }
 756
 757       // Call read callback on POLLHUP, to give it a chance to read (and act
 758       // on) end of file
 759       if (events & (POLLIN | POLLHUP)) {
 760         DCHECK(!(events & POLLOUT));
 761         if (readCallback(p.parentFd, p.childFd)) {
 762           toClose.push_back(i);
 763           closed = true;
 764         }
 765       }
 766
 767       if ((events & (POLLHUP | POLLERR)) && !closed) {
 768         toClose.push_back(i);
 769         closed = true;
 770       }
 771     }
 772
 773     // Close the fds in reverse order so the indexes hold after erase()
 774     for (int idx : boost::adaptors::reverse(toClose)) {
 775       auto pos = pipes_.begin() + idx;
 776       closeChecked(pos->parentFd);
 777       pipes_.erase(pos);
 778     }
 779   }
 780 }
 781
 782 void Subprocess::enableNotifications(int childFd, bool enabled) {
 783   pipes_[findByChildFd(childFd)].enabled = enabled;
 784 }
 785
 786 bool Subprocess::notificationsEnabled(int childFd) const {
 787   return pipes_[findByChildFd(childFd)].enabled;
 788 }
 789
 790 int Subprocess::findByChildFd(int childFd) const {
 791   auto pos = std::lower_bound(
 792       pipes_.begin(), pipes_.end(), childFd,
 793       [] (const PipeInfo& info, int fd) { return info.childFd < fd; });
 794   if (pos == pipes_.end() || pos->childFd != childFd) {
 795     throw std::invalid_argument(folly::to<std::string>(
 796         "child fd not found ", childFd));
 797   }
 798   return pos - pipes_.begin();
 799 }
 800
 801 void Subprocess::closeParentFd(int childFd) {
 802   int idx = findByChildFd(childFd);
 803   closeChecked(pipes_[idx].parentFd);
 804   pipes_.erase(pipes_.begin() + idx);
 805 }
 806
 807 namespace {
 808
 809 class Initializer {
 810  public:
 811   Initializer() {
 812     // We like EPIPE, thanks.
 813     ::signal(SIGPIPE, SIG_IGN);
 814   }
 815 };
 816
 817 Initializer initializer;
 818
 819 }  // namespace
 820
 821 }  // namespace folly
 822