Rewrite Worker in terms of Thread<>

include/mbgl/util/async_queue.hpp

@@ -1,95 +0,0 @@
-#ifndef MBGL_UTIL_ASYNC_QUEUE
-#define MBGL_UTIL_ASYNC_QUEUE
-
-#include "std.hpp"
-
-#include <uv.h>
-
-#include <thread>
-#include <mutex>
-#include <functional>
-#include <queue>
-#include <string>
-
-
-#if UV_VERSION_MAJOR == 0 && UV_VERSION_MINOR <= 10
-#define UV_ASYNC_PARAMS(handle) uv_async_t *handle, int
-#else
-#define UV_ASYNC_PARAMS(handle) uv_async_t *handle
-#endif
-
-namespace mbgl {
-namespace util {
-
-template <typename T>
-class AsyncQueue {
-public:
-    AsyncQueue(uv_loop_t *loop, std::function<void(T &)> fn) :
-          callback(fn) {
-        async.data = this;
-        uv_async_init(loop, &async, [](UV_ASYNC_PARAMS(handle)) {
-            auto q = reinterpret_cast<AsyncQueue *>(handle->data);
-            q->process();
-        });
-    }
-
-    void send(T &&data) {
-        {
-            std::lock_guard<std::mutex> lock(mutex);
-            queue.push(util::make_unique<T>(std::move(data)));
-        }
-        uv_async_send(&async);
-    }
-
-    void send(std::unique_ptr<T> data) {
-        {
-            std::lock_guard<std::mutex> lock(mutex);
-            queue.push(std::move(data));
-        }
-        uv_async_send(&async);
-    }
-
-    void stop() {
-        uv_close((uv_handle_t *)&async, [](uv_handle_t *handle) {
-            delete reinterpret_cast<AsyncQueue *>(handle->data);
-        });
-    }
-
-    void ref() {
-        uv_ref((uv_handle_t *)&async);
-    }
-
-    void unref() {
-        uv_unref((uv_handle_t *)&async);
-    }
-
-private:
-    ~AsyncQueue() {
-    }
-
-    void process() {
-        std::unique_ptr<T> item;
-        while (true) {
-            mutex.lock();
-            if (queue.empty()) {
-                mutex.unlock();
-                break;
-            }
-            item = std::move(queue.front());
-            queue.pop();
-            mutex.unlock();
-            callback(*item);
-        }
-    }
-
-private:
-    std::mutex mutex;
-    uv_async_t async;
-    std::queue<std::unique_ptr<T>> queue;
-    std::function<void(T &)> callback;
-};
-
-}
-}
-
-#endif

src/mbgl/map/map_context.cpp

@@ -37,7 +37,7 @@ MapContext::MapContext(uv_loop_t* loop, View& view_, FileSource& fileSource, Map
       envScope(env, ThreadType::Map, "Map"),
       updated(static_cast<UpdateType>(Update::Nothing)),
       asyncUpdate(util::make_unique<uv::async>(loop, [this] { update(); })),
-      workers(util::make_unique<Worker>(loop, 4)),
+      workers(util::make_unique<Worker>(4)),
       glyphStore(util::make_unique<GlyphStore>(env)),
       glyphAtlas(util::make_unique<GlyphAtlas>(1024, 1024)),
       spriteAtlas(util::make_unique<SpriteAtlas>(512, 512)),

src/mbgl/util/run_loop.hpp

@@ -51,6 +51,18 @@ public:
         });
     }
 
+    // Invoke fn() in the runloop thread, then invoke callback() in the current thread.
+    template <class Fn>
+    void invokeWithResult(Fn&& fn, std::function<void ()> callback) {
+        RunLoop* outer = current.get();
+        assert(outer);
+
+        invoke([fn, callback, outer] {
+            fn();
+            outer->invoke(std::move(callback));
+        });
+    }
+
     uv_loop_t* get() { return *loop; }
 
     static uv::tls<RunLoop> current;

src/mbgl/util/thread.hpp

