From 6186df3562e33e92f04ed8c850204ceabc4746e1 Mon Sep 17 00:00:00 2001
From: "Justin M. Keyes" <justinkz@gmail.com>
Date: Thu, 29 Sep 2016 17:09:37 +0200
Subject: event/multiqueue.c: Rename "queue" to "multiqueue".

`lib/queue.h` implements a basic queue. `event/queue.c` implements
a specialized data structure on top of lib/queue.h; it is not a "normal"
queue.

Rename the specialized multi-level queue implemented in event/queue.c to
"multiqueue", to avoid confusion when reading the code.

Before this change one can eventually notice that "macros (uppercase
symbols) are for the normal queue, lowercase operations are for the
multi-level queue", but that is unnecessary friction for new developers
(or existing developers just visiting this part of the codebase).
---
 src/.asan-blacklist            |   4 +-
 src/nvim/edit.c                |   2 +-
 src/nvim/eval.c                |  28 +++---
 src/nvim/event/loop.c          |  24 ++---
 src/nvim/event/loop.h          |  20 ++--
 src/nvim/event/multiqueue.c    | 209 +++++++++++++++++++++++++++++++++++++++++
 src/nvim/event/multiqueue.h    |  19 ++++
 src/nvim/event/process.c       |  11 ++-
 src/nvim/event/process.h       |   2 +-
 src/nvim/event/queue.c         | 208 ----------------------------------------
 src/nvim/event/queue.h         |  19 ----
 src/nvim/event/signal.h        |   2 +-
 src/nvim/event/socket.h        |   2 +-
 src/nvim/event/stream.h        |   2 +-
 src/nvim/event/time.c          |   2 +-
 src/nvim/event/time.h          |   2 +-
 src/nvim/ex_getln.c            |   2 +-
 src/nvim/main.c                |   2 +-
 src/nvim/msgpack_rpc/channel.c |  13 +--
 src/nvim/normal.c              |   6 +-
 src/nvim/os/input.c            |   6 +-
 src/nvim/os/shell.c            |   8 +-
 src/nvim/state.c               |   6 +-
 src/nvim/terminal.c            |   6 +-
 src/nvim/tui/input.c           |   2 +-
 src/nvim/tui/tui.c             |   4 +-
 26 files changed, 307 insertions(+), 304 deletions(-)
 create mode 100644 src/nvim/event/multiqueue.c
 create mode 100644 src/nvim/event/multiqueue.h
 delete mode 100644 src/nvim/event/queue.c
 delete mode 100644 src/nvim/event/queue.h

(limited to 'src')

diff --git a/src/.asan-blacklist b/src/.asan-blacklist
index 63558170b3..7636f8fa82 100644
--- a/src/.asan-blacklist
+++ b/src/.asan-blacklist
@@ -1,3 +1,3 @@
-# libuv queue.h pointer arithmetic is not accepted by asan
-fun:queue_node_data
+# multiqueue.h pointer arithmetic is not accepted by asan
+fun:multiqueue_node_data
 fun:dictwatcher_node_data
diff --git a/src/nvim/edit.c b/src/nvim/edit.c
index ca5b6f02ce..892748ff5c 100644
--- a/src/nvim/edit.c
+++ b/src/nvim/edit.c
@@ -962,7 +962,7 @@ static int insert_handle_key(InsertState *s)
     break;
 
   case K_EVENT:       // some event
-    queue_process_events(main_loop.events);
+    multiqueue_process_events(main_loop.events);
     break;
 
   case K_FOCUSGAINED:  // Neovim has been given focus
diff --git a/src/nvim/eval.c b/src/nvim/eval.c
index b05b4c4f7e..0b1fd6670e 100644
--- a/src/nvim/eval.c
+++ b/src/nvim/eval.c
@@ -415,7 +415,7 @@ typedef struct {
   dict_T *self;
   int *status_ptr;
   uint64_t id;
-  Queue *events;
+  MultiQueue *events;
 } TerminalJobData;
 
 typedef struct dict_watcher {
@@ -11692,7 +11692,7 @@ static void f_jobwait(typval_T *argvars, typval_T *rettv, FunPtr fptr)
   list_T *rv = list_alloc();
 
   ui_busy_start();
-  Queue *waiting_jobs = queue_new_parent(loop_on_put, &main_loop);
+  MultiQueue *waiting_jobs = multiqueue_new_parent(loop_on_put, &main_loop);
   // For each item in the input list append an integer to the output list. -3
   // is used to represent an invalid job id, -2 is for a interrupted job and
   // -1 for jobs that were skipped or timed out.
@@ -11708,8 +11708,8 @@ static void f_jobwait(typval_T *argvars, typval_T *rettv, FunPtr fptr)
       data->status_ptr = &rv->lv_last->li_tv.vval.v_number;
       // Process any pending events for the job because we'll temporarily
       // replace the parent queue
-      queue_process_events(data->events);
-      queue_replace_parent(data->events, waiting_jobs);
+      multiqueue_process_events(data->events);
+      multiqueue_replace_parent(data->events, waiting_jobs);
     }
   }
 
@@ -11769,11 +11769,11 @@ static void f_jobwait(typval_T *argvars, typval_T *rettv, FunPtr fptr)
       continue;
     }
     // restore the parent queue for the job
-    queue_process_events(data->events);
-    queue_replace_parent(data->events, main_loop.events);
+    multiqueue_process_events(data->events);
+    multiqueue_replace_parent(data->events, main_loop.events);
   }
 
-  queue_free(waiting_jobs);
+  multiqueue_free(waiting_jobs);
   ui_busy_stop();
   rv->lv_refcount++;
   rettv->v_type = VAR_LIST;
