feat(daemon): pull from input pipes during each flush round

Author: OldUser101

daemon/buffer.cpp

@@ -56,12 +56,12 @@ size_t HopperBuffer::write(void *src, size_t len) {
     return done_len;
 }
 
-size_t HopperBuffer::write(HopperPipe *pipe) {
+size_t HopperBuffer::write(HopperPipe *pipe, bool *more) {
     size_t max_len = max_write();
     size_t done_len = 0;
 
     size_t next_len = std::min((m_buf.size() - m_edge), max_len);
-    size_t res = pipe->read_pipe(&m_buf[m_edge], next_len);
+    size_t res = pipe->read_pipe(&m_buf[m_edge], next_len, more);
     if (res == (size_t)-1)
         // -1 indicates read error
         return -1;
@@ -74,7 +74,7 @@ size_t HopperBuffer::write(HopperPipe *pipe) {
         return done_len;
 
     next_len = max_len - next_len;
-    res = pipe->read_pipe(&m_buf[m_edge], next_len);
+    res = pipe->read_pipe(&m_buf[m_edge], next_len, more);
     if (res == (size_t)-1)
         return -1;
 
@@ -106,13 +106,13 @@ size_t HopperBuffer::read(BufferMarker *m, void *dst, size_t len) {
     return done_len;
 }
 
-size_t HopperBuffer::read(HopperPipe *pipe) {
+size_t HopperBuffer::read(HopperPipe *pipe, bool *more) {
     BufferMarker *m = pipe->marker();
     size_t max_len = max_read(m);
     size_t done_len = 0;
 
     size_t next_len = std::min((m_buf.size() - m->pos()), max_len);
-    size_t res = pipe->write_pipe(&m_buf[m->pos()], next_len);
+    size_t res = pipe->write_pipe(&m_buf[m->pos()], next_len, more);
     if (res == (size_t)-1)
         return -1;
 
@@ -122,7 +122,7 @@ size_t HopperBuffer::read(HopperPipe *pipe) {
         return done_len;
 
     next_len = max_len - next_len;
-    res = pipe->write_pipe(&m_buf[m->pos()], next_len);
+    res = pipe->write_pipe(&m_buf[m->pos()], next_len, more);
     if (res == (size_t)-1)
         return -1;
 
@@ -134,7 +134,7 @@ size_t HopperBuffer::read(HopperPipe *pipe) {
 size_t HopperBuffer::max_write() {
     size_t cap = m_buf.size();
     if (m_markers.empty())
-        return cap;
+        return std::max((size_t)0, cap - 1);
 
     size_t min_dist = cap;
 
@@ -148,7 +148,7 @@ size_t HopperBuffer::max_write() {
             min_dist = d;
     }
 
-    return min_dist;
+    return std::max((size_t)0, min_dist - 1);
 }
 
 size_t HopperBuffer::max_read(BufferMarker *m) {

daemon/endpoint.cpp

@@ -1,4 +1,7 @@
+#include <fcntl.h>
 #include <filesystem>
+#include <sys/ioctl.h>
+#include <unistd.h>
 
 #include "hopper/daemon/endpoint.hpp"
 #include "hopper/daemon/pipe.hpp"
@@ -25,21 +28,41 @@ void HopperEndpoint::on_pipe_readable(uint64_t id) {
 
     HopperPipe *pipe = m_inputs[id];
 
-    size_t res = m_buffer.write(pipe);
+/*
+ *  Calculation for "pipe pressure"
+    int p_sz, p_nb;
+    if ((p_sz = fcntl(pipe->fd(), F_GETPIPE_SZ)) != -1 &&
+        (ioctl(pipe->fd(), FIONREAD, &p_nb) > -1)) {
+        m_logger.trace(*pipe, ": ", (float)p_nb / (float)p_sz);
+    }
+*/
+
+    bool more = false;
+    size_t res = m_buffer.write(pipe, &more);
     if (res == (size_t)-1)
         throw_errno("read");
 
-    m_logger.trace(*pipe, " -> ", res, " bytes");
+    if (more && !m_more.contains(id))
+        m_more.insert(id);
+
+    m_logger.trace(*pipe, " -> ", res, " bytes", (more ? " ++" : ""));
 }
 
 void HopperEndpoint::flush_pipes() {
+    auto c_more = std::unordered_set<uint64_t>(m_more.begin(), m_more.end());
+    for (const uint64_t id : c_more) {
+        m_more.erase(id);
+        on_pipe_readable(id);
+    }
+
     for (const auto &[_, pipe] : m_outputs) {
         if (pipe->status() == PipeStatus::INACTIVE)
             continue;
 
-        size_t res = m_buffer.read(pipe);
+        bool more = false;
+        size_t res = m_buffer.read(pipe, &more);
         if (res > 0)
-            m_logger.trace(*pipe, " <- ", res, " bytes");
+            m_logger.trace(*pipe, " <- ", res, " bytes", (more ? " ++" : ""));
     }
 }
 
@@ -98,6 +121,7 @@ void HopperEndpoint::remove_by_id(uint64_t pipe_id) {
 
         delete pipe;
         m_inputs.erase(pipe_id);
+        m_more.erase(pipe_id);
     } else if (type == PipeType::OUT && m_outputs.contains(pipe_id)) {
         HopperPipe *pipe = m_outputs[pipe_id];
         m_buffer.delete_marker(pipe->marker());

daemon/pipe.cpp

@@ -58,10 +58,13 @@ void HopperPipe::close_pipe() {
         close(m_fd);
 }
 
-size_t HopperPipe::write_pipe(void *src, size_t len) {
+size_t HopperPipe::write_pipe(void *src, size_t len, bool *more) {
     if (m_type == PipeType::IN)
         return -1;
 
+    if (more)
+        *more = true;
+
     if (len == 0)
         return 0;
 
@@ -71,9 +74,11 @@ size_t HopperPipe::write_pipe(void *src, size_t len) {
         ssize_t res = write(m_fd, reinterpret_cast<char *>(src) + done_len,
                             len - done_len);
 
-        if (res == -1 && (errno == EWOULDBLOCK || errno == EINTR))
+        if (res == -1 && (errno == EWOULDBLOCK || errno == EINTR)) {
+            if (more)
+                *more = false;
             return done_len;
-        else if (res == -1) {
+        } else if (res == -1) {
             perror("write");
             return -1;
         }
@@ -84,10 +89,13 @@ size_t HopperPipe::write_pipe(void *src, size_t len) {
     return done_len;
 }
 
-size_t HopperPipe::read_pipe(void *dst, size_t len) {
+size_t HopperPipe::read_pipe(void *dst, size_t len, bool *more) {
     if (m_type == PipeType::OUT)
         return -1;
 
+    if (more)
+        *more = true;
+
     if (len == 0)
         return 0;
 
@@ -97,13 +105,18 @@ size_t HopperPipe::read_pipe(void *dst, size_t len) {
         ssize_t res = read(m_fd, reinterpret_cast<char *>(dst) + done_len,
                            len - done_len);
 
-        if (res == -1 && (errno == EWOULDBLOCK || errno == EINTR))
+        if (res == -1 && (errno == EWOULDBLOCK || errno == EINTR)) {
+            if (more)
+                *more = false;
             return done_len;
-        else if (res == -1) {
+        } else if (res == -1) {
             perror("read");
             return -1;
-        } else if (res == 0)
+        } else if (res == 0) {
+            if (more)
+                *more = false;
             break;
+        }
 
         done_len += res;
     }

include/hopper/daemon/buffer.hpp

@@ -25,10 +25,10 @@ class HopperBuffer {
     void delete_marker(BufferMarker *marker);
 
     size_t write(void *src, size_t len);
-    size_t write(HopperPipe *pipe);
+    size_t write(HopperPipe *pipe, bool *more);
 
     size_t read(BufferMarker *marker, void *dst, size_t len);
-    size_t read(HopperPipe *pipe);
+    size_t read(HopperPipe *pipe, bool *more);
 
     size_t max_write();
     size_t max_read(BufferMarker *marker);

include/hopper/daemon/daemon.hpp

@@ -34,7 +34,7 @@ class HopperDaemon {
     int m_epoll_fd = -1;
 
     int m_max_events = 64;
-    int m_timeout = 250;
+    int m_timeout = 100;
 
     std::filesystem::path m_path;
     Logger &m_logger;

include/hopper/daemon/endpoint.hpp

@@ -2,6 +2,7 @@
 #define endpoint_hpp_INCLUDED
 
 #include <unordered_map>
+#include <unordered_set>
 
 #include "hopper/daemon/buffer.hpp"
 #include "hopper/daemon/logging.hpp"
@@ -14,6 +15,9 @@ class HopperEndpoint {
     std::unordered_map<uint64_t, HopperPipe *> m_inputs;
     std::unordered_map<uint64_t, HopperPipe *> m_outputs;
 
+    // inputs that may have more data
+    std::unordered_set<uint64_t> m_more;
+
     HopperBuffer m_buffer{};
 
     uint64_t m_last_pipe_id = 1;

include/hopper/daemon/pipe.hpp

@@ -40,8 +40,8 @@ class HopperPipe {
 
     int open_pipe();
     void close_pipe();
-    size_t write_pipe(void *src, size_t len);
-    size_t read_pipe(void *dst, size_t len);
+    size_t write_pipe(void *src, size_t len, bool *more);
+    size_t read_pipe(void *dst, size_t len, bool *more);
 
     const std::filesystem::path &path() { return m_path; }
     BufferMarker *marker() { return m_marker; }