process.c: Prevent data loss for process output streams

For a terminating process, it's output streams could be closed,
before all data is read.

5 files changed, 45 insertions, 5 deletions

diff --git a/src/nvim/event/process.c b/src/nvim/event/process.c
index 9bb62891c7..2e6511a167 100644
--- a/src/nvim/event/process.c
+++ b/src/nvim/event/process.c
@@ -333,9 +333,47 @@ static void process_close(Process *proc)
   }
 }
 
+/// Flush output stream.
+///
+/// @param proc     Process, for which an output stream should be flushed.
+/// @param stream   Stream to flush.
+static void flush_stream(Process *proc, Stream *stream)
+  FUNC_ATTR_NONNULL_ARG(1)
+{
+  if (!stream || stream->closed) {
+    return;
+  }
+
+  // Limit amount of data we accept after process terminated.
+  size_t max_bytes = stream->num_bytes + rbuffer_capacity(stream->buffer);
+
+  while (!stream->closed && stream->num_bytes < max_bytes) {
+    // Remember number of bytes before polling
+    size_t num_bytes = stream->num_bytes;
+
+    // Poll for data and process the generated events.
+    loop_poll_events(&loop, 0);
+    if (proc->events && !queue_empty(proc->events)) {
+      queue_process_events(proc->events);
+    }
+
+    // Stream can be closed if it is empty.
+    if (num_bytes == stream->num_bytes) {
+      break;
+    }
+  }
+}
+
 static void process_close_handles(void **argv)
 {
   Process *proc = argv[0];
+
+  // Did our process forked a child that keeps the output streams open?
+  if (!process_is_tearing_down) {
+    flush_stream(proc, proc->out);
+    flush_stream(proc, proc->err);
+  }
+
   process_close_streams(proc);
   process_close(proc);
 }
diff --git a/src/nvim/event/rstream.c b/src/nvim/event/rstream.c
index 9f3fbc25ff..a520143064 100644
--- a/src/nvim/event/rstream.c
+++ b/src/nvim/event/rstream.c
@@ -100,6 +100,10 @@ static void read_cb(uv_stream_t *uvstream, ssize_t cnt, const uv_buf_t *buf)
 {
   Stream *stream = uvstream->data;
 
+  if (cnt > 0) {
+    stream->num_bytes += (size_t)cnt;
+  }
+
   if (cnt <= 0) {
     if (cnt != UV_ENOBUFS
         // cnt == 0 means libuv asked for a buffer and decided it wasn't needed:
@@ -185,10 +189,6 @@ static void read_event(void **argv)
 
 static void invoke_read_cb(Stream *stream, size_t count, bool eof)
 {
-  if (stream->closed) {
-    return;
-  }
-
   // Don't let the stream be closed before the event is processed.
   stream->pending_reqs++;
 
diff --git a/src/nvim/event/stream.c b/src/nvim/event/stream.c
index 71582ab357..33404158cf 100644
--- a/src/nvim/event/stream.c
+++ b/src/nvim/event/stream.c
@@ -71,6 +71,7 @@ void stream_init(Loop *loop, Stream *stream, int fd, uv_stream_t *uvstream,
   stream->closed = false;
   stream->buffer = NULL;
   stream->events = NULL;
+  stream->num_bytes = 0;
 }
 
 void stream_close(Stream *stream, stream_close_cb on_stream_close)
diff --git a/src/nvim/event/stream.h b/src/nvim/event/stream.h
index c6baac0db7..ad4e24775b 100644
--- a/src/nvim/event/stream.h
+++ b/src/nvim/event/stream.h
@@ -49,6 +49,7 @@ struct stream {
   size_t curmem;
   size_t maxmem;
   size_t pending_reqs;
+  size_t num_bytes;
   void *data, *internal_data;
   bool closed;
   Queue *events;
diff --git a/src/nvim/rbuffer.c b/src/nvim/rbuffer.c
index b3805a3a28..36f388700a 100644
--- a/src/nvim/rbuffer.c
+++ b/src/nvim/rbuffer.c
@@ -15,7 +15,7 @@ RBuffer *rbuffer_new(size_t capacity)
   FUNC_ATTR_WARN_UNUSED_RESULT FUNC_ATTR_NONNULL_RET
 {
   if (!capacity) {
-    capacity = 0xffff;
+    capacity = 0x10000;
   }
 
   RBuffer *rv = xmalloc(sizeof(RBuffer) + capacity);