Fix reading of output from subprocess

CMakeLists.txt

@@ -2,7 +2,7 @@ cmake_minimum_required(VERSION 3.2)
 
 project(linuxdeploy C CXX)
 
-set(CMAKE_CXX_STANDARD 11)
+set(CMAKE_CXX_STANDARD 17)
 set(CMAKE_CXX_STANDARD_REQUIRED ON)
 
 set(CMAKE_MODULE_PATH ${CMAKE_MODULE_PATH} "${PROJECT_SOURCE_DIR}/cmake/Modules/")

include/linuxdeploy/subprocess/pipe_reader.h

@@ -4,15 +4,24 @@
 #include <set>
 #include <vector>
 #include <poll.h>
+#include <chrono>
 
 /**
  * Reads from a pipe when data is available, and hands data to registered callbacks.
  */
 class pipe_reader {
 private:
-    const int pipe_fd_;
+    struct pollfd pollfd_;
 
 public:
+    static constexpr std::chrono::milliseconds READ_TIMEOUT{50};
+
+    enum class result {
+        SUCCESS = 0,
+        TIMEOUT,
+        END_OF_FILE,
+    };
+
     /**
      * Construct new instance from pipe file descriptor.
      * @param pipe_fd file descriptor for pipe we will read from (e.g., a subprocess's stdout, stderr pipes)
@@ -27,10 +36,12 @@ public:
      *     - no more data left in the pipe to be read
      *     - buffer is completely filled
      *
+     * The buffer will be resized to the number of bytes read from the pipe.
+     *
      * On errors, a subprocess_error is thrown.
      *
      * @param buffer buffer to store read data into
-     * @returns amount of characters read from the pipe
+     * @returns
      */
-    size_t read(std::vector<std::string::value_type>& buffer) const;
+    result read(std::vector<std::string::value_type>& buffer, std::chrono::milliseconds read_timeout = READ_TIMEOUT);
 };

src/plugin/plugin_process_handler.cpp