@@ -16400,7 +16400,7 @@ static void f_timer_start(typval_T *argvars, typval_T *rettv, FunPtr fptr)
   timer->callback = func;
 
   time_watcher_init(&main_loop, &timer->tw, timer);
-  timer->tw.events = queue_new_child(main_loop.events);
+  timer->tw.events = multiqueue_new_child(main_loop.events);
   // if main loop is blocked, don't queue up multiple events
   timer->tw.blockable = true;
   time_watcher_start(&timer->tw, timer_due_cb, timeout,
@@ -16477,7 +16477,7 @@ static void timer_stop(timer_T *timer)
 static void timer_close_cb(TimeWatcher *tw, void *data)
 {
   timer_T *timer = (timer_T *)data;
-  queue_free(timer->tw.events);
+  multiqueue_free(timer->tw.events);
   user_func_unref(timer->callback);
   pmap_del(uint64_t)(timers, timer->timer_id);
   timer_decref(timer);
@@ -21725,7 +21725,7 @@ static inline TerminalJobData *common_job_init(char **argv,
   data->on_stderr = on_stderr;
   data->on_exit = on_exit;
   data->self = self;
-  data->events = queue_new_child(main_loop.events);
+  data->events = multiqueue_new_child(main_loop.events);
   data->rpc = rpc;
   if (pty) {
     data->proc.pty = pty_process_init(&main_loop, data);
@@ -21834,7 +21834,7 @@ static inline void free_term_job_data_event(void **argv)
   if (data->self) {
     dict_unref(data->self);
   }
-  queue_free(data->events);
+  multiqueue_free(data->events);
   pmap_del(uint64_t)(jobs, data->id);
   xfree(data);
 }
@@ -21843,7 +21843,7 @@ static inline void free_term_job_data(TerminalJobData *data)
 {
   // data->queue may still be used after this function returns(process_wait), so
   // only free in the next event loop iteration
-  queue_put(main_loop.fast_events, free_term_job_data_event, 1, data);
+  multiqueue_put(main_loop.fast_events, free_term_job_data_event, 1, data);
 }
 
 // vimscript job callbacks must be executed on Nvim main loop
@@ -21962,7 +21962,7 @@ static inline void term_delayed_free(void **argv)
 {
   TerminalJobData *j = argv[0];
   if (j->in.pending_reqs || j->out.pending_reqs || j->err.pending_reqs) {
-    queue_put(j->events, term_delayed_free, 1, j);
+    multiqueue_put(j->events, term_delayed_free, 1, j);
     return;
   }
 
@@ -21977,7 +21977,7 @@ static void term_close(void *d)
     data->exited = true;
     process_stop((Process *)&data->proc);
   }
-  queue_put(data->events, term_delayed_free, 1, data);
+  multiqueue_put(data->events, term_delayed_free, 1, data);
 }
 
 static void term_job_data_decref(TerminalJobData *data)
diff --git a/src/nvim/event/loop.c b/src/nvim/event/loop.c
index a4c4e5ac9c..d562ac1ed3 100644
--- a/src/nvim/event/loop.c
+++ b/src/nvim/event/loop.c
@@ -18,9 +18,9 @@ void loop_init(Loop *loop, void *data)
   loop->uv.data = loop;
   loop->children = kl_init(WatcherPtr);
   loop->children_stop_requests = 0;
-  loop->events = queue_new_parent(loop_on_put, loop);
-  loop->fast_events = queue_new_child(loop->events);
-  loop->thread_events = queue_new_parent(NULL, NULL);
+  loop->events = multiqueue_new_parent(loop_on_put, loop);
+  loop->fast_events = multiqueue_new_child(loop->events);
+  loop->thread_events = multiqueue_new_parent(NULL, NULL);
   uv_mutex_init(&loop->mutex);
   uv_async_init(&loop->uv, &loop->async, async_cb);
   uv_signal_init(&loop->uv, &loop->children_watcher);
@@ -53,19 +53,19 @@ void loop_poll_events(Loop *loop, int ms)
   }
 
   loop->recursive--;  // Can re-enter uv_run now
-  queue_process_events(loop->fast_events);
+  multiqueue_process_events(loop->fast_events);
 }
 
 // Schedule an event from another thread
 void loop_schedule(Loop *loop, Event event)
 {
   uv_mutex_lock(&loop->mutex);
-  queue_put_event(loop->thread_events, event);
+  multiqueue_put_event(loop->thread_events, event);
   uv_async_send(&loop->async);
   uv_mutex_unlock(&loop->mutex);
 }
 
-void loop_on_put(Queue *queue, void *data)
+void loop_on_put(MultiQueue *queue, void *data)
 {
   Loop *loop = data;
   // Sometimes libuv will run pending callbacks(timer for example) before
@@ -86,9 +86,9 @@ void loop_close(Loop *loop, bool wait)
   do {
     uv_run(&loop->uv, wait ? UV_RUN_DEFAULT : UV_RUN_NOWAIT);
   } while (uv_loop_close(&loop->uv) && wait);
-  queue_free(loop->fast_events);
-  queue_free(loop->thread_events);
-  queue_free(loop->events);
+  multiqueue_free(loop->fast_events);
+  multiqueue_free(loop->thread_events);
+  multiqueue_free(loop->events);
   kl_destroy(WatcherPtr, loop->children);
 }
 
