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, 9 insertions, 21 deletions

diff --git a/src/nvim/os/channel.c b/src/nvim/os/channel.c
index 11a58f246a..d5f29aa667 100644
--- a/src/nvim/os/channel.c
+++ b/src/nvim/os/channel.c
@@ -118,7 +118,7 @@ void channel_from_stream(uv_stream_t *stream)
   stream->data = NULL;
   channel->is_job = false;
   // read stream
-  channel->data.streams.read = rstream_new(parse_msgpack, 1024, channel, true);
+  channel->data.streams.read = rstream_new(parse_msgpack, 1024, channel, NULL);
   rstream_set_stream(channel->data.streams.read, stream);
   rstream_start(channel->data.streams.read);
   // write stream
@@ -189,16 +189,10 @@ bool channel_send_call(uint64_t id,
   // Send the msgpack-rpc request
   send_request(channel, request_id, name, arg);
 
-  if (!kv_size(channel->call_stack)) {
-    // This is the first frame, we must disable event deferral for this
-    // channel because we won't be returning until the client sends a
-    // response
-    if (channel->is_job) {
-      job_set_defer(channel->data.job, false);
-    } else {
-      rstream_set_defer(channel->data.streams.read, false);
-    }
-  }
+  EventSource channel_source = channel->is_job
+    ? job_event_source(channel->data.job)
+    : rstream_event_source(channel->data.streams.read);
+  EventSource sources[] = {channel_source, NULL};
 
   // Push the frame
   ChannelCallFrame frame = {request_id, false, NIL};
@@ -206,24 +200,18 @@ bool channel_send_call(uint64_t id,
   size_t size = kv_size(channel->call_stack);
 
   do {
-    event_poll(-1);
+    event_poll(-1, sources);
   } while (
       // Continue running if ...
       channel->enabled &&  // the channel is still enabled
       kv_size(channel->call_stack) >= size);  // the call didn't return
 
-  if (!kv_size(channel->call_stack)) {
-    // Popped last frame, restore event deferral
-    if (channel->is_job) {
-      job_set_defer(channel->data.job, true);
-    } else {
-      rstream_set_defer(channel->data.streams.read, true);
-    }
-    if (!channel->enabled && !channel->rpc_call_level) {
+  if (!(kv_size(channel->call_stack)
+        || channel->enabled
+        || channel->rpc_call_level)) {
       // Close the channel if it has been disabled and we have not been called
       // by `parse_msgpack`(It would be unsafe to close the channel otherwise)
       close_channel(channel);
-    }
   }
 
   *errored = frame.errored;