events: Refactor how event deferral is handled

- Remove all *_set_defer methods and the 'defer' flag from rstream/jobs
- Added {signal,rstream,job}_event_source functions. Each return a pointer that
  represent the event source for the object in question(For signals, a static
  pointer is returned)
- Added a 'source' field to the Event struct, which is set to the appropriate
  value by the code that created the event.
- Added a 'sources' parameter to `event_poll`. It should point to a
  NULL-terminated array of event sources that will be used to decide which
  events should be processed immediately
- Added a 'source_override' parameter to `rstream_new`. This was required to use
  jobs as event sources of RStream instances(When "focusing" on a job, for
  example).
- Extracted `process_from` static function from `event_process`.
- Remove 'defer' parameter from `event_process`, which now operates only on
  deferred events.
- Refactor `channel_send_call` to use the new lock mechanism

What changed in a single sentence: Code that calls `event_poll` have to specify
which event sources should NOT be deferred. This change was necessary for a
number of reasons:

- To fix a bug where due to race conditions, a client request
  could end in the deferred queue in the middle of a `channel_send_call`
  invocation, resulting in a deadlock since the client process would never
  receive a response, and channel_send_call would never return because
  the client would still be waiting for the response.
- To handle "event locking" correctly in recursive `channel_send_call`
  invocations when the frames are waiting for responses from different
  clients. Not much of an issue now since there's only a python client, but
  could break things later.
- To simplify the process of implementing synchronous functions that depend on
  asynchronous events.

1 files changed, 73 insertions, 15 deletions

diff --git a/src/nvim/os/event.c b/src/nvim/os/event.c
index 6367a60af7..a460b2db96 100644
--- a/src/nvim/os/event.c
+++ b/src/nvim/os/event.c
@@ -33,6 +33,11 @@ typedef struct {
 # include "os/event.c.generated.h"
 #endif
 static klist_t(Event) *deferred_events, *immediate_events;
+// NULL-terminated array of event sources that we should process immediately.
+//
+// Events from sources that are not contained in this array are processed
+// later when `event_process` is called
+static EventSource *immediate_sources = NULL;
 
 void event_init(void)
 {
@@ -63,7 +68,8 @@ void event_teardown(void)
 }
 
 // Wait for some event
-bool event_poll(int32_t ms)
+bool event_poll(int32_t ms, EventSource sources[])
+  FUNC_ATTR_NONNULL_ARG(2)
 {
   uv_run_mode run_mode = UV_RUN_ONCE;
 
@@ -99,10 +105,7 @@ bool event_poll(int32_t ms)
   do {
     // Run one event loop iteration, blocking for events if run_mode is
     // UV_RUN_ONCE
-    DLOG("Entering event loop");
-    uv_run(uv_default_loop(), run_mode);
-    processed_events = event_process(false);
-    DLOG("Exited event loop, processed %u events", processed_events);
+    processed_events = loop(run_mode, sources);
   } while (
       // Continue running if ...
       !processed_events &&   // we didn't process any immediate events
@@ -120,8 +123,7 @@ bool event_poll(int32_t ms)
     // once more to let libuv perform it's cleanup
     uv_close((uv_handle_t *)&timer, NULL);
     uv_close((uv_handle_t *)&timer_prepare, NULL);
-    uv_run(uv_default_loop(), UV_RUN_NOWAIT);
-    event_process(false);
+    processed_events += loop(UV_RUN_NOWAIT, sources);
   }
 
   return !timer_data.timed_out && (processed_events || event_has_deferred());
@@ -129,22 +131,41 @@ bool event_poll(int32_t ms)
 
 bool event_has_deferred(void)
 {
-  return !kl_empty(get_queue(true));
+  return !kl_empty(deferred_events);
 }
 
-// Push an event to the queue
-void event_push(Event event, bool deferred)
+// Queue an event
+void event_push(Event event)
 {
-  *kl_pushp(Event, get_queue(deferred)) = event;
+  bool defer = true;
+
+  if (immediate_sources) {
+    size_t i;
+    EventSource src;
+
+    for (src = immediate_sources[i = 0]; src; src = immediate_sources[++i]) {
+      if (src == event.source) {
+        defer = false;
+        break;
+      }
+    }
+  }
+
+  *kl_pushp(Event, defer ? deferred_events : immediate_events) = event;
+}
+
+void event_process(void)
+{
+  process_from(deferred_events);
 }
 
 // Runs the appropriate action for each queued event
-size_t event_process(bool deferred)
+static size_t process_from(klist_t(Event) *queue)
 {
   size_t count = 0;
   Event event;
 
-  while (kl_shift(Event, get_queue(deferred), &event) == 0) {
+  while (kl_shift(Event, queue, &event) == 0) {
     switch (event.type) {
       case kEventSignal:
         signal_handle(event);
@@ -161,6 +182,8 @@ size_t event_process(bool deferred)
     count++;
   }
 
+  DLOG("Processed %u events", count);
+
   return count;
 }
 
@@ -179,7 +202,42 @@ static void timer_prepare_cb(uv_prepare_t *handle)
   uv_prepare_stop(handle);
 }
 
-static klist_t(Event) *get_queue(bool deferred)
+static void requeue_deferred_events(void)
 {
-  return deferred ? deferred_events : immediate_events;
+  size_t remaining = deferred_events->size;
+
+  DLOG("Number of deferred events: %u", remaining);
+
+  while (remaining--) {
+    // Re-push each deferred event to ensure it will be in the right queue
+    Event event;
+    kl_shift(Event, deferred_events, &event);
+    event_push(event);
+    DLOG("Re-queueing event");
+  }
+
+  DLOG("Number of deferred events: %u", deferred_events->size);
+}
+
+static size_t loop(uv_run_mode run_mode, EventSource *sources)
+{
+  size_t count;
+  immediate_sources = sources;
+  // It's possible that some events from the immediate sources are waiting
+  // in the deferred queue. If so, move them to the immediate queue so they
+  // will be processed in order of arrival by the next `process_from` call.
+  requeue_deferred_events();
+  count = process_from(immediate_events);
+
+  if (count) {
+    // No need to enter libuv, events were already processed
+    return count;
+  }
+
+  DLOG("Enter event loop");
+  uv_run(uv_default_loop(), run_mode);
+  DLOG("Exit event loop");
+  immediate_sources = NULL;
+  count = process_from(immediate_events);
+  return count;
 }