@@ -96,9 +96,9 @@ static void async_cb(uv_async_t *handle)
 {
   Loop *l = handle->loop->data;
   uv_mutex_lock(&l->mutex);
-  while (!queue_empty(l->thread_events)) {
-    Event ev = queue_get(l->thread_events);
-    queue_put_event(l->fast_events, ev);
+  while (!multiqueue_empty(l->thread_events)) {
+    Event ev = multiqueue_get(l->thread_events);
+    multiqueue_put_event(l->fast_events, ev);
   }
   uv_mutex_unlock(&l->mutex);
 }
diff --git a/src/nvim/event/loop.h b/src/nvim/event/loop.h
index 407aa4245f..e7d7bdd483 100644
--- a/src/nvim/event/loop.h
+++ b/src/nvim/event/loop.h
@@ -7,7 +7,7 @@
 
 #include "nvim/lib/klist.h"
 #include "nvim/os/time.h"
-#include "nvim/event/queue.h"
+#include "nvim/event/multiqueue.h"
 
 typedef void * WatcherPtr;
 
@@ -16,7 +16,7 @@ KLIST_INIT(WatcherPtr, WatcherPtr, _noop)
 
 typedef struct loop {
   uv_loop_t uv;
-  Queue *events, *fast_events, *thread_events;
+  MultiQueue *events, *fast_events, *thread_events;
   klist_t(WatcherPtr) *children;
   uv_signal_t children_watcher;
   uv_timer_t children_kill_timer, poll_timer;
@@ -26,10 +26,10 @@ typedef struct loop {
   int recursive;
 } Loop;
 
-#define CREATE_EVENT(queue, handler, argc, ...) \
+#define CREATE_EVENT(multiqueue, handler, argc, ...) \
   do { \
-    if (queue) { \
-      queue_put((queue), (handler), argc, __VA_ARGS__); \
+    if (multiqueue) { \
+      multiqueue_put((multiqueue), (handler), argc, __VA_ARGS__); \
     } else { \
       void *argv[argc] = { __VA_ARGS__ }; \
       (handler)(argv); \
@@ -37,12 +37,12 @@ typedef struct loop {
   } while (0)
 
 // Poll for events until a condition or timeout
-#define LOOP_PROCESS_EVENTS_UNTIL(loop, queue, timeout, condition) \
+#define LOOP_PROCESS_EVENTS_UNTIL(loop, multiqueue, timeout, condition) \
   do { \
     int remaining = timeout; \
     uint64_t before = (remaining > 0) ? os_hrtime() : 0; \
     while (!(condition)) { \
-      LOOP_PROCESS_EVENTS(loop, queue, remaining); \
+      LOOP_PROCESS_EVENTS(loop, multiqueue, remaining); \
       if (remaining == 0) { \
         break; \
       } else if (remaining > 0) { \
@@ -56,10 +56,10 @@ typedef struct loop {
     } \
   } while (0)
 
-#define LOOP_PROCESS_EVENTS(loop, queue, timeout) \
+#define LOOP_PROCESS_EVENTS(loop, multiqueue, timeout) \
   do { \
-    if (queue && !queue_empty(queue)) { \
-      queue_process_events(queue); \
+    if (multiqueue && !multiqueue_empty(multiqueue)) { \
+      multiqueue_process_events(multiqueue); \
     } else { \
       loop_poll_events(loop, timeout); \
     } \
diff --git a/src/nvim/event/multiqueue.c b/src/nvim/event/multiqueue.c
new file mode 100644
index 0000000000..7efdfc4cad
--- /dev/null
+++ b/src/nvim/event/multiqueue.c
@@ -0,0 +1,209 @@
+// Multi-level queue for selective async event processing. Multiqueue supports
+// a parent-child relationship with the following properties:
+//
+// - pushing a node to a child queue will push a corresponding link node to the
+//   parent queue
+// - removing a link node from a parent queue will remove the next node
+//   in the linked child queue
+// - removing a node from a child queue will remove the corresponding link node
+//   in the parent queue
+//
+// These properties allow Nvim to organize and process events from different
+// sources with a certain degree of control. How the multiqueue is used:
+//
+//                         +----------------+
+//                         |   Main loop    |
+//                         +----------------+
+//                                  ^
+//                                  |
+//                         +----------------+
+//         +-------------->|   Event loop   |<------------+
+//         |               +--+-------------+             |
+//         |                  ^           ^               |
+//         |                  |           |               |
+//    +-----------+   +-----------+    +---------+    +---------+
+//    | Channel 1 |   | Channel 2 |    |  Job 1  |    |  Job 2  |
+//    +-----------+   +-----------+    +---------+    +---------+
+//
+//
+// The lower boxes represent event emitters, each with its own private queue
+// having the event loop queue as the parent.
+//
+// When idle, the main loop spins the event loop which queues events from many
+// sources (channels, jobs, user...). Each event emitter pushes events to its
+// private queue which is propagated to the event loop queue. When the main loop
+// consumes an event, the corresponding event is removed from the emitter's
+// queue.
+//
+// The main reason for this queue hierarchy is to allow focusing on a single
+// event emitter while blocking the main loop. For example, if the `jobwait`
+// VimL function is called on job1, the main loop will temporarily stop polling
+// the event loop queue and poll job1 queue instead. Same with channels, when
+// calling `rpcrequest` we want to temporarily stop processing events from
+// other sources and focus on a specific channel.
+
+#include <assert.h>
+#include <stdarg.h>
+#include <stdbool.h>
+#include <stdint.h>
+
+
+#include <uv.h>
+
+#include "nvim/event/multiqueue.h"
+#include "nvim/memory.h"
+#include "nvim/os/time.h"
+
+typedef struct multiqueue_item MultiQueueItem;
+struct multiqueue_item {
+  union {
+    MultiQueue *queue;
+    struct {
+      Event event;
+      MultiQueueItem *parent;
+    } item;
+  } data;
+  bool link;  // true: current item is just a link to a node in a child queue
+  QUEUE node;
+};
+
+struct multiqueue {
+  MultiQueue *parent;
+  QUEUE headtail;
+  put_callback put_cb;
+  void *data;
+};
+
+#ifdef INCLUDE_GENERATED_DECLARATIONS
+# include "event/multiqueue.c.generated.h"
+#endif
+
+static Event NILEVENT = { .handler = NULL, .argv = {NULL} };
+
+MultiQueue *multiqueue_new_parent(put_callback put_cb, void *data)
+{
+  return multiqueue_new(NULL, put_cb, data);
+}
+
+MultiQueue *multiqueue_new_child(MultiQueue *parent)
+  FUNC_ATTR_NONNULL_ALL
+{
+  assert(!parent->parent);
+  return multiqueue_new(parent, NULL, NULL);
+}
+
+static MultiQueue *multiqueue_new(MultiQueue *parent, put_callback put_cb,
+                                  void *data)
+{
+  MultiQueue *rv = xmalloc(sizeof(MultiQueue));
+  QUEUE_INIT(&rv->headtail);
+  rv->parent = parent;
+  rv->put_cb = put_cb;
+  rv->data = data;
+  return rv;
+}
+
+void multiqueue_free(MultiQueue *this)
+{
+  assert(this);
+  while (!QUEUE_EMPTY(&this->headtail)) {
+    QUEUE *q = QUEUE_HEAD(&this->headtail);
+    MultiQueueItem *item = multiqueue_node_data(q);
+    if (this->parent) {
+      QUEUE_REMOVE(&item->data.item.parent->node);
+      xfree(item->data.item.parent);
+    }
+    QUEUE_REMOVE(q);
+    xfree(item);
+  }
+
+  xfree(this);
+}
+
+Event multiqueue_get(MultiQueue *this)
+{
+  return multiqueue_empty(this) ? NILEVENT : multiqueue_remove(this);
+}
+
+void multiqueue_put_event(MultiQueue *this, Event event)
+{
+  assert(this);
+  multiqueue_push(this, event);
+  if (this->parent && this->parent->put_cb) {
+    this->parent->put_cb(this->parent, this->parent->data);
+  }
+}
+
+void multiqueue_process_events(MultiQueue *this)
+{
+  assert(this);
+  while (!multiqueue_empty(this)) {
+    Event event = multiqueue_get(this);
+    if (event.handler) {
+      event.handler(event.argv);
+    }
+  }
+}
+
+bool multiqueue_empty(MultiQueue *this)
+{
+  assert(this);
+  return QUEUE_EMPTY(&this->headtail);
+}
+
+void multiqueue_replace_parent(MultiQueue *this, MultiQueue *new_parent)
+{
+  assert(multiqueue_empty(this));
+  this->parent = new_parent;
+}
+
+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->node);
+      xfree(item->data.item.parent);
+    }
+    rv = item->data.item.event;
+  }
+
+  xfree(item);
+  return rv;
+}
+
+static void multiqueue_push(MultiQueue *this, Event event)
+{
+  MultiQueueItem *item = xmalloc(sizeof(MultiQueueItem));
+  item->link = false;
+  item->data.item.event = event;
+  QUEUE_INSERT_TAIL(&this->headtail, &item->node);
+  if (this->parent) {
+    // push link node to the parent queue
+    item->data.item.parent = xmalloc(sizeof(MultiQueueItem));
+    item->data.item.parent->link = true;
+    item->data.item.parent->data.queue = this;
+    QUEUE_INSERT_TAIL(&this->parent->headtail, &item->data.item.parent->node);
+  }
+}
+
+static MultiQueueItem *multiqueue_node_data(QUEUE *q)
+{
+  return QUEUE_DATA(q, MultiQueueItem, node);
+}
diff --git a/src/nvim/event/multiqueue.h b/src/nvim/event/multiqueue.h
new file mode 100644
index 0000000000..def6b95a10
--- /dev/null
+++ b/src/nvim/event/multiqueue.h
@@ -0,0 +1,19 @@
+#ifndef NVIM_EVENT_MULTIQUEUE_H
+#define NVIM_EVENT_MULTIQUEUE_H
+
+#include <uv.h>
+
+#include "nvim/event/defs.h"
+#include "nvim/lib/queue.h"
+
+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__));
+
+
+#ifdef INCLUDE_GENERATED_DECLARATIONS
+# include "event/multiqueue.h.generated.h"
+#endif
+#endif  // NVIM_EVENT_MULTIQUEUE_H
diff --git a/src/nvim/event/process.c b/src/nvim/event/process.c
index 51f20b7eac..39dd5fd55a 100644
--- a/src/nvim/event/process.c
+++ b/src/nvim/event/process.c
@@ -126,7 +126,7 @@ void process_teardown(Loop *loop) FUNC_ATTR_NONNULL_ALL
   // Wait until all children exit and all close events are processed.
   LOOP_PROCESS_EVENTS_UNTIL(
       loop, loop->events, -1,
-      kl_empty(loop->children) && queue_empty(loop->events));
+      kl_empty(loop->children) && multiqueue_empty(loop->events));
   pty_process_teardown(loop);
 }
 
@@ -163,7 +163,8 @@ void process_close_err(Process *proc) FUNC_ATTR_NONNULL_ALL
 ///         indistinguishable from the process returning -1 by itself. Which
 ///         is possible on some OS. Returns -2 if an user has interruped the
 ///         wait.
-int process_wait(Process *proc, int ms, Queue *events) FUNC_ATTR_NONNULL_ARG(1)
+int process_wait(Process *proc, int ms, MultiQueue *events)
+  FUNC_ATTR_NONNULL_ARG(1)
 {
   // The default status is -1, which represents a timeout
   int status = -1;
@@ -208,7 +209,7 @@ int process_wait(Process *proc, int ms, Queue *events) FUNC_ATTR_NONNULL_ARG(1)
     decref(proc);
     if (events) {
       // the decref call created an exit event, process it now
-      queue_process_events(events);
+      multiqueue_process_events(events);
     }
   } else {
     proc->refcount--;
@@ -362,7 +363,7 @@ static void flush_stream(Process *proc, Stream *stream)
     // Poll for data and process the generated events.
     loop_poll_events(proc->loop, 0);
     if (proc->events) {
-        queue_process_events(proc->events);
+        multiqueue_process_events(proc->events);
     }
 
     // Stream can be closed if it is empty.
@@ -402,7 +403,7 @@ static void on_process_exit(Process *proc)
   // OS. We are still in the libuv loop, so we cannot call code that polls for
   // more data directly. Instead delay the reading after the libuv loop by
   // queueing process_close_handles() as an event.
-  Queue *queue = proc->events ? proc->events : loop->events;
+  MultiQueue *queue = proc->events ? proc->events : loop->events;
   CREATE_EVENT(queue, process_close_handles, 1, proc);
 }
 
diff --git a/src/nvim/event/process.h b/src/nvim/event/process.h
index a4c6e7eeb2..5cbf7f9ce7 100644
--- a/src/nvim/event/process.h
+++ b/src/nvim/event/process.h
@@ -27,7 +27,7 @@ struct process {
   process_exit_cb cb;
   internal_process_cb internal_exit_cb, internal_close_cb;
   bool closed, term_sent, detach;
-  Queue *events;
+  MultiQueue *events;
 };
 
 static inline Process process_init(Loop *loop, ProcessType type, void *data)
diff --git a/src/nvim/event/queue.c b/src/nvim/event/queue.c
deleted file mode 100644
index ee224b0a25..0000000000
--- a/src/nvim/event/queue.c
+++ /dev/null
@@ -1,208 +0,0 @@
-// Queue for selective async event processing. Instances of this queue support a
-// parent/child relationship with the following properties:
-//
-// - pushing a node to a child queue will push a corresponding link node to the
-//   parent queue
-// - removing a link node from a parent queue will remove the next node
-//   in the linked child queue
-// - removing a node from a child queue will remove the corresponding link node
-//   in the parent queue
-//
-// These properties allow Nvim to organize and process events from different
-// sources with a certain degree of control. Here's how the queue is used:
-//
-//                         +----------------+
-//                         |   Main loop    |
-//                         +----------------+
-//                                  ^
-//                                  |
-//                         +----------------+
-//         +-------------->|   Event loop   |<------------+
-//         |               +--+-------------+             |
-//         |                  ^           ^               |
-//         |                  |           |               |
-//    +-----------+   +-----------+    +---------+    +---------+
-//    | Channel 1 |   | Channel 2 |    |  Job 1  |    |  Job 2  |
-//    +-----------+   +-----------+    +---------+    +---------+
-//
-//
-// The lower boxes represents event emitters, each with its own private queue
-// having the event loop queue as the parent.
-//
-// When idle, the main loop spins the event loop which queues events from many
-// sources (channels, jobs, user...). Each event emitter pushes events to its
-// private queue which is propagated to the event loop queue. When the main loop
-// consumes an event, the corresponding event is removed from the emitter's
-// queue.
-//
-// The main reason for this queue hierarchy is to allow focusing on a single
-// event emitter while blocking the main loop. For example, if the `jobwait`
-// VimL function is called on job1, the main loop will temporarily stop polling
-// the event loop queue and poll job1 queue instead. Same with channels, when
-// calling `rpcrequest` we want to temporarily stop processing events from
-// other sources and focus on a specific channel.
-
-#include <assert.h>
-#include <stdarg.h>
-#include <stdbool.h>
-#include <stdint.h>
-
-
-#include <uv.h>
-
-#include "nvim/event/queue.h"
-#include "nvim/memory.h"
-#include "nvim/os/time.h"
-
-typedef struct queue_item QueueItem;
-struct queue_item {
-  union {
-    Queue *queue;
-    struct {
-      Event event;
-      QueueItem *parent;
-    } item;
-  } data;
-  bool link;  // this is just a link to a node in a child queue
-  QUEUE node;
-};
-
-struct queue {
-  Queue *parent;
-  QUEUE headtail;
-  put_callback put_cb;
-  void *data;
-};
-
-#ifdef INCLUDE_GENERATED_DECLARATIONS
-# include "event/queue.c.generated.h"
-#endif
-
-static Event NILEVENT = {.handler = NULL, .argv = {NULL}};
-
-Queue *queue_new_parent(put_callback put_cb, void *data)
-{
-  return queue_new(NULL, put_cb, data);
-}
-
-Queue *queue_new_child(Queue *parent)
-  FUNC_ATTR_NONNULL_ALL
-{
-  assert(!parent->parent);
-  return queue_new(parent, NULL, NULL);
-}
-
-static Queue *queue_new(Queue *parent, put_callback put_cb, void *data)
-{
-  Queue *rv = xmalloc(sizeof(Queue));
-  QUEUE_INIT(&rv->headtail);
-  rv->parent = parent;
-  rv->put_cb = put_cb;
-  rv->data = data;
-  return rv;
-}
-
-void queue_free(Queue *queue)
-{
-  assert(queue);
-  while (!QUEUE_EMPTY(&queue->headtail)) {
-    QUEUE *q = QUEUE_HEAD(&queue->headtail);
-    QueueItem *item = queue_node_data(q);
-    if (queue->parent) {
-      QUEUE_REMOVE(&item->data.item.parent->node);
-      xfree(item->data.item.parent);
-    }
-    QUEUE_REMOVE(q);
-    xfree(item);
-  }
-
-  xfree(queue);
-}
-
-Event queue_get(Queue *queue)
-{
-  return queue_empty(queue) ? NILEVENT : queue_remove(queue);
-}
-
-void queue_put_event(Queue *queue, Event event)
-{
-  assert(queue);
-  queue_push(queue, event);
-  if (queue->parent && queue->parent->put_cb) {
-    queue->parent->put_cb(queue->parent, queue->parent->data);
-  }
-}
-
-void queue_process_events(Queue *queue)
-{
-  assert(queue);
-  while (!queue_empty(queue)) {
-    Event event = queue_get(queue);
-    if (event.handler) {
-      event.handler(event.argv);
-    }
-  }
-}
-
-bool queue_empty(Queue *queue)
-{
-  assert(queue);
-  return QUEUE_EMPTY(&queue->headtail);
-}
-
-void queue_replace_parent(Queue *queue, Queue *new_parent)
-{
-  assert(queue_empty(queue));
-  queue->parent = new_parent;
-}
-
-static Event queue_remove(Queue *queue)
-{
-  assert(!queue_empty(queue));
-  QUEUE *h = QUEUE_HEAD(&queue->headtail);
-  QUEUE_REMOVE(h);
-  QueueItem *item = queue_node_data(h);
-  Event rv;
-
-  if (item->link) {
-    assert(!queue->parent);
-    // remove the next node in the linked queue
-    Queue *linked = item->data.queue;
-    assert(!queue_empty(linked));
-    QueueItem *child =
-      queue_node_data(QUEUE_HEAD(&linked->headtail));
-    QUEUE_REMOVE(&child->node);
-    rv = child->data.item.event;
-    xfree(child);
-  } else {
-    if (queue->parent) {
-      // remove the corresponding link node in the parent queue
-      QUEUE_REMOVE(&item->data.item.parent->node);
-      xfree(item->data.item.parent);
-    }
-    rv = item->data.item.event;
-  }
-
-  xfree(item);
-  return rv;
-}
-
-static void queue_push(Queue *queue, Event event)
-{
-  QueueItem *item = xmalloc(sizeof(QueueItem));
-  item->link = false;
-  item->data.item.event = event;
-  QUEUE_INSERT_TAIL(&queue->headtail, &item->node);
-  if (queue->parent) {
-    // push link node to the parent queue
-    item->data.item.parent = xmalloc(sizeof(QueueItem));
-    item->data.item.parent->link = true;
-    item->data.item.parent->data.queue = queue;
-    QUEUE_INSERT_TAIL(&queue->parent->headtail, &item->data.item.parent->node);
-  }
-}
-
-static QueueItem *queue_node_data(QUEUE *q)
-{
-  return QUEUE_DATA(q, QueueItem, node);
-}
diff --git a/src/nvim/event/queue.h b/src/nvim/event/queue.h
deleted file mode 100644
index 85fc59f8b2..0000000000
--- a/src/nvim/event/queue.h
+++ /dev/null
@@ -1,19 +0,0 @@
-#ifndef NVIM_EVENT_QUEUE_H
-#define NVIM_EVENT_QUEUE_H
-
-#include <uv.h>
-
-#include "nvim/event/defs.h"
-#include "nvim/lib/queue.h"
-
-typedef struct queue Queue;
-typedef void (*put_callback)(Queue *queue, void *data);
-
-#define queue_put(q, h, ...) \
-  queue_put_event(q, event_create(1, h, __VA_ARGS__));
-
-
-#ifdef INCLUDE_GENERATED_DECLARATIONS
-# include "event/queue.h.generated.h"
-#endif
-#endif  // NVIM_EVENT_QUEUE_H
diff --git a/src/nvim/event/signal.h b/src/nvim/event/signal.h
index e32608acc0..7fe352edef 100644
--- a/src/nvim/event/signal.h
+++ b/src/nvim/event/signal.h
@@ -14,7 +14,7 @@ struct signal_watcher {
   void *data;
   signal_cb cb;
   signal_close_cb close_cb;
-  Queue *events;
+  MultiQueue *events;
 };
 
 #ifdef INCLUDE_GENERATED_DECLARATIONS
diff --git a/src/nvim/event/socket.h b/src/nvim/event/socket.h
index ad59fdbe3a..eb0823c76d 100644
--- a/src/nvim/event/socket.h
+++ b/src/nvim/event/socket.h
@@ -30,7 +30,7 @@ struct socket_watcher {
   void *data;
   socket_cb cb;
   socket_close_cb close_cb;
-  Queue *events;
+  MultiQueue *events;
 };
 
 #ifdef INCLUDE_GENERATED_DECLARATIONS
diff --git a/src/nvim/event/stream.h b/src/nvim/event/stream.h
index a176fac1c0..d27497e4a4 100644
--- a/src/nvim/event/stream.h
+++ b/src/nvim/event/stream.h
@@ -53,7 +53,7 @@ struct stream {
   size_t pending_reqs;
   size_t num_bytes;
   bool closed;
-  Queue *events;
+  MultiQueue *events;
 };
 
 #ifdef INCLUDE_GENERATED_DECLARATIONS
diff --git a/src/nvim/event/time.c b/src/nvim/event/time.c
index f68a66345f..77260546db 100644
--- a/src/nvim/event/time.c
+++ b/src/nvim/event/time.c
@@ -51,7 +51,7 @@ static void time_watcher_cb(uv_timer_t *handle)
   FUNC_ATTR_NONNULL_ALL
 {
   TimeWatcher *watcher = handle->data;
-  if (watcher->blockable && !queue_empty(watcher->events)) {
+  if (watcher->blockable && !multiqueue_empty(watcher->events)) {
     // the timer blocked and there already is an unprocessed event waiting
     return;
   }
diff --git a/src/nvim/event/time.h b/src/nvim/event/time.h
index 14df176ea3..a6de89ad6e 100644
--- a/src/nvim/event/time.h
+++ b/src/nvim/event/time.h
@@ -12,7 +12,7 @@ struct time_watcher {
   uv_timer_t uv;
   void *data;
   time_cb cb, close_cb;
-  Queue *events;
+  MultiQueue *events;
   bool blockable;
 };
 
diff --git a/src/nvim/ex_getln.c b/src/nvim/ex_getln.c
index 4c997844ea..4254697241 100644
--- a/src/nvim/ex_getln.c
+++ b/src/nvim/ex_getln.c
@@ -357,7 +357,7 @@ static int command_line_execute(VimState *state, int key)
   s->c = key;
 
   if (s->c == K_EVENT) {
-    queue_process_events(main_loop.events);
+    multiqueue_process_events(main_loop.events);
     redrawcmdline();
     return 1;
   }
diff --git a/src/nvim/main.c b/src/nvim/main.c
index 86bef782f7..005f4dcc77 100644
--- a/src/nvim/main.c
+++ b/src/nvim/main.c
@@ -156,7 +156,7 @@ void event_teardown(void)
     return;
   }
 
-  queue_process_events(main_loop.events);
+  multiqueue_process_events(main_loop.events);
   input_stop();
   channel_teardown();
   process_teardown(&main_loop);
diff --git a/src/nvim/msgpack_rpc/channel.c b/src/nvim/msgpack_rpc/channel.c
index cd12f258b6..fef1d08db7 100644
--- a/src/nvim/msgpack_rpc/channel.c
+++ b/src/nvim/msgpack_rpc/channel.c
@@ -65,7 +65,7 @@ typedef struct {
   uint64_t next_request_id;
   kvec_t(ChannelCallFrame *) call_stack;
   kvec_t(WBuffer *) delayed_notifications;
-  Queue *events;
+  MultiQueue *events;
 } Channel;
 
 typedef struct {
@@ -440,7 +440,7 @@ static void handle_request(Channel *channel, msgpack_object *request)
   if (handler.async) {
     on_request_event((void **)&event_data);
   } else {
-    queue_put(channel->events, on_request_event, 1, event_data);
+    multiqueue_put(channel->events, on_request_event, 1, event_data);
   }
 }
 
@@ -620,7 +620,7 @@ static void close_channel(Channel *channel)
     case kChannelTypeStdio:
       stream_close(&channel->data.std.in, NULL, NULL);
       stream_close(&channel->data.std.out, NULL, NULL);
-      queue_put(main_loop.fast_events, exit_event, 1, channel);
+      multiqueue_put(main_loop.fast_events, exit_event, 1, channel);
       return;
     default:
       abort();
@@ -654,7 +654,7 @@ static void free_channel(Channel *channel)
   kv_destroy(channel->call_stack);
   kv_destroy(channel->delayed_notifications);
   if (channel->type != kChannelTypeProc) {
-    queue_free(channel->events);
+    multiqueue_free(channel->events);
   }
   xfree(channel);
 }
@@ -664,10 +664,11 @@ static void close_cb(Stream *stream, void *data)
   decref(data);
 }
 
