use separate workers for tile parsing to avoid blocking the threadpool for other libuv actions

include/mbgl/map/map.hpp

@@ -126,6 +126,7 @@ public:
     util::ptr<Sprite> getSprite();
     inline util::ptr<Texturepool> getTexturepool() { return texturepool; }
     inline util::ptr<uv::loop> 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<uv::loop> loop;
+    std::unique_ptr<uv::worker> workers;
     std::unique_ptr<uv::thread> thread;
     std::unique_ptr<uv_async_t> async_terminate;
     std::unique_ptr<uv_async_t> async_render;

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);

include/mbgl/util/uv.hpp

@@ -14,6 +14,7 @@ std::string cwd();
 class thread;
 class rwlock;
 class loop;
+class worker;
 
 }
 

include/mbgl/util/uv_detail.hpp

@@ -2,6 +2,7 @@
 #define MBGL_UTIL_UV_DETAIL
 
 #include <mbgl/util/ptr.hpp>
+#include <mbgl/util/uv-worker.h>
 
 #include <uv.h>
 
@@ -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 <typename T>
 class work {
 public:
@@ -115,30 +132,26 @@ public:
     typedef void (*after_work_callback)(T &object);
 
     template<typename... Args>
-    work(const mbgl::util::ptr<loop> &loop, work_callback work_cb, after_work_callback after_work_cb, Args&&... args)
-        : loop(loop),
-          data(std::forward<Args>(args)...),
+    work(worker &worker, work_callback work_cb, after_work_callback after_work_cb, Args&&... args)
+        : data(std::forward<Args>(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<T> *w = static_cast<work<T> *>(req->data);
+    static void do_work(void *data) {
+        work<T> *w = reinterpret_cast<work<T> *>(data);
         w->work_cb(w->data);
     }
 
-    static void after_work(uv_work_t *req, int) {
-        work<T> *w = static_cast<work<T> *>(req->data);
+    static void after_work(void *data) {
+        work<T> *w = reinterpret_cast<work<T> *>(data);
         w->after_work_cb(w->data);
         delete w;
     }
 
 private:
-    mbgl::util::ptr<uv::loop> loop;
-    uv_work_t req;
     T data;
     work_callback work_cb;
     after_work_callback after_work_cb;

src/map/map.cpp

@@ -60,6 +60,13 @@ Map::~Map() {
     }
 }
 
+uv::worker &Map::getWorker() {
+    if (!workers) {
+        workers = std::make_unique<uv::worker>(**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);

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<util::ptr<TileData>>(
-        map.getLoop(),
+        map.getWorker(),
         [](util::ptr<TileData> &tile) {
             tile->parse();
         },

src/storage/sqlite_store.cpp

@@ -32,7 +32,7 @@ SQLiteStore::SQLiteStore(uv_loop_t *loop, const std::string &path)
       db(std::make_shared<Database>(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() {

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);
 }

src/util/uv-worker.c

@@ -1,5 +1,6 @@
 #include <mbgl/util/uv-worker.h>
 #include <mbgl/util/uv-messenger.h>
+#include <mbgl/util/queue.h>
 
 #include <stdio.h>
 
@@ -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);
 }