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).
This commit is contained in:
Justin M. Keyes
2016-09-29 17:09:37 +02:00
parent c8b6ec2e6a
commit 6186df3562
30 changed files with 191 additions and 188 deletions

View File

@@ -1,3 +1,3 @@
# libuv queue.h pointer arithmetic is not accepted by asan # multiqueue.h pointer arithmetic is not accepted by asan
fun:queue_node_data fun:multiqueue_node_data
fun:dictwatcher_node_data fun:dictwatcher_node_data

View File

@@ -962,7 +962,7 @@ static int insert_handle_key(InsertState *s)
break; break;
case K_EVENT: // some event case K_EVENT: // some event
queue_process_events(main_loop.events); multiqueue_process_events(main_loop.events);
break; break;
case K_FOCUSGAINED: // Neovim has been given focus case K_FOCUSGAINED: // Neovim has been given focus

View File

@@ -415,7 +415,7 @@ typedef struct {
dict_T *self; dict_T *self;
int *status_ptr; int *status_ptr;
uint64_t id; uint64_t id;
Queue *events; MultiQueue *events;
} TerminalJobData; } TerminalJobData;
typedef struct dict_watcher { 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(); list_T *rv = list_alloc();
ui_busy_start(); 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 // 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 // is used to represent an invalid job id, -2 is for a interrupted job and
// -1 for jobs that were skipped or timed out. // -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; data->status_ptr = &rv->lv_last->li_tv.vval.v_number;
// Process any pending events for the job because we'll temporarily // Process any pending events for the job because we'll temporarily
// replace the parent queue // replace the parent queue
queue_process_events(data->events); multiqueue_process_events(data->events);
queue_replace_parent(data->events, waiting_jobs); multiqueue_replace_parent(data->events, waiting_jobs);
} }
} }
@@ -11769,11 +11769,11 @@ static void f_jobwait(typval_T *argvars, typval_T *rettv, FunPtr fptr)
continue; continue;
} }
// restore the parent queue for the job // restore the parent queue for the job
queue_process_events(data->events); multiqueue_process_events(data->events);
queue_replace_parent(data->events, main_loop.events); multiqueue_replace_parent(data->events, main_loop.events);
} }
queue_free(waiting_jobs); multiqueue_free(waiting_jobs);
ui_busy_stop(); ui_busy_stop();
rv->lv_refcount++; rv->lv_refcount++;
rettv->v_type = VAR_LIST; 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; timer->callback = func;
time_watcher_init(&main_loop, &timer->tw, timer); 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 // if main loop is blocked, don't queue up multiple events
timer->tw.blockable = true; timer->tw.blockable = true;
time_watcher_start(&timer->tw, timer_due_cb, timeout, 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) static void timer_close_cb(TimeWatcher *tw, void *data)
{ {
timer_T *timer = (timer_T *)data; timer_T *timer = (timer_T *)data;
queue_free(timer->tw.events); multiqueue_free(timer->tw.events);
user_func_unref(timer->callback); user_func_unref(timer->callback);
pmap_del(uint64_t)(timers, timer->timer_id); pmap_del(uint64_t)(timers, timer->timer_id);
timer_decref(timer); timer_decref(timer);
@@ -21725,7 +21725,7 @@ static inline TerminalJobData *common_job_init(char **argv,
data->on_stderr = on_stderr; data->on_stderr = on_stderr;
data->on_exit = on_exit; data->on_exit = on_exit;
data->self = self; data->self = self;
data->events = queue_new_child(main_loop.events); data->events = multiqueue_new_child(main_loop.events);
data->rpc = rpc; data->rpc = rpc;
if (pty) { if (pty) {
data->proc.pty = pty_process_init(&main_loop, data); 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) { if (data->self) {
dict_unref(data->self); dict_unref(data->self);
} }
queue_free(data->events); multiqueue_free(data->events);
pmap_del(uint64_t)(jobs, data->id); pmap_del(uint64_t)(jobs, data->id);
xfree(data); 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 // data->queue may still be used after this function returns(process_wait), so
// only free in the next event loop iteration // 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 // 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]; TerminalJobData *j = argv[0];
if (j->in.pending_reqs || j->out.pending_reqs || j->err.pending_reqs) { 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; return;
} }
@@ -21977,7 +21977,7 @@ static void term_close(void *d)
data->exited = true; data->exited = true;
process_stop((Process *)&data->proc); 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) static void term_job_data_decref(TerminalJobData *data)

View File

@@ -18,9 +18,9 @@ void loop_init(Loop *loop, void *data)
loop->uv.data = loop; loop->uv.data = loop;
loop->children = kl_init(WatcherPtr); loop->children = kl_init(WatcherPtr);
loop->children_stop_requests = 0; loop->children_stop_requests = 0;
loop->events = queue_new_parent(loop_on_put, loop); loop->events = multiqueue_new_parent(loop_on_put, loop);
loop->fast_events = queue_new_child(loop->events); loop->fast_events = multiqueue_new_child(loop->events);
loop->thread_events = queue_new_parent(NULL, NULL); loop->thread_events = multiqueue_new_parent(NULL, NULL);
uv_mutex_init(&loop->mutex); uv_mutex_init(&loop->mutex);
uv_async_init(&loop->uv, &loop->async, async_cb); uv_async_init(&loop->uv, &loop->async, async_cb);
uv_signal_init(&loop->uv, &loop->children_watcher); 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 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 // Schedule an event from another thread
void loop_schedule(Loop *loop, Event event) void loop_schedule(Loop *loop, Event event)
{ {
uv_mutex_lock(&loop->mutex); 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_async_send(&loop->async);
uv_mutex_unlock(&loop->mutex); uv_mutex_unlock(&loop->mutex);
} }
void loop_on_put(Queue *queue, void *data) void loop_on_put(MultiQueue *queue, void *data)
{ {
Loop *loop = data; Loop *loop = data;
// Sometimes libuv will run pending callbacks(timer for example) before // Sometimes libuv will run pending callbacks(timer for example) before
@@ -86,9 +86,9 @@ void loop_close(Loop *loop, bool wait)
do { do {
uv_run(&loop->uv, wait ? UV_RUN_DEFAULT : UV_RUN_NOWAIT); uv_run(&loop->uv, wait ? UV_RUN_DEFAULT : UV_RUN_NOWAIT);
} while (uv_loop_close(&loop->uv) && wait); } while (uv_loop_close(&loop->uv) && wait);
queue_free(loop->fast_events); multiqueue_free(loop->fast_events);
queue_free(loop->thread_events); multiqueue_free(loop->thread_events);
queue_free(loop->events); multiqueue_free(loop->events);
kl_destroy(WatcherPtr, loop->children); kl_destroy(WatcherPtr, loop->children);
} }
@@ -96,9 +96,9 @@ static void async_cb(uv_async_t *handle)
{ {
Loop *l = handle->loop->data; Loop *l = handle->loop->data;
uv_mutex_lock(&l->mutex); uv_mutex_lock(&l->mutex);
while (!queue_empty(l->thread_events)) { while (!multiqueue_empty(l->thread_events)) {
Event ev = queue_get(l->thread_events); Event ev = multiqueue_get(l->thread_events);
queue_put_event(l->fast_events, ev); multiqueue_put_event(l->fast_events, ev);
} }
uv_mutex_unlock(&l->mutex); uv_mutex_unlock(&l->mutex);
} }

View File

@@ -7,7 +7,7 @@
#include "nvim/lib/klist.h" #include "nvim/lib/klist.h"
#include "nvim/os/time.h" #include "nvim/os/time.h"
#include "nvim/event/queue.h" #include "nvim/event/multiqueue.h"
typedef void * WatcherPtr; typedef void * WatcherPtr;
@@ -16,7 +16,7 @@ KLIST_INIT(WatcherPtr, WatcherPtr, _noop)
typedef struct loop { typedef struct loop {
uv_loop_t uv; uv_loop_t uv;
Queue *events, *fast_events, *thread_events; MultiQueue *events, *fast_events, *thread_events;
klist_t(WatcherPtr) *children; klist_t(WatcherPtr) *children;
uv_signal_t children_watcher; uv_signal_t children_watcher;
uv_timer_t children_kill_timer, poll_timer; uv_timer_t children_kill_timer, poll_timer;
@@ -26,10 +26,10 @@ typedef struct loop {
int recursive; int recursive;
} Loop; } Loop;
#define CREATE_EVENT(queue, handler, argc, ...) \ #define CREATE_EVENT(multiqueue, handler, argc, ...) \
do { \ do { \
if (queue) { \ if (multiqueue) { \
queue_put((queue), (handler), argc, __VA_ARGS__); \ multiqueue_put((multiqueue), (handler), argc, __VA_ARGS__); \
} else { \ } else { \
void *argv[argc] = { __VA_ARGS__ }; \ void *argv[argc] = { __VA_ARGS__ }; \
(handler)(argv); \ (handler)(argv); \
@@ -37,12 +37,12 @@ typedef struct loop {
} while (0) } while (0)
// Poll for events until a condition or timeout // 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 { \ do { \
int remaining = timeout; \ int remaining = timeout; \
uint64_t before = (remaining > 0) ? os_hrtime() : 0; \ uint64_t before = (remaining > 0) ? os_hrtime() : 0; \
while (!(condition)) { \ while (!(condition)) { \
LOOP_PROCESS_EVENTS(loop, queue, remaining); \ LOOP_PROCESS_EVENTS(loop, multiqueue, remaining); \
if (remaining == 0) { \ if (remaining == 0) { \
break; \ break; \
} else if (remaining > 0) { \ } else if (remaining > 0) { \
@@ -56,10 +56,10 @@ typedef struct loop {
} \ } \
} while (0) } while (0)
#define LOOP_PROCESS_EVENTS(loop, queue, timeout) \ #define LOOP_PROCESS_EVENTS(loop, multiqueue, timeout) \
do { \ do { \
if (queue && !queue_empty(queue)) { \ if (multiqueue && !multiqueue_empty(multiqueue)) { \
queue_process_events(queue); \ multiqueue_process_events(multiqueue); \
} else { \ } else { \
loop_poll_events(loop, timeout); \ loop_poll_events(loop, timeout); \
} \ } \

View File

@@ -1,5 +1,5 @@
// Queue for selective async event processing. Instances of this queue support a // Multi-level queue for selective async event processing. Multiqueue supports
// parent/child relationship with the following properties: // a parent-child relationship with the following properties:
// //
// - pushing a node to a child queue will push a corresponding link node to the // - pushing a node to a child queue will push a corresponding link node to the
// parent queue // parent queue
@@ -9,7 +9,7 @@
// in the parent queue // in the parent queue
// //
// These properties allow Nvim to organize and process events from different // 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: // sources with a certain degree of control. How the multiqueue is used:
// //
// +----------------+ // +----------------+
// | Main loop | // | Main loop |
@@ -26,7 +26,7 @@
// +-----------+ +-----------+ +---------+ +---------+ // +-----------+ +-----------+ +---------+ +---------+
// //
// //
// The lower boxes represents event emitters, each with its own private queue // The lower boxes represent event emitters, each with its own private queue
// having the event loop queue as the parent. // having the event loop queue as the parent.
// //
// When idle, the main loop spins the event loop which queues events from many // When idle, the main loop spins the event loop which queues events from many
@@ -50,51 +50,52 @@
#include <uv.h> #include <uv.h>
#include "nvim/event/queue.h" #include "nvim/event/multiqueue.h"
#include "nvim/memory.h" #include "nvim/memory.h"
#include "nvim/os/time.h" #include "nvim/os/time.h"
typedef struct queue_item QueueItem; typedef struct multiqueue_item MultiQueueItem;
struct queue_item { struct multiqueue_item {
union { union {
Queue *queue; MultiQueue *queue;
struct { struct {
Event event; Event event;
QueueItem *parent; MultiQueueItem *parent;
} item; } item;
} data; } data;
bool link; // this is just a link to a node in a child queue bool link; // true: current item is just a link to a node in a child queue
QUEUE node; QUEUE node;
}; };
struct queue { struct multiqueue {
Queue *parent; MultiQueue *parent;
QUEUE headtail; QUEUE headtail;
put_callback put_cb; put_callback put_cb;
void *data; void *data;
}; };
#ifdef INCLUDE_GENERATED_DECLARATIONS #ifdef INCLUDE_GENERATED_DECLARATIONS
# include "event/queue.c.generated.h" # include "event/multiqueue.c.generated.h"
#endif #endif
static Event NILEVENT = {.handler = NULL, .argv = {NULL}}; static Event NILEVENT = { .handler = NULL, .argv = {NULL} };
Queue *queue_new_parent(put_callback put_cb, void *data) MultiQueue *multiqueue_new_parent(put_callback put_cb, void *data)
{ {
return queue_new(NULL, put_cb, data); return multiqueue_new(NULL, put_cb, data);
} }
Queue *queue_new_child(Queue *parent) MultiQueue *multiqueue_new_child(MultiQueue *parent)
FUNC_ATTR_NONNULL_ALL FUNC_ATTR_NONNULL_ALL
{ {
assert(!parent->parent); assert(!parent->parent);
return queue_new(parent, NULL, NULL); return multiqueue_new(parent, NULL, NULL);
} }
static Queue *queue_new(Queue *parent, put_callback put_cb, void *data) static MultiQueue *multiqueue_new(MultiQueue *parent, put_callback put_cb,
void *data)
{ {
Queue *rv = xmalloc(sizeof(Queue)); MultiQueue *rv = xmalloc(sizeof(MultiQueue));
QUEUE_INIT(&rv->headtail); QUEUE_INIT(&rv->headtail);
rv->parent = parent; rv->parent = parent;
rv->put_cb = put_cb; rv->put_cb = put_cb;
@@ -102,13 +103,13 @@ static Queue *queue_new(Queue *parent, put_callback put_cb, void *data)
return rv; return rv;
} }
void queue_free(Queue *queue) void multiqueue_free(MultiQueue *this)
{ {
assert(queue); assert(this);
while (!QUEUE_EMPTY(&queue->headtail)) { while (!QUEUE_EMPTY(&this->headtail)) {
QUEUE *q = QUEUE_HEAD(&queue->headtail); QUEUE *q = QUEUE_HEAD(&this->headtail);
QueueItem *item = queue_node_data(q); MultiQueueItem *item = multiqueue_node_data(q);
if (queue->parent) { if (this->parent) {
QUEUE_REMOVE(&item->data.item.parent->node); QUEUE_REMOVE(&item->data.item.parent->node);
xfree(item->data.item.parent); xfree(item->data.item.parent);
} }
@@ -116,66 +117,66 @@ void queue_free(Queue *queue)
xfree(item); xfree(item);
} }
xfree(queue); xfree(this);
} }
Event queue_get(Queue *queue) Event multiqueue_get(MultiQueue *this)
{ {
return queue_empty(queue) ? NILEVENT : queue_remove(queue); return multiqueue_empty(this) ? NILEVENT : multiqueue_remove(this);
} }
void queue_put_event(Queue *queue, Event event) void multiqueue_put_event(MultiQueue *this, Event event)
{ {
assert(queue); assert(this);
queue_push(queue, event); multiqueue_push(this, event);
if (queue->parent && queue->parent->put_cb) { if (this->parent && this->parent->put_cb) {
queue->parent->put_cb(queue->parent, queue->parent->data); this->parent->put_cb(this->parent, this->parent->data);
} }
} }
void queue_process_events(Queue *queue) void multiqueue_process_events(MultiQueue *this)
{ {
assert(queue); assert(this);
while (!queue_empty(queue)) { while (!multiqueue_empty(this)) {
Event event = queue_get(queue); Event event = multiqueue_get(this);
if (event.handler) { if (event.handler) {
event.handler(event.argv); event.handler(event.argv);
} }
} }
} }
bool queue_empty(Queue *queue) bool multiqueue_empty(MultiQueue *this)
{ {
assert(queue); assert(this);
return QUEUE_EMPTY(&queue->headtail); return QUEUE_EMPTY(&this->headtail);
} }
void queue_replace_parent(Queue *queue, Queue *new_parent) void multiqueue_replace_parent(MultiQueue *this, MultiQueue *new_parent)
{ {
assert(queue_empty(queue)); assert(multiqueue_empty(this));
queue->parent = new_parent; this->parent = new_parent;
} }
static Event queue_remove(Queue *queue) static Event multiqueue_remove(MultiQueue *this)
{ {
assert(!queue_empty(queue)); assert(!multiqueue_empty(this));
QUEUE *h = QUEUE_HEAD(&queue->headtail); QUEUE *h = QUEUE_HEAD(&this->headtail);
QUEUE_REMOVE(h); QUEUE_REMOVE(h);
QueueItem *item = queue_node_data(h); MultiQueueItem *item = multiqueue_node_data(h);
Event rv; Event rv;
if (item->link) { if (item->link) {
assert(!queue->parent); assert(!this->parent);
// remove the next node in the linked queue // remove the next node in the linked queue
Queue *linked = item->data.queue; MultiQueue *linked = item->data.queue;
assert(!queue_empty(linked)); assert(!multiqueue_empty(linked));
QueueItem *child = MultiQueueItem *child =
queue_node_data(QUEUE_HEAD(&linked->headtail)); multiqueue_node_data(QUEUE_HEAD(&linked->headtail));
QUEUE_REMOVE(&child->node); QUEUE_REMOVE(&child->node);
rv = child->data.item.event; rv = child->data.item.event;
xfree(child); xfree(child);
} else { } else {
if (queue->parent) { if (this->parent) {
// remove the corresponding link node in the parent queue // remove the corresponding link node in the parent queue
QUEUE_REMOVE(&item->data.item.parent->node); QUEUE_REMOVE(&item->data.item.parent->node);
xfree(item->data.item.parent); xfree(item->data.item.parent);
@@ -187,22 +188,22 @@ static Event queue_remove(Queue *queue)
return rv; return rv;
} }
static void queue_push(Queue *queue, Event event) static void multiqueue_push(MultiQueue *this, Event event)
{ {
QueueItem *item = xmalloc(sizeof(QueueItem)); MultiQueueItem *item = xmalloc(sizeof(MultiQueueItem));
item->link = false; item->link = false;
item->data.item.event = event; item->data.item.event = event;
QUEUE_INSERT_TAIL(&queue->headtail, &item->node); QUEUE_INSERT_TAIL(&this->headtail, &item->node);
if (queue->parent) { if (this->parent) {
// push link node to the parent queue // push link node to the parent queue
item->data.item.parent = xmalloc(sizeof(QueueItem)); item->data.item.parent = xmalloc(sizeof(MultiQueueItem));
item->data.item.parent->link = true; item->data.item.parent->link = true;
item->data.item.parent->data.queue = queue; item->data.item.parent->data.queue = this;
QUEUE_INSERT_TAIL(&queue->parent->headtail, &item->data.item.parent->node); QUEUE_INSERT_TAIL(&this->parent->headtail, &item->data.item.parent->node);
} }
} }
static QueueItem *queue_node_data(QUEUE *q) static MultiQueueItem *multiqueue_node_data(QUEUE *q)
{ {
return QUEUE_DATA(q, QueueItem, node); return QUEUE_DATA(q, MultiQueueItem, node);
} }

View File

@@ -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

View File

@@ -126,7 +126,7 @@ void process_teardown(Loop *loop) FUNC_ATTR_NONNULL_ALL
// Wait until all children exit and all close events are processed. // Wait until all children exit and all close events are processed.
LOOP_PROCESS_EVENTS_UNTIL( LOOP_PROCESS_EVENTS_UNTIL(
loop, loop->events, -1, loop, loop->events, -1,
kl_empty(loop->children) && queue_empty(loop->events)); kl_empty(loop->children) && multiqueue_empty(loop->events));
pty_process_teardown(loop); 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 /// indistinguishable from the process returning -1 by itself. Which
/// is possible on some OS. Returns -2 if an user has interruped the /// is possible on some OS. Returns -2 if an user has interruped the
/// wait. /// 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 // The default status is -1, which represents a timeout
int status = -1; int status = -1;
@@ -208,7 +209,7 @@ int process_wait(Process *proc, int ms, Queue *events) FUNC_ATTR_NONNULL_ARG(1)
decref(proc); decref(proc);
if (events) { if (events) {
// the decref call created an exit event, process it now // the decref call created an exit event, process it now
queue_process_events(events); multiqueue_process_events(events);
} }
} else { } else {
proc->refcount--; proc->refcount--;
@@ -362,7 +363,7 @@ static void flush_stream(Process *proc, Stream *stream)
// Poll for data and process the generated events. // Poll for data and process the generated events.
loop_poll_events(proc->loop, 0); loop_poll_events(proc->loop, 0);
if (proc->events) { if (proc->events) {
queue_process_events(proc->events); multiqueue_process_events(proc->events);
} }
// Stream can be closed if it is empty. // 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 // 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 // more data directly. Instead delay the reading after the libuv loop by
// queueing process_close_handles() as an event. // 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); CREATE_EVENT(queue, process_close_handles, 1, proc);
} }

View File

@@ -27,7 +27,7 @@ struct process {
process_exit_cb cb; process_exit_cb cb;
internal_process_cb internal_exit_cb, internal_close_cb; internal_process_cb internal_exit_cb, internal_close_cb;
bool closed, term_sent, detach; bool closed, term_sent, detach;
Queue *events; MultiQueue *events;
}; };
static inline Process process_init(Loop *loop, ProcessType type, void *data) static inline Process process_init(Loop *loop, ProcessType type, void *data)

View File

@@ -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

View File

@@ -14,7 +14,7 @@ struct signal_watcher {
void *data; void *data;
signal_cb cb; signal_cb cb;
signal_close_cb close_cb; signal_close_cb close_cb;
Queue *events; MultiQueue *events;
}; };
#ifdef INCLUDE_GENERATED_DECLARATIONS #ifdef INCLUDE_GENERATED_DECLARATIONS

View File

@@ -30,7 +30,7 @@ struct socket_watcher {
void *data; void *data;
socket_cb cb; socket_cb cb;
socket_close_cb close_cb; socket_close_cb close_cb;
Queue *events; MultiQueue *events;
}; };
#ifdef INCLUDE_GENERATED_DECLARATIONS #ifdef INCLUDE_GENERATED_DECLARATIONS

View File

@@ -53,7 +53,7 @@ struct stream {
size_t pending_reqs; size_t pending_reqs;
size_t num_bytes; size_t num_bytes;
bool closed; bool closed;
Queue *events; MultiQueue *events;
}; };
#ifdef INCLUDE_GENERATED_DECLARATIONS #ifdef INCLUDE_GENERATED_DECLARATIONS

View File

@@ -51,7 +51,7 @@ static void time_watcher_cb(uv_timer_t *handle)
FUNC_ATTR_NONNULL_ALL FUNC_ATTR_NONNULL_ALL
{ {
TimeWatcher *watcher = handle->data; 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 // the timer blocked and there already is an unprocessed event waiting
return; return;
} }

View File

@@ -12,7 +12,7 @@ struct time_watcher {
uv_timer_t uv; uv_timer_t uv;
void *data; void *data;
time_cb cb, close_cb; time_cb cb, close_cb;
Queue *events; MultiQueue *events;
bool blockable; bool blockable;
}; };

View File

@@ -357,7 +357,7 @@ static int command_line_execute(VimState *state, int key)
s->c = key; s->c = key;
if (s->c == K_EVENT) { if (s->c == K_EVENT) {
queue_process_events(main_loop.events); multiqueue_process_events(main_loop.events);
redrawcmdline(); redrawcmdline();
return 1; return 1;
} }

View File

@@ -156,7 +156,7 @@ void event_teardown(void)
return; return;
} }
queue_process_events(main_loop.events); multiqueue_process_events(main_loop.events);
input_stop(); input_stop();
channel_teardown(); channel_teardown();
process_teardown(&main_loop); process_teardown(&main_loop);

View File

@@ -65,7 +65,7 @@ typedef struct {
uint64_t next_request_id; uint64_t next_request_id;
kvec_t(ChannelCallFrame *) call_stack; kvec_t(ChannelCallFrame *) call_stack;
kvec_t(WBuffer *) delayed_notifications; kvec_t(WBuffer *) delayed_notifications;
Queue *events; MultiQueue *events;
} Channel; } Channel;
typedef struct { typedef struct {
@@ -440,7 +440,7 @@ static void handle_request(Channel *channel, msgpack_object *request)
if (handler.async) { if (handler.async) {
on_request_event((void **)&event_data); on_request_event((void **)&event_data);
} else { } 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: case kChannelTypeStdio:
stream_close(&channel->data.std.in, NULL, NULL); stream_close(&channel->data.std.in, NULL, NULL);
stream_close(&channel->data.std.out, 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; return;
default: default:
abort(); abort();
@@ -654,7 +654,7 @@ static void free_channel(Channel *channel)
kv_destroy(channel->call_stack); kv_destroy(channel->call_stack);
kv_destroy(channel->delayed_notifications); kv_destroy(channel->delayed_notifications);
if (channel->type != kChannelTypeProc) { if (channel->type != kChannelTypeProc) {
queue_free(channel->events); multiqueue_free(channel->events);
} }
xfree(channel); xfree(channel);
} }
@@ -664,10 +664,11 @@ static void close_cb(Stream *stream, void *data)
decref(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)); 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->type = type;
rv->refcount = 1; rv->refcount = 1;
rv->closed = false; rv->closed = false;

View File

@@ -7880,15 +7880,15 @@ static void nv_event(cmdarg_T *cap)
{ {
// Garbage collection should have been executed before blocking for events in // 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 // 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 // `os_inchar` branch was not executed (!multiqueue_empty(loop.events), which
// have `may_garbage_collect` set to true in `normal_check`). // could have `may_garbage_collect` set to true in `normal_check`).
// //
// That is because here we may run code that calls `os_inchar` // 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 // 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 // not safe to perform garbage collection because there could be unreferenced
// lists or dicts being used. // lists or dicts being used.
may_garbage_collect = false; 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 cap->retval |= CA_COMMAND_BUSY; // don't call edit() now
} }

View File

@@ -87,8 +87,8 @@ static void create_cursorhold_event(void)
// have been called(inbuf_poll would return kInputAvail) // have been called(inbuf_poll would return kInputAvail)
// TODO(tarruda): Cursorhold should be implemented as a timer set during the // TODO(tarruda): Cursorhold should be implemented as a timer set during the
// `state_check` callback for the states where it can be triggered. // `state_check` callback for the states where it can be triggered.
assert(!events_enabled || queue_empty(main_loop.events)); assert(!events_enabled || multiqueue_empty(main_loop.events));
queue_put(main_loop.events, cursorhold_event, 0); multiqueue_put(main_loop.events, cursorhold_event, 0);
} }
// Low level input function // Low level input function
@@ -422,5 +422,5 @@ static void read_error_exit(void)
static bool pending_events(void) static bool pending_events(void)
{ {
return events_enabled && !queue_empty(main_loop.events); return events_enabled && !multiqueue_empty(main_loop.events);
} }

View File

@@ -207,7 +207,7 @@ static int do_os_system(char **argv,
Stream in, out, err; Stream in, out, err;
LibuvProcess uvproc = libuv_process_init(&main_loop, &buf); LibuvProcess uvproc = libuv_process_init(&main_loop, &buf);
Process *proc = &uvproc.process; Process *proc = &uvproc.process;
Queue *events = queue_new_child(main_loop.events); MultiQueue *events = multiqueue_new_child(main_loop.events);
proc->events = events; proc->events = events;
proc->argv = argv; proc->argv = argv;
proc->in = input != NULL ? &in : NULL; proc->in = input != NULL ? &in : NULL;
@@ -221,7 +221,7 @@ static int do_os_system(char **argv,
msg_outtrans((char_u *)prog); msg_outtrans((char_u *)prog);
msg_putchar('\n'); msg_putchar('\n');
} }
queue_free(events); multiqueue_free(events);
return -1; return -1;
} }
@@ -277,8 +277,8 @@ static int do_os_system(char **argv,
} }
} }
assert(queue_empty(events)); assert(multiqueue_empty(events));
queue_free(events); multiqueue_free(events);
return status; return status;
} }

