input.c: Process only safe events before blocking.

Introduce multiqueue_process_priority() to process only events at or
above a certain priority.

src/nvim/api/vim.c

@@ -714,7 +714,6 @@ Dictionary nvim_get_mode(void)
   Dictionary rv = ARRAY_DICT_INIT;
   char *modestr = get_mode();
   bool blocked = input_blocking();
-  ILOG("blocked=%d", blocked);
 
   PUT(rv, "mode", STRING_OBJ(cstr_as_string(modestr)));
   PUT(rv, "blocking", BOOLEAN_OBJ(blocked));

src/nvim/event/defs.h

@@ -6,6 +6,11 @@
 
 #define EVENT_HANDLER_MAX_ARGC 6
 
+typedef enum {
+  kEvPriorityNormal = 1,
+  kEvPriorityAsync  = 2,   // safe to run in any state
+} EventPriority;
+
 typedef void (*argv_callback)(void **argv);
 typedef struct message {
   int priority;

src/nvim/event/multiqueue.c

@@ -127,6 +127,7 @@ void multiqueue_free(MultiQueue *this)
   xfree(this);
 }
 
+/// Removes the next item and returns its Event.
 Event multiqueue_get(MultiQueue *this)
 {
   return multiqueue_empty(this) ? NILEVENT : multiqueue_remove(this);
@@ -145,45 +146,38 @@ void multiqueue_process_events(MultiQueue *this)
 {
   assert(this);
   while (!multiqueue_empty(this)) {
-    Event event = multiqueue_get(this);
+    Event event = multiqueue_remove(this);
     if (event.handler) {
       event.handler(event.argv);
     }
   }
 }
 
-void multiqueue_process_debug(MultiQueue *this)
+void multiqueue_process_priority(MultiQueue *this, int priority)
 {
   assert(this);
   QUEUE *start = QUEUE_HEAD(&this->headtail);
   QUEUE *cur = start;
-  // MultiQueue *start = this;
-  // MultiQueue *cur   = start;
-  do {
+  while (!multiqueue_empty(this)) {
     MultiQueueItem *item = multiqueue_node_data(cur);
-    Event ev;
-    if (item->link) {
-      assert(!this->parent);
-      // get the next node in the linked queue
-      MultiQueue *linked = item->data.queue;
-      assert(!multiqueue_empty(linked));
-      MultiQueueItem *child =
-        multiqueue_node_data(QUEUE_HEAD(&linked->headtail));
-      ev = child->data.item.event;
-    } else {
-      ev = item->data.item.event;
+    assert(!item->link || !this->parent);  // Only a parent queue has link-nodes
+    Event ev = multiqueueitem_get_event(item, false);
+
+    if (ev.priority >= priority) {
+      if (ev.handler) {
+        ev.handler(ev.argv);
+      }
+      // Processed. Remove this item and get the new head.
+      (void)multiqueue_remove(this);
+      cur = QUEUE_HEAD(&this->headtail);
+    } else {
+      // Not processed. Skip this item and get the next one.
+      cur = cur->next->next;
+      if (!cur || cur == start) {
+        break;
+      }
+    }
   }
-
-    // Event event = multiqueue_get(this);
-    // if (event.handler) {
-    //   event.handler(event.argv);
-    // }
-
-    ILOG("ev: priority=%d, handler=%p arg1=%s", ev.priority, ev.handler,
-         ev.argv ? ev.argv[0] : "(null)");
-
-    cur = cur->next;
-  } while (cur && cur != start);
 }
 
 /// Removes all events without processing them.
@@ -213,36 +207,48 @@ size_t multiqueue_size(MultiQueue *this)
   return this->size;
 }
 
+/// Gets an Event from an item.
+///
+/// @param remove   Remove the node from its queue, and free it.
+static Event multiqueueitem_get_event(MultiQueueItem *item, bool remove)
+{
+  assert(item != NULL);
+  Event ev;
+  if (item->link) {
+    // get the next node in the linked queue
+    MultiQueue *linked = item->data.queue;
+    assert(!multiqueue_empty(linked));
+    MultiQueueItem *child =
+      multiqueue_node_data(QUEUE_HEAD(&linked->headtail));
+    ev = child->data.item.event;
+    // remove the child node
+    if (remove) {
+      QUEUE_REMOVE(&child->node);
+      xfree(child);
+    }
+  } else {
+    // remove the corresponding link node in the parent queue
+    if (remove && item->data.item.parent_item) {
+      QUEUE_REMOVE(&item->data.item.parent_item->node);
+      xfree(item->data.item.parent_item);
+      item->data.item.parent_item = NULL;
+    }
+    ev = item->data.item.event;
+  }
+  return ev;
+}
+
 static Event multiqueue_remove(MultiQueue *this)
 {
   assert(!multiqueue_empty(this));
   QUEUE *h = QUEUE_HEAD(&this->headtail);
   QUEUE_REMOVE(h);
   MultiQueueItem *item = multiqueue_node_data(h);
-  Event rv;
-
-  if (item->link) {
-    assert(!this->parent);
-    // remove the next node in the linked queue
-    MultiQueue *linked = item->data.queue;
-    assert(!multiqueue_empty(linked));
-    MultiQueueItem *child =
-      multiqueue_node_data(QUEUE_HEAD(&linked->headtail));
-    QUEUE_REMOVE(&child->node);
-    rv = child->data.item.event;
-    xfree(child);
-  } else {
-    if (this->parent) {
-      // remove the corresponding link node in the parent queue
-      QUEUE_REMOVE(&item->data.item.parent_item->node);
-      xfree(item->data.item.parent_item);
-    }
-    rv = item->data.item.event;
-  }
-
+  assert(!item->link || !this->parent);  // Only a parent queue has link-nodes
+  Event ev = multiqueueitem_get_event(item, true);
   this->size--;
   xfree(item);
-  return rv;
+  return ev;
 }
 
 static void multiqueue_push(MultiQueue *this, Event event)
@@ -250,6 +256,7 @@ static void multiqueue_push(MultiQueue *this, Event event)
   MultiQueueItem *item = xmalloc(sizeof(MultiQueueItem));
   item->link = false;
   item->data.item.event = event;
+  item->data.item.parent_item = NULL;
   QUEUE_INSERT_TAIL(&this->headtail, &item->node);
   if (this->parent) {
     // push link node to the parent queue

src/nvim/event/multiqueue.h

@@ -10,7 +10,7 @@ typedef struct multiqueue MultiQueue;
 typedef void (*put_callback)(MultiQueue *multiq, void *data);
 
 #define multiqueue_put(q, h, ...) \
-  multiqueue_put_event(q, event_create(1, h, __VA_ARGS__));
+  multiqueue_put_event(q, event_create(kEvPriorityNormal, h, __VA_ARGS__));
 
 
 #ifdef INCLUDE_GENERATED_DECLARATIONS

src/nvim/msgpack_rpc/channel.c

@@ -435,24 +435,26 @@ static void handle_request(Channel *channel, msgpack_object *request)
     handler.async = true;
   }
 
-  if (handler.async) {
-    char buf[256] = { 0 };
-    memcpy(buf, method->via.bin.ptr, MIN(255, method->via.bin.size));
-    if (strcmp("nvim_get_mode", buf) == 0) {
-      handler.async = input_blocking();
-    }
-  }
-
-  RequestEvent *event_data = xmalloc(sizeof(RequestEvent));
-  event_data->channel = channel;
-  event_data->handler = handler;
-  event_data->args = args;
-  event_data->request_id = request_id;
+  RequestEvent *evdata = xmalloc(sizeof(RequestEvent));
+  evdata->channel = channel;
+  evdata->handler = handler;
+  evdata->args = args;
+  evdata->request_id = request_id;
   incref(channel);
   if (handler.async) {
-    on_request_event((void **)&event_data);
+    bool is_get_mode = sizeof("nvim_get_mode") - 1 == method->via.bin.size
+      && !strncmp("nvim_get_mode", method->via.bin.ptr, method->via.bin.size);
+
+    if (is_get_mode && !input_blocking()) {
+      // Schedule on the main loop with special priority. #6247
+      Event ev = event_create(kEvPriorityAsync, on_request_event, 1, evdata);
+      multiqueue_put_event(channel->events, ev);
     } else {
-    multiqueue_put(channel->events, on_request_event, 1, event_data);
+      // Invoke immediately.
+      on_request_event((void **)&evdata);
+    }
+  } else {
+    multiqueue_put(channel->events, on_request_event, 1, evdata);
   }
 }
 

