event: Refactor async event processing

- Improve the implementation of deferred/immediate events.
- Use the new queue module to change how/when events are queued/processed by
  giving a private queue to each emitter.
- Immediate events(which only exist to break uv_run recursion) are now
  represented in the `loop->fast_events` queue.
- Events pushed to child queues are propagated to the event loop main queue and
  processed as K_EVENT keys.

28 files changed, 215 insertions, 210 deletions

diff --git a/src/nvim/edit.c b/src/nvim/edit.c
index 3982e32c2f..6bcf5e804a 100644
--- a/src/nvim/edit.c
+++ b/src/nvim/edit.c
@@ -254,7 +254,7 @@ edit (
 )
 {
   if (curbuf->terminal) {
-    terminal_enter(true);
+    terminal_enter();
     return false;
   }
 
@@ -609,7 +609,7 @@ edit (
 
     if (c == K_EVENT) {
       c = lastc;
-      loop_process_event(&loop);
+      queue_process_events(loop.events);
       continue;
     }
 
diff --git a/src/nvim/eval.c b/src/nvim/eval.c
index e2f095dbad..d61cccd41c 100644
--- a/src/nvim/eval.c
+++ b/src/nvim/eval.c
@@ -468,6 +468,7 @@ typedef struct {
   dict_T *self;
   int *status_ptr;
   uint64_t id;
+  Queue *events;
 } TerminalJobData;
 
 /// Structure representing current VimL to messagepack conversion state
@@ -493,6 +494,13 @@ typedef struct {
 /// Stack used to convert VimL values to messagepack.
 typedef kvec_t(MPConvStackVal) MPConvStack;
 
+typedef struct {
+  TerminalJobData *data;
+  ufunc_T *callback;
+  const char *type;
+  list_T *received;
+  int status;
+} JobEvent;
 
 #ifdef INCLUDE_GENERATED_DECLARATIONS
 # include "eval.c.generated.h"
@@ -501,15 +509,6 @@ typedef kvec_t(MPConvStackVal) MPConvStack;
 #define FNE_INCL_BR     1       /* find_name_end(): include [] in name */
 #define FNE_CHECK_START 2       /* find_name_end(): check name starts with
                                    valid character */
-// Memory pool for reusing JobEvent structures
-typedef struct {
-  TerminalJobData *data;
-  ufunc_T *callback;
-  const char *type;
-  list_T *received;
-  int status;
-} JobEvent;
-static int disable_job_defer = 0;
 static uint64_t current_job_id = 1;
 static PMap(uint64_t) *jobs = NULL; 
 
@@ -11038,15 +11037,6 @@ static void f_jobwait(typval_T *argvars, typval_T *rettv)
   list_T *rv = list_alloc();
 
   ui_busy_start();
-  // disable breakchecks, which could result in job callbacks being executed
-  // at unexpected places
-  disable_breakcheck++;
-  // disable job event deferring so the callbacks are processed while waiting.
-  if (!disable_job_defer++) {
-    // process any pending job events in the deferred queue, but only do this if
-    // deferred is not disabled(at the top-level `jobwait()` call)
-    loop_process_event(&loop);
-  }
   // For each item in the input list append an integer to the output list. -3
   // is used to represent an invalid job id, -2 is for a interrupted job and
   // -1 for jobs that were skipped or timed out.
@@ -11113,8 +11103,6 @@ static void f_jobwait(typval_T *argvars, typval_T *rettv)
     // job exits
     data->status_ptr = NULL;
   }
-  disable_job_defer--;
-  disable_breakcheck--;
   ui_busy_stop();
 
   rv->lv_refcount++;
@@ -21053,6 +21041,7 @@ static inline TerminalJobData *common_job_init(char **argv, ufunc_T *on_stdout,
   data->on_stderr = on_stderr;
   data->on_exit = on_exit;
   data->self = self;
+  data->events = queue_new_child(loop.events);
   if (pty) {
     data->proc.pty = pty_process_init(&loop, data);
   } else {
@@ -21064,6 +21053,7 @@ static inline TerminalJobData *common_job_init(char **argv, ufunc_T *on_stdout,
   proc->out = &data->out;
   proc->err = &data->err;
   proc->cb = on_process_exit;
+  proc->events = data->events;
   return data;
 }
 
@@ -21118,7 +21108,9 @@ static inline bool common_job_start(TerminalJobData *data, typval_T *rettv)
   return true;
 }
 
-static inline void free_term_job_data(TerminalJobData *data) {
+static inline void free_term_job_data_event(void **argv)
+{
+  TerminalJobData *data = argv[0];
   if (data->on_stdout) {
     user_func_unref(data->on_stdout);
   }
@@ -21133,17 +21125,25 @@ static inline void free_term_job_data(TerminalJobData *data) {
     data->self->internal_refcount--;
     dict_unref(data->self);
   }
+  queue_free(data->events);
   xfree(data);
 }
 
+static inline void free_term_job_data(TerminalJobData *data)
+{
+  // data->queue may still be used after this function returns(process_wait), so
+  // only free in the next event loop iteration
+  queue_put(loop.fast_events, free_term_job_data_event, 1, data);
+}
+
 // vimscript job callbacks must be executed on Nvim main loop
-static inline void push_job_event(TerminalJobData *data, ufunc_T *callback,
+static inline void process_job_event(TerminalJobData *data, ufunc_T *callback,
     const char *type, char *buf, size_t count, int status)
 {
-  JobEvent *event_data = xmalloc(sizeof(JobEvent));
-  event_data->received = NULL;
+  JobEvent event_data;
+  event_data.received = NULL;
   if (buf) {
-    event_data->received = list_alloc();
+    event_data.received = list_alloc();
     char *ptr = buf;
     size_t remaining = count;
     size_t off = 0;
@@ -21151,7 +21151,7 @@ static inline void push_job_event(TerminalJobData *data, ufunc_T *callback,
     while (off < remaining) {
       // append the line
       if (ptr[off] == NL) {
-        list_append_string(event_data->received, (uint8_t *)ptr, off);
+        list_append_string(event_data.received, (uint8_t *)ptr, off);
         size_t skip = off + 1;
         ptr += skip;
         remaining -= skip;
@@ -21164,17 +21164,14 @@ static inline void push_job_event(TerminalJobData *data, ufunc_T *callback,
       }
       off++;
     }
-    list_append_string(event_data->received, (uint8_t *)ptr, off);
+    list_append_string(event_data.received, (uint8_t *)ptr, off);
   } else {
-    event_data->status = status;
+    event_data.status = status;
   }
-  event_data->data = data;
-  event_data->callback = callback;
-  event_data->type = type;
-  loop_push_event(&loop, (Event) {
-    .handler = on_job_event,
-    .data = event_data
-  }, !disable_job_defer);
+  event_data.data = data;
+  event_data.callback = callback;
+  event_data.type = type;
+  on_job_event(&event_data);
 }
 
 static void on_job_stdout(Stream *stream, RBuffer *buf, void *job, bool eof)
@@ -21198,13 +21195,13 @@ static void on_job_output(Stream *stream, TerminalJobData *data, RBuffer *buf,
 
   RBUFFER_UNTIL_EMPTY(buf, ptr, len) {
     // The order here matters, the terminal must receive the data first because
-    // push_job_event will modify the read buffer(convert NULs into NLs)
+    // process_job_event will modify the read buffer(convert NULs into NLs)
     if (data->term) {
       terminal_receive(data->term, ptr, len);
     }
 
     if (callback) {
-      push_job_event(data, callback, type, ptr, len, 0);
+      process_job_event(data, callback, type, ptr, len, 0);
     }
 
     rbuffer_consumed(buf, len);
@@ -21224,7 +21221,7 @@ static void on_process_exit(Process *proc, int status, void *d)
     *data->status_ptr = status;
   }
 
-  push_job_event(data, data->on_exit, "exit", NULL, 0, status);
+  process_job_event(data, data->on_exit, "exit", NULL, 0, status);
 }
 
 static void term_write(char *buf, size_t size, void *d)
@@ -21258,10 +21255,8 @@ static void term_job_data_decref(TerminalJobData *data)
   }
 }
 
-static void on_job_event(Event event)
+static void on_job_event(JobEvent *ev)
 {
-  JobEvent *ev = event.data;
-
   if (!ev->callback) {
     goto end;
   }
@@ -21306,7 +21301,6 @@ end:
     pmap_del(uint64_t)(jobs, ev->data->id);
     term_job_data_decref(ev->data);
   }
-  xfree(ev);
 }
 
 static TerminalJobData *find_job(uint64_t id)
diff --git a/src/nvim/event/loop.c b/src/nvim/event/loop.c
index 67572c4f30..1a50ec0d9a 100644
--- a/src/nvim/event/loop.c
+++ b/src/nvim/event/loop.c
@@ -1,3 +1,4 @@
+#include <stdarg.h>
 #include <stdint.h>
 
 #include <uv.h>
@@ -9,15 +10,20 @@
 # include "event/loop.c.generated.h"
 #endif
 
+typedef struct idle_event {
+  uv_idle_t idle;
+  Event event;
+} IdleEvent;
+
 
 void loop_init(Loop *loop, void *data)
 {
   uv_loop_init(&loop->uv);
   loop->uv.data = loop;
-  loop->deferred_events = kl_init(Event);
-  loop->immediate_events = kl_init(Event);
   loop->children = kl_init(WatcherPtr);
   loop->children_stop_requests = 0;
+  loop->events = queue_new_parent(on_put, loop);
+  loop->fast_events = queue_new_child(loop->events);
   uv_signal_init(&loop->uv, &loop->children_watcher);
   uv_timer_init(&loop->uv, &loop->children_kill_timer);
   uv_timer_init(&loop->uv, &loop->poll_timer);
@@ -50,29 +56,20 @@ void loop_poll_events(Loop *loop, int ms)
   }
 
   recursive--;  // Can re-enter uv_run now
-  process_events_from(loop->immediate_events);
+  queue_process_events(loop->fast_events);
 }
 
-
-// Queue an event
-void loop_push_event(Loop *loop, Event event, bool deferred)
+static void on_put(Queue *queue, void *data)
 {
+  Loop *loop = data;
   // Sometimes libuv will run pending callbacks(timer for example) before
   // blocking for a poll. If this happens and the callback pushes a event to one
   // of the queues, the event would only be processed after the poll
   // returns(user hits a key for example). To avoid this scenario, we call
   // uv_stop when a event is enqueued.
   uv_stop(&loop->uv);
-  kl_push(Event, deferred ? loop->deferred_events : loop->immediate_events,
-      event);
 }
 
-void loop_process_event(Loop *loop)
-{
-  process_events_from(loop->deferred_events);
-}
-
-
 void loop_close(Loop *loop)
 {
   uv_close((uv_handle_t *)&loop->children_watcher, NULL);
@@ -83,20 +80,6 @@ void loop_close(Loop *loop)
   } while (uv_loop_close(&loop->uv));
 }
 
-void loop_process_all_events(Loop *loop)
-{
-  process_events_from(loop->immediate_events);
-  process_events_from(loop->deferred_events);
-}
-
-static void process_events_from(klist_t(Event) *queue)
-{
-  while (!kl_empty(queue)) {
-    Event event = kl_shift(Event, queue);
-    event.handler(event);
-  }
-}
-
 static void timer_cb(uv_timer_t *handle)
 {
 }
diff --git a/src/nvim/event/loop.h b/src/nvim/event/loop.h
index 340e096fec..9212a45aa4 100644
--- a/src/nvim/event/loop.h
+++ b/src/nvim/event/loop.h
@@ -7,37 +7,39 @@
 
 #include "nvim/lib/klist.h"
 #include "nvim/os/time.h"
-
-typedef struct event Event;
-typedef void (*event_handler)(Event event);
-
-struct event {
-  void *data;
-  event_handler handler;
-};
+#include "nvim/event/queue.h"
 
 typedef void * WatcherPtr;
 
 #define _noop(x)
 KLIST_INIT(WatcherPtr, WatcherPtr, _noop)
-KLIST_INIT(Event, Event, _noop)
 
 typedef struct loop {
   uv_loop_t uv;
-  klist_t(Event) *deferred_events, *immediate_events;
+  Queue *events, *fast_events;
   klist_t(WatcherPtr) *children;
   uv_signal_t children_watcher;
   uv_timer_t children_kill_timer, poll_timer;
   size_t children_stop_requests;
 } Loop;
 
+#define CREATE_EVENT(queue, handler, argc, ...)                  \
+  do {                                                           \
+    if (queue) {                                                 \
+      queue_put((queue), (handler), argc, __VA_ARGS__);          \
+    } else {                                                     \
+      void *argv[argc] = {__VA_ARGS__};                          \
+      (handler)(argv);                                           \
+    }                                                            \
+  } while (0)
+
 // Poll for events until a condition or timeout
-#define LOOP_POLL_EVENTS_UNTIL(loop, timeout, condition)                     \
+#define LOOP_PROCESS_EVENTS_UNTIL(loop, queue, timeout, condition)           \
   do {                                                                       \
     int remaining = timeout;                                                 \
     uint64_t before = (remaining > 0) ? os_hrtime() : 0;                     \
     while (!(condition)) {                                                   \
-      loop_poll_events(loop, remaining);                                     \
+      LOOP_PROCESS_EVENTS(loop, queue, remaining);                           \
       if (remaining == 0) {                                                  \
         break;                                                               \
       } else if (remaining > 0) {                                            \
@@ -51,6 +53,16 @@ typedef struct loop {
     }                                                                        \
   } while (0)
 
+#define LOOP_PROCESS_EVENTS(loop, queue, timeout)                            \
+  do {                                                                       \
+    if (queue && !queue_empty(queue)) {                                      \
+      queue_process_events(queue);                                           \
+    } else {                                                                 \
+      loop_poll_events(loop, timeout);                                       \
+    }                                                                        \
+  } while (0)
+
+
 #ifdef INCLUDE_GENERATED_DECLARATIONS
 # include "event/loop.h.generated.h"
 #endif
diff --git a/src/nvim/event/process.c b/src/nvim/event/process.c
index c7360b8614..54dbc11a03 100644
--- a/src/nvim/event/process.c
+++ b/src/nvim/event/process.c
@@ -22,7 +22,7 @@
 #define TERM_TIMEOUT 1000000000
 #define KILL_TIMEOUT (TERM_TIMEOUT * 2)
 
-#define CLOSE_PROC_STREAM(proc, stream)                          \
+#define CLOSE_PROC_STREAM(proc, stream)                             \
   do {                                                              \
     if (proc->stream && !proc->stream->closed) {                    \
       stream_close(proc->stream, NULL);                             \
@@ -76,6 +76,7 @@ bool process_spawn(Process *proc) FUNC_ATTR_NONNULL_ALL
 
   if (proc->in) {
     stream_init(NULL, proc->in, -1, (uv_stream_t *)&proc->in->uv.pipe, data);
+    proc->in->events = proc->events;
     proc->in->internal_data = proc;
     proc->in->internal_close_cb = on_process_stream_close;
     proc->refcount++;
@@ -83,6 +84,7 @@ bool process_spawn(Process *proc) FUNC_ATTR_NONNULL_ALL
 
   if (proc->out) {
     stream_init(NULL, proc->out, -1, (uv_stream_t *)&proc->out->uv.pipe, data);
+    proc->out->events = proc->events;
     proc->out->internal_data = proc;
     proc->out->internal_close_cb = on_process_stream_close;
     proc->refcount++;
@@ -90,6 +92,7 @@ bool process_spawn(Process *proc) FUNC_ATTR_NONNULL_ALL
 
   if (proc->err) {
     stream_init(NULL, proc->err, -1, (uv_stream_t *)&proc->err->uv.pipe, data);
+    proc->err->events = proc->events;
     proc->err->internal_data = proc;
     proc->err->internal_close_cb = on_process_stream_close;
     proc->refcount++;
@@ -112,7 +115,7 @@ void process_teardown(Loop *loop) FUNC_ATTR_NONNULL_ALL
   }
 
   // Wait until all children exit
-  LOOP_POLL_EVENTS_UNTIL(loop, -1, kl_empty(loop->children));
+  LOOP_PROCESS_EVENTS_UNTIL(loop, loop->events, -1, kl_empty(loop->children));
   pty_process_teardown(loop);
 }
 
@@ -154,11 +157,15 @@ int process_wait(Process *proc, int ms) FUNC_ATTR_NONNULL_ALL
   // The default status is -1, which represents a timeout
   int status = -1;
   bool interrupted = false;
+  if (!proc->refcount) {
+    LOOP_PROCESS_EVENTS(proc->loop, proc->events, 0);
+    return proc->status;
+  }
 
   // Increase refcount to stop the exit callback from being called(and possibly
   // being freed) before we have a chance to get the status.
   proc->refcount++;
-  LOOP_POLL_EVENTS_UNTIL(proc->loop, ms,
+  LOOP_PROCESS_EVENTS_UNTIL(proc->loop, proc->events, ms,
       // Until...
       got_int ||             // interrupted by the user
       proc->refcount == 1);  // job exited
@@ -170,12 +177,12 @@ int process_wait(Process *proc, int ms) FUNC_ATTR_NONNULL_ALL
     got_int = false;
     process_stop(proc);
     if (ms == -1) {
-      // We can only return, if all streams/handles are closed and the job
-
+      // We can only return if all streams/handles are closed and the job
       // exited.
-      LOOP_POLL_EVENTS_UNTIL(proc->loop, -1, proc->refcount == 1);
+      LOOP_PROCESS_EVENTS_UNTIL(proc->loop, proc->events, -1,
+          proc->refcount == 1);
     } else {
-      loop_poll_events(proc->loop, 0);
+      LOOP_PROCESS_EVENTS(proc->loop, proc->events, 0);
     }
   }
 
@@ -184,6 +191,10 @@ int process_wait(Process *proc, int ms) FUNC_ATTR_NONNULL_ALL
     // resources
     status = interrupted ? -2 : proc->status;
     decref(proc);
+    if (proc->events) {
+      // the decref call created an exit event, process it now
+      queue_process_events(proc->events);
+    }
   } else {
     proc->refcount--;
   }
@@ -249,6 +260,18 @@ static void children_kill_cb(uv_timer_t *handle)
   }
 }
 
+static void process_close_event(void **argv)
+{
+  Process *proc = argv[0];
+  shell_free_argv(proc->argv);
+  if (proc->type == kProcessTypePty) {
+    xfree(((PtyProcess *)proc)->term_name);
+  }
+  if (proc->cb) {
+    proc->cb(proc, proc->status, proc->data);
+  }
+}
+
 static void decref(Process *proc)
 {
   if (--proc->refcount != 0) {
@@ -263,16 +286,9 @@ static void decref(Process *proc)
       break;
     }
   }
-
   assert(node);
   kl_shift_at(WatcherPtr, loop->children, node);
-  shell_free_argv(proc->argv);
-  if (proc->type == kProcessTypePty) {
-    xfree(((PtyProcess *)proc)->term_name);
-  }
-  if (proc->cb) {
-    proc->cb(proc, proc->status, proc->data);
-  }
+  CREATE_EVENT(proc->events, process_close_event, 1, proc);
 }
 
 static void process_close(Process *proc)
@@ -292,28 +308,27 @@ static void process_close(Process *proc)
   }
 }
 
-static void on_process_exit(Process *proc)
+static void process_close_handles(void **argv)
 {
-  if (exiting) {
-    on_process_exit_event((Event) {.data = proc});
-  } else {
-    loop_push_event(proc->loop,
-        (Event) {.handler = on_process_exit_event, .data = proc}, false);
-  }
+  Process *proc = argv[0];
+  process_close_streams(proc);
+  process_close(proc);
+}
 
+static void on_process_exit(Process *proc)
+{
   Loop *loop = proc->loop;
-  if (loop->children_stop_requests && !--loop->children_stop_requests) {
+  if (proc->stopped_time && loop->children_stop_requests
+      && !--loop->children_stop_requests) {
     // Stop the timer if no more stop requests are pending
     DLOG("Stopping process kill timer");
     uv_timer_stop(&loop->children_kill_timer);
   }
-}
-
-static void on_process_exit_event(Event event)
-{
-  Process *proc = event.data;
-  process_close_streams(proc);
-  process_close(proc);
+  // 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);
 }
 
 static void on_process_stream_close(Stream *stream, void *data)
diff --git a/src/nvim/event/process.h b/src/nvim/event/process.h
index 7ef2b24b7f..45edc46b95 100644
--- a/src/nvim/event/process.h
+++ b/src/nvim/event/process.h
@@ -26,6 +26,7 @@ struct process {
   process_exit_cb cb;
   internal_process_cb internal_exit_cb, internal_close_cb;
   bool closed, term_sent;
+  Queue *events;
 };
 
 static inline Process process_init(Loop *loop, ProcessType type, void *data)
@@ -34,6 +35,7 @@ static inline Process process_init(Loop *loop, ProcessType type, void *data)
     .type = type,
     .data = data,
     .loop = loop,
+    .events = NULL,
     .pid = 0,
     .status = 0,
     .refcount = 0,
diff --git a/src/nvim/event/rstream.c b/src/nvim/event/rstream.c
index 7283cca02b..94853f616a 100644
--- a/src/nvim/event/rstream.c
+++ b/src/nvim/event/rstream.c
@@ -169,9 +169,16 @@ static void fread_idle_cb(uv_idle_t *handle)
   invoke_read_cb(stream, false);
 }
 
-static void invoke_read_cb(Stream *stream, bool eof)
+static void read_event(void **argv)
 {
+  Stream *stream = argv[0];
   if (stream->read_cb) {
+    bool eof = (uintptr_t)argv[1];
     stream->read_cb(stream, stream->buffer, stream->data, eof);
   }
 }
+
+static void invoke_read_cb(Stream *stream, bool eof)
+{
+  CREATE_EVENT(stream->events, read_event, 2, stream, (void *)(uintptr_t)eof);
+}
diff --git a/src/nvim/event/signal.c b/src/nvim/event/signal.c
index 63133b4f57..11ce15a882 100644
--- a/src/nvim/event/signal.c
+++ b/src/nvim/event/signal.c
@@ -15,6 +15,7 @@ void signal_watcher_init(Loop *loop, SignalWatcher *watcher, void *data)
   watcher->uv.data = watcher;
   watcher->data = data;
   watcher->cb = NULL;
+  watcher->events = loop->fast_events;
 }
 
 void signal_watcher_start(SignalWatcher *watcher, signal_cb cb, int signum)
