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