Merge #4646 from oni-link/fix.issue.4569.3

Fix for missing output (#4569, ...)

6 files changed, 123 insertions, 23 deletions

diff --git a/src/nvim/event/process.c b/src/nvim/event/process.c
index 1fffdbf957..1c4c9737c3 100644
--- a/src/nvim/event/process.c
+++ b/src/nvim/event/process.c
@@ -333,9 +333,61 @@ 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;
+  }
+
+  // Maximal remaining data size of terminated process is system
+  // buffer size.
+  // Also helps with a child process that keeps the output streams open. If it
+  // keeps sending data, we only accept as much data as the system buffer size.
+  // Otherwise this would block cleanup/teardown.
+  int system_buffer_size = 0;
+  int err = uv_recv_buffer_size((uv_handle_t *)&stream->uv.pipe,
+                                &system_buffer_size);
+  if (err) {
+    system_buffer_size = (int)rbuffer_capacity(stream->buffer);
+  }
+
+  size_t max_bytes = stream->num_bytes + (size_t)system_buffer_size;
+
+  // Read remaining data.
+  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(proc->loop, 0);
+    if (proc->events) {
+        queue_process_events(proc->events);
+    }
+
+    // Stream can be closed if it is empty.
+    if (num_bytes == stream->num_bytes) {
+      if (stream->read_cb) {
+        // Stream callback could miss EOF handling if a child keeps the stream
+        // open.
+        stream->read_cb(stream, stream->buffer, 0, stream->data, true);
+      }
+      break;
+    }
+  }
+}
+
 static void process_close_handles(void **argv)
 {
   Process *proc = argv[0];
+
+  flush_stream(proc, proc->out);
+  flush_stream(proc, proc->err);
+
   process_close_streams(proc);
   process_close(proc);
 }
@@ -350,11 +402,12 @@ static void on_process_exit(Process *proc)
     uv_timer_stop(&loop->children_kill_timer);
   }
 
-  // Process handles are closed in the next event loop tick. This is done to
-  // give libuv more time to read data from the OS after the process exits(If
-  // process_close_streams is called with data still in the OS buffer, we lose
-  // it)
-  CREATE_EVENT(proc->events, process_close_handles, 1, proc);
+  // Process has terminated, but there could still be data to be read from the
+  // OS. We are still in the libuv loop, so we cannot call code that polls for
+  // more data directly. Instead delay the reading after the libuv loop by
+  // queueing process_close_handles() as an event.
+  Queue *queue = proc->events ? proc->events : loop->events;
+  CREATE_EVENT(queue, process_close_handles, 1, proc);
 }
 
 static void on_process_stream_close(Stream *stream, void *data)
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/os/shell.c b/src/nvim/os/shell.c
index 988620b145..64c673930a 100644
--- a/src/nvim/os/shell.c
+++ b/src/nvim/os/shell.c
@@ -310,25 +310,70 @@ static void system_data_cb(Stream *stream, RBuffer *buf, size_t count,
   dbuf->len += nread;
 }
 
+/// Continue to append data to last screen line.
+///
+/// @param output       Data to append to screen lines.
+/// @param remaining    Size of data.
+/// @param new_line     If true, next data output will be on a new line.
+static void append_to_screen_end(char *output, size_t remaining, bool new_line)
+{
+  // Column of last row to start appending data to.
+  static colnr_T last_col = 0;
+
+  size_t off = 0;
+  int last_row = (int)Rows - 1;
+
+  while (off < remaining) {
+    // Found end of line?
+    if (output[off] == NL) {
+      // Can we start a new line or do we need to continue the last one?
+      if (last_col == 0) {
+        screen_del_lines(0, 0, 1, (int)Rows, NULL);
+      }
+      screen_puts_len((char_u *)output, (int)off, last_row, last_col, 0);
+      last_col = 0;
+
+      size_t skip = off + 1;
+      output += skip;
+      remaining -= skip;
+      off = 0;
+      continue;
+    }
+
+    // Translate NUL to SOH
+    if (output[off] == NUL) {
+      output[off] = 1;
+    }
+
+    off++;
+  }
+
+  if (remaining) {
+    if (last_col == 0) {
+      screen_del_lines(0, 0, 1, (int)Rows, NULL);
+    }
+    screen_puts_len((char_u *)output, (int)remaining, last_row, last_col, 0);
+    last_col += (colnr_T)remaining;
+  }
+
+  if (new_line) {
+    last_col = 0;
+  }
+
+  ui_flush();
+}
+
 static void out_data_cb(Stream *stream, RBuffer *buf, size_t count, void *data,
     bool eof)
 {
+  // We always output the whole buffer, so the buffer can never
+  // wrap around.
   size_t cnt;
   char *ptr = rbuffer_read_ptr(buf, &cnt);
 
-  if (!cnt) {
-    return;
-  }
-
-  size_t written = write_output(ptr, cnt, false, eof);
-  // No output written, force emptying the Rbuffer if it is full.
-  if (!written && rbuffer_size(buf) == rbuffer_capacity(buf)) {
-    screen_del_lines(0, 0, 1, (int)Rows, NULL);
-    screen_puts_len((char_u *)ptr, (int)cnt, (int)Rows - 1, 0, 0);
-    written = cnt;
-  }
-  if (written) {
-    rbuffer_consumed(buf, written);
+  append_to_screen_end(ptr, cnt, eof);
+  if (cnt) {
+    rbuffer_consumed(buf, cnt);
   }
 }
 
diff --git a/src/nvim/rbuffer.c b/src/nvim/rbuffer.c
index d4cc8c0d5d..a2cc432eca 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);