-static Channel *register_channel(ChannelType type, uint64_t id, Queue *events)
+static Channel *register_channel(ChannelType type, uint64_t id,
+                                 MultiQueue *events)
 {
   Channel *rv = xmalloc(sizeof(Channel));
-  rv->events = events ? events : queue_new_child(main_loop.events);
+  rv->events = events ? events : multiqueue_new_child(main_loop.events);
   rv->type = type;
   rv->refcount = 1;
   rv->closed = false;
diff --git a/src/nvim/normal.c b/src/nvim/normal.c
index c7c112200a..6dcbf50750 100644
--- a/src/nvim/normal.c
+++ b/src/nvim/normal.c
@@ -7880,15 +7880,15 @@ static void nv_event(cmdarg_T *cap)
 {
   // Garbage collection should have been executed before blocking for events in
   // the `os_inchar` in `state_enter`, but we also disable it here in case the
-  // `os_inchar` branch was not executed(!queue_empty(loop.events), which could
-  // have `may_garbage_collect` set to true in `normal_check`).
+  // `os_inchar` branch was not executed (!multiqueue_empty(loop.events), which
+  // could have `may_garbage_collect` set to true in `normal_check`).
   //
   // That is because here we may run code that calls `os_inchar`
   // later(`f_confirm` or `get_keystroke` for example), but in these cases it is
   // not safe to perform garbage collection because there could be unreferenced
   // lists or dicts being used.
   may_garbage_collect = false;
-  queue_process_events(main_loop.events);
+  multiqueue_process_events(main_loop.events);
   cap->retval |= CA_COMMAND_BUSY;       // don't call edit() now
 }
 
diff --git a/src/nvim/os/input.c b/src/nvim/os/input.c
index 6873e32f34..1307ab5e5a 100644
--- a/src/nvim/os/input.c
+++ b/src/nvim/os/input.c
@@ -87,8 +87,8 @@ static void create_cursorhold_event(void)
   // have been called(inbuf_poll would return kInputAvail)
   // TODO(tarruda): Cursorhold should be implemented as a timer set during the
   // `state_check` callback for the states where it can be triggered.
-  assert(!events_enabled || queue_empty(main_loop.events));
-  queue_put(main_loop.events, cursorhold_event, 0);
+  assert(!events_enabled || multiqueue_empty(main_loop.events));
+  multiqueue_put(main_loop.events, cursorhold_event, 0);
 }
 
 // Low level input function