src/nvim/os/input.c

@@ -341,12 +341,12 @@ static bool input_poll(int ms)
     prof_inchar_enter();
   }
 
-  if ((ms == - 1 || ms > 0)
-      && !(events_enabled || input_ready() || input_eof)
-      ) {
+  if ((ms == - 1 || ms > 0) && !events_enabled && !input_eof) {
+    // We have discovered that the pending input will provoke a blocking wait.
+    // Process any events marked with priority `kEvPriorityAsync`: these events
+    // must be handled after flushing input. See channel.c:handle_request #6247
     blocking = true;
-    multiqueue_process_debug(main_loop.events);
-    multiqueue_process_events(main_loop.events);
+    multiqueue_process_priority(main_loop.events, kEvPriorityAsync);
   }
   LOOP_PROCESS_EVENTS_UNTIL(&main_loop, NULL, ms, input_ready() || input_eof);
   blocking = false;

test/functional/api/vim_spec.lua

@@ -9,7 +9,6 @@ local funcs = helpers.funcs
 local request = helpers.request
 local meth_pcall = helpers.meth_pcall
 local command = helpers.command
-local wait = helpers.wait
 
 describe('api', function()
   before_each(clear)
@@ -222,13 +221,6 @@ describe('api', function()
     end)
   end)
 
-  local function appendfile(fname, text)
-    local file = io.open(fname, 'a')
-    file:write(text)
-    file:flush()
-    file:close()
-  end
-
   describe('nvim_get_mode', function()
     it("during normal-mode `g` returns blocking=true", function()
       nvim("input", "o")                -- add a line