tei/MapLibreParticleLab
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Stricter Thread<> interface

Browse Source 
Implements a Thread<>::invoke interface that only permits calling
member functions with bound arguments.

Also introduce invokeWithResult, which accepts a callback and invokes
it, passing the result, on the invoking thread's RunLoop. This closes
the loop of two evented threads doing message passing.
This commit is contained in:
John Firebaugh
2015-04-13 16:41:37 -07:00
parent 1066504008
commit 1eaf6b8c18
9 changed files with 237 additions and 137 deletions
Download Patch File Download Diff File Expand all files Collapse all files

21
platform/default/sqlite_cache.cpp
View File

@@ -5,7 +5,6 @@
#include <mbgl/util/compression.hpp>
#include <mbgl/util/io.hpp>
#include <mbgl/util/thread.hpp>
#include <mbgl/util/run_loop.hpp>
#include <mbgl/platform/log.hpp>
#include "sqlite3.hpp"
@@ -62,16 +61,14 @@ std::string unifyMapboxURLs(const std::string &url) {
using namespace mapbox::sqlite;
SQLiteCache::SQLiteCache(const std::string& path_) : thread(util::make_unique<util::Thread<Impl>>(path_)) {
SQLiteCache::SQLiteCache(const std::string& path_)
: thread(util::make_unique<util::Thread<Impl>>("SQLite Cache", path_)) {
}
SQLiteCache::~SQLiteCache() = default;
SQLiteCache::Impl::Impl(const std::string& path_)
: path(path_) {
#ifdef __APPLE__
pthread_setname_np("SQLite Cache");
#endif
}
SQLiteCache::Impl::~Impl() {
@@ -134,10 +131,10 @@ void SQLiteCache::get(const Resource &resource, Callback callback) {
// Will try to load the URL from the SQLite database and call the callback when done.
// Note that the callback is probably going to invoked from another thread, so the caller
// must make sure that it can run in that thread.
(*thread)->invoke([=] { (*thread)->processGet(resource, callback); });
thread->invokeWithResult(&Impl::processGet, callback, resource);
}
void SQLiteCache::Impl::processGet(const Resource &resource, Callback callback) {
std::unique_ptr<Response> SQLiteCache::Impl::processGet(const Resource &resource) {
try {
// This is called in the SQLite event loop.
if (!db) {
@@ -170,14 +167,14 @@ void SQLiteCache::Impl::processGet(const Resource &resource, Callback callback)
if (getStmt->get<int>(5)) { // == compressed
response->data = util::decompress(response->data);
}
callback(std::move(response));
return std::move(response);
} else {
// There is no data.
callback(nullptr);
return nullptr;
}
} catch (mapbox::sqlite::Exception& ex) {
Log::Error(Event::Database, ex.code, ex.what());
callback(nullptr);
return nullptr;
}
}
@@ -186,9 +183,9 @@ void SQLiteCache::put(const Resource &resource, std::shared_ptr<const Response>
// storing a new response or updating the currently stored response, potentially setting a new
// expiry date.
if (hint == Hint::Full) {
(*thread)->invoke([=] { (*thread)->processPut(resource, response); });
thread->invoke(&Impl::processPut, resource, std::move(response));
} else if (hint == Hint::Refresh) {
(*thread)->invoke([=] { (*thread)->processRefresh(resource, response->expires); });
thread->invoke(&Impl::processRefresh, resource, int64_t(response->expires));
}
}

9
platform/default/sqlite_cache_impl.hpp
View File

@@ -2,7 +2,6 @@
#define MBGL_STORAGE_DEFAULT_SQLITE_CACHE_IMPL
#include <mbgl/storage/sqlite_cache.hpp>
#include <mbgl/util/run_loop.hpp>
namespace mapbox {
namespace sqlite {
@@ -13,15 +12,12 @@ class Statement;
namespace mbgl {
class SQLiteCache::Impl : public util::RunLoop {
friend class util::Thread<SQLiteCache::Impl>;
class SQLiteCache::Impl {
public:
Impl(const std::string &path = ":memory:");
~Impl();
public:
void processGet(const Resource& resource, Callback callback);
std::unique_ptr<Response> processGet(const Resource&);
void processPut(const Resource& resource, std::shared_ptr<const Response> response);
void processRefresh(const Resource& resource, int64_t expires);
@@ -29,7 +25,6 @@ private:
void createDatabase();
void createSchema();
private:
const std::string path;
std::unique_ptr<::mapbox::sqlite::Database> db;
std::unique_ptr<::mapbox::sqlite::Statement> getStmt;

27
src/mbgl/storage/default_file_source.cpp
View File

@@ -10,6 +10,7 @@
#include <mbgl/util/chrono.hpp>
#include <mbgl/util/thread.hpp>
#include <mbgl/platform/log.hpp>
#include <mbgl/map/environment.hpp>
#pragma GCC diagnostic push
#ifndef __clang__
@@ -29,13 +30,10 @@ namespace mbgl {
DefaultFileSource::Impl::Impl(FileCache* cache_, const std::string& root)
: assetRoot(root.empty() ? platform::assetRoot() : root), cache(cache_) {
#ifdef __APPLE__
pthread_setname_np("FileSource");
#endif
}
DefaultFileSource::DefaultFileSource(FileCache* cache, const std::string& root)
: thread(util::make_unique<util::Thread<Impl>>(cache, root)) {
: thread(util::make_unique<util::Thread<Impl>>("FileSource", cache, root)) {
}
DefaultFileSource::~DefaultFileSource() {
@@ -62,7 +60,7 @@ Request* DefaultFileSource::request(const Resource& resource,
// This function can be called from any thread. Make sure we're executing the actual call in the
// file source loop by sending it over the queue.
(*thread)->invoke([=] { (*thread)->processAdd(req); });
thread->invoke(&Impl::processAdd, std::move(req), thread->get());
return req;
}
@@ -76,14 +74,14 @@ void DefaultFileSource::cancel(Request *req) {
// This function can be called from any thread. Make sure we're executing the actual call in the
// file source loop by sending it over the queue.
(*thread)->invoke([=] { (*thread)->processCancel(req); });
thread->invoke(&Impl::processCancel, std::move(req));
}
void DefaultFileSource::abort(const Environment &env) {
(*thread)->invoke([=, &env] { (*thread)->processAbort(env); });
thread->invoke(&Impl::processAbort, std::ref(env));
}
void DefaultFileSource::Impl::processAdd(Request* req) {
void DefaultFileSource::Impl::processAdd(Request* req, uv_loop_t* loop) {
const Resource &resource = req->resource;
// We're adding a new Request.
@@ -102,13 +100,12 @@ void DefaultFileSource::Impl::processAdd(Request* req) {
// But first, we're going to start querying the database if it exists.
if (!cache) {
sharedRequest->start(loop().get());
sharedRequest->start(loop);
} else {
// Otherwise, first check the cache for existing data so that we can potentially
// revalidate the information without having to redownload everything.
cache->get(resource, [this, resource](std::unique_ptr<Response> response) {
std::shared_ptr<const Response> sharedResponse = std::move(response);
invoke([this, resource, sharedResponse] { processResult(resource, sharedResponse); });
cache->get(resource, [this, resource, loop](std::unique_ptr<Response> response) {
processResult(resource, std::move(response), loop);
});
}
}
@@ -132,7 +129,7 @@ void DefaultFileSource::Impl::processCancel(Request* req) {
req->destruct();
}
void DefaultFileSource::Impl::processResult(const Resource& resource, std::shared_ptr<const Response> response) {
void DefaultFileSource::Impl::processResult(const Resource& resource, std::shared_ptr<const Response> response, uv_loop_t* loop) {
SharedRequestBase *sharedRequest = find(resource);
if (sharedRequest) {
if (response) {
@@ -146,11 +143,11 @@ void DefaultFileSource::Impl::processResult(const Resource& resource, std::share
return;
} else {
// The cached response is stale. Now run the real request.
sharedRequest->start(loop().get(), response);
sharedRequest->start(loop, response);
}
} else {
// There is no response. Now run the real request.
sharedRequest->start(loop().get());
sharedRequest->start(loop);
}
} else {
// There is no request for this URL anymore. Likely, the request was canceled

12
src/mbgl/storage/default_file_source_impl.hpp
View File

@@ -2,7 +2,6 @@
#define MBGL_STORAGE_DEFAULT_DEFAULT_FILE_SOURCE_IMPL
#include <mbgl/storage/default_file_source.hpp>
#include <mbgl/util/run_loop.hpp>
#include <set>
#include <unordered_map>
@@ -11,26 +10,23 @@ namespace mbgl {
class SharedRequestBase;
class DefaultFileSource::Impl : public util::RunLoop {
friend class util::Thread<DefaultFileSource::Impl>;
class DefaultFileSource::Impl {
public:
Impl(FileCache *cache, const std::string &root = "");
public:
void notify(SharedRequestBase *sharedRequest, const std::set<Request *> &observers,
std::shared_ptr<const Response> response, FileCache::Hint hint);
SharedRequestBase *find(const Resource &resource);
void processAdd(Request* request);
void processAdd(Request* request, uv_loop_t* loop);
void processCancel(Request* request);
void processResult(const Resource& resource, std::shared_ptr<const Response> response);
void processAbort(const Environment& env);
public:
const std::string assetRoot;
private:
void processResult(const Resource& resource, std::shared_ptr<const Response> response, uv_loop_t* loop);
std::unordered_map<Resource, SharedRequestBase *, Resource::Hash> pending;
FileCache *cache = nullptr;
};

58
src/mbgl/util/run_loop.cpp
View File

@@ -1,54 +1,38 @@
#include <mbgl/util/run_loop.hpp>
#include <mbgl/util/uv_detail.hpp>
#include <mbgl/util/std.hpp>
#include <uv.h>
namespace /* anonymous */ {
inline void critical_section(std::mutex& mutex, std::function<void()> fn) {
std::lock_guard<std::mutex> lock(mutex);
fn();
}
}
namespace mbgl {
namespace util {
uv::tls<RunLoop> RunLoop::current;
RunLoop::RunLoop()
: runloop(util::make_unique<uv::loop>()),
runloopAsync(util::make_unique<uv::async>(runloop->get(), std::bind(&RunLoop::process, this))) {
: async(*loop, std::bind(&RunLoop::process, this)) {
}
// Define here since we can't destroy the uv::* objects from just the header file.
RunLoop::~RunLoop() = default;
void RunLoop::start() {
runloop->run();
}
void RunLoop::stop() {
critical_section(runloopMutex, [this] { runloopQueue.push(nullptr); });
runloopAsync->send();
void RunLoop::withMutex(std::function<void()>&& fn) {
std::lock_guard<std::mutex> lock(mutex);
fn();
}
void RunLoop::process() {
std::queue<std::function<void()>> queue;
critical_section(runloopMutex, [this, &queue] { queue.swap(runloopQueue); });
while (!queue.empty()) {
if (queue.front()) {
queue.front()();
} else {
runloopAsync->unref();
}
queue.pop();
Queue queue_;
withMutex([&] { queue_.swap(queue); });
while (!queue_.empty()) {
(queue_.front())();
queue_.pop();
}
}
void RunLoop::invoke(std::function<void()>&& fn) {
if (fn) {
critical_section(runloopMutex, [this, &fn] { runloopQueue.emplace(std::move(fn)); });
runloopAsync->send();
}
void RunLoop::run() {
assert(!current.get());
current.set(this);
loop.run();
current.set(nullptr);
}
void RunLoop::stop() {
invoke([&] { async.unref(); });
}
}

106
src/mbgl/util/run_loop.hpp
View File

@@ -1,53 +1,97 @@
#ifndef MBGL_UTIL_RUN_LOOP
#define MBGL_UTIL_RUN_LOOP
#include <memory>
#include <mutex>
#include <mbgl/util/uv_detail.hpp>
#include <functional>
#include <queue>
namespace uv {
class async;
class loop;
}
#include <mutex>
namespace mbgl {
namespace util {
template <typename T> class Thread;
class RunLoop {
friend Thread<RunLoop>;
protected:
// These are called by the Thread<> wrapper.
public:
RunLoop();
~RunLoop();
// Called by the Thread<> wrapper to start the loop. When you implement this
// method in a child class, you *must* call this function as the last action.
void start();
protected:
// Called by the Thread<> wrapper to terminate this loop.
void run();
void stop();
// Obtain the underlying loop object in case you want to attach additional listeners.
uv::loop& loop() { return *runloop; };
// Invoke fn() in the runloop thread.
template <class Fn>
void invoke(Fn&& fn) {
withMutex([&] { queue.push(Message(std::move(fn))); });
async.send();
}
// Invoke fn() in the runloop thread, then invoke callback(result) in the current thread.
template <class Fn, class R>
void invokeWithResult(Fn&& fn, std::function<void (R)> callback) {
RunLoop* outer = current.get();
assert(outer);
invoke([fn, callback, outer] {
/*
With C++14, we could write:
outer->invoke([callback, result = std::move(fn())] () mutable {
callback(std::move(result));
});
Instead we're using a workaround with std::bind
to obtain move-capturing semantics with C++11:
http://stackoverflow.com/a/12744730/52207
*/
outer->invoke(std::bind([callback] (R& result) {
callback(std::move(result));
}, std::move(fn())));
});
}
uv_loop_t* get() { return *loop; }
private:
// Invokes function in the run loop.
struct Message {
struct Base {
virtual void operator()() = 0;
virtual ~Base() = default;
};
template <class F>
struct Invoker : Base {
Invoker(F&& f) : func(std::move(f)) {}
void operator()() override { func(); }
F func;
};
Message() = default;
Message(Message&&) = default;
~Message() = default;
Message& operator=(Message&&) = default;
// copy members implicitly deleted
template <class Fn>
Message(Fn fn)
: p_fn(new Invoker<Fn>(std::move(fn))) {
}
void operator()() const { (*p_fn)(); }
std::unique_ptr<Base> p_fn;
};
using Queue = std::queue<Message>;
static uv::tls<RunLoop> current;
void withMutex(std::function<void()>&&);
void process();
public:
// Schedules a function to be executed as part of this run loop.
void invoke(std::function<void()>&& fn);
Queue queue;
std::mutex mutex;
private:
const std::unique_ptr<uv::loop> runloop;
const std::unique_ptr<uv::async> runloopAsync;
std::mutex runloopMutex;
std::queue<std::function<void()>> runloopQueue;
uv::loop loop;
uv::async async;
};
}

88
src/mbgl/util/thread.hpp
View File

@@ -3,6 +3,9 @@
#include <future>
#include <thread>
#include <functional>
#include <mbgl/util/run_loop.hpp>
namespace {
@@ -34,49 +37,80 @@ template <class Object>
class Thread {
public:
template <class... Args>
Thread(Args&&... args);
Thread(const std::string& name, Args&&... args);
~Thread();
// Invoke object->fn(args...) in the runloop thread.
template <typename Fn, class... Args>
void invoke(Fn fn, Args&&... args) {
loop->invoke(std::bind(fn, object, args...));
}
// Invoke object->fn(args...) in the runloop thread, then invoke callback(result) in the current thread.
template <typename Fn, class R, class... Args>
void invokeWithResult(Fn fn, std::function<void (R)> callback, Args&&... args) {
loop->invokeWithResult(std::bind(fn, object, args...), callback);
}
uv_loop_t* get() { return loop->get(); }
private:
Thread(const Thread&) = delete;
Thread(Thread&&) = delete;
Thread& operator=(const Thread&) = delete;
Thread& operator=(Thread&&) = delete;
~Thread();
inline Object* operator->() const { return &object; }
inline operator Object*() const { return &object; }
private:
template <typename P, std::size_t... I>
void run(std::promise<Object&>& promise, P&& params, index_sequence<I...>) {
Object context(std::get<I>(std::forward<P>(params))...);
promise.set_value(context);
context.start();
joinable.get_future().get();
}
void run(P&& params, index_sequence<I...>);
private:
std::thread thread;
std::promise<void> running;
std::promise<void> joinable;
Object& object;
std::thread thread;
Object* object;
RunLoop* loop;
};
template <class Object>
template <class... Args>
Thread<Object>::Thread(Args&&... args)
: object([&](std::tuple<Args...>&& params) -> Object& {
// Note: We're using std::tuple<> to store the arguments because GCC 4.9 has a bug
// when expanding parameters packs captured in lambdas.
std::promise<Object&> promise;
constexpr auto seq = typename integer_sequence<sizeof...(Args)>::type();
thread = std::thread([&] {
run(promise, std::move(params), seq);
});
return promise.get_future().get();
}(std::move(std::forward_as_tuple(::std::forward<Args>(args)...)))) {
Thread<Object>::Thread(const std::string& name, Args&&... args) {
// Note: We're using std::tuple<> to store the arguments because GCC 4.9 has a bug
// when expanding parameters packs captured in lambdas.
std::tuple<Args...> params = std::forward_as_tuple(::std::forward<Args>(args)...);
thread = std::thread([&] {
#ifdef __APPLE__
pthread_setname_np(name.c_str());
#else
(void(name));
#endif
constexpr auto seq = typename integer_sequence<sizeof...(Args)>::type();
run(std::move(params), seq);
});
running.get_future().get();
}
template <class Object>
template <typename P, std::size_t... I>
void Thread<Object>::run(P&& params, index_sequence<I...>) {
Object object_(std::get<I>(std::forward<P>(params))...);
object = &object_;
RunLoop loop_;
loop = &loop_;
running.set_value();
loop_.run();
joinable.get_future().get();
}
template <class Object>
Thread<Object>::~Thread() {
object.stop();
loop->stop();
joinable.set_value();
thread.join();
}

22
src/mbgl/util/uv.cpp
View File

@@ -3,6 +3,28 @@
#include <uv.h>
#if UV_VERSION_MAJOR == 0 && UV_VERSION_MINOR <= 10
int uv_key_create(uv_key_t* key) {
return -pthread_key_create(key, NULL);
}
void uv_key_delete(uv_key_t* key) {
if (pthread_key_delete(*key))
abort();
}
void* uv_key_get(uv_key_t* key) {
return pthread_getspecific(*key);
}
void uv_key_set(uv_key_t* key, void* value) {
if (pthread_setspecific(*key, value))
abort();
}
#endif
namespace uv {
std::string cwd() {

31
src/mbgl/util/uv_detail.hpp
View File

@@ -11,6 +11,21 @@
#include <memory>
#include <string>
#if UV_VERSION_MAJOR == 0 && UV_VERSION_MINOR <= 10
// Add thread local storage to libuv API:
// https://github.com/joyent/libuv/commit/5d2434bf71e47802841bad218d521fa254d1ca2d
typedef pthread_key_t uv_key_t;
UV_EXTERN int uv_key_create(uv_key_t* key);
UV_EXTERN void uv_key_delete(uv_key_t* key);
UV_EXTERN void* uv_key_get(uv_key_t* key);
UV_EXTERN void uv_key_set(uv_key_t* key, void* value);
#endif
namespace uv {
template <class T>
@@ -133,6 +148,22 @@ private:
uv_rwlock_t mtx;
};
template <class T>
class tls : public mbgl::util::noncopyable {
public:
inline tls() {
if (uv_key_create(&key) != 0) {
throw std::runtime_error("failed to initialize thread local storage key");
}
}
inline ~tls() { uv_key_delete(&key); }
inline T* get() { return reinterpret_cast<T*>(uv_key_get(&key)); }
inline void set(T* val) { uv_key_set(&key, val); }
private:
uv_key_t key;
};
}
#endif