1 files changed, 87 insertions, 52 deletions

diff --git a/src/nvim/channel.c b/src/nvim/channel.c
index e8fe80a3b6..ebeaffe5a1 100644
--- a/src/nvim/channel.c
+++ b/src/nvim/channel.c
@@ -10,16 +10,21 @@
 #include "nvim/api/private/defs.h"
 #include "nvim/api/private/helpers.h"
 #include "nvim/autocmd.h"
+#include "nvim/autocmd_defs.h"
 #include "nvim/buffer_defs.h"
 #include "nvim/channel.h"
 #include "nvim/eval.h"
 #include "nvim/eval/encode.h"
 #include "nvim/eval/typval.h"
+#include "nvim/event/loop.h"
+#include "nvim/event/multiqueue.h"
+#include "nvim/event/process.h"
 #include "nvim/event/rstream.h"
 #include "nvim/event/socket.h"
+#include "nvim/event/stream.h"
 #include "nvim/event/wstream.h"
 #include "nvim/garray.h"
-#include "nvim/gettext.h"
+#include "nvim/gettext_defs.h"
 #include "nvim/globals.h"
 #include "nvim/log.h"
 #include "nvim/lua/executor.h"
@@ -28,10 +33,14 @@
 #include "nvim/message.h"
 #include "nvim/msgpack_rpc/channel.h"
 #include "nvim/msgpack_rpc/server.h"
+#include "nvim/os/fs.h"
 #include "nvim/os/os_defs.h"
 #include "nvim/os/shell.h"
 #include "nvim/path.h"
 #include "nvim/rbuffer.h"
+#include "nvim/rbuffer_defs.h"
+#include "nvim/terminal.h"
+#include "nvim/types_defs.h"
 
 #ifdef MSWIN
 # include "nvim/os/fs.h"
@@ -53,12 +62,24 @@ static uint64_t next_chan_id = CHAN_STDERR + 1;
 /// Teardown the module
 void channel_teardown(void)
 {
-  Channel *channel;
+  Channel *chan;
+  map_foreach_value(&channels, chan, {
+    channel_close(chan->id, kChannelPartAll, NULL);
+  });
+}
 
-  map_foreach_value(&channels, channel, {
-    channel_close(channel->id, kChannelPartAll, NULL);
+#ifdef EXITFREE
+void channel_free_all_mem(void)
+{
+  Channel *chan;
+  map_foreach_value(&channels, chan, {
+    channel_destroy(chan);
   });
+  map_destroy(uint64_t, &channels);
+
+  callback_free(&on_print);
 }
+#endif
 
 /// Closes a channel
 ///
@@ -217,16 +238,17 @@ void channel_create_event(Channel *chan, const char *ext_source)
   }
 
   assert(chan->id <= VARNUMBER_MAX);
-  Dictionary info = channel_info(chan->id);
+  Arena arena = ARENA_EMPTY;
+  Dictionary info = channel_info(chan->id, &arena);
   typval_T tv = TV_INITIAL_VALUE;
   // TODO(bfredl): do the conversion in one step. Also would be nice
   // to pretty print top level dict in defined order
-  (void)object_to_vim(DICTIONARY_OBJ(info), &tv, NULL);
+  object_to_vim(DICTIONARY_OBJ(info), &tv, NULL);
   assert(tv.v_type == VAR_DICT);
   char *str = encode_tv2json(&tv, NULL);
   ILOG("new channel %" PRIu64 " (%s) : %s", chan->id, source, str);
   xfree(str);
-  api_free_dictionary(info);
+  arena_mem_free(arena_finish(&arena));
 
 #else
   (void)ext_source;
@@ -244,7 +266,7 @@ void channel_decref(Channel *chan)
 {
   if (!(--chan->refcount)) {
     // delay free, so that libuv is done with the handles
-    multiqueue_put(main_loop.events, free_channel_event, 1, chan);
+    multiqueue_put(main_loop.events, free_channel_event, chan);
   }
 }
 
@@ -260,9 +282,8 @@ void callback_reader_start(CallbackReader *reader, const char *type)
   reader->type = type;
 }
 
-static void free_channel_event(void **argv)
+static void channel_destroy(Channel *chan)
 {
-  Channel *chan = argv[0];
   if (chan->is_rpc) {
     rpc_free(chan);
   }
@@ -275,11 +296,17 @@ static void free_channel_event(void **argv)
   callback_reader_free(&chan->on_stderr);
   callback_free(&chan->on_exit);
 
-  pmap_del(uint64_t)(&channels, chan->id, NULL);
   multiqueue_free(chan->events);
   xfree(chan);
 }
 
