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.

4 files changed, 17 insertions, 15 deletions

diff --git a/src/nvim/os/input.c b/src/nvim/os/input.c
index b0263f18d0..8bc713bcff 100644
--- a/src/nvim/os/input.c
+++ b/src/nvim/os/input.c
@@ -132,7 +132,7 @@ bool os_char_avail(void)
 // Check for CTRL-C typed by reading all available characters.
 void os_breakcheck(void)
 {
-  if (!disable_breakcheck && !got_int) {
+  if (!got_int) {
     loop_poll_events(&loop, 0);
   }
 }
@@ -292,7 +292,7 @@ static bool input_poll(int ms)
     prof_inchar_enter();
   }
 
-  LOOP_POLL_EVENTS_UNTIL(&loop, ms, input_ready() || input_eof);
+  LOOP_PROCESS_EVENTS_UNTIL(&loop, NULL, ms, input_ready() || input_eof);
 
   if (do_profiling == PROF_YES && ms) {
     prof_inchar_exit();
@@ -383,5 +383,5 @@ static void read_error_exit(void)
 
 static bool pending_events(void)
 {
-  return events_enabled && !kl_empty(loop.deferred_events);
+  return events_enabled && !queue_empty(loop.events);
 }
diff --git a/src/nvim/os/shell.c b/src/nvim/os/shell.c
index 958b4483e8..8faa46dd63 100644
--- a/src/nvim/os/shell.c
+++ b/src/nvim/os/shell.c
@@ -207,6 +207,8 @@ static int do_os_system(char **argv,
   Stream in, out, err;
   UvProcess uvproc = uv_process_init(&loop, &buf);
   Process *proc = &uvproc.process;
+  Queue *events = queue_new_child(loop.events);
+  proc->events = events;
   proc->argv = argv;
   proc->in = input != NULL ? &in : NULL;
   proc->out = &out;
@@ -219,14 +221,22 @@ static int do_os_system(char **argv,
       msg_outtrans((char_u *)prog);
       msg_putchar('\n');
     }
+    queue_free(events);
     return -1;
   }
 
+  // We want to deal with stream events as fast a possible while queueing
+  // process events, so reset everything to NULL. It prevents closing the
+  // streams while there's still data in the OS buffer(due to the process
+  // exiting before all data is read).
   if (input != NULL) {
+    proc->in->events = NULL;
     wstream_init(proc->in, 0);
   }
+  proc->out->events = NULL;
   rstream_init(proc->out, 0);
   rstream_start(proc->out, data_cb);
+  proc->err->events = NULL;
   rstream_init(proc->err, 0);
   rstream_start(proc->err, data_cb);
 
@@ -267,6 +277,9 @@ static int do_os_system(char **argv,
     }
   }
 
+  assert(queue_empty(events));
+  queue_free(events);
+
   return status;
 }
 
diff --git a/src/nvim/os/signal.c b/src/nvim/os/signal.c
index 6de3435c4c..7158721433 100644
--- a/src/nvim/os/signal.c
+++ b/src/nvim/os/signal.c
@@ -115,16 +115,6 @@ static void deadly_signal(int signum)
 static void on_signal(SignalWatcher *handle, int signum, void *data)
 {
   assert(signum >= 0);
-  loop_push_event(&loop, (Event) {
-    .handler = on_signal_event,
-    .data = (void *)(uintptr_t)signum
-  }, false);
-}
-
-static void on_signal_event(Event event)
-{
-  int signum = (int)(uintptr_t)event.data;
-
   switch (signum) {
 #ifdef SIGPWR
     case SIGPWR:
@@ -148,4 +138,3 @@ static void on_signal_event(Event event)
       break;
   }
 }
-
diff --git a/src/nvim/os/time.c b/src/nvim/os/time.c
index 6b5d4359db..ee17938afc 100644
--- a/src/nvim/os/time.c
+++ b/src/nvim/os/time.c
@@ -43,7 +43,7 @@ void os_delay(uint64_t milliseconds, bool ignoreinput)
     if (milliseconds > INT_MAX) {
       milliseconds = INT_MAX;
     }
-    LOOP_POLL_EVENTS_UNTIL(&loop, (int)milliseconds, got_int);
+    LOOP_PROCESS_EVENTS_UNTIL(&loop, NULL, (int)milliseconds, got_int);
   } else {
     os_microdelay(milliseconds * 1000);
   }