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, 13 insertions, 18 deletions

diff --git a/src/nvim/os/job.c b/src/nvim/os/job.c
index 2f5b257b91..203aa2c990 100644
--- a/src/nvim/os/job.c
+++ b/src/nvim/os/job.c
@@ -214,8 +214,8 @@ Job *job_start(char **argv,
   job->in = wstream_new(maxmem);
   wstream_set_stream(job->in, (uv_stream_t *)&job->proc_stdin);
   // Start the readable streams
-  job->out = rstream_new(read_cb, JOB_BUFFER_SIZE, job, defer);
-  job->err = rstream_new(read_cb, JOB_BUFFER_SIZE, job, defer);
+  job->out = rstream_new(read_cb, JOB_BUFFER_SIZE, job, job_event_source(job));
+  job->err = rstream_new(read_cb, JOB_BUFFER_SIZE, job, job_event_source(job));
   rstream_set_stream(job->out, (uv_stream_t *)&job->proc_stdout);
   rstream_set_stream(job->err, (uv_stream_t *)&job->proc_stderr);
   rstream_start(job->out);
@@ -269,18 +269,6 @@ bool job_write(Job *job, WBuffer *buffer)
   return wstream_write(job->in, buffer);
 }
 
-/// Sets the `defer` flag for a Job instance
-///
-/// @param rstream The Job id
-/// @param defer The new value for the flag
-void job_set_defer(Job *job, bool defer)
-{
-  job->defer = defer;
-  rstream_set_defer(job->out, defer);
-  rstream_set_defer(job->err, defer);
-}
-
-
 /// Runs the read callback associated with the job exit event
 ///
 /// @param event Object containing data necessary to invoke the callback
@@ -307,6 +295,11 @@ void *job_data(Job *job)
   return job->data;
 }
 
+EventSource job_event_source(Job *job)
+{
+  return job;
+}
+
 static void job_exit_callback(Job *job)
 {
   // Free the slot now, 'exit_cb' may want to start another job to replace
@@ -391,10 +384,12 @@ static void exit_cb(uv_process_t *proc, int64_t status, int term_signal)
 
 static void emit_exit_event(Job *job)
 {
-  Event event;
-  event.type = kEventJobExit;
-  event.data.job = job;
-  event_push(event, true);
+  Event event = {
+    .source = job_event_source(job),
+    .type = kEventJobExit,
+    .data.job = job
+  };
+  event_push(event);
 }
 
 static void close_cb(uv_handle_t *handle)