Fixing things for programs with very large output.

daggy/include/daggy/executors/ForkingExecutor.hpp

@@ -1,5 +1,6 @@
 #pragma once
 
+#include <iostream>
 #include "../Executor.hpp"
 
 namespace daggy {

daggy/src/executors/ForkingExecutor.cpp

@@ -1,5 +1,9 @@
 #include <daggy/executors/ForkingExecutor.hpp>
 
+#include <array>
+#include <utility>
+
+#include <fcntl.h>
 #include <unistd.h>
 #include <wait.h>
 #include <poll.h>
@@ -9,27 +13,21 @@ using namespace daggy::executor;
 std::string slurp(int fd)  {
   std::string result;
 
-  const size_t BUFFER_SIZE = 4096;
+  const ssize_t BUFFER_SIZE = 4096;
   char buffer[BUFFER_SIZE];
 
   struct pollfd pfd{ .fd = fd, .events = POLLIN, .revents = 0 };
-  poll(&pfd, 1, 0);
+  poll(&pfd, 1, 1);
 
   while (pfd.revents & POLLIN) {
     ssize_t bytes = read(fd, buffer, BUFFER_SIZE);
-    if (bytes == -1) {
-      if (errno == EINTR) {
-        continue;
-      } else {
-        perror("read");
-        exit(1);
-      }
-    } else if (bytes == 0) {
+    if (bytes == 0) {
       break;
     } else {
-      result += buffer;
-      if (bytes < BUFFER_SIZE) break;
+      result.append(buffer, bytes);
     }
+    pfd.revents = 0;
+    poll(&pfd, 1, 1);
   }
 
   return result;
@@ -51,8 +49,8 @@ daggy::AttemptRecord
   argv.push_back(nullptr);
 
   // Create the pipe
-  int stdoutPipe[2]; pipe(stdoutPipe);
-  int stderrPipe[2]; pipe(stderrPipe);
+  int stdoutPipe[2]; pipe2(stdoutPipe, O_DIRECT);
+  int stderrPipe[2]; pipe2(stderrPipe, O_DIRECT);
 
   pid_t child = fork();
   if (child < 0) {
@@ -66,8 +64,13 @@ daggy::AttemptRecord
     exit(-1);
   }
 
+  std::atomic<bool> running = true;
+  std::thread stdoutReader([&]() { while(running) rec.output.append(slurp(stdoutPipe[0])); });
+  std::thread stderrReader([&]() { while(running) rec.error.append(slurp(stderrPipe[0])); });
+
   int rc = 0;
   waitpid(child, &rc, 0);
+  running = false;
 
   rec.stopTime = Clock::now();
   if (WIFEXITED(rc)) {
@@ -76,9 +79,8 @@ daggy::AttemptRecord
     rec.rc = -1;
   }
 
-
-  rec.output = slurp(stdoutPipe[0]);
-  rec.error = slurp(stderrPipe[0]);
+  stdoutReader.join();
+  stderrReader.join();
 
   close(stdoutPipe[0]);
   close(stderrPipe[0]);