From fd2b2e208229259630f92bd7615f9197c163d27f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Konstantin=20Ka=CC=88fer?= Date: Tue, 16 Sep 2014 18:23:42 +0200 Subject: [PATCH] use separate workers for tile parsing to avoid blocking the threadpool for other libuv actions --- include/mbgl/map/map.hpp | 2 ++ include/mbgl/util/uv-worker.h | 5 ++-- include/mbgl/util/uv.hpp | 1 + include/mbgl/util/uv_detail.hpp | 35 ++++++++++++++++++--------- src/map/map.cpp | 8 ++++++ src/map/tile_data.cpp | 2 +- src/storage/sqlite_store.cpp | 2 +- src/util/uv-channel.c | 14 +++++++++++ src/util/uv-worker.c | 43 +++++++++++++++++++++++++++------ 9 files changed, 89 insertions(+), 23 deletions(-) diff --git a/include/mbgl/map/map.hpp b/include/mbgl/map/map.hpp index 786c7196a..c389d898b 100644 --- a/include/mbgl/map/map.hpp +++ b/include/mbgl/map/map.hpp @@ -126,6 +126,7 @@ public: util::ptr getSprite(); inline util::ptr getTexturepool() { return texturepool; } inline util::ptr getLoop() { return loop; } + uv::worker &getWorker(); inline timestamp getAnimationTime() const { return animationTime; } inline timestamp getTime() const { return animationTime; } void updateTiles(); @@ -159,6 +160,7 @@ private: private: bool async = false; util::ptr loop; + std::unique_ptr workers; std::unique_ptr thread; std::unique_ptr async_terminate; std::unique_ptr async_render; diff --git a/include/mbgl/util/uv-worker.h b/include/mbgl/util/uv-worker.h index b9eb10fb7..a66feb4d0 100644 --- a/include/mbgl/util/uv-worker.h +++ b/include/mbgl/util/uv-worker.h @@ -14,17 +14,16 @@ typedef struct uv_messenger_s uv_messenger_t; typedef struct uv_worker_s uv_worker_t; struct uv_worker_s { - uv_thread_t thread; uv_messenger_t *msgr; uv_chan_t chan; const char *name; + void *threads[2]; }; 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 uv_worker_init_named(uv_worker_t *worker, uv_loop_t *loop, const char *name); +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); diff --git a/include/mbgl/util/uv.hpp b/include/mbgl/util/uv.hpp index 4ceb8b873..80fff9c96 100644 --- a/include/mbgl/util/uv.hpp +++ b/include/mbgl/util/uv.hpp @@ -14,6 +14,7 @@ std::string cwd(); class thread; class rwlock; class loop; +class worker; } diff --git a/include/mbgl/util/uv_detail.hpp b/include/mbgl/util/uv_detail.hpp index 272ca4149..b0b4fe583 100644 --- a/include/mbgl/util/uv_detail.hpp +++ b/include/mbgl/util/uv_detail.hpp @@ -2,6 +2,7 @@ #define MBGL_UTIL_UV_DETAIL #include +#include #include @@ -108,6 +109,22 @@ private: uv_once_t o = UV_ONCE_INIT; }; +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_worker_close(&w); + } + 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); + } + +private: + uv_worker_t w; +}; + template class work { public: @@ -115,30 +132,26 @@ public: typedef void (*after_work_callback)(T &object); template - work(const mbgl::util::ptr &loop, work_callback work_cb, after_work_callback after_work_cb, Args&&... args) - : loop(loop), - data(std::forward(args)...), + work(worker &worker, work_callback work_cb, after_work_callback after_work_cb, Args&&... args) + : data(std::forward(args)...), work_cb(work_cb), after_work_cb(after_work_cb) { - req.data = this; - uv_queue_work(**loop, &req, do_work, after_work); + worker.add(this, do_work, after_work); } private: - static void do_work(uv_work_t *req) { - work *w = static_cast *>(req->data); + static void do_work(void *data) { + work *w = reinterpret_cast *>(data); w->work_cb(w->data); } - static void after_work(uv_work_t *req, int) { - work *w = static_cast *>(req->data); + static void after_work(void *data) { + work *w = reinterpret_cast *>(data); w->after_work_cb(w->data); delete w; } private: - mbgl::util::ptr loop; - uv_work_t req; T data; work_callback work_cb; after_work_callback after_work_cb; diff --git a/src/map/map.cpp b/src/map/map.cpp index ecf4eb283..93e1bea53 100644 --- a/src/map/map.cpp +++ b/src/map/map.cpp @@ -60,6 +60,13 @@ Map::~Map() { } } +uv::worker &Map::getWorker() { + if (!workers) { + workers = std::make_unique(**loop, 4, "Tile Worker"); + } + return *workers; +} + void Map::start() { assert(uv_thread_self() == main_thread); assert(!async); @@ -198,6 +205,7 @@ void Map::terminate(uv_async_t *async) { map->glyphStore.reset(); map->fileSource.reset(); map->style.reset(); + map->workers.reset(); map->activeSources.clear(); uv_close((uv_handle_t *)map->async_cleanup.get(), nullptr); diff --git a/src/map/tile_data.cpp b/src/map/tile_data.cpp index 2a0e96cad..ecb05d5b2 100644 --- a/src/map/tile_data.cpp +++ b/src/map/tile_data.cpp @@ -87,7 +87,7 @@ void TileData::reparse() { // We're creating a new work request. The work request deletes itself after it executed // the after work handler new uv::work>( - map.getLoop(), + map.getWorker(), [](util::ptr &tile) { tile->parse(); }, diff --git a/src/storage/sqlite_store.cpp b/src/storage/sqlite_store.cpp index baa599eef..d44520a95 100644 --- a/src/storage/sqlite_store.cpp +++ b/src/storage/sqlite_store.cpp @@ -32,7 +32,7 @@ SQLiteStore::SQLiteStore(uv_loop_t *loop, const std::string &path) db(std::make_shared(path.c_str(), ReadWrite | Create)) { createSchema(); worker = new uv_worker_t; - uv_worker_init_named(worker, loop, "SQLite"); + uv_worker_init(worker, loop, 1, "SQLite"); } SQLiteStore::~SQLiteStore() { diff --git a/src/util/uv-channel.c b/src/util/uv-channel.c index 3d0483fd7..2936b01c5 100644 --- a/src/util/uv-channel.c +++ b/src/util/uv-channel.c @@ -49,7 +49,21 @@ void *uv_chan_receive(uv_chan_t *chan) { return data; } +void uv_chan_clear(uv_chan_t *chan) { + uv_mutex_lock(&chan->mutex); + uv__chan_item_t *item = NULL; + QUEUE *head = NULL; + while (!QUEUE_EMPTY(&chan->q)) { + head = QUEUE_HEAD(&chan->q); + item = QUEUE_DATA(head, uv__chan_item_t, active_queue); + QUEUE_REMOVE(head); + free(item); + } + uv_mutex_unlock(&chan->mutex); +} + void uv_chan_destroy(uv_chan_t *chan) { + uv_chan_clear(chan); uv_cond_destroy(&chan->cond); uv_mutex_destroy(&chan->mutex); } diff --git a/src/util/uv-worker.c b/src/util/uv-worker.c index d0e2550b7..c47b0ffed 100644 --- a/src/util/uv-worker.c +++ b/src/util/uv-worker.c @@ -1,5 +1,6 @@ #include #include +#include #include @@ -10,6 +11,12 @@ 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 { + uv_thread_t thread; + void *queue[2]; +}; + void uv__worker_after(void *ptr) { uv__worker_item_t *item = (uv__worker_item_t *)ptr; item->after_work_cb(item->data); @@ -30,20 +37,29 @@ void uv__worker_thread_loop(void *ptr) { item->work_cb(item->data); uv_messenger_send(worker->msgr, item); } + + // Make sure to close all other workers too. + uv_chan_send(&worker->chan, NULL); } -int uv_worker_init(uv_worker_t *worker, uv_loop_t *loop) { - return uv_worker_init_named(worker, loop, NULL); -} - -int uv_worker_init_named(uv_worker_t *worker, uv_loop_t *loop, const char *name) { +int uv_worker_init(uv_worker_t *worker, uv_loop_t *loop, int count, const char *name) { worker->name = name; worker->msgr = malloc(sizeof(uv_messenger_t)); int ret = uv_messenger_init(loop, worker->msgr, uv__worker_after); if (ret < 0) return ret; ret = uv_chan_init(&worker->chan); if (ret < 0) return ret; - return uv_thread_create(&worker->thread, uv__worker_thread_loop, worker); + + // Initialize worker threads. + QUEUE_INIT(&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); + if (ret < 0) return ret; + QUEUE_INSERT_TAIL(&worker->threads, &work_thread->queue); + } + + return 0; } void uv_worker_send(uv_worker_t *worker, void *data, uv_worker_cb work_cb, @@ -55,8 +71,21 @@ 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_chan_send(&worker->chan, NULL); - uv_thread_join(&worker->thread); + uv_worker_join_all(worker); + uv_chan_destroy(&worker->chan); uv_messenger_stop(worker->msgr); }