@@ -422,5 +422,5 @@ static void read_error_exit(void)
 
 static bool pending_events(void)
 {
-  return events_enabled && !queue_empty(main_loop.events);
+  return events_enabled && !multiqueue_empty(main_loop.events);
 }
diff --git a/src/nvim/os/shell.c b/src/nvim/os/shell.c
index 617a505367..e9a3dcbff8 100644
--- a/src/nvim/os/shell.c
+++ b/src/nvim/os/shell.c
@@ -207,7 +207,7 @@ static int do_os_system(char **argv,
   Stream in, out, err;
   LibuvProcess uvproc = libuv_process_init(&main_loop, &buf);
   Process *proc = &uvproc.process;
-  Queue *events = queue_new_child(main_loop.events);
+  MultiQueue *events = multiqueue_new_child(main_loop.events);
   proc->events = events;
   proc->argv = argv;
   proc->in = input != NULL ? &in : NULL;
@@ -221,7 +221,7 @@ static int do_os_system(char **argv,
       msg_outtrans((char_u *)prog);
       msg_putchar('\n');
     }
-    queue_free(events);
+    multiqueue_free(events);
     return -1;
   }
 
@@ -277,8 +277,8 @@ static int do_os_system(char **argv,
     }
   }
 
-  assert(queue_empty(events));
-  queue_free(events);
+  assert(multiqueue_empty(events));
+  multiqueue_free(events);
 
   return status;
 }
diff --git a/src/nvim/state.c b/src/nvim/state.c
index f792ec00a4..44c6441e40 100644
--- a/src/nvim/state.c
+++ b/src/nvim/state.c
@@ -35,8 +35,8 @@ getkey:
       // processing. Characters can come from mappings, scripts and other
       // sources, so this scenario is very common.
       key = safe_vgetc();
-    } else if (!queue_empty(main_loop.events)) {
-      // Event was made available after the last queue_process_events call
+    } else if (!multiqueue_empty(main_loop.events)) {
+      // Event was made available after the last multiqueue_process_events call
       key = K_EVENT;
     } else {
       input_enable_events();
@@ -48,7 +48,7 @@ getkey:
       // directly.
       (void)os_inchar(NULL, 0, -1, 0);
       input_disable_events();
-      key = !queue_empty(main_loop.events) ? K_EVENT : safe_vgetc();
+      key = !multiqueue_empty(main_loop.events) ? K_EVENT : safe_vgetc();
     }
 
     if (key == K_EVENT) {
diff --git a/src/nvim/terminal.c b/src/nvim/terminal.c
index ff98dc9f22..8401343d7a 100644
--- a/src/nvim/terminal.c
+++ b/src/nvim/terminal.c
@@ -166,7 +166,7 @@ void terminal_init(void)
   invalidated_terminals = pmap_new(ptr_t)();
   time_watcher_init(&main_loop, &refresh_timer, NULL);
   // refresh_timer_cb will redraw the screen which can call vimscript
-  refresh_timer.events = queue_new_child(main_loop.events);
+  refresh_timer.events = multiqueue_new_child(main_loop.events);
 
   // initialize a rgb->color index map for cterm attributes(VTermScreenCell
   // only has RGB information and we need color indexes for terminal UIs)
@@ -201,7 +201,7 @@ void terminal_init(void)
 void terminal_teardown(void)
 {
   time_watcher_stop(&refresh_timer);
-  queue_free(refresh_timer.events);
+  multiqueue_free(refresh_timer.events);
   time_watcher_close(&refresh_timer, NULL);
   pmap_free(ptr_t)(invalidated_terminals);
   map_free(int, int)(color_indexes);
@@ -445,7 +445,7 @@ static int terminal_execute(VimState *state, int key)
     case K_EVENT:
       // We cannot let an event free the terminal yet. It is still needed.
       s->term->refcount++;
-      queue_process_events(main_loop.events);
+      multiqueue_process_events(main_loop.events);
       s->term->refcount--;
       if (s->term->buf_handle == 0) {
         s->close = true;
diff --git a/src/nvim/tui/input.c b/src/nvim/tui/input.c
index 68ea00ee63..740716f0ef 100644
--- a/src/nvim/tui/input.c
+++ b/src/nvim/tui/input.c
@@ -340,7 +340,7 @@ static void read_cb(Stream *stream, RBuffer *buf, size_t c, void *data,
       // ls *.md | xargs nvim
       input->in_fd = 2;
       stream_close(&input->read_stream, NULL, NULL);
-      queue_put(input->loop->fast_events, restart_reading, 1, input);
+      multiqueue_put(input->loop->fast_events, restart_reading, 1, input);
     } else {
       loop_schedule(&main_loop, event_create(1, input_done_event, 0));
     }
diff --git a/src/nvim/tui/tui.c b/src/nvim/tui/tui.c
index 6f90742928..de0f121b46 100644
--- a/src/nvim/tui/tui.c
+++ b/src/nvim/tui/tui.c
@@ -628,8 +628,8 @@ static void tui_suspend(UI *ui)
   // kill(0, SIGTSTP) won't stop the UI thread, so we must poll for SIGCONT
   // before continuing. This is done in another callback to avoid
   // loop_poll_events recursion
-  queue_put_event(data->loop->fast_events,
-      event_create(1, suspend_event, 1, ui));
+  multiqueue_put_event(data->loop->fast_events,
+                       event_create(1, suspend_event, 1, ui));
 }
 
 static void tui_set_title(UI *ui, char *title)
-- 
cgit