Commit 675a137d authored by Alexey Spiridonov's avatar Alexey Spiridonov Committed by Praveen Kumar Ramakrishnan

Factor out pipe methods, add takeOwnershipOfPipes

Summary:
In order to create an EventBase'd Suprocess class, I'd like to be able to manage the lifetime of pipes independently of the lifetime of the process. To this effect, I factored out basic Pipe handling, and provided a function that detaches the pipe vector from the Subprocess object.

#6996492 a push-blocking test is broken in trunk

Test Plan: added a unit test, fbconfig -r folly && fbmake runtests && fbmake runtests_opt

Reviewed By: dancol@fb.com

Subscribers: yfeldblum, chalfant, dancol, wez, anarayanan, trunkagent, net-systems@, njormrod, folly-diffs@

FB internal diff: D1699969

Signature: t1:1699969:1430975299:30d291ab7fcc555edddf098b33095a5b29500e76
parent ec3c8f79
...@@ -96,6 +96,7 @@ std::string ProcessReturnCode::str() const { ...@@ -96,6 +96,7 @@ std::string ProcessReturnCode::str() const {
(coreDumped() ? " (core dumped)" : "")); (coreDumped() ? " (core dumped)" : ""));
} }
CHECK(false); // unreached CHECK(false); // unreached
return ""; // silence GCC warning
} }
CalledProcessError::CalledProcessError(ProcessReturnCode rc) CalledProcessError::CalledProcessError(ProcessReturnCode rc)
...@@ -189,13 +190,9 @@ Subprocess::Subprocess( ...@@ -189,13 +190,9 @@ Subprocess::Subprocess(
Subprocess::~Subprocess() { Subprocess::~Subprocess() {
CHECK_NE(returnCode_.state(), ProcessReturnCode::RUNNING) CHECK_NE(returnCode_.state(), ProcessReturnCode::RUNNING)
<< "Subprocess destroyed without reaping child"; << "Subprocess destroyed without reaping child";
closeAll();
} }
namespace { namespace {
void closeChecked(int fd) {
checkUnixError(::close(fd), "close");
}
struct ChildErrorInfo { struct ChildErrorInfo {
int errCode; int errCode;
...@@ -214,16 +211,9 @@ void childError(int errFd, int errCode, int errnoValue) { ...@@ -214,16 +211,9 @@ void childError(int errFd, int errCode, int errnoValue) {
} // namespace } // namespace
void Subprocess::closeAll() {
for (auto& p : pipes_) {
closeChecked(p.parentFd);
}
pipes_.clear();
}
void Subprocess::setAllNonBlocking() { void Subprocess::setAllNonBlocking() {
for (auto& p : pipes_) { for (auto& p : pipes_) {
int fd = p.parentFd; int fd = p.pipe.fd();
int flags = ::fcntl(fd, F_GETFL); int flags = ::fcntl(fd, F_GETFL);
checkUnixError(flags, "fcntl"); checkUnixError(flags, "fcntl");
int r = ::fcntl(fd, F_SETFL, flags | O_NONBLOCK); int r = ::fcntl(fd, F_SETFL, flags | O_NONBLOCK);
...@@ -244,12 +234,8 @@ void Subprocess::spawn( ...@@ -244,12 +234,8 @@ void Subprocess::spawn(
// Make a copy, we'll mutate options // Make a copy, we'll mutate options
Options options(optionsIn); Options options(optionsIn);
// On error, close all of the pipes_ // On error, close all pipes_ (ignoring errors, but that seems fine here).
auto pipesGuard = makeGuard([&] { auto pipesGuard = makeGuard([this] { pipes_.clear(); });
for (auto& p : this->pipes_) {
CHECK_ERR(::close(p.parentFd));
}
});
// Create a pipe to use to receive error information from the child, // Create a pipe to use to receive error information from the child,
// in case it fails before calling exec() // in case it fails before calling exec()
...@@ -325,6 +311,9 @@ void Subprocess::spawnInternal( ...@@ -325,6 +311,9 @@ void Subprocess::spawnInternal(
// doesn't need to reset the flag on its end, as we always dup2() the fd, // doesn't need to reset the flag on its end, as we always dup2() the fd,
// and dup2() fds don't share the close-on-exec flag. // and dup2() fds don't share the close-on-exec flag.
#if FOLLY_HAVE_PIPE2 #if FOLLY_HAVE_PIPE2
// If possible, set close-on-exec atomically. Otherwise, a concurrent
// Subprocess invocation can fork() between "pipe" and "fnctl",
// causing FDs to leak.
r = ::pipe2(fds, O_CLOEXEC); r = ::pipe2(fds, O_CLOEXEC);
checkUnixError(r, "pipe2"); checkUnixError(r, "pipe2");
#else #else
...@@ -335,21 +324,21 @@ void Subprocess::spawnInternal( ...@@ -335,21 +324,21 @@ void Subprocess::spawnInternal(
r = fcntl(fds[1], F_SETFD, FD_CLOEXEC); r = fcntl(fds[1], F_SETFD, FD_CLOEXEC);
checkUnixError(r, "set FD_CLOEXEC"); checkUnixError(r, "set FD_CLOEXEC");
#endif #endif
PipeInfo pinfo; pipes_.emplace_back();
pinfo.direction = p.second; Pipe& pipe = pipes_.back();
pipe.direction = p.second;
int cfd; int cfd;
if (p.second == PIPE_IN) { if (p.second == PIPE_IN) {
// Child gets reading end // Child gets reading end
pinfo.parentFd = fds[1]; pipe.pipe = folly::File(fds[1], /*owns_fd=*/ true);
cfd = fds[0]; cfd = fds[0];
} else { } else {
pinfo.parentFd = fds[0]; pipe.pipe = folly::File(fds[0], /*owns_fd=*/ true);
cfd = fds[1]; cfd = fds[1];
} }
p.second = cfd; // ensure it gets dup2()ed p.second = cfd; // ensure it gets dup2()ed
pinfo.childFd = p.first; pipe.childFd = p.first;
childFds.push_back(cfd); childFds.push_back(cfd);
pipes_.push_back(pinfo);
} }
} }
...@@ -543,6 +532,8 @@ ProcessReturnCode Subprocess::poll() { ...@@ -543,6 +532,8 @@ ProcessReturnCode Subprocess::poll() {
pid_t found = ::waitpid(pid_, &status, WNOHANG); pid_t found = ::waitpid(pid_, &status, WNOHANG);
checkUnixError(found, "waitpid"); checkUnixError(found, "waitpid");
if (found != 0) { if (found != 0) {
// Though the child process had quit, this call does not close the pipes
// since its descendants may still be using them.
returnCode_ = ProcessReturnCode(status); returnCode_ = ProcessReturnCode(status);
pid_ = -1; pid_ = -1;
} }
...@@ -566,6 +557,8 @@ ProcessReturnCode Subprocess::wait() { ...@@ -566,6 +557,8 @@ ProcessReturnCode Subprocess::wait() {
found = ::waitpid(pid_, &status, 0); found = ::waitpid(pid_, &status, 0);
} while (found == -1 && errno == EINTR); } while (found == -1 && errno == EINTR);
checkUnixError(found, "waitpid"); checkUnixError(found, "waitpid");
// Though the child process had quit, this call does not close the pipes
// since its descendants may still be using them.
DCHECK_EQ(found, pid_); DCHECK_EQ(found, pid_);
returnCode_ = ProcessReturnCode(status); returnCode_ = ProcessReturnCode(status);
pid_ = -1; pid_ = -1;
...@@ -709,12 +702,14 @@ std::pair<IOBufQueue, IOBufQueue> Subprocess::communicateIOBuf( ...@@ -709,12 +702,14 @@ std::pair<IOBufQueue, IOBufQueue> Subprocess::communicateIOBuf(
void Subprocess::communicate(FdCallback readCallback, void Subprocess::communicate(FdCallback readCallback,
FdCallback writeCallback) { FdCallback writeCallback) {
// This serves to prevent wait() followed by communicate(), but if you
// legitimately need that, send a patch to delete this line.
returnCode_.enforce(ProcessReturnCode::RUNNING); returnCode_.enforce(ProcessReturnCode::RUNNING);
setAllNonBlocking(); setAllNonBlocking();
std::vector<pollfd> fds; std::vector<pollfd> fds;
fds.reserve(pipes_.size()); fds.reserve(pipes_.size());
std::vector<int> toClose; std::vector<size_t> toClose; // indexes into pipes_
toClose.reserve(pipes_.size()); toClose.reserve(pipes_.size());
while (!pipes_.empty()) { while (!pipes_.empty()) {
...@@ -723,7 +718,7 @@ void Subprocess::communicate(FdCallback readCallback, ...@@ -723,7 +718,7 @@ void Subprocess::communicate(FdCallback readCallback,
for (auto& p : pipes_) { for (auto& p : pipes_) {
pollfd pfd; pollfd pfd;
pfd.fd = p.parentFd; pfd.fd = p.pipe.fd();
// Yes, backwards, PIPE_IN / PIPE_OUT are defined from the // Yes, backwards, PIPE_IN / PIPE_OUT are defined from the
// child's point of view. // child's point of view.
if (!p.enabled) { if (!p.enabled) {
...@@ -746,13 +741,14 @@ void Subprocess::communicate(FdCallback readCallback, ...@@ -746,13 +741,14 @@ void Subprocess::communicate(FdCallback readCallback,
for (size_t i = 0; i < pipes_.size(); ++i) { for (size_t i = 0; i < pipes_.size(); ++i) {
auto& p = pipes_[i]; auto& p = pipes_[i];
DCHECK_EQ(fds[i].fd, p.parentFd); auto parentFd = p.pipe.fd();
DCHECK_EQ(fds[i].fd, parentFd);
short events = fds[i].revents; short events = fds[i].revents;
bool closed = false; bool closed = false;
if (events & POLLOUT) { if (events & POLLOUT) {
DCHECK(!(events & POLLIN)); DCHECK(!(events & POLLIN));
if (writeCallback(p.parentFd, p.childFd)) { if (writeCallback(parentFd, p.childFd)) {
toClose.push_back(i); toClose.push_back(i);
closed = true; closed = true;
} }
...@@ -762,7 +758,7 @@ void Subprocess::communicate(FdCallback readCallback, ...@@ -762,7 +758,7 @@ void Subprocess::communicate(FdCallback readCallback,
// on) end of file // on) end of file
if (events & (POLLIN | POLLHUP)) { if (events & (POLLIN | POLLHUP)) {
DCHECK(!(events & POLLOUT)); DCHECK(!(events & POLLOUT));
if (readCallback(p.parentFd, p.childFd)) { if (readCallback(parentFd, p.childFd)) {
toClose.push_back(i); toClose.push_back(i);
closed = true; closed = true;
} }
...@@ -777,7 +773,7 @@ void Subprocess::communicate(FdCallback readCallback, ...@@ -777,7 +773,7 @@ void Subprocess::communicate(FdCallback readCallback,
// Close the fds in reverse order so the indexes hold after erase() // Close the fds in reverse order so the indexes hold after erase()
for (int idx : boost::adaptors::reverse(toClose)) { for (int idx : boost::adaptors::reverse(toClose)) {
auto pos = pipes_.begin() + idx; auto pos = pipes_.begin() + idx;
closeChecked(pos->parentFd); pos->pipe.close(); // Throws on error
pipes_.erase(pos); pipes_.erase(pos);
} }
} }
...@@ -791,10 +787,10 @@ bool Subprocess::notificationsEnabled(int childFd) const { ...@@ -791,10 +787,10 @@ bool Subprocess::notificationsEnabled(int childFd) const {
return pipes_[findByChildFd(childFd)].enabled; return pipes_[findByChildFd(childFd)].enabled;
} }
int Subprocess::findByChildFd(int childFd) const { size_t Subprocess::findByChildFd(int childFd) const {
auto pos = std::lower_bound( auto pos = std::lower_bound(
pipes_.begin(), pipes_.end(), childFd, pipes_.begin(), pipes_.end(), childFd,
[] (const PipeInfo& info, int fd) { return info.childFd < fd; }); [] (const Pipe& pipe, int fd) { return pipe.childFd < fd; });
if (pos == pipes_.end() || pos->childFd != childFd) { if (pos == pipes_.end() || pos->childFd != childFd) {
throw std::invalid_argument(folly::to<std::string>( throw std::invalid_argument(folly::to<std::string>(
"child fd not found ", childFd)); "child fd not found ", childFd));
...@@ -804,10 +800,19 @@ int Subprocess::findByChildFd(int childFd) const { ...@@ -804,10 +800,19 @@ int Subprocess::findByChildFd(int childFd) const {
void Subprocess::closeParentFd(int childFd) { void Subprocess::closeParentFd(int childFd) {
int idx = findByChildFd(childFd); int idx = findByChildFd(childFd);
closeChecked(pipes_[idx].parentFd); pipes_[idx].pipe.close(); // May throw
pipes_.erase(pipes_.begin() + idx); pipes_.erase(pipes_.begin() + idx);
} }
std::vector<Subprocess::ChildPipe> Subprocess::takeOwnershipOfPipes() {
std::vector<Subprocess::ChildPipe> pipes;
for (auto& p : pipes_) {
pipes.emplace_back(ChildPipe{p.childFd, std::move(p.pipe)});
}
pipes_.clear();
return pipes;
}
namespace { namespace {
class Initializer { class Initializer {
......
This diff is collapsed.
...@@ -444,3 +444,21 @@ TEST(CommunicateSubprocessTest, Chatty) { ...@@ -444,3 +444,21 @@ TEST(CommunicateSubprocessTest, Chatty) {
EXPECT_EQ(0, proc.wait().exitStatus()); EXPECT_EQ(0, proc.wait().exitStatus());
}); });
} }
TEST(CommunicateSubprocessTest, TakeOwnershipOfPipes) {
std::vector<Subprocess::ChildPipe> pipes;
{
Subprocess proc(
"echo $'oh\\nmy\\ncat' | wc -l &", Subprocess::pipeStdout()
);
pipes = proc.takeOwnershipOfPipes();
proc.waitChecked();
}
EXPECT_EQ(1, pipes.size());
EXPECT_EQ(1, pipes[0].childFd);
char buf[10];
EXPECT_EQ(2, readFull(pipes[0].pipe.fd(), buf, 10));
buf[2] = 0;
EXPECT_EQ("3\n", std::string(buf));
}
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment