event: Extract event_poll loops to event_poll_until macro

A pattern that is becoming common across the project is to poll for events until
a certain condition is true, optionally passing a timeout. To address this
scenario, the event_poll_until macro was created and the job/channel/input
modules were refactored to use it on their blocking functions.

src/nvim/msgpack_rpc/channel.c

@@ -203,11 +203,7 @@ Object channel_send_call(uint64_t id,
   // Push the frame
   ChannelCallFrame frame = {request_id, false, false, NIL};
   kv_push(ChannelCallFrame *, channel->call_stack, &frame);
-
-  do {
-    event_poll(-1);
-  } while (!frame.returned);
-
+  event_poll_until(-1, frame.returned);
   (void)kv_pop(channel->call_stack);
 
   if (frame.errored) {

src/nvim/os/event.c

@@ -27,7 +27,7 @@ KLIST_INIT(Event, Event, _destroy_event)
 
 typedef struct {
   bool timed_out;
-  int32_t ms;
+  int ms;
   uv_timer_t *timer;
 } TimerData;
 
@@ -66,7 +66,7 @@ void event_teardown(void)
 }
 
 // Wait for some event
-bool event_poll(int32_t ms)
+void event_poll(int ms)
 {
   uv_run_mode run_mode = UV_RUN_ONCE;
 
@@ -111,8 +111,6 @@ bool event_poll(int32_t ms)
     uv_close((uv_handle_t *)&timer_prepare, NULL);
     loop(UV_RUN_NOWAIT);
   }
-
-  return !timer_data.timed_out && event_has_deferred();
 }
 
 bool event_has_deferred(void)

src/nvim/os/event.h

@@ -6,6 +6,27 @@
 
 #include "nvim/os/event_defs.h"
 #include "nvim/os/job_defs.h"
+#include "nvim/os/time.h"
+
+// Poll for events until a condition is true or a timeout has passed
+#define event_poll_until(timeout, condition)                                 \
+  do {                                                                       \
+    int remaining = timeout;                                                 \
+    uint64_t before = (remaining > 0) ? os_hrtime() : 0;                     \
+    while (!(condition)) {                                                   \
+      event_poll(remaining);                                                 \
+      if (remaining == 0) {                                                  \
+        break;                                                               \
+      } else if (remaining > 0) {                                            \
+        uint64_t now = os_hrtime();                                          \
+        remaining -= (int) ((now - before) / 1000000);                       \
+        before = now;                                                        \
+        if (remaining <= 0) {                                                \
+          break;                                                             \
+        }                                                                    \
+      }                                                                      \
+    }                                                                        \
+  } while (0)
 
 #ifdef INCLUDE_GENERATED_DECLARATIONS
 # include "os/event.h.generated.h"

src/nvim/os/input.c

@@ -163,13 +163,10 @@ void input_buffer_restore(String str)
   free(str.data);
 }
 
-static bool input_poll(int32_t ms)
+static bool input_poll(int ms)
 {
-  if (embedded_mode) {
-    return event_poll(ms);
-  }
-
-  return input_ready() || event_poll(ms) || input_ready();
+  event_poll_until(ms, input_ready());
+  return input_ready();
 }
 
 // This is a replacement for the old `WaitForChar` function in os_unix.c
@@ -294,6 +291,10 @@ static int push_event_key(uint8_t *buf, int maxlen)
 // Check if there's pending input
 static bool input_ready(void)
 {
-  return rstream_pending(read_stream) > 0 || eof;
+  return typebuf_was_filled ||                   // API call filled typeahead
+         event_has_deferred() ||                 // Events must be processed
+         (!embedded_mode && (
+            rstream_pending(read_stream) > 0 ||  // Stdin input
+            eof));                               // Stdin closed
 }
 

src/nvim/os/job.c

@@ -12,7 +12,6 @@
 #include "nvim/os/wstream_defs.h"
 #include "nvim/os/event.h"
 #include "nvim/os/event_defs.h"
-#include "nvim/os/time.h"
 #include "nvim/os/shell.h"
 #include "nvim/vim.h"
 #include "nvim/memory.h"
@@ -273,45 +272,33 @@ int job_wait(Job *job, int ms) FUNC_ATTR_NONNULL_ALL
   int old_mode = cur_tmode;
   settmode(TMODE_COOK);
 
-  // keep track of the elapsed time if ms > 0
-  uint64_t before = (ms > 0) ? os_hrtime() : 0;
+  // Increase pending_refs to stop the exit_cb from being called, which
+  // could result in the job being freed before we have a chance
+  // to get the status.
+  job->pending_refs++;
+  event_poll_until(ms,
+      // Until...
+      got_int ||                // interrupted by the user
+      job->pending_refs == 1);  // job exited
+  job->pending_refs--;
 
-  while (1) {
-    // check if the job has exited (and the status is available).
-    if (job->pending_refs == 0) {
-      break;
-    }
-
-    event_poll(ms);
-
-    // we'll assume that a user frantically hitting interrupt doesn't like
-    // the current job. Signal that it has to be killed.
-    if (got_int) {
-      job_stop(job);
-    }
-
-    if (ms == 0) {
-      break;
-    }
-
-    // check if the poll timed out, if not, decrease the ms to wait for the
-    // next run
-    if (ms > 0) {
-      uint64_t now = os_hrtime();
-      ms -= (int) ((now - before) / 1000000);
-      before = now;
-
-      // if the time elapsed is greater than the `ms` wait time, break
-      if (ms <= 0) {
-        break;
-      }
-    }
+  // we'll assume that a user frantically hitting interrupt doesn't like
+  // the current job. Signal that it has to be killed.
+  if (got_int) {
+    job_stop(job);
+    event_poll(0);
   }
 
   settmode(old_mode);
 
-  // return -1 for a timeout, the job status otherwise
-  return (job->pending_refs) ? -1 : (int) job->status;
+  if (!job->pending_refs) {
+    int status = (int) job->status;
+    job_exit_callback(job);
+    return status;
+  }
+
+  // return -1 for a timeout
+  return  -1;
 }
 
 /// Close the pipe used to write to the job.