loop: Improvements for thread-safety

- Implement `loop_schedule` method for queueing events from other threads
- Make `loop_poll_events` `recursive` static variable a field of the Loop
  structure
This commit is contained in:
Thiago de Arruda 2015-09-01 10:11:55 -03:00
parent 203a4d5650
commit c20b802511
3 changed files with 39 additions and 11 deletions

View File

@ -4,7 +4,7 @@
#include <assert.h> #include <assert.h>
#include <stdarg.h> #include <stdarg.h>
#define EVENT_HANDLER_MAX_ARGC 4 #define EVENT_HANDLER_MAX_ARGC 6
typedef void (*argv_callback)(void **argv); typedef void (*argv_callback)(void **argv);
typedef struct message { typedef struct message {
@ -12,6 +12,7 @@ typedef struct message {
argv_callback handler; argv_callback handler;
void *argv[EVENT_HANDLER_MAX_ARGC]; void *argv[EVENT_HANDLER_MAX_ARGC];
} Event; } Event;
typedef void(*event_scheduler)(Event event, void *data);
#define VA_EVENT_INIT(event, p, h, a) \ #define VA_EVENT_INIT(event, p, h, a) \
do { \ do { \

View File

@ -10,20 +10,19 @@
# include "event/loop.c.generated.h" # include "event/loop.c.generated.h"
#endif #endif
typedef struct idle_event {
uv_idle_t idle;
Event event;
} IdleEvent;
void loop_init(Loop *loop, void *data) void loop_init(Loop *loop, void *data)
{ {
uv_loop_init(&loop->uv); uv_loop_init(&loop->uv);
loop->recursive = 0;
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 = queue_new_parent(loop_on_put, loop);
loop->fast_events = queue_new_child(loop->events); loop->fast_events = queue_new_child(loop->events);
loop->thread_events = queue_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); uv_signal_init(&loop->uv, &loop->children_watcher);
uv_timer_init(&loop->uv, &loop->children_kill_timer); uv_timer_init(&loop->uv, &loop->children_kill_timer);
uv_timer_init(&loop->uv, &loop->poll_timer); uv_timer_init(&loop->uv, &loop->poll_timer);
@ -31,9 +30,7 @@ void loop_init(Loop *loop, void *data)
void loop_poll_events(Loop *loop, int ms) void loop_poll_events(Loop *loop, int ms)
{ {
static int recursive = 0; if (loop->recursive++) {
if (recursive++) {
abort(); // Should not re-enter uv_run abort(); // Should not re-enter uv_run
} }
@ -55,10 +52,19 @@ void loop_poll_events(Loop *loop, int ms)
uv_timer_stop(&loop->poll_timer); uv_timer_stop(&loop->poll_timer);
} }
recursive--; // Can re-enter uv_run now loop->recursive--; // Can re-enter uv_run now
queue_process_events(loop->fast_events); queue_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);
uv_async_send(&loop->async);
uv_mutex_unlock(&loop->mutex);
}
void loop_on_put(Queue *queue, void *data) void loop_on_put(Queue *queue, void *data)
{ {
Loop *loop = data; Loop *loop = data;
@ -72,14 +78,32 @@ void loop_on_put(Queue *queue, void *data)
void loop_close(Loop *loop) void loop_close(Loop *loop)
{ {
uv_mutex_destroy(&loop->mutex);
uv_close((uv_handle_t *)&loop->children_watcher, NULL); uv_close((uv_handle_t *)&loop->children_watcher, NULL);
uv_close((uv_handle_t *)&loop->children_kill_timer, NULL); uv_close((uv_handle_t *)&loop->children_kill_timer, NULL);
uv_close((uv_handle_t *)&loop->poll_timer, NULL); uv_close((uv_handle_t *)&loop->poll_timer, NULL);
uv_close((uv_handle_t *)&loop->async, NULL);
do { do {
uv_run(&loop->uv, UV_RUN_DEFAULT); uv_run(&loop->uv, UV_RUN_DEFAULT);
} while (uv_loop_close(&loop->uv)); } while (uv_loop_close(&loop->uv));
queue_free(loop->events);
queue_free(loop->fast_events);
queue_free(loop->thread_events);
kl_destroy(WatcherPtr, loop->children);
}
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);
}
uv_mutex_unlock(&l->mutex);
} }
static void timer_cb(uv_timer_t *handle) static void timer_cb(uv_timer_t *handle)
{ {
} }

View File

@ -16,11 +16,14 @@ KLIST_INIT(WatcherPtr, WatcherPtr, _noop)
typedef struct loop { typedef struct loop {
uv_loop_t uv; uv_loop_t uv;
Queue *events, *fast_events; Queue *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;
size_t children_stop_requests; size_t children_stop_requests;
uv_async_t async;
uv_mutex_t mutex;
int recursive;
} Loop; } Loop;
#define CREATE_EVENT(queue, handler, argc, ...) \ #define CREATE_EVENT(queue, handler, argc, ...) \