View File

@@ -35,8 +35,8 @@ getkey:
// processing. Characters can come from mappings, scripts and other // processing. Characters can come from mappings, scripts and other
// sources, so this scenario is very common. // sources, so this scenario is very common.
key = safe_vgetc(); key = safe_vgetc();
} else if (!queue_empty(main_loop.events)) { } else if (!multiqueue_empty(main_loop.events)) {
// Event was made available after the last queue_process_events call // Event was made available after the last multiqueue_process_events call
key = K_EVENT; key = K_EVENT;
} else { } else {
input_enable_events(); input_enable_events();
@@ -48,7 +48,7 @@ getkey:
// directly. // directly.
(void)os_inchar(NULL, 0, -1, 0); (void)os_inchar(NULL, 0, -1, 0);
input_disable_events(); 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) { if (key == K_EVENT) {

View File

@@ -166,7 +166,7 @@ void terminal_init(void)
invalidated_terminals = pmap_new(ptr_t)(); invalidated_terminals = pmap_new(ptr_t)();
time_watcher_init(&main_loop, &refresh_timer, NULL); time_watcher_init(&main_loop, &refresh_timer, NULL);
// refresh_timer_cb will redraw the screen which can call vimscript // 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 // initialize a rgb->color index map for cterm attributes(VTermScreenCell
// only has RGB information and we need color indexes for terminal UIs) // only has RGB information and we need color indexes for terminal UIs)
@@ -201,7 +201,7 @@ void terminal_init(void)
void terminal_teardown(void) void terminal_teardown(void)
{ {
time_watcher_stop(&refresh_timer); time_watcher_stop(&refresh_timer);
queue_free(refresh_timer.events); multiqueue_free(refresh_timer.events);
time_watcher_close(&refresh_timer, NULL); time_watcher_close(&refresh_timer, NULL);
pmap_free(ptr_t)(invalidated_terminals); pmap_free(ptr_t)(invalidated_terminals);
map_free(int, int)(color_indexes); map_free(int, int)(color_indexes);
@@ -445,7 +445,7 @@ static int terminal_execute(VimState *state, int key)
case K_EVENT: case K_EVENT:
// We cannot let an event free the terminal yet. It is still needed. // We cannot let an event free the terminal yet. It is still needed.
s->term->refcount++; s->term->refcount++;
queue_process_events(main_loop.events); multiqueue_process_events(main_loop.events);
s->term->refcount--; s->term->refcount--;
if (s->term->buf_handle == 0) { if (s->term->buf_handle == 0) {
s->close = true; s->close = true;

View File

@@ -340,7 +340,7 @@ static void read_cb(Stream *stream, RBuffer *buf, size_t c, void *data,
// ls *.md | xargs nvim // ls *.md | xargs nvim
input->in_fd = 2; input->in_fd = 2;
stream_close(&input->read_stream, NULL, NULL); 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 { } else {
loop_schedule(&main_loop, event_create(1, input_done_event, 0)); loop_schedule(&main_loop, event_create(1, input_done_event, 0));
} }

View File

@@ -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 // 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 // before continuing. This is done in another callback to avoid
// loop_poll_events recursion // loop_poll_events recursion
queue_put_event(data->loop->fast_events, multiqueue_put_event(data->loop->fast_events,
event_create(1, suspend_event, 1, ui)); event_create(1, suspend_event, 1, ui));
} }
static void tui_set_title(UI *ui, char *title) static void tui_set_title(UI *ui, char *title)

View File

@@ -0,0 +1,16 @@
#include <string.h>
#include <stdlib.h>
#include "nvim/event/multiqueue.h"
#include "multiqueue.h"
void ut_multiqueue_put(MultiQueue *this, const char *str)
{
multiqueue_put(this, NULL, 1, str);
}
const char *ut_multiqueue_get(MultiQueue *this)
{
Event event = multiqueue_get(this);
return event.argv[0];
}

View File

@@ -0,0 +1,4 @@
#include "nvim/event/multiqueue.h"
void ut_multiqueue_put(MultiQueue *queue, const char *str);
const char *ut_multiqueue_get(MultiQueue *queue);

View File

@@ -1,16 +0,0 @@
#include <string.h>
#include <stdlib.h>
#include "nvim/event/queue.h"
#include "queue.h"
void ut_queue_put(Queue *queue, const char *str)
{
queue_put(queue, NULL, 1, str);
}
const char *ut_queue_get(Queue *queue)
{
Event event = queue_get(queue);
return event.argv[0];
}

View File

@@ -1,4 +0,0 @@
#include "nvim/event/queue.h"
void ut_queue_put(Queue *queue, const char *str);
const char *ut_queue_get(Queue *queue);

View File

@@ -3,28 +3,28 @@ local helpers = require("test.unit.helpers")
local ffi = helpers.ffi local ffi = helpers.ffi
local eq = helpers.eq local eq = helpers.eq
local queue = helpers.cimport("./test/unit/fixtures/queue.h") local multiqueue = helpers.cimport("./test/unit/fixtures/multiqueue.h")
describe('queue', function() describe("multiqueue (multi-level event-queue)", function()
local parent, child1, child2, child3 local parent, child1, child2, child3
local function put(q, str) local function put(q, str)
queue.ut_queue_put(q, str) multiqueue.ut_multiqueue_put(q, str)
end end
local function get(q) local function get(q)
return ffi.string(queue.ut_queue_get(q)) return ffi.string(multiqueue.ut_multiqueue_get(q))
end end
local function free(q) local function free(q)
queue.queue_free(q) multiqueue.multiqueue_free(q)
end end
before_each(function() before_each(function()
parent = queue.queue_new_parent(ffi.NULL, ffi.NULL) parent = multiqueue.multiqueue_new_parent(ffi.NULL, ffi.NULL)
child1 = queue.queue_new_child(parent) child1 = multiqueue.multiqueue_new_child(parent)
child2 = queue.queue_new_child(parent) child2 = multiqueue.multiqueue_new_child(parent)
child3 = queue.queue_new_child(parent) child3 = multiqueue.multiqueue_new_child(parent)
put(child1, 'c1i1') put(child1, 'c1i1')
put(child1, 'c1i2') put(child1, 'c1i2')
put(child2, 'c2i1') put(child2, 'c2i1')