@@ -53,6 +53,12 @@ public:
         loop->invokeWithResult(std::bind(fn, object, args...), callback);
     }
 
+    // Invoke object->fn(args...) in the runloop thread, then invoke callback() in the current thread.
+    template <typename Fn, class... Args>
+    void invokeWithResult(Fn fn, std::function<void ()> callback, Args&&... args) {
+        loop->invokeWithResult(std::bind(fn, object, args...), callback);
+    }
+
     // Invoke object->fn(args...) in the runloop thread, and wait for the result.
     template <class R, typename Fn, class... Args>
     R invokeSync(Fn fn, Args&&... args) {

src/mbgl/util/worker.cpp

@@ -4,70 +4,26 @@
 
 namespace mbgl {
 
-Worker::Worker(uv_loop_t* loop, std::size_t count)
-    : queue(new Queue(loop, [this](Fn after) { afterWork(after); }))
-{
-    queue->unref();
+class Worker::Impl {
+public:
+    Impl(uv_loop_t*) {}
 
+    void doWork(Fn work) {
+        work();
+    }
+};
+
+Worker::Worker(std::size_t count) {
     for (std::size_t i = 0; i < count; i++) {
-        threads.emplace_back(&Worker::workLoop, this);
+        threads.emplace_back(util::make_unique<util::Thread<Impl>>("Worker"));
     }
 }
 
-Worker::~Worker() {
-    MBGL_VERIFY_THREAD(tid);
-
-    if (active++ == 0) {
-        queue->ref();
-    }
-
-    channel.send(Work());
-
-    for (auto& thread : threads) {
-        thread.join();
-    }
-
-    queue->stop();
-}
+Worker::~Worker() = default;
 
 void Worker::send(Fn work, Fn after) {
-    MBGL_VERIFY_THREAD(tid);
-    assert(work);
-
-    if (active++ == 0) {
-        queue->ref();
-    }
-
-    channel.send({work, after});
-}
-
-void Worker::workLoop() {
-#ifdef __APPLE__
-    pthread_setname_np("Worker");
-#endif
-
-    while (true) {
-        Work item = channel.receive();
-
-        if (!item.work)
-            break;
-
-        item.work();
-        queue->send(std::move(item.after));
-    }
-
-    // Make sure to close all other workers too.
-    channel.send(Work());
-}
-
-void Worker::afterWork(Fn after) {
-    if (after) {
-        after();
-    }
-
-    if (--active == 0) {
-        queue->unref();
-    }
+    threads[current]->invokeWithResult(&Worker::Impl::doWork, after, work);
+    current = (current + 1) % threads.size();
 }
 
 }

src/mbgl/util/worker.hpp

@@ -2,11 +2,8 @@
 #define MBGL_UTIL_WORKER
 
 #include <mbgl/util/noncopyable.hpp>
-#include <mbgl/util/async_queue.hpp>
-#include <mbgl/util/channel.hpp>
-#include <mbgl/util/util.hpp>
+#include <mbgl/util/thread.hpp>
 
-#include <thread>
 #include <functional>
 
 namespace mbgl {
@@ -15,28 +12,15 @@ class Worker : public mbgl::util::noncopyable {
 public:
     using Fn = std::function<void ()>;
 
-    Worker(uv_loop_t* loop, std::size_t count);
+    Worker(std::size_t count);
     ~Worker();
 
     void send(Fn work, Fn after);
 
 private:
-    void workLoop();
-    void afterWork(Fn after);
-
-    struct Work {
-        Fn work;
-        Fn after;
-    };
-
-    using Queue = util::AsyncQueue<std::function<void ()>>;
-
-    std::size_t active = 0;
-    Queue* queue = nullptr;
-    Channel<Work> channel;
-    std::vector<std::thread> threads;
-
-    MBGL_STORE_THREAD(tid)
+    class Impl;
+    std::vector<std::unique_ptr<util::Thread<Impl>>> threads;
+    std::size_t current = 0;
 };
 
 }