folly/Subprocess.cpp

   1 /*
   2  * Copyright 2012 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 <fcntl.h>
  20 #include <poll.h>
  21 #include <unistd.h>
  22 #include <wait.h>
  23
  24 #include <array>
  25 #include <algorithm>
  26 #include <system_error>
  27
  28 #include <boost/container/flat_set.hpp>
  29 #include <boost/range/adaptors.hpp>
  30
  31 #include <glog/logging.h>
  32
  33 #include "folly/Conv.h"
  34 #include "folly/ScopeGuard.h"
  35 #include "folly/String.h"
  36 #include "folly/experimental/io/Cursor.h"
  37
  38 extern char** environ;
  39
  40 namespace folly {
  41
  42 ProcessReturnCode::State ProcessReturnCode::state() const {
  43   if (rawStatus_ == RV_NOT_STARTED) return NOT_STARTED;
  44   if (rawStatus_ == RV_RUNNING) return RUNNING;
  45   if (WIFEXITED(rawStatus_)) return EXITED;
  46   if (WIFSIGNALED(rawStatus_)) return KILLED;
  47   throw std::runtime_error(to<std::string>(
  48       "Invalid ProcessReturnCode: ", rawStatus_));
  49 }
  50
  51 void ProcessReturnCode::enforce(State s) const {
  52   if (state() != s) {
  53     throw std::logic_error(to<std::string>("Invalid state ", s));
  54   }
  55 }
  56
  57 int ProcessReturnCode::exitStatus() const {
  58   enforce(EXITED);
  59   return WEXITSTATUS(rawStatus_);
  60 }
  61
  62 int ProcessReturnCode::killSignal() const {
  63   enforce(KILLED);
  64   return WTERMSIG(rawStatus_);
  65 }
  66
  67 bool ProcessReturnCode::coreDumped() const {
  68   enforce(KILLED);
  69   return WCOREDUMP(rawStatus_);
  70 }
  71
  72 std::string ProcessReturnCode::str() const {
  73   switch (state()) {
  74   case NOT_STARTED:
  75     return "not started";
  76   case RUNNING:
  77     return "running";
  78   case EXITED:
  79     return to<std::string>("exited with status ", exitStatus());
  80   case KILLED:
  81     return to<std::string>("killed by signal ", killSignal(),
  82                            (coreDumped() ? " (core dumped)" : ""));
  83   }
  84   CHECK(false);  // unreached
  85 }
  86
  87 CalledProcessError::CalledProcessError(ProcessReturnCode rc)
  88   : returnCode_(rc),
  89     what_(returnCode_.str()) {
  90 }
  91
  92 namespace {
  93
  94 // Copy pointers to the given strings in a format suitable for posix_spawn
  95 std::unique_ptr<const char*[]> cloneStrings(const std::vector<std::string>& s) {
  96   std::unique_ptr<const char*[]> d(new const char*[s.size() + 1]);
  97   for (int i = 0; i < s.size(); i++) {
  98     d[i] = s[i].c_str();
  99   }
 100   d[s.size()] = nullptr;
 101   return d;
 102 }
 103
 104 // Helper to throw std::system_error
 105 void throwSystemError(int err, const char* msg) __attribute__((noreturn));
 106 void throwSystemError(int err, const char* msg) {
 107   throw std::system_error(err, std::system_category(), msg);
 108 }
 109
 110 // Helper to throw std::system_error from errno
 111 void throwSystemError(const char* msg) __attribute__((noreturn));
 112 void throwSystemError(const char* msg) {
 113   throwSystemError(errno, msg);
 114 }
 115
 116 // Check a Posix return code (0 on success, error number on error), throw
 117 // on error.
 118 void checkPosixError(int err, const char* msg) {
 119   if (err != 0) {
 120     throwSystemError(err, msg);
 121   }
 122 }
 123
 124 // Check a traditional Uinx return code (-1 and sets errno on error), throw
 125 // on error.
 126 void checkUnixError(ssize_t ret, const char* msg) {
 127   if (ret == -1) {
 128     throwSystemError(msg);
 129   }
 130 }
 131
 132 // Check a wait() status, throw on non-successful
 133 void checkStatus(ProcessReturnCode returnCode) {
 134   if (returnCode.state() != ProcessReturnCode::EXITED ||
 135       returnCode.exitStatus() != 0) {
 136     throw CalledProcessError(returnCode);
 137   }
 138 }
 139
 140 }  // namespace
 141
 142 Subprocess::Options& Subprocess::Options::fd(int fd, int action) {
 143   if (action == Subprocess::PIPE) {
 144     if (fd == 0) {
 145       action = Subprocess::PIPE_IN;
 146     } else if (fd == 1 || fd == 2) {
 147       action = Subprocess::PIPE_OUT;
 148     } else {
 149       throw std::invalid_argument(
 150           to<std::string>("Only fds 0, 1, 2 are valid for action=PIPE: ", fd));
 151     }
 152   }
 153   fdActions_[fd] = action;
 154   return *this;
 155 }
 156
 157 Subprocess::Subprocess(
 158     const std::vector<std::string>& argv,
 159     const Options& options,
 160     const char* executable,
 161     const std::vector<std::string>* env)
 162   : pid_(-1),
 163     returnCode_(RV_NOT_STARTED) {
 164   if (argv.empty()) {
 165     throw std::invalid_argument("argv must not be empty");
 166   }
 167   if (!executable) executable = argv[0].c_str();
 168   spawn(cloneStrings(argv), executable, options, env);
 169 }
 170
 171 Subprocess::Subprocess(
 172     const std::string& cmd,
 173     const Options& options,
 174     const std::vector<std::string>* env)
 175   : pid_(-1),
 176     returnCode_(RV_NOT_STARTED) {
 177   if (options.usePath_) {
 178     throw std::invalid_argument("usePath() not allowed when running in shell");
 179   }
 180   const char* shell = getenv("SHELL");
 181   if (!shell) {
 182     shell = "/bin/sh";
 183   }
 184
 185   std::unique_ptr<const char*[]> argv(new const char*[4]);
 186   argv[0] = shell;
 187   argv[1] = "-c";
 188   argv[2] = cmd.c_str();
 189   argv[3] = nullptr;
 190   spawn(std::move(argv), shell, options, env);
 191 }
 192
 193 Subprocess::~Subprocess() {
 194   if (returnCode_.state() == ProcessReturnCode::RUNNING) {
 195     LOG(ERROR) << "Subprocess destroyed without reaping; killing child.";
 196     try {
 197       kill();
 198       wait();
 199     } catch (...) {
 200       LOG(FATAL) << "Killing child failed, terminating: "
 201                  << exceptionStr(std::current_exception());
 202     }
 203   }
 204   try {
 205     closeAll();
 206   } catch (...) {
 207     LOG(FATAL) << "close failed, terminating: "
 208                << exceptionStr(std::current_exception());
 209   }
 210 }
 211
 212 namespace {
 213 void closeChecked(int fd) {
 214   checkUnixError(::close(fd), "close");
 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   // Parent work, pre-fork: create pipes
 249   std::vector<int> childFds;
 250   for (auto& p : options.fdActions_) {
 251     if (p.second == PIPE_IN || p.second == PIPE_OUT) {
 252       int fds[2];
 253       int r = ::pipe(fds);
 254       checkUnixError(r, "pipe");
 255       PipeInfo pinfo;
 256       pinfo.direction = p.second;
 257       int cfd;
 258       if (p.second == PIPE_IN) {
 259         // Child gets reading end
 260         pinfo.parentFd = fds[1];
 261         cfd = fds[0];
 262       } else {
 263         pinfo.parentFd = fds[0];
 264         cfd = fds[1];
 265       }
 266       p.second = cfd;  // ensure it gets dup2()ed
 267       pinfo.childFd = p.first;
 268       childFds.push_back(cfd);
 269       pipes_.push_back(pinfo);
 270     }
 271   }
 272
 273   // This should already be sorted, as options.fdActions_ is
 274   DCHECK(std::is_sorted(pipes_.begin(), pipes_.end()));
 275
 276   // Note that the const casts below are legit, per
 277   // http://pubs.opengroup.org/onlinepubs/009695399/functions/exec.html
 278
 279   char** argVec = const_cast<char**>(argv.get());
 280
 281   // Set up environment
 282   std::unique_ptr<const char*[]> envHolder;
 283   char** envVec;
 284   if (env) {
 285     envHolder = cloneStrings(*env);
 286     envVec = const_cast<char**>(envHolder.get());
 287   } else {
 288     envVec = environ;
 289   }
 290
 291   pid_t pid = vfork();
 292   if (pid == 0) {
 293     runChild(executable, argVec, envVec, options);
 294     // This should never return, but there's nothing else we can do here.
 295     abort();
 296   }
 297
 298   // In parent
 299   checkUnixError(pid, "vfork");
 300   pid_ = pid;
 301   returnCode_ = ProcessReturnCode(RV_RUNNING);
 302
 303   // Parent work, post-fork: close child's ends of pipes
 304   for (int f : childFds) {
 305     closeChecked(f);
 306   }
 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   // Now, finally, exec.
 356   int r;
 357   if (options.usePath_) {
 358     ::execvp(executable, argv);
 359   } else {
 360     ::execve(executable, argv, env);
 361   }
 362
 363   // If we're here, something's wrong.
 364   abort();
 365 }
 366
 367 ProcessReturnCode Subprocess::poll() {
 368   returnCode_.enforce(ProcessReturnCode::RUNNING);
 369   DCHECK_GT(pid_, 0);
 370   int status;
 371   pid_t found = ::waitpid(pid_, &status, WNOHANG);
 372   checkUnixError(found, "waitpid");
 373   if (found != 0) {
 374     returnCode_ = ProcessReturnCode(status);
 375     pid_ = -1;
 376   }
 377   return returnCode_;
 378 }
 379
 380 bool Subprocess::pollChecked() {
 381   if (poll().state() == ProcessReturnCode::RUNNING) {
 382     return false;
 383   }
 384   checkStatus(returnCode_);
 385   return true;
 386 }
 387
 388 ProcessReturnCode Subprocess::wait() {
 389   returnCode_.enforce(ProcessReturnCode::RUNNING);
 390   DCHECK_GT(pid_, 0);
 391   int status;
 392   pid_t found = ::waitpid(pid_, &status, 0);
 393   checkUnixError(found, "waitpid");
 394   returnCode_ = ProcessReturnCode(status);
 395   return returnCode_;
 396 }
 397
 398 void Subprocess::waitChecked() {
 399   wait();
 400   checkStatus(returnCode_);
 401 }
 402
 403 void Subprocess::sendSignal(int signal) {
 404   returnCode_.enforce(ProcessReturnCode::RUNNING);
 405   int r = ::kill(pid_, signal);
 406   checkUnixError(r, "kill");
 407 }
 408
 409 namespace {
 410 void setNonBlocking(int fd) {
 411   int flags = ::fcntl(fd, F_GETFL);
 412   checkUnixError(flags, "fcntl");
 413   int r = ::fcntl(fd, F_SETFL, flags | O_NONBLOCK);
 414   checkUnixError(r, "fcntl");
 415 }
 416
 417 std::pair<const uint8_t*, size_t> queueFront(const IOBufQueue& queue) {
 418   auto* p = queue.front();
 419   if (!p) return std::make_pair(nullptr, 0);
 420   return io::Cursor(p).peek();
 421 }
 422
 423 // fd write
 424 bool handleWrite(int fd, IOBufQueue& queue) {
 425   for (;;) {
 426     auto p = queueFront(queue);
 427     if (p.second == 0) {
 428       return true;  // EOF
 429     }
 430
 431     ssize_t n;
 432     do {
 433       n = ::write(fd, p.first, p.second);
 434     } while (n == -1 && errno == EINTR);
 435     if (n == -1 && errno == EAGAIN) {
 436       return false;
 437     }
 438     checkUnixError(n, "write");
 439     queue.trimStart(n);
 440   }
 441 }
 442
 443 // fd read
 444 bool handleRead(int fd, IOBufQueue& queue) {
 445   for (;;) {
 446     auto p = queue.preallocate(100, 65000);
 447     ssize_t n;
 448     do {
 449       n = ::read(fd, p.first, p.second);
 450     } while (n == -1 && errno == EINTR);
 451     if (n == -1 && errno == EAGAIN) {
 452       return false;
 453     }
 454     checkUnixError(n, "read");
 455     if (n == 0) {
 456       return true;
 457     }
 458     queue.postallocate(n);
 459   }
 460 }
 461
 462 bool discardRead(int fd) {
 463   static const size_t bufSize = 65000;
 464   // Thread unsafe, but it doesn't matter.
 465   static std::unique_ptr<char[]> buf(new char[bufSize]);
 466
 467   for (;;) {
 468     ssize_t n;
 469     do {
 470       n = ::read(fd, buf.get(), bufSize);
 471     } while (n == -1 && errno == EINTR);
 472     if (n == -1 && errno == EAGAIN) {
 473       return false;
 474     }
 475     checkUnixError(n, "read");
 476     if (n == 0) {
 477       return true;
 478     }
 479   }
 480 }
 481
 482 }  // namespace
 483
 484 std::pair<std::string, std::string> Subprocess::communicate(
 485     int flags,
 486     StringPiece data) {
 487   IOBufQueue dataQueue;
 488   dataQueue.wrapBuffer(data.data(), data.size());
 489
 490   auto outQueues = communicateIOBuf(flags, std::move(dataQueue));
 491   auto outBufs = std::make_pair(outQueues.first.move(),
 492                                 outQueues.second.move());
 493   std::pair<std::string, std::string> out;
 494   if (outBufs.first) {
 495     outBufs.first->coalesce();
 496     out.first.assign(reinterpret_cast<const char*>(outBufs.first->data()),
 497                      outBufs.first->length());
 498   }
 499   if (outBufs.second) {
 500     outBufs.second->coalesce();
 501     out.second.assign(reinterpret_cast<const char*>(outBufs.second->data()),
 502                      outBufs.second->length());
 503   }
 504   return out;
 505 }
 506
 507 std::pair<IOBufQueue, IOBufQueue> Subprocess::communicateIOBuf(
 508     int flags,
 509     IOBufQueue data) {
 510   std::pair<IOBufQueue, IOBufQueue> out;
 511
 512   auto readCallback = [&, flags] (int pfd, int cfd) {
 513     if (cfd == 1 && (flags & READ_STDOUT)) {
 514       return handleRead(pfd, out.first);
 515     } else if (cfd == 2 && (flags & READ_STDERR)) {
 516       return handleRead(pfd, out.second);
 517     } else {
 518       // Don't close the file descriptor, the child might not like SIGPIPE,
 519       // just read and throw the data away.
 520       return discardRead(pfd);
 521     }
 522   };
 523
 524   auto writeCallback = [&, flags] (int pfd, int cfd) {
 525     if (cfd == 0 && (flags & WRITE_STDIN)) {
 526       return handleWrite(pfd, data);
 527     } else {
 528       // If we don't want to write to this fd, just close it.
 529       return false;
 530     }
 531   };
 532
 533   communicate(std::move(readCallback), std::move(writeCallback));
 534
 535   return out;
 536 }
 537
 538 void Subprocess::communicate(FdCallback readCallback,
 539                              FdCallback writeCallback) {
 540   returnCode_.enforce(ProcessReturnCode::RUNNING);
 541   setAllNonBlocking();
 542
 543   std::vector<pollfd> fds;
 544   fds.reserve(pipes_.size());
 545   std::vector<int> toClose;
 546   toClose.reserve(pipes_.size());
 547
 548   while (!pipes_.empty()) {
 549     fds.clear();
 550     toClose.clear();
 551
 552     for (auto& p : pipes_) {
 553       pollfd pfd;
 554       pfd.fd = p.parentFd;
 555       // Yes, backwards, PIPE_IN / PIPE_OUT are defined from the
 556       // child's point of view.
 557       pfd.events = (p.direction == PIPE_IN ?  POLLOUT : POLLIN);
 558       fds.push_back(pfd);
 559     }
 560
 561     int r;
 562     do {
 563       r = ::poll(fds.data(), fds.size(), -1);
 564     } while (r == -1 && errno == EINTR);
 565     checkUnixError(r, "poll");
 566
 567     for (int i = 0; i < pipes_.size(); ++i) {
 568       auto& p = pipes_[i];
 569       DCHECK_EQ(fds[i].fd, p.parentFd);
 570       short events = fds[i].revents;
 571
 572       bool closed = false;
 573       if (events & POLLOUT) {
 574         DCHECK(!(events & POLLIN));
 575         if (writeCallback(p.parentFd, p.childFd)) {
 576           toClose.push_back(i);
 577           closed = true;
 578         }
 579       }
 580
 581       if (events & POLLIN) {
 582         DCHECK(!(events & POLLOUT));
 583         if (readCallback(p.parentFd, p.childFd)) {
 584           toClose.push_back(i);
 585           closed = true;
 586         }
 587       }
 588
 589       if ((events & (POLLHUP | POLLERR)) && !closed) {
 590         toClose.push_back(i);
 591         closed = true;
 592       }
 593     }
 594
 595     // Close the fds in reverse order so the indexes hold after erase()
 596     for (int idx : boost::adaptors::reverse(toClose)) {
 597       auto pos = pipes_.begin() + idx;
 598       closeChecked(pos->parentFd);
 599       pipes_.erase(pos);
 600     }
 601   }
 602 }
 603
 604 int Subprocess::findByChildFd(int childFd) const {
 605   auto pos = std::lower_bound(
 606       pipes_.begin(), pipes_.end(), childFd,
 607       [] (const PipeInfo& info, int fd) { return info.childFd < fd; });
 608   if (pos == pipes_.end() || pos->childFd != childFd) {
 609     throw std::invalid_argument(folly::to<std::string>(
 610         "child fd not found ", childFd));
 611   }
 612   return pos - pipes_.begin();
 613 }
 614
 615 void Subprocess::closeParentFd(int childFd) {
 616   int idx = findByChildFd(childFd);
 617   closeChecked(pipes_[idx].parentFd);
 618   pipes_.erase(pipes_.begin() + idx);
 619 }
 620
 621 namespace {
 622
 623 class Initializer {
 624  public:
 625   Initializer() {
 626     // We like EPIPE, thanks.
 627     ::signal(SIGPIPE, SIG_IGN);
 628   }
 629 };
 630
 631 Initializer initializer;
 632
 633 }  // namespace
 634
 635 }  // namespace folly
 636