From 4ccf246c10b8def2707fec11e8cff565bba90db7 Mon Sep 17 00:00:00 2001 From: John Firebaugh Date: Thu, 23 Apr 2015 16:38:46 -0700 Subject: [PATCH] Rewrite Worker in terms of Thread<> --- include/mbgl/util/async_queue.hpp | 95 ------------------------------- src/mbgl/map/map_context.cpp | 2 +- src/mbgl/util/run_loop.hpp | 12 ++++ src/mbgl/util/thread.hpp | 6 ++ src/mbgl/util/worker.cpp | 70 +++++------------------ src/mbgl/util/worker.hpp | 26 ++------- 6 files changed, 37 insertions(+), 174 deletions(-) delete mode 100644 include/mbgl/util/async_queue.hpp diff --git a/include/mbgl/util/async_queue.hpp b/include/mbgl/util/async_queue.hpp deleted file mode 100644 index b3eaabc31..000000000 --- a/include/mbgl/util/async_queue.hpp +++ /dev/null @@ -1,95 +0,0 @@ -#ifndef MBGL_UTIL_ASYNC_QUEUE -#define MBGL_UTIL_ASYNC_QUEUE - -#include "std.hpp" - -#include - -#include -#include -#include -#include -#include - - -#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 -class AsyncQueue { -public: - AsyncQueue(uv_loop_t *loop, std::function fn) : - callback(fn) { - async.data = this; - uv_async_init(loop, &async, [](UV_ASYNC_PARAMS(handle)) { - auto q = reinterpret_cast(handle->data); - q->process(); - }); - } - - void send(T &&data) { - { - std::lock_guard lock(mutex); - queue.push(util::make_unique(std::move(data))); - } - uv_async_send(&async); - } - - void send(std::unique_ptr data) { - { - std::lock_guard 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(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 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> queue; - std::function callback; -}; - -} -} - -#endif diff --git a/src/mbgl/map/map_context.cpp b/src/mbgl/map/map_context.cpp index da71b787d..ccd239253 100644 --- a/src/mbgl/map/map_context.cpp +++ b/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(Update::Nothing)), asyncUpdate(util::make_unique(loop, [this] { update(); })), - workers(util::make_unique(loop, 4)), + workers(util::make_unique(4)), glyphStore(util::make_unique(env)), glyphAtlas(util::make_unique(1024, 1024)), spriteAtlas(util::make_unique(512, 512)), diff --git a/src/mbgl/util/run_loop.hpp b/src/mbgl/util/run_loop.hpp index a192fe179..971ad2cac 100644 --- a/src/mbgl/util/run_loop.hpp +++ b/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 + void invokeWithResult(Fn&& fn, std::function 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 current; diff --git a/src/mbgl/util/thread.hpp b/src/mbgl/util/thread.hpp index 2150f5bc3..850b35c9e 100644 --- a/src/mbgl/util/thread.hpp +++ b/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 + void invokeWithResult(Fn fn, std::function 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 R invokeSync(Fn fn, Args&&... args) { diff --git a/src/mbgl/util/worker.cpp b/src/mbgl/util/worker.cpp index 3559cdd71..fcbe6f5df 100644 --- a/src/mbgl/util/worker.cpp +++ b/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>("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(); } } diff --git a/src/mbgl/util/worker.hpp b/src/mbgl/util/worker.hpp index 86c2e6acf..b71b4fd5b 100644 --- a/src/mbgl/util/worker.hpp +++ b/src/mbgl/util/worker.hpp @@ -2,11 +2,8 @@ #define MBGL_UTIL_WORKER #include -#include -#include -#include +#include -#include #include namespace mbgl { @@ -15,28 +12,15 @@ class Worker : public mbgl::util::noncopyable { public: using Fn = std::function; - 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::size_t active = 0; - Queue* queue = nullptr; - Channel channel; - std::vector threads; - - MBGL_STORE_THREAD(tid) + class Impl; + std::vector>> threads; + std::size_t current = 0; }; }