channels: improvements to buffering

src/nvim/channel.c

@@ -257,7 +257,8 @@ void callback_reader_free(CallbackReader *reader)
 void callback_reader_start(CallbackReader *reader)
 {
   if (reader->buffered) {
-    ga_init(&reader->buffer, sizeof(char *), 1);
+    ga_init(&reader->buffer, sizeof(char *), 32);
+    ga_grow(&reader->buffer, 32);
   }
 }
 
@@ -521,16 +522,10 @@ err:
   return 0;
 }
 
-// vimscript job callbacks must be executed on Nvim main loop
+/// NB: mutates buf in place!
-static inline void process_channel_event(Channel *chan, Callback *callback,
+static list_T *buffer_to_tv_list(char *buf, size_t count)
-                                         const char *type, char *buf,
-                                         size_t count, int status)
 {
-  assert(callback);
+  list_T *ret = tv_list_alloc();
-  ChannelEvent *event_data = xmalloc(sizeof(*event_data));
-  event_data->received = NULL;
-  if (buf) {
-    event_data->received = tv_list_alloc();
   char *ptr = buf;
   size_t remaining = count;
   size_t off = 0;
@@ -538,7 +533,7 @@ static inline void process_channel_event(Channel *chan, Callback *callback,
   while (off < remaining) {
     // append the line
     if (ptr[off] == NL) {
-        tv_list_append_string(event_data->received, ptr, (ssize_t)off);
+      tv_list_append_string(ret, ptr, (ssize_t)off);
       size_t skip = off + 1;
       ptr += skip;
       remaining -= skip;
@@ -551,7 +546,20 @@ static inline void process_channel_event(Channel *chan, Callback *callback,
     }
     off++;
   }
-    tv_list_append_string(event_data->received, ptr, (ssize_t)off);
+  tv_list_append_string(ret, ptr, (ssize_t)off);
+  return ret;
+}
+
+// vimscript job callbacks must be executed on Nvim main loop
+static inline void process_channel_event(Channel *chan, Callback *callback,
+                                         const char *type, char *buf,
+                                         size_t count, int status)
+{
+  assert(callback);
+  ChannelEvent *event_data = xmalloc(sizeof(*event_data));
+  event_data->received = NULL;
+  if (buf) {
+    event_data->received = buffer_to_tv_list(buf, count);
   } else {
     event_data->status = status;
   }
@@ -602,10 +610,18 @@ static void on_channel_output(Stream *stream, Channel *chan, RBuffer *buf,
 
   if (eof) {
     if (reader->buffered) {
+      if (reader->cb.type != kCallbackNone) {
         process_channel_event(chan, &reader->cb, type, reader->buffer.ga_data,
                               (size_t)reader->buffer.ga_len, 0);
         ga_clear(&reader->buffer);
-    } else if (callback_reader_set(*reader)) {
+      } else if (reader->self) {
+        list_T *data = buffer_to_tv_list(reader->buffer.ga_data,
+                                         (size_t)reader->buffer.ga_len);
+        tv_dict_add_list(reader->self, type, strlen(type), data);
+      } else {
+        abort();
+      }
+    } else if (reader->cb.type != kCallbackNone) {
       process_channel_event(chan, &reader->cb, type, ptr, 0, 0);
     }
     return;

src/nvim/channel.h

@@ -40,16 +40,18 @@ typedef struct {
 
 typedef struct {
   Callback cb;
+  dict_T *self;
   garray_T buffer;
   bool buffered;
 } CallbackReader;
 
 #define CALLBACK_READER_INIT ((CallbackReader){ .cb = CALLBACK_NONE, \
+                                                .self = NULL, \
                                                 .buffer = GA_EMPTY_INIT_VALUE, \
                                                 .buffered = false })
 static inline bool callback_reader_set(CallbackReader reader)
 {
-  return reader.cb.type != kCallbackNone;
+  return reader.cb.type != kCallbackNone || reader.self;
 }
 
 struct Channel {

src/nvim/eval.c

@@ -15090,6 +15090,9 @@ static void f_sockconnect(typval_T *argvars, typval_T *rettv, FunPtr fptr)
       return;
     }
     on_data.buffered = tv_dict_get_number(opts, "data_buffered");
+    if (on_data.buffered && on_data.cb.type == kCallbackNone) {
+      on_data.self = opts;
+    }
   }
 
   const char *error = NULL;
@@ -15490,6 +15493,10 @@ static void f_stdioopen(typval_T *argvars, typval_T *rettv, FunPtr fptr)
   if (!tv_dict_get_callback(opts, S_LEN("on_stdin"), &on_stdin.cb)) {
     return;
   }
+  on_stdin.buffered = tv_dict_get_number(opts, "stdin_buffered");
+  if (on_stdin.buffered && on_stdin.cb.type == kCallbackNone) {
+    on_stdin.self = opts;
+  }
 
   const char *error;
   uint64_t id = channel_from_stdio(rpc, on_stdin, &error);
@@ -16764,7 +16771,17 @@ static bool set_ref_in_callback_reader(CallbackReader *reader, int copyID,
                                        ht_stack_T **ht_stack,
                                        list_stack_T **list_stack)
 {
-  return set_ref_in_callback(&reader->cb, copyID, ht_stack, list_stack);
+  if (set_ref_in_callback(&reader->cb, copyID, ht_stack, list_stack)) {
+    return true;
+  }
+
+  if (reader->self) {
+    typval_T tv;
+    tv.v_type = VAR_DICT;
+    tv.vval.v_dict = reader->self;
+    return set_ref_in_item(&tv, copyID, ht_stack, list_stack);
+  }
+  return false;
 }
 
 static void add_timer_info(typval_T *rettv, timer_T *timer)
@@ -22344,6 +22361,12 @@ static inline bool common_job_callbacks(dict_T *vopts,
       && tv_dict_get_callback(vopts, S_LEN("on_exit"), on_exit)) {
     on_stdout->buffered = tv_dict_get_number(vopts, "stdout_buffered");
     on_stderr->buffered = tv_dict_get_number(vopts, "stderr_buffered");
+    if (on_stdout->buffered && on_stdout->cb.type == kCallbackNone) {
+      on_stdout->self = vopts;
+    }
+    if (on_stderr->buffered && on_stderr->cb.type == kCallbackNone) {
+      on_stderr->self = vopts;
+    }
     vopts->dv_refcount++;
     return true;
   }