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/ScopeGuard.h"
  37 #include "folly/String.h"
  38 #include "folly/io/Cursor.h"
  39
  40 extern char** environ;
  41
  42 constexpr int kExecFailure = 127;
  43 constexpr int kChildFailure = 126;
  44
  45 namespace folly {
  46
  47 ProcessReturnCode::State ProcessReturnCode::state() const {
  48   if (rawStatus_ == RV_NOT_STARTED) return NOT_STARTED;
  49   if (rawStatus_ == RV_RUNNING) return RUNNING;
  50   if (WIFEXITED(rawStatus_)) return EXITED;
  51   if (WIFSIGNALED(rawStatus_)) return KILLED;
  52   throw std::runtime_error(to<std::string>(
  53       "Invalid ProcessReturnCode: ", rawStatus_));
  54 }
  55
  56 void ProcessReturnCode::enforce(State s) const {
  57   if (state() != s) {
  58     throw std::logic_error(to<std::string>("Invalid state ", s));
  59   }
  60 }
  61
  62 int ProcessReturnCode::exitStatus() const {
  63   enforce(EXITED);
  64   return WEXITSTATUS(rawStatus_);
  65 }
  66
  67 int ProcessReturnCode::killSignal() const {
  68   enforce(KILLED);
  69   return WTERMSIG(rawStatus_);
  70 }
  71
  72 bool ProcessReturnCode::coreDumped() const {
  73   enforce(KILLED);
  74   return WCOREDUMP(rawStatus_);
  75 }
  76
  77 std::string ProcessReturnCode::str() const {
  78   switch (state()) {
  79   case NOT_STARTED:
  80     return "not started";
  81   case RUNNING:
  82     return "running";
  83   case EXITED:
  84     return to<std::string>("exited with status ", exitStatus());
  85   case KILLED:
  86     return to<std::string>("killed by signal ", killSignal(),
  87                            (coreDumped() ? " (core dumped)" : ""));
  88   }
  89   CHECK(false);  // unreached
  90 }
  91
  92 CalledProcessError::CalledProcessError(ProcessReturnCode rc)
  93   : returnCode_(rc),
  94     what_(returnCode_.str()) {
  95 }
  96
  97 namespace {
  98
  99 // Copy pointers to the given strings in a format suitable for posix_spawn
 100 std::unique_ptr<const char*[]> cloneStrings(const std::vector<std::string>& s) {
 101   std::unique_ptr<const char*[]> d(new const char*[s.size() + 1]);
 102   for (int i = 0; i < s.size(); i++) {
 103     d[i] = s[i].c_str();
 104   }
 105   d[s.size()] = nullptr;
 106   return d;
 107 }
 108
 109 // Check a wait() status, throw on non-successful
 110 void checkStatus(ProcessReturnCode returnCode) {
 111   if (returnCode.state() != ProcessReturnCode::EXITED ||
 112       returnCode.exitStatus() != 0) {
 113     throw CalledProcessError(returnCode);
 114   }
 115 }
 116
 117 }  // namespace
 118
 119 Subprocess::Options& Subprocess::Options::fd(int fd, int action) {
 120   if (action == Subprocess::PIPE) {
 121     if (fd == 0) {
 122       action = Subprocess::PIPE_IN;
 123     } else if (fd == 1 || fd == 2) {
 124       action = Subprocess::PIPE_OUT;
 125     } else {
 126       throw std::invalid_argument(
 127           to<std::string>("Only fds 0, 1, 2 are valid for action=PIPE: ", fd));
 128     }
 129   }
 130   fdActions_[fd] = action;
 131   return *this;
 132 }
 133
 134 Subprocess::Subprocess(
 135     const std::vector<std::string>& argv,
 136     const Options& options,
 137     const char* executable,
 138     const std::vector<std::string>* env)
 139   : pid_(-1),
 140     returnCode_(RV_NOT_STARTED) {
 141   if (argv.empty()) {
 142     throw std::invalid_argument("argv must not be empty");
 143   }
 144   if (!executable) executable = argv[0].c_str();
 145   spawn(cloneStrings(argv), executable, options, env);
 146 }
 147
 148 Subprocess::Subprocess(
 149     const std::string& cmd,
 150     const Options& options,
 151     const std::vector<std::string>* env)
 152   : pid_(-1),
 153     returnCode_(RV_NOT_STARTED) {
 154   if (options.usePath_) {
 155     throw std::invalid_argument("usePath() not allowed when running in shell");
 156   }
 157   const char* shell = getenv("SHELL");
 158   if (!shell) {
 159     shell = "/bin/sh";
 160   }
 161
 162   std::unique_ptr<const char*[]> argv(new const char*[4]);
 163   argv[0] = shell;
 164   argv[1] = "-c";
 165   argv[2] = cmd.c_str();
 166   argv[3] = nullptr;
 167   spawn(std::move(argv), shell, options, env);
 168 }
 169
 170 Subprocess::~Subprocess() {
 171   CHECK_NE(returnCode_.state(), ProcessReturnCode::RUNNING)
 172     << "Subprocess destroyed without reaping child";
 173 }
 174
 175 namespace {
 176 void closeChecked(int fd) {
 177   checkUnixError(::close(fd), "close");
 178 }
 179 }  // namespace
 180
 181 void Subprocess::closeAll() {
 182   for (auto& p : pipes_) {
 183     closeChecked(p.parentFd);
 184   }
 185   pipes_.clear();
 186 }
 187
 188 void Subprocess::setAllNonBlocking() {
 189   for (auto& p : pipes_) {
 190     int fd = p.parentFd;
 191     int flags = ::fcntl(fd, F_GETFL);
 192     checkUnixError(flags, "fcntl");
 193     int r = ::fcntl(fd, F_SETFL, flags | O_NONBLOCK);
 194     checkUnixError(r, "fcntl");
 195   }
 196 }
 197
 198 void Subprocess::spawn(
 199     std::unique_ptr<const char*[]> argv,
 200     const char* executable,
 201     const Options& optionsIn,
 202     const std::vector<std::string>* env) {
 203   if (optionsIn.usePath_ && env) {
 204     throw std::invalid_argument(
 205         "usePath() not allowed when overriding environment");
 206   }
 207
 208   // Make a copy, we'll mutate options
 209   Options options(optionsIn);
 210
 211   // Parent work, pre-fork: create pipes
 212   std::vector<int> childFds;
 213
 214   // If we throw, don't leak file descriptors
 215   auto guard = makeGuard([&] {
 216     // These are only pipes, closing them shouldn't fail
 217     for (int cfd : childFds) {
 218       CHECK_ERR(::close(cfd));
 219     }
 220     for (auto& p : this->pipes_) {
 221       CHECK_ERR(::close(p.parentFd));
 222     }
 223   });
 224
 225   for (auto& p : options.fdActions_) {
 226     if (p.second == PIPE_IN || p.second == PIPE_OUT) {
 227       int fds[2];
 228       int r = ::pipe(fds);
 229       checkUnixError(r, "pipe");
 230       PipeInfo pinfo;
 231       pinfo.direction = p.second;
 232       int cfd;
 233       if (p.second == PIPE_IN) {
 234         // Child gets reading end
 235         pinfo.parentFd = fds[1];
 236         cfd = fds[0];
 237       } else {
 238         pinfo.parentFd = fds[0];
 239         cfd = fds[1];
 240       }
 241       p.second = cfd;  // ensure it gets dup2()ed
 242       pinfo.childFd = p.first;
 243       childFds.push_back(cfd);
 244       pipes_.push_back(pinfo);
 245     }
 246   }
 247
 248   // This should already be sorted, as options.fdActions_ is
 249   DCHECK(std::is_sorted(pipes_.begin(), pipes_.end()));
 250
 251   // Note that the const casts below are legit, per
 252   // http://pubs.opengroup.org/onlinepubs/009695399/functions/exec.html
 253
 254   char** argVec = const_cast<char**>(argv.get());
 255
 256   // Set up environment
 257   std::unique_ptr<const char*[]> envHolder;
 258   char** envVec;
 259   if (env) {
 260     envHolder = cloneStrings(*env);
 261     envVec = const_cast<char**>(envHolder.get());
 262   } else {
 263     envVec = environ;
 264   }
 265
 266   // Block all signals around vfork; see http://ewontfix.com/7/.
 267   //
 268   // As the child may run in the same address space as the parent until
 269   // the actual execve() system call, any (custom) signal handlers that
 270   // the parent has might alter parent's memory if invoked in the child,
 271   // with undefined results.  So we block all signals in the parent before
 272   // vfork(), which will cause them to be blocked in the child as well (we
 273   // rely on the fact that Linux, just like all sane implementations, only
 274   // clones the calling thread).  Then, in the child, we reset all signals
 275   // to their default dispositions (while still blocked), and unblock them
 276   // (so the exec()ed process inherits the parent's signal mask)
 277   //
 278   // The parent also unblocks all signals as soon as vfork() returns.
 279   sigset_t allBlocked;
 280   int r = ::sigfillset(&allBlocked);
 281   checkUnixError(r, "sigfillset");
 282   sigset_t oldSignals;
 283   r = pthread_sigmask(SIG_SETMASK, &allBlocked, &oldSignals);
 284   checkPosixError(r, "pthread_sigmask");
 285
 286   pid_t pid = vfork();
 287   if (pid == 0) {
 288     // While all signals are blocked, we must reset their
 289     // dispositions to default.
 290     for (int sig = 1; sig < NSIG; ++sig) {
 291       ::signal(sig, SIG_DFL);
 292     }
 293     // Unblock signals; restore signal mask.
 294     int r = pthread_sigmask(SIG_SETMASK, &oldSignals, nullptr);
 295     if (r != 0) _exit(kChildFailure);
 296
 297     runChild(executable, argVec, envVec, options);
 298     // This should never return, but there's nothing else we can do here.
 299     _exit(kExecFailure);
 300   }
 301   // In parent.  We want to restore the signal mask even if vfork fails,
 302   // so we'll save errno here, restore the signal mask, and only then
 303   // throw.
 304   int savedErrno = errno;
 305
 306   // Restore signal mask; do this even if vfork fails!
 307   r = pthread_sigmask(SIG_SETMASK, &oldSignals, nullptr);
 308   CHECK_EQ(r, 0) << "pthread_sigmask: " << errnoStr(r);  // shouldn't fail
 309   checkUnixError(pid, savedErrno, "vfork");
 310
 311   // Child is alive.  We can't throw any more, as we can't figure out
 312   // what to do with the child.
 313   guard.dismiss();
 314   pid_ = pid;
 315   returnCode_ = ProcessReturnCode(RV_RUNNING);
 316
 317   // Parent work, post-fork: close child's ends of pipes; closing them
 318   // shouldn't fail.
 319   for (int f : childFds) {
 320     CHECK_ERR(::close(f));
 321   }
 322 }
 323
 324 namespace {
 325
 326 // Checked version of close() to use in the child: _exit(126) on error
 327 void childClose(int fd) {
 328   int r = ::close(fd);
 329   if (r == -1) _exit(kChildFailure);
 330 }
 331
 332 // Checked version of dup2() to use in the child: _exit(126) on error
 333 void childDup2(int oldfd, int newfd) {
 334   int r = ::dup2(oldfd, newfd);
 335   if (r == -1) _exit(kChildFailure);
 336 }
 337
 338 }  // namespace
 339
 340 void Subprocess::runChild(const char* executable,
 341                           char** argv, char** env,
 342                           const Options& options) const {
 343   // Close parent's ends of all pipes
 344   for (auto& p : pipes_) {
 345     childClose(p.parentFd);
 346   }
 347
 348   // Close all fds that we're supposed to close.
 349   // Note that we're ignoring errors here, in case some of these
 350   // fds were set to close on exec.
 351   for (auto& p : options.fdActions_) {
 352     if (p.second == CLOSE) {
 353       ::close(p.first);
 354     } else {
 355       childDup2(p.second, p.first);
 356     }
 357   }
 358
 359   // If requested, close all other file descriptors.  Don't close
 360   // any fds in options.fdActions_, and don't touch stdin, stdout, stderr.
 361   // Ignore errors.
 362   if (options.closeOtherFds_) {
 363     for (int fd = getdtablesize() - 1; fd >= 3; --fd) {
 364       if (options.fdActions_.count(fd) == 0) {
 365         ::close(fd);
 366       }
 367     }
 368   }
 369
 370   // Opt to receive signal on parent death, if requested
 371   if (options.parentDeathSignal_ != 0) {
 372     int r = prctl(PR_SET_PDEATHSIG, options.parentDeathSignal_, 0, 0, 0);
 373     if (r == -1) {
 374       _exit(kChildFailure);
 375     }
 376   }
 377
 378   // Now, finally, exec.
 379   int r;
 380   if (options.usePath_) {
 381     ::execvp(executable, argv);
 382   } else {
 383     ::execve(executable, argv, env);
 384   }
 385 }
 386
 387 ProcessReturnCode Subprocess::poll() {
 388   returnCode_.enforce(ProcessReturnCode::RUNNING);
 389   DCHECK_GT(pid_, 0);
 390   int status;
 391   pid_t found = ::waitpid(pid_, &status, WNOHANG);
 392   checkUnixError(found, "waitpid");
 393   if (found != 0) {
 394     returnCode_ = ProcessReturnCode(status);
 395     pid_ = -1;
 396   }
 397   return returnCode_;
 398 }
 399
 400 bool Subprocess::pollChecked() {
 401   if (poll().state() == ProcessReturnCode::RUNNING) {
 402     return false;
 403   }
 404   checkStatus(returnCode_);
 405   return true;
 406 }
 407
 408 ProcessReturnCode Subprocess::wait() {
 409   returnCode_.enforce(ProcessReturnCode::RUNNING);
 410   DCHECK_GT(pid_, 0);
 411   int status;
 412   pid_t found;
 413   do {
 414     found = ::waitpid(pid_, &status, 0);
 415   } while (found == -1 && errno == EINTR);
 416   checkUnixError(found, "waitpid");
 417   DCHECK_EQ(found, pid_);
 418   returnCode_ = ProcessReturnCode(status);
 419   pid_ = -1;
 420   return returnCode_;
 421 }
 422
 423 void Subprocess::waitChecked() {
 424   wait();
 425   checkStatus(returnCode_);
 426 }
 427
 428 void Subprocess::sendSignal(int signal) {
 429   returnCode_.enforce(ProcessReturnCode::RUNNING);
 430   int r = ::kill(pid_, signal);
 431   checkUnixError(r, "kill");
 432 }
 433
 434 pid_t Subprocess::pid() const {
 435   return pid_;
 436 }
 437
 438 namespace {
 439
 440 std::pair<const uint8_t*, size_t> queueFront(const IOBufQueue& queue) {
 441   auto* p = queue.front();
 442   if (!p) return std::make_pair(nullptr, 0);
 443   return io::Cursor(p).peek();
 444 }
 445
 446 // fd write
 447 bool handleWrite(int fd, IOBufQueue& queue) {
 448   for (;;) {
 449     auto p = queueFront(queue);
 450     if (p.second == 0) {
 451       return true;  // EOF
 452     }
 453
 454     ssize_t n;
 455     do {
 456       n = ::write(fd, p.first, p.second);
 457     } while (n == -1 && errno == EINTR);
 458     if (n == -1 && errno == EAGAIN) {
 459       return false;
 460     }
 461     checkUnixError(n, "write");
 462     queue.trimStart(n);
 463   }
 464 }
 465
 466 // fd read
 467 bool handleRead(int fd, IOBufQueue& queue) {
 468   for (;;) {
 469     auto p = queue.preallocate(100, 65000);
 470     ssize_t n;
 471     do {
 472       n = ::read(fd, p.first, p.second);
 473     } while (n == -1 && errno == EINTR);
 474     if (n == -1 && errno == EAGAIN) {
 475       return false;
 476     }
 477     checkUnixError(n, "read");
 478     if (n == 0) {
 479       return true;
 480     }
 481     queue.postallocate(n);
 482   }
 483 }
 484
 485 bool discardRead(int fd) {
 486   static const size_t bufSize = 65000;
 487   // Thread unsafe, but it doesn't matter.
 488   static std::unique_ptr<char[]> buf(new char[bufSize]);
 489
 490   for (;;) {
 491     ssize_t n;
 492     do {
 493       n = ::read(fd, buf.get(), bufSize);
 494     } while (n == -1 && errno == EINTR);
 495     if (n == -1 && errno == EAGAIN) {
 496       return false;
 497     }
 498     checkUnixError(n, "read");
 499     if (n == 0) {
 500       return true;
 501     }
 502   }
 503 }
 504
 505 }  // namespace
 506
 507 std::pair<std::string, std::string> Subprocess::communicate(
 508     const CommunicateFlags& flags,
 509     StringPiece data) {
 510   IOBufQueue dataQueue;
 511   dataQueue.wrapBuffer(data.data(), data.size());
 512
 513   auto outQueues = communicateIOBuf(flags, std::move(dataQueue));
 514   auto outBufs = std::make_pair(outQueues.first.move(),
 515                                 outQueues.second.move());
 516   std::pair<std::string, std::string> out;
 517   if (outBufs.first) {
 518     outBufs.first->coalesce();
 519     out.first.assign(reinterpret_cast<const char*>(outBufs.first->data()),
 520                      outBufs.first->length());
 521   }
 522   if (outBufs.second) {
 523     outBufs.second->coalesce();
 524     out.second.assign(reinterpret_cast<const char*>(outBufs.second->data()),
 525                      outBufs.second->length());
 526   }
 527   return out;
 528 }
 529
 530 std::pair<IOBufQueue, IOBufQueue> Subprocess::communicateIOBuf(
 531     const CommunicateFlags& flags,
 532     IOBufQueue data) {
 533   std::pair<IOBufQueue, IOBufQueue> out;
 534
 535   auto readCallback = [&] (int pfd, int cfd) -> bool {
 536     if (cfd == 1 && flags.readStdout_) {
 537       return handleRead(pfd, out.first);
 538     } else if (cfd == 2 && flags.readStderr_) {
 539       return handleRead(pfd, out.second);
 540     } else {
 541       // Don't close the file descriptor, the child might not like SIGPIPE,
 542       // just read and throw the data away.
 543       return discardRead(pfd);
 544     }
 545   };
 546
 547   auto writeCallback = [&] (int pfd, int cfd) -> bool {
 548     if (cfd == 0 && flags.writeStdin_) {
 549       return handleWrite(pfd, data);
 550     } else {
 551       // If we don't want to write to this fd, just close it.
 552       return false;
 553     }
 554   };
 555
 556   communicate(std::move(readCallback), std::move(writeCallback));
 557
 558   return out;
 559 }
 560
 561 void Subprocess::communicate(FdCallback readCallback,
 562                              FdCallback writeCallback) {
 563   returnCode_.enforce(ProcessReturnCode::RUNNING);
 564   setAllNonBlocking();
 565
 566   std::vector<pollfd> fds;
 567   fds.reserve(pipes_.size());
 568   std::vector<int> toClose;
 569   toClose.reserve(pipes_.size());
 570
 571   while (!pipes_.empty()) {
 572     fds.clear();
 573     toClose.clear();
 574
 575     for (auto& p : pipes_) {
 576       pollfd pfd;
 577       pfd.fd = p.parentFd;
 578       // Yes, backwards, PIPE_IN / PIPE_OUT are defined from the
 579       // child's point of view.
 580       pfd.events = (p.direction == PIPE_IN ?  POLLOUT : POLLIN);
 581       fds.push_back(pfd);
 582     }
 583
 584     int r;
 585     do {
 586       r = ::poll(fds.data(), fds.size(), -1);
 587     } while (r == -1 && errno == EINTR);
 588     checkUnixError(r, "poll");
 589
 590     for (int i = 0; i < pipes_.size(); ++i) {
 591       auto& p = pipes_[i];
 592       DCHECK_EQ(fds[i].fd, p.parentFd);
 593       short events = fds[i].revents;
 594
 595       bool closed = false;
 596       if (events & POLLOUT) {
 597         DCHECK(!(events & POLLIN));
 598         if (writeCallback(p.parentFd, p.childFd)) {
 599           toClose.push_back(i);
 600           closed = true;
 601         }
 602       }
 603
 604       if (events & POLLIN) {
 605         DCHECK(!(events & POLLOUT));
 606         if (readCallback(p.parentFd, p.childFd)) {
 607           toClose.push_back(i);
 608           closed = true;
 609         }
 610       }
 611
 612       if ((events & (POLLHUP | POLLERR)) && !closed) {
 613         toClose.push_back(i);
 614         closed = true;
 615       }
 616     }
 617
 618     // Close the fds in reverse order so the indexes hold after erase()
 619     for (int idx : boost::adaptors::reverse(toClose)) {
 620       auto pos = pipes_.begin() + idx;
 621       closeChecked(pos->parentFd);
 622       pipes_.erase(pos);
 623     }
 624   }
 625 }
 626
 627 int Subprocess::findByChildFd(int childFd) const {
 628   auto pos = std::lower_bound(
 629       pipes_.begin(), pipes_.end(), childFd,
 630       [] (const PipeInfo& info, int fd) { return info.childFd < fd; });
 631   if (pos == pipes_.end() || pos->childFd != childFd) {
 632     throw std::invalid_argument(folly::to<std::string>(
 633         "child fd not found ", childFd));
 634   }
 635   return pos - pipes_.begin();
 636 }
 637
 638 void Subprocess::closeParentFd(int childFd) {
 639   int idx = findByChildFd(childFd);
 640   closeChecked(pipes_[idx].parentFd);
 641   pipes_.erase(pipes_.begin() + idx);
 642 }
 643
 644 namespace {
 645
 646 class Initializer {
 647  public:
 648   Initializer() {
 649     // We like EPIPE, thanks.
 650     ::signal(SIGPIPE, SIG_IGN);
 651   }
 652 };
 653
 654 Initializer initializer;
 655
 656 }  // namespace
 657
 658 }  // namespace folly
 659