+TEST(CommunicateSubprocessTest, ProcessGroupLeader) {
+ const auto testIsLeader = "test $(cut -d ' ' -f 5 /proc/$$/stat) = $$";
+ Subprocess nonLeader(testIsLeader);
+ EXPECT_THROW(nonLeader.waitChecked(), CalledProcessError);
+ Subprocess leader(testIsLeader, Subprocess::Options().processGroupLeader());
+ leader.waitChecked();
+}
+
+TEST(CommunicateSubprocessTest, Duplex2) {
+ checkFdLeak([] {
+ // Pipe 200,000 lines through sed
+ const size_t numCopies = 100000;
+ auto iobuf = IOBuf::copyBuffer("this is a test\nanother line\n");
+ IOBufQueue input;
+ for (size_t n = 0; n < numCopies; ++n) {
+ input.append(iobuf->clone());
+ }
+
+ std::vector<std::string> cmd({
+ "sed", "-u",
+ "-e", "s/a test/a successful test/",
+ "-e", "/^another line/w/dev/stderr",
+ });
+ auto options = Subprocess::pipeStdin().pipeStdout().pipeStderr().usePath();
+ Subprocess proc(cmd, options);
+ auto out = proc.communicateIOBuf(std::move(input));
+ proc.waitChecked();
+
+ // Convert stdout and stderr to strings so we can call split() on them.
+ fbstring stdoutStr;
+ if (out.first.front()) {
+ stdoutStr = out.first.move()->moveToFbString();
+ }
+ fbstring stderrStr;
+ if (out.second.front()) {
+ stderrStr = out.second.move()->moveToFbString();
+ }
+
+ // stdout should be a copy of stdin, with "a test" replaced by
+ // "a successful test"
+ std::vector<StringPiece> stdoutLines;
+ split('\n', stdoutStr, stdoutLines);
+ EXPECT_EQ(numCopies * 2 + 1, stdoutLines.size());
+ // Strip off the trailing empty line
+ if (!stdoutLines.empty()) {
+ EXPECT_EQ("", stdoutLines.back());
+ stdoutLines.pop_back();
+ }
+ size_t linenum = 0;
+ for (const auto& line : stdoutLines) {
+ if ((linenum & 1) == 0) {
+ EXPECT_EQ("this is a successful test", line);
+ } else {
+ EXPECT_EQ("another line", line);
+ }
+ ++linenum;
+ }
+
+ // stderr should only contain the lines containing "another line"
+ std::vector<StringPiece> stderrLines;
+ split('\n', stderrStr, stderrLines);
+ EXPECT_EQ(numCopies + 1, stderrLines.size());
+ // Strip off the trailing empty line
+ if (!stderrLines.empty()) {
+ EXPECT_EQ("", stderrLines.back());
+ stderrLines.pop_back();
+ }
+ for (const auto& line : stderrLines) {
+ EXPECT_EQ("another line", line);
+ }
+ });
+}
+
+namespace {
+
+bool readToString(int fd, std::string& buf, size_t maxSize) {
+ buf.resize(maxSize);
+ char* dest = &buf.front();
+ size_t remaining = maxSize;
+
+ ssize_t n = -1;
+ while (remaining) {
+ n = ::read(fd, dest, remaining);
+ if (n == -1) {
+ if (errno == EINTR) {
+ continue;
+ }
+ if (errno == EAGAIN) {
+ break;
+ }
+ PCHECK("read failed");
+ } else if (n == 0) {
+ break;
+ }
+ dest += n;
+ remaining -= n;
+ }
+
+ buf.resize(dest - buf.data());
+ return (n == 0);
+}
+
+} // namespace
+
+TEST(CommunicateSubprocessTest, Chatty) {
+ checkFdLeak([] {
+ const int lineCount = 1000;
+
+ int wcount = 0;
+ int rcount = 0;
+
+ auto options = Subprocess::pipeStdin().pipeStdout().pipeStderr().usePath();
+ std::vector<std::string> cmd {
+ "sed",
+ "-u",
+ "-e",
+ "s/a test/a successful test/",
+ };
+
+ Subprocess proc(cmd, options);
+
+ auto writeCallback = [&] (int pfd, int cfd) -> bool {
+ EXPECT_EQ(0, cfd); // child stdin
+ EXPECT_EQ(rcount, wcount); // chatty, one read for every write
+
+ auto msg = folly::to<std::string>("a test ", wcount, "\n");
+
+ // Not entirely kosher, we should handle partial writes, but this is
+ // fine for writes <= PIPE_BUF
+ EXPECT_EQ(msg.size(), writeFull(pfd, msg.data(), msg.size()));
+
+ ++wcount;
+ proc.enableNotifications(0, false);
+
+ return (wcount == lineCount);
+ };
+
+ bool eofSeen = false;
+
+ auto readCallback = [&] (int pfd, int cfd) -> bool {
+ std::string lineBuf;
+
+ if (cfd != 1) {
+ EXPECT_EQ(2, cfd);
+ EXPECT_TRUE(readToString(pfd, lineBuf, 1));
+ EXPECT_EQ(0, lineBuf.size());
+ return true;
+ }
+
+ EXPECT_FALSE(eofSeen);
+
+ std::string expected;
+
+ if (rcount < lineCount) {
+ expected = folly::to<std::string>("a successful test ", rcount++, "\n");
+ }
+
+ EXPECT_EQ(wcount, rcount);
+
+ // Not entirely kosher, we should handle partial reads, but this is
+ // fine for reads <= PIPE_BUF
+ bool atEof = readToString(pfd, lineBuf, expected.size() + 1);
+ if (atEof) {
+ // EOF only expected after we finished reading
+ EXPECT_EQ(lineCount, rcount);
+ eofSeen = true;
+ }
+
+ EXPECT_EQ(expected, lineBuf);
+
+ if (wcount != lineCount) { // still more to write...
+ proc.enableNotifications(0, true);
+ }
+
+ return eofSeen;
+ };
+
+ proc.communicate(readCallback, writeCallback);
+
+ EXPECT_EQ(lineCount, wcount);
+ EXPECT_EQ(lineCount, rcount);
+ EXPECT_TRUE(eofSeen);
+
+ 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));
+}