+static void free_channel_event(void **argv)
+{
+  Channel *chan = argv[0];
+  pmap_del(uint64_t)(&channels, chan->id, NULL);
+  channel_destroy(chan);
+}
+
 static void channel_destroy_early(Channel *chan)
 {
   if ((chan->id != --next_chan_id)) {
@@ -293,7 +320,7 @@ static void channel_destroy_early(Channel *chan)
   }
 
   // uv will keep a reference to handles until next loop tick, so delay free
-  multiqueue_put(main_loop.events, free_channel_event, 1, chan);
+  multiqueue_put(main_loop.events, free_channel_event, chan);
 }
 
 static void close_cb(Stream *stream, void *data)
@@ -553,7 +580,10 @@ size_t channel_send(uint64_t id, char *data, size_t len, bool data_owned, const
       goto retfree;
     }
     // unbuffered write
-    written = len * fwrite(data, len, 1, stderr);
+    ptrdiff_t wres = os_write(STDERR_FILENO, data, len, false);
+    if (wres >= 0) {
+      written = (size_t)wres;
+    }
     goto retfree;
   }
 
@@ -657,7 +687,7 @@ static void schedule_channel_event(Channel *chan)
 {
   if (!chan->callback_scheduled) {
     if (!chan->callback_busy) {
-      multiqueue_put(chan->events, on_channel_event, 1, chan);
+      multiqueue_put(chan->events, on_channel_event, chan);
       channel_incref(chan);
     }
     chan->callback_scheduled = true;
@@ -682,7 +712,7 @@ static void on_channel_event(void **args)
   chan->callback_busy = false;
   if (chan->callback_scheduled) {
     // further callback was deferred to avoid recursion.
-    multiqueue_put(chan->events, on_channel_event, 1, chan);
+    multiqueue_put(chan->events, on_channel_event, chan);
     channel_incref(chan);
   }
 
@@ -777,19 +807,21 @@ static void channel_callback_call(Channel *chan, CallbackReader *reader)
 /// and `buf` is assumed to be a new, unmodified buffer.
 void channel_terminal_open(buf_T *buf, Channel *chan)
 {
-  TerminalOptions topts;
-  topts.data = chan;
-  topts.width = chan->stream.pty.width;
-  topts.height = chan->stream.pty.height;
-  topts.write_cb = term_write;
-  topts.resize_cb = term_resize;
-  topts.close_cb = term_close;
+  TerminalOptions topts = {
+    .data = chan,
+    .width = chan->stream.pty.width,
+    .height = chan->stream.pty.height,
+    .write_cb = term_write,
+    .resize_cb = term_resize,
+    .close_cb = term_close,
+    .force_crlf = false,
+  };
   buf->b_p_channel = (OptInt)chan->id;  // 'channel' option
   channel_incref(chan);
   terminal_open(&chan->term, buf, topts);
 }
 
-static void term_write(char *buf, size_t size, void *data)
+static void term_write(const char *buf, size_t size, void *data)
 {
   Channel *chan = data;
   if (chan->stream.proc.in.closed) {
@@ -812,7 +844,7 @@ static inline void term_delayed_free(void **argv)
 {
   Channel *chan = argv[0];
   if (chan->stream.proc.in.pending_reqs || chan->stream.proc.out.pending_reqs) {
-    multiqueue_put(chan->events, term_delayed_free, 1, chan);
+    multiqueue_put(chan->events, term_delayed_free, chan);
     return;
   }
 
@@ -826,7 +858,7 @@ static void term_close(void *data)
 {
   Channel *chan = data;
   process_stop(&chan->stream.proc);
-  multiqueue_put(chan->events, term_delayed_free, 1, data);
+  multiqueue_put(chan->events, term_delayed_free, data);
 }
 
 void channel_info_changed(Channel *chan, bool new_chan)
@@ -834,7 +866,7 @@ void channel_info_changed(Channel *chan, bool new_chan)
   event_T event = new_chan ? EVENT_CHANOPEN : EVENT_CHANINFO;
   if (has_event(event)) {
     channel_incref(chan);
-    multiqueue_put(main_loop.events, set_info_event, 2, chan, event);
+    multiqueue_put(main_loop.events, set_info_event, chan, (void *)(intptr_t)event);
   }
 }
 
@@ -845,9 +877,10 @@ static void set_info_event(void **argv)
 
   save_v_event_T save_v_event;
   dict_T *dict = get_v_event(&save_v_event);
-  Dictionary info = channel_info(chan->id);
+  Arena arena = ARENA_EMPTY;
+  Dictionary info = channel_info(chan->id, &arena);
   typval_T retval;
-  (void)object_to_vim(DICTIONARY_OBJ(info), &retval, NULL);
+  object_to_vim(DICTIONARY_OBJ(info), &retval, NULL);
   assert(retval.v_type == VAR_DICT);
   tv_dict_add_dict(dict, S_LEN("info"), retval.vval.v_dict);
   tv_dict_set_keys_readonly(dict);
@@ -855,7 +888,7 @@ static void set_info_event(void **argv)
   apply_autocmds(event, NULL, NULL, false, curbuf);
 
   restore_v_event(dict, &save_v_event);
-  api_free_dictionary(info);
+  arena_mem_free(arena_finish(&arena));
   channel_decref(chan);
 }
 
@@ -867,15 +900,15 @@ bool channel_job_running(uint64_t id)
           && !process_is_stopped(&chan->stream.proc));
 }
 
-Dictionary channel_info(uint64_t id)
+Dictionary channel_info(uint64_t id, Arena *arena)
 {
   Channel *chan = find_channel(id);
   if (!chan) {
     return (Dictionary)ARRAY_DICT_INIT;
   }
 
-  Dictionary info = ARRAY_DICT_INIT;
-  PUT(info, "id", INTEGER_OBJ((Integer)chan->id));
+  Dictionary info = arena_dict(arena, 8);
+  PUT_C(info, "id", INTEGER_OBJ((Integer)chan->id));
 
   const char *stream_desc, *mode_desc;
   switch (chan->streamtype) {
@@ -883,18 +916,20 @@ Dictionary channel_info(uint64_t id)
     stream_desc = "job";
     if (chan->stream.proc.type == kProcessTypePty) {
       const char *name = pty_process_tty_name(&chan->stream.pty);
-      PUT(info, "pty", CSTR_TO_OBJ(name));
+      PUT_C(info, "pty", CSTR_TO_ARENA_OBJ(arena, name));
     }
 
-    char **p = chan->stream.proc.argv;
+    char **args = chan->stream.proc.argv;
     Array argv = ARRAY_DICT_INIT;
-    if (p != NULL) {
-      while (*p != NULL) {
-        ADD(argv, CSTR_TO_OBJ(*p));
-        p++;
+    if (args != NULL) {
+      size_t n;
+      for (n = 0; args[n] != NULL; n++) {}
+      argv = arena_array(arena, n);
+      for (size_t i = 0; i < n; i++) {
+        ADD_C(argv, CSTR_AS_OBJ(args[i]));
       }
     }
-    PUT(info, "argv", ARRAY_OBJ(argv));
+    PUT_C(info, "argv", ARRAY_OBJ(argv));
     break;
   }
 
@@ -907,51 +942,51 @@ Dictionary channel_info(uint64_t id)
     break;
 
   case kChannelStreamInternal:
-    PUT(info, "internal", BOOLEAN_OBJ(true));
+    PUT_C(info, "internal", BOOLEAN_OBJ(true));
     FALLTHROUGH;
 
   case kChannelStreamSocket:
     stream_desc = "socket";
     break;
   }
-  PUT(info, "stream", CSTR_TO_OBJ(stream_desc));
+  PUT_C(info, "stream", CSTR_AS_OBJ(stream_desc));
 
   if (chan->is_rpc) {
     mode_desc = "rpc";
-    PUT(info, "client", DICTIONARY_OBJ(rpc_client_info(chan)));
+    PUT_C(info, "client", DICTIONARY_OBJ(chan->rpc.info));
   } else if (chan->term) {
     mode_desc = "terminal";
-    PUT(info, "buffer", BUFFER_OBJ(terminal_buf(chan->term)));
+    PUT_C(info, "buffer", BUFFER_OBJ(terminal_buf(chan->term)));
   } else {
     mode_desc = "bytes";
   }
-  PUT(info, "mode", CSTR_TO_OBJ(mode_desc));
+  PUT_C(info, "mode", CSTR_AS_OBJ(mode_desc));
 
   return info;
 }
 
 /// Simple int64_t comparison function for use with qsort()
-static int int64_t_cmp(const void *a, const void *b)
+static int int64_t_cmp(const void *pa, const void *pb)
 {
-  int64_t diff = *(int64_t *)a - *(int64_t *)b;
-  return (diff < 0) ? -1 : (diff > 0);
+  const int64_t a = *(const int64_t *)pa;
+  const int64_t b = *(const int64_t *)pb;
+  return a == b ? 0 : a > b ? 1 : -1;
 }
 
-Array channel_all_info(void)
+Array channel_all_info(Arena *arena)
 {
   // order the items in the array by channel number, for Determinism™
   kvec_t(int64_t) ids = KV_INITIAL_VALUE;
-  kv_resize(ids, map_size(&channels));
+  kv_fixsize_arena(arena, ids, map_size(&channels));
   uint64_t id;
   map_foreach_key(&channels, id, {
     kv_push(ids, (int64_t)id);
   });
   qsort(ids.items, ids.size, sizeof ids.items[0], int64_t_cmp);
 
-  Array ret = ARRAY_DICT_INIT;
+  Array ret = arena_array(arena, ids.size);
   for (size_t i = 0; i < ids.size; i++) {
-    ADD(ret, DICTIONARY_OBJ(channel_info((uint64_t)ids.items[i])));
+    ADD_C(ret, DICTIONARY_OBJ(channel_info((uint64_t)ids.items[i], arena)));
   }
-  kv_destroy(ids);
   return ret;
 }