@@ -68,17 +68,8 @@ namespace linuxdeploy {
                     subprocess::subprocess_result_buffer_t intermediate_buffer(4096);
 
                     // (try to) read from pipe
-                    const auto bytes_read = pipe_to_be_logged.reader_.read(intermediate_buffer);
+                    switch (pipe_to_be_logged.reader_.read(intermediate_buffer)) {
-
+                        case pipe_reader::result::SUCCESS: {
-                    // 0 means EOF
-                    if (bytes_read == 0) {
-                        pipe_to_be_logged.eof = true;
-                        break;
-                    }
-
-                    // we just trim the buffer to the bytes we read (makes the code below easier)
-                    intermediate_buffer.resize(bytes_read);
-
                             // all we have to do now is to look for CR or LF, send everything up to that location into the ldLog instance,
                             // write our prefix and then repeat
                             for (auto it = intermediate_buffer.begin(); it != intermediate_buffer.end(); ++it) {
@@ -119,12 +110,15 @@ namespace linuxdeploy {
                                     break;
                                 }
                             }
+                            break;
+                        }
+                        case pipe_reader::result::END_OF_FILE: {
+                            pipe_to_be_logged.eof = true;
+                            break;
+                        }
+                        case pipe_reader::result::TIMEOUT:
+                            break;
                     }
-
-                // do-while might be a little more elegant, but we can save this one unnecessary sleep, so...
-                if (proc.is_running()) {
-                    // reduce load on CPU
-                    std::this_thread::sleep_for(std::chrono::milliseconds(50));
                 }
 
                 // once all buffers are EOF, we can stop reading

src/subprocess/pipe_reader.cpp

@@ -1,37 +1,59 @@
 // system headers
 #include <algorithm>
-#include <fcntl.h>
-#include <unistd.h>
-#include <functional>
 #include <cstring>
+#include <poll.h>
+#include <unistd.h>
 #include <stdexcept>
 
 // local headers
 #include "linuxdeploy/subprocess/pipe_reader.h"
 
-pipe_reader::pipe_reader(int pipe_fd) : pipe_fd_(pipe_fd) {
+pipe_reader::pipe_reader(int pipe_fd) : pollfd_(pollfd{pipe_fd, POLLIN | POLLHUP}) {}
-    // add O_NONBLOCK TO fd's flags to be able to read
-    auto flags = fcntl(pipe_fd_, F_GETFL, 0);
-    flags |= O_NONBLOCK;
-    fcntl(pipe_fd_, F_SETFL, flags);
-}
 
-size_t pipe_reader::read(std::vector<std::string::value_type>& buffer) const {
+pipe_reader::result pipe_reader::read(std::vector<std::string::value_type>& buffer, std::chrono::milliseconds read_timeout) {
-    for (;;) {
+    const auto timeout_msec = std::chrono::duration_cast<std::chrono::milliseconds>(read_timeout).count();
-        ssize_t rv = ::read(pipe_fd_, buffer.data(), buffer.size());
 
-        if (rv == -1) {
+    // we could (and probably should) be using poll on multiple fds at once
-            switch (errno) {
+    // however given the low bandwidth of data to handle, this should be fine, given we use a small-enough timeout
-                // retry in case data is currently not available
+    // also the read buffer sizes could be further increased to improve the overall performance
-                case EINTR:
+    switch (poll(&pollfd_, 1, static_cast<int>(timeout_msec))) {
-                case EAGAIN:
+        case -1:
-                    continue;
-                default:
             // TODO: introduce custom subprocess_error
             throw std::runtime_error{"unexpected error reading from pipe: " + std::string(strerror(errno))};
+        case 0:
+            return result::TIMEOUT;
+        case 1: {
+            if ((pollfd_.revents & POLLIN) != 0) {
+                ssize_t rv = ::read(pollfd_.fd, buffer.data(), buffer.size());
+
+                switch (rv) {
+                    case -1: {
+                        throw std::runtime_error{"unexpected error reading from pipe: " + std::string(strerror(errno))};
+                    }
+                    case 0: {
+                        return result::END_OF_FILE;
+                    }
+                    default: {
+                        // set the size correctly so the caller can just query the vector's size if the number of read chars is needed
+                        buffer.resize(rv);
+                        return result::SUCCESS;
+                    }
                 }
             }
 
-        return rv;
+            if ((pollfd_.revents & POLLHUP) != 0) {
+                // appears like this can be considered eof
+                return result::END_OF_FILE;
+            }
+
+            if ((pollfd_.revents & POLLERR) != 0 || (pollfd_.revents & POLLNVAL) != 0) {
+                throw std::runtime_error{"poll() failed unexpectedly"};
+            }
+
+            break;
+        }
+        default:
+            // this is a should-never-ever-happen case, a return value not handled by the lines above is actually not possible
+            throw std::runtime_error{"unexpected return value from pollfd"};
     }
 }

src/subprocess/subprocess.cpp

@@ -51,23 +51,19 @@ namespace linuxdeploy {
                     subprocess_result_buffer_t intermediate_buffer(4096);
 
                     // (try to) read all available data from pipe
-                    for (;;) {
+                    for (; !pipe_state.eof; ) {
-                        if (pipe_state.eof) {
+                        switch (pipe_state.reader.read(intermediate_buffer)) {
-                            break;
+                            case pipe_reader::result::SUCCESS: {
-                        }
-
-                        const auto bytes_read = pipe_state.reader.read(intermediate_buffer);
-
-                        // 0 means EOF
-                        if (bytes_read == 0) {
-                            pipe_state.eof = true;
-                            break;
-                        }
-
                                 // append to main buffer
-                        pipe_state.buffer.reserve(pipe_state.buffer.size() + bytes_read);
+                                pipe_state.buffer.reserve(pipe_state.buffer.size() + intermediate_buffer.size());
-                        std::copy(intermediate_buffer.begin(), (intermediate_buffer.begin() + bytes_read),
+                                std::copy(intermediate_buffer.begin(), intermediate_buffer.end(), std::back_inserter(pipe_state.buffer));
-                                  std::back_inserter(pipe_state.buffer));
+                            }
+                            case pipe_reader::result::END_OF_FILE: {
+                                pipe_state.eof = true;
+                            }
+                            default:
+                                break;
+                        }
                     }
                 }