diff --git a/include/mbgl/util/uv-worker.h b/include/mbgl/util/uv-worker.h index a66feb4d0..a9c9aff50 100644 --- a/include/mbgl/util/uv-worker.h +++ b/include/mbgl/util/uv-worker.h @@ -13,20 +13,25 @@ typedef struct uv_messenger_s uv_messenger_t; typedef struct uv_worker_s uv_worker_t; +typedef void (*uv_worker_cb)(void *data); +typedef void (*uv_worker_after_cb)(void *data); +typedef void (*uv_worker_close_cb)(uv_worker_t *worker); + struct uv_worker_s { +#ifndef NDEBUG + unsigned long thread_id; +#endif uv_messenger_t *msgr; uv_chan_t chan; const char *name; - void *threads[2]; + int count; + uv_worker_close_cb close_cb; }; -typedef void (*uv_worker_cb)(void *data); -typedef void (*uv_worker_after_cb)(void *data); - int uv_worker_init(uv_worker_t *worker, uv_loop_t *loop, int count, const char *name); void uv_worker_send(uv_worker_t *worker, void *data, uv_worker_cb work_cb, uv_worker_after_cb after_work_cb); -void uv_worker_close(uv_worker_t *worker); +void uv_worker_close(uv_worker_t *worker, uv_worker_close_cb close_cb); #ifdef __cplusplus } diff --git a/include/mbgl/util/uv_detail.hpp b/include/mbgl/util/uv_detail.hpp index b0b4fe583..e0a57ce65 100644 --- a/include/mbgl/util/uv_detail.hpp +++ b/include/mbgl/util/uv_detail.hpp @@ -111,18 +111,20 @@ private: class worker { public: - inline worker(uv_loop_t *loop, unsigned int count, const char *name = nullptr) { - uv_worker_init(&w, loop, count, name); + inline worker(uv_loop_t *loop, unsigned int count, const char *name = nullptr) : w(new uv_worker_t) { + uv_worker_init(w, loop, count, name); } inline ~worker() { - uv_worker_close(&w); + uv_worker_close(w, [](uv_worker_t *worker) { + delete worker; + }); } inline void add(void *data, uv_worker_cb work_cb, uv_worker_after_cb after_work_cb) { - uv_worker_send(&w, data, work_cb, after_work_cb); + uv_worker_send(w, data, work_cb, after_work_cb); } private: - uv_worker_t w; + uv_worker_t *w; }; template diff --git a/src/storage/sqlite_store.cpp b/src/storage/sqlite_store.cpp index d44520a95..6aba92e91 100644 --- a/src/storage/sqlite_store.cpp +++ b/src/storage/sqlite_store.cpp @@ -39,8 +39,9 @@ SQLiteStore::~SQLiteStore() { // Nothing to do. This function needs to be here because we're forward-declaring // Database, so we need the actual definition here to be able to properly destruct it. if (worker) { - uv_worker_close(worker); - delete worker; + uv_worker_close(worker, [](uv_worker_t *worker) { + delete worker; + }); } } diff --git a/src/util/uv-worker.c b/src/util/uv-worker.c index c47b0ffed..53ccb828c 100644 --- a/src/util/uv-worker.c +++ b/src/util/uv-worker.c @@ -1,8 +1,8 @@ #include #include -#include #include +#include typedef struct uv__worker_item_s uv__worker_item_t; struct uv__worker_item_s { @@ -11,20 +11,56 @@ struct uv__worker_item_s { uv_worker_after_cb after_work_cb; }; -typedef struct uv__work_thread_s uv__work_thread_t; -struct uv__work_thread_s { +typedef struct uv__worker_thread_s uv__worker_thread_t; +struct uv__worker_thread_s { + uv_worker_t *worker; uv_thread_t thread; - void *queue[2]; }; +void uv__worker_thread_finished(uv__worker_thread_t *worker_thread) { + uv_worker_t *worker = worker_thread->worker; + +#ifndef NDEBUG + assert(uv_thread_self() == worker->thread_id); +#endif + + // This should at most block very briefly. We are sending the termination + // notification as the last thing in the worker thread, so by now the thread + // has probably terminated already. If not, the waiting time should be + // extremely short. + uv_thread_join(&worker_thread->thread); + + assert(worker->count > 0); + worker->count--; + if (worker->count == 0) { + uv_chan_destroy(&worker->chan); + uv_messenger_stop(worker->msgr); + if (worker->close_cb) { + worker->close_cb(worker); + } + } +} + void uv__worker_after(void *ptr) { uv__worker_item_t *item = (uv__worker_item_t *)ptr; - item->after_work_cb(item->data); + + if (item->work_cb) { + // We are finishing a regular work request. + assert(item->after_work_cb); + item->after_work_cb(item->data); + } else { + // This is a worker thread termination. + uv__worker_thread_t *worker_thread = (uv__worker_thread_t *)item->data; + uv__worker_thread_finished(worker_thread); + free(worker_thread); + } + free(item); } void uv__worker_thread_loop(void *ptr) { - uv_worker_t *worker = (uv_worker_t *)ptr; + uv__worker_thread_t *worker_thread = (uv__worker_thread_t *)ptr; + uv_worker_t *worker = worker_thread->worker; #ifdef __APPLE__ if (worker->name) { @@ -34,29 +70,52 @@ void uv__worker_thread_loop(void *ptr) { uv__worker_item_t *item = NULL; while ((item = uv_chan_receive(&worker->chan)) != NULL) { + assert(item->work_cb); item->work_cb(item->data); - uv_messenger_send(worker->msgr, item); + + if (item->after_work_cb) { + // Trigger the after callback in the main thread. + uv_messenger_send(worker->msgr, item); + } else { + // There is no after work callback, so it wouldn't do anything anyway. + free(item); + } } // Make sure to close all other workers too. uv_chan_send(&worker->chan, NULL); + + // Create a new worker item that acts as a terminate flag for this thread. + item = malloc(sizeof(uv__worker_item_t)); + item->data = worker_thread; + item->work_cb = NULL; + item->after_work_cb = NULL; + uv_messenger_send(worker->msgr, item); } int uv_worker_init(uv_worker_t *worker, uv_loop_t *loop, int count, const char *name) { +#ifndef NDEBUG + worker->thread_id = uv_thread_self(); +#endif worker->name = name; + worker->count = 0; + worker->close_cb = NULL; worker->msgr = malloc(sizeof(uv_messenger_t)); int ret = uv_messenger_init(loop, worker->msgr, uv__worker_after); - if (ret < 0) return ret; + if (ret < 0) { + free(worker->msgr); + return ret; + } ret = uv_chan_init(&worker->chan); if (ret < 0) return ret; - // Initialize worker threads. - QUEUE_INIT(&worker->threads); + // Initialize all worker threads. for (int i = 0; i < count; i++) { - uv__work_thread_t *work_thread = malloc(sizeof(uv__work_thread_t)); - ret = uv_thread_create(&work_thread->thread, uv__worker_thread_loop, worker); + uv__worker_thread_t *worker_thread = malloc(sizeof(uv__worker_thread_t)); + worker_thread->worker = worker; + ret = uv_thread_create(&worker_thread->thread, uv__worker_thread_loop, worker_thread); if (ret < 0) return ret; - QUEUE_INSERT_TAIL(&worker->threads, &work_thread->queue); + worker->count++; } return 0; @@ -64,6 +123,15 @@ int uv_worker_init(uv_worker_t *worker, uv_loop_t *loop, int count, const char * void uv_worker_send(uv_worker_t *worker, void *data, uv_worker_cb work_cb, uv_worker_after_cb after_work_cb) { +#ifndef NDEBUG + assert(uv_thread_self() == worker->thread_id); +#endif + + // It doesn't make sense to not provide a work callback. On the other hand, the after_work_cb + // may be NULL. In that case, there will be no callback called in the current thread and the + // worker item will instead be freed in the worker thread. + assert(work_cb); + uv__worker_item_t *item = malloc(sizeof(uv__worker_item_t)); item->work_cb = work_cb; item->after_work_cb = after_work_cb; @@ -71,21 +139,14 @@ void uv_worker_send(uv_worker_t *worker, void *data, uv_worker_cb work_cb, uv_chan_send(&worker->chan, item); } -void uv_worker_join_all(uv_worker_t *worker) { - QUEUE *head = NULL; - uv__work_thread_t *item = NULL; - while (!QUEUE_EMPTY(&worker->threads)) { - head = QUEUE_HEAD(&worker->threads); - item = QUEUE_DATA(head, uv__work_thread_t, queue); - QUEUE_REMOVE(head); - uv_thread_join(&item->thread); - free(item); - } -} +void uv_worker_close(uv_worker_t *worker, uv_worker_close_cb close_cb) { +#ifndef NDEBUG + assert(uv_thread_self() == worker->thread_id); +#endif -void uv_worker_close(uv_worker_t *worker) { + // Prevent double calling. + assert(worker->close_cb == NULL); + + worker->close_cb = close_cb; uv_chan_send(&worker->chan, NULL); - uv_worker_join_all(worker); - uv_chan_destroy(&worker->chan); - uv_messenger_stop(worker->msgr); }