events: Refactor how events are queued for processing

To make it possible reuse `event_poll` recursively and in other blocking
function calls, this changes how deferred/immediate events are processed:

- There are two queues in event.c, one for immediate events and another for
  deferred events. The queue used when pushing/processing events is determined
  with boolean arguments passed to `event_push`/`event_process` respectively.
- Events pushed to the immediate queue are processed inside `event_poll` but
  after the `uv_run` call. This is required because libuv event loop does not
  support recursion, and processing events may result in other `event_poll`
  calls.
- Events pushed to the deferred queue are processed later by calling
  `event_process(true)`. This is required to "trick" vim into treating all
  asynchronous events as special keypresses, which is the least obtrusive
  way of introducing asynchronicity into the editor.
- RStream instances will now forward the `defer` flag to the `event_push` call.

src/nvim/edit.c

@@ -939,7 +939,7 @@ doESCkey:
       break;
 
     case K_EVENT:
-      event_process();
+      event_process(true);
       break;
 
     case K_HOME:        /* <Home> */

src/nvim/ex_getln.c

@@ -758,7 +758,7 @@ getcmdline (
      */
     switch (c) {
     case K_EVENT:
-      event_process();
+      event_process(true);
       // Force a redraw even though the command line didn't change
       shell_resized();
       goto cmdline_not_changed;
@@ -1873,8 +1873,8 @@ redraw:
       }
 
       if (IS_SPECIAL(c1)) {
-        // Process pending events
-        event_process();
+        // Process deferred events
+        event_process(true);
         // Ignore other special key codes
         continue;
       }

src/nvim/message.c

@@ -2064,7 +2064,7 @@ static int do_more_prompt(int typed_char)
     toscroll = 0;
     switch (c) {
     case K_EVENT:
-      event_process();
+      event_process(true);
       break;
     case BS:                    /* scroll one line back */
     case K_BS:

src/nvim/normal.c

@@ -7368,5 +7368,5 @@ static void nv_cursorhold(cmdarg_T *cap)
 
 static void nv_event(cmdarg_T *cap)
 {
-  event_process();
+  event_process(true);
 }

src/nvim/os/event.c

@@ -30,12 +30,13 @@ typedef struct {
 #ifdef INCLUDE_GENERATED_DECLARATIONS
 # include "os/event.c.generated.h"
 #endif
-static klist_t(Event) *event_queue;
+static klist_t(Event) *deferred_events, *immediate_events;
 
 void event_init()
 {
-  // Initialize the event queue
-  event_queue = kl_init(Event);
+  // Initialize the event queues
+  deferred_events = kl_init(Event);
+  immediate_events = kl_init(Event);
   // Initialize input events
   input_init();
   // Timer to wake the event loop if a timeout argument is passed to
@@ -67,7 +68,12 @@ bool event_poll(int32_t ms)
     return true;
   }
 
-  input_start();
+  static int recursive = 0;
+
+  if (!(recursive++)) {
+    // Only needs to start the libuv handle the first time we enter here
+    input_start();
+  }
 
   uv_timer_t timer;
   uv_prepare_t timer_prepare;
@@ -93,14 +99,21 @@ bool event_poll(int32_t ms)
     // Run one event loop iteration, blocking for events if run_mode is
     // UV_RUN_ONCE
     uv_run(uv_default_loop(), run_mode);
+    // Process immediate events outside uv_run since libuv event loop not
+    // support recursion(processing events may cause a recursive event_poll
+    // call)
+    event_process(false);
   } while (
       // Continue running if ...
       !input_ready() &&   // we have no input
-      kl_empty(event_queue) &&   // no events are waiting to be processed
+      !event_has_deferred() &&   // no events are waiting to be processed
       run_mode != UV_RUN_NOWAIT &&   // ms != 0
       !timer_data.timed_out);  // we didn't get a timeout
 
-  input_stop();
+  if (!(--recursive)) {
+    // Again, only stop when we leave the top-level invocation
+    input_stop();
+  }
 
   if (ms > 0) {
     // Ensure the timer-related handles are closed and run the event loop
@@ -111,26 +124,26 @@ bool event_poll(int32_t ms)
     event_process(false);
   }
 
-  return input_ready() || event_is_pending();
+  return input_ready() || event_has_deferred();
 }
 
-bool event_is_pending()
+bool event_has_deferred()
 {
-  return !kl_empty(event_queue);
+  return !kl_empty(get_queue(true));
 }
 
 // Push an event to the queue
-void event_push(Event event)
+void event_push(Event event, bool deferred)
 {
-  *kl_pushp(Event, event_queue) = event;
+  *kl_pushp(Event, get_queue(deferred)) = event;
 }
 
 // Runs the appropriate action for each queued event
-void event_process()
+void event_process(bool deferred)
 {
   Event event;
 
-  while (kl_shift(Event, event_queue, &event) == 0) {
+  while (kl_shift(Event, get_queue(deferred), &event) == 0) {
     switch (event.type) {
       case kEventSignal:
         signal_handle(event);
@@ -161,3 +174,8 @@ static void timer_prepare_cb(uv_prepare_t *handle)
   uv_timer_start(data->timer, timer_cb, (uint32_t)data->ms, 0);
   uv_prepare_stop(handle);
 }
+
+static klist_t(Event) *get_queue(bool deferred)
+{
+  return deferred ? deferred_events : immediate_events;
+}

src/nvim/os/input.c

@@ -67,7 +67,7 @@ int os_inchar(uint8_t *buf, int maxlen, int32_t ms, int tb_change_cnt)
 {
   InbufPollResult result;
 
-  if (event_is_pending()) {
+  if (event_has_deferred()) {
     // Return pending event bytes
     return push_event_key(buf, maxlen);
   }
@@ -91,8 +91,8 @@ int os_inchar(uint8_t *buf, int maxlen, int32_t ms, int tb_change_cnt)
     }
   }
 
-  // If there are pending events, return the keys directly
-  if (event_is_pending()) {
+  // If there are deferred events, return the keys directly
+  if (event_has_deferred()) {
     return push_event_key(buf, maxlen);
   }
 

src/nvim/os/job.c

@@ -390,14 +390,10 @@ static void exit_cb(uv_process_t *proc, int64_t status, int term_signal)
 
 static void emit_exit_event(Job *job)
 {
-  if (job->defer) {
-    Event event;
-    event.type = kEventJobExit;
-    event.data.job = job;
-    event_push(event);
-  } else {
-    job_exit_callback(job);
-  }
+  Event event;
+  event.type = kEventJobExit;
+  event.data.job = job;
+  event_push(event, true);
 }
 
 static void close_cb(uv_handle_t *handle)

src/nvim/os/rstream.c

@@ -340,16 +340,9 @@ static void close_cb(uv_handle_t *handle)
 
 static void emit_read_event(RStream *rstream, bool eof)
 {
-  if (rstream->defer) {
-    Event event;
-
-    event.type = kEventRStreamData;
-    event.data.rstream.ptr = rstream;
-    event.data.rstream.eof = eof;
-    event_push(event);
-  } else {
-    // Invoke the callback passing in the number of bytes available and data
-    // associated with the stream
-    rstream->cb(rstream, rstream->data, eof);
-  }
+  Event event;
+  event.type = kEventRStreamData;
+  event.data.rstream.ptr = rstream;
+  event.data.rstream.eof = eof;
+  event_push(event, rstream->defer);
 }

src/nvim/os/signal.c

@@ -159,5 +159,5 @@ static void signal_cb(uv_signal_t *handle, int signum)
       .signum = signum
     }
   };
-  event_push(event);
+  event_push(event, true);
 }