@@ -37,10 +38,16 @@ void signal_watcher_close(SignalWatcher *watcher, signal_close_cb cb)
   uv_close((uv_handle_t *)&watcher->uv, close_cb);
 }
 
+static void signal_event(void **argv)
+{
+  SignalWatcher *watcher = argv[0];
+  watcher->cb(watcher, watcher->uv.signum, watcher->data);
+}
+
 static void signal_watcher_cb(uv_signal_t *handle, int signum)
 {
   SignalWatcher *watcher = handle->data;
-  watcher->cb(watcher, signum, watcher->data);
+  CREATE_EVENT(watcher->events, signal_event, 1, watcher);
 }
 
 static void close_cb(uv_handle_t *handle)
diff --git a/src/nvim/event/signal.h b/src/nvim/event/signal.h
index c269fa9d95..e32608acc0 100644
--- a/src/nvim/event/signal.h
+++ b/src/nvim/event/signal.h
@@ -14,6 +14,7 @@ struct signal_watcher {
   void *data;
   signal_cb cb;
   signal_close_cb close_cb;
+  Queue *events;
 };
 
 #ifdef INCLUDE_GENERATED_DECLARATIONS
diff --git a/src/nvim/event/socket.c b/src/nvim/event/socket.c
index 2a618d290d..347e464d25 100644
--- a/src/nvim/event/socket.c
+++ b/src/nvim/event/socket.c
@@ -77,6 +77,7 @@ void socket_watcher_init(Loop *loop, SocketWatcher *watcher,
   watcher->stream->data = watcher;
   watcher->cb = NULL;
   watcher->close_cb = NULL;
+  watcher->events = NULL;
 }
 
 int socket_watcher_start(SocketWatcher *watcher, int backlog, socket_cb cb)
@@ -143,10 +144,18 @@ void socket_watcher_close(SocketWatcher *watcher, socket_close_cb cb)
   uv_close((uv_handle_t *)watcher->stream, close_cb);
 }
 
+static void connection_event(void **argv)
+{
+  SocketWatcher *watcher = argv[0];
+  int status = (int)(uintptr_t)(argv[1]);
+  watcher->cb(watcher, status, watcher->data);
+}
+
 static void connection_cb(uv_stream_t *handle, int status)
 {
   SocketWatcher *watcher = handle->data;
-  watcher->cb(watcher, status, watcher->data);
+  CREATE_EVENT(watcher->events, connection_event, 2, watcher,
+      (void *)(uintptr_t)status);
 }
 
 static void close_cb(uv_handle_t *handle)
diff --git a/src/nvim/event/socket.h b/src/nvim/event/socket.h
index 17fd39f33b..ad59fdbe3a 100644
--- a/src/nvim/event/socket.h
+++ b/src/nvim/event/socket.h
@@ -30,6 +30,7 @@ struct socket_watcher {
   void *data;
   socket_cb cb;
   socket_close_cb close_cb;
+  Queue *events;
 };
 
 #ifdef INCLUDE_GENERATED_DECLARATIONS
diff --git a/src/nvim/event/stream.c b/src/nvim/event/stream.c
index 72dabc3ce7..6caad6fdcc 100644
--- a/src/nvim/event/stream.c
+++ b/src/nvim/event/stream.c
@@ -56,6 +56,7 @@ void stream_init(Loop *loop, Stream *stream, int fd, uv_stream_t *uvstream,
 
   if (stream->uvstream) {
     stream->uvstream->data = stream;
+    loop = stream->uvstream->loop->data;
   }
 
   stream->data = data;
@@ -70,17 +71,13 @@ void stream_init(Loop *loop, Stream *stream, int fd, uv_stream_t *uvstream,
   stream->internal_close_cb = NULL;
   stream->closed = false;
   stream->buffer = NULL;
+  stream->events = NULL;
 }
 
 void stream_close(Stream *stream, stream_close_cb on_stream_close)
   FUNC_ATTR_NONNULL_ARG(1)
 {
   assert(!stream->closed);
-
-  if (stream->buffer) {
-    rbuffer_free(stream->buffer);
-  }
-
   stream->closed = true;
   stream->close_cb = on_stream_close;
 
@@ -102,6 +99,9 @@ void stream_close_handle(Stream *stream)
 static void close_cb(uv_handle_t *handle)
 {
   Stream *stream = handle->data;
+  if (stream->buffer) {
+    rbuffer_free(stream->buffer);
+  }
   if (stream->close_cb) {
     stream->close_cb(stream, stream->data);
   }
diff --git a/src/nvim/event/stream.h b/src/nvim/event/stream.h
index 37410b2036..eaf4b010f5 100644
--- a/src/nvim/event/stream.h
+++ b/src/nvim/event/stream.h
@@ -47,6 +47,7 @@ struct stream {
   size_t pending_reqs;
   void *data, *internal_data;
   bool closed;
+  Queue *events;
 };
 
 #ifdef INCLUDE_GENERATED_DECLARATIONS
diff --git a/src/nvim/event/time.c b/src/nvim/event/time.c
index ce33cdfc10..7bf333bcea 100644
--- a/src/nvim/event/time.c
+++ b/src/nvim/event/time.c
@@ -16,6 +16,7 @@ void time_watcher_init(Loop *loop, TimeWatcher *watcher, void *data)
   uv_timer_init(&loop->uv, &watcher->uv);
   watcher->uv.data = watcher;
   watcher->data = data;
+  watcher->events = loop->fast_events;
 }
 
 void time_watcher_start(TimeWatcher *watcher, time_cb cb, uint64_t timeout,
@@ -39,11 +40,17 @@ void time_watcher_close(TimeWatcher *watcher, time_cb cb)
   uv_close((uv_handle_t *)&watcher->uv, close_cb);
 }
 
+static void time_event(void **argv)
+{
+  TimeWatcher *watcher = argv[0];
+  watcher->cb(watcher, watcher->data);
+}
+
 static void time_watcher_cb(uv_timer_t *handle)
   FUNC_ATTR_NONNULL_ALL
 {
   TimeWatcher *watcher = handle->data;
-  watcher->cb(watcher, watcher->data);
+  CREATE_EVENT(watcher->events, time_event, 1, watcher);
 }
 
 static void close_cb(uv_handle_t *handle)
diff --git a/src/nvim/event/time.h b/src/nvim/event/time.h
index ee50e53d11..7882b2b627 100644
--- a/src/nvim/event/time.h
+++ b/src/nvim/event/time.h
@@ -12,6 +12,7 @@ struct time_watcher {
   uv_timer_t uv;
   void *data;
   time_cb cb, close_cb;
+  Queue *events;
 };
 
 #ifdef INCLUDE_GENERATED_DECLARATIONS
diff --git a/src/nvim/ex_getln.c b/src/nvim/ex_getln.c
index f32aee809a..03116d454f 100644
--- a/src/nvim/ex_getln.c
+++ b/src/nvim/ex_getln.c
@@ -306,7 +306,7 @@ getcmdline (
     input_disable_events();
 
     if (c == K_EVENT) {
-      loop_process_event(&loop);
+      queue_process_events(loop.events);
       continue;
     }
 
diff --git a/src/nvim/globals.h b/src/nvim/globals.h
index e4dcad9afb..68cb923e42 100644
--- a/src/nvim/globals.h
+++ b/