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

Move WorkerTask functionality into RunLoop core

Browse Source 
As a byproduct, this makes FileCache get requests properly
cancelable.
This commit is contained in:
John Firebaugh
2015-06-08 18:40:11 -04:00
parent d596d9dc38
commit 5fadda53e5
11 changed files with 201 additions and 152 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
include/mbgl/storage/file_cache.hpp
View File

@@ -10,6 +10,7 @@ namespace mbgl {
struct Resource;
class Response;
class WorkRequest;
class FileCache : private util::noncopyable {
public:
@@ -18,7 +19,7 @@ public:
enum class Hint : uint8_t { Full, Refresh, No };
using Callback = std::function<void(std::unique_ptr<Response>)>;
virtual void get(const Resource &resource, Callback callback) = 0;
virtual std::unique_ptr<WorkRequest> get(const Resource &resource, Callback callback) = 0;
virtual void put(const Resource &resource, std::shared_ptr<const Response> response, Hint hint) = 0;
};

2
include/mbgl/storage/sqlite_cache.hpp
View File

@@ -17,7 +17,7 @@ public:
~SQLiteCache() override;
// FileCache API
void get(const Resource &resource, Callback callback) override;
std::unique_ptr<WorkRequest> get(const Resource &resource, Callback callback) override;
void put(const Resource &resource, std::shared_ptr<const Response> response, Hint hint) override;
class Impl;

4
platform/default/sqlite_cache.cpp
View File

@@ -126,12 +126,12 @@ void SQLiteCache::Impl::createSchema() {
}
}
void SQLiteCache::get(const Resource &resource, Callback callback) {
std::unique_ptr<WorkRequest> SQLiteCache::get(const Resource &resource, Callback callback) {
// Can be called from any thread, but most likely from the file source thread.
// 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->invokeWithResult(&Impl::get, std::move(callback), resource);
return thread->invokeWithResult(&Impl::get, std::move(callback), resource);
}
std::unique_ptr<Response> SQLiteCache::Impl::get(const Resource &resource) {

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

@@ -100,26 +100,16 @@ void DefaultFileSource::Impl::add(Request* req) {
request->observers.insert(req);
if (cache) {
startCacheRequest(resource);
startCacheRequest(request);
} else {
startRealRequest(resource);
startRealRequest(request);
}
}
void DefaultFileSource::Impl::startCacheRequest(const Resource& resource) {
void DefaultFileSource::Impl::startCacheRequest(DefaultFileRequest* request) {
// 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) {
DefaultFileRequest* request = find(resource);
if (!request || request->request) {
// There is no request for this URL anymore. Likely, the request was canceled
// before we got around to process the cache result.
// The second possibility is that a request has already been started by another cache
// request. In this case, we don't have to do anything either.
return;
}
request->cacheRequest = cache->get(request->resource, [this, request](std::unique_ptr<Response> response) {
auto expired = [&response] {
const int64_t now = std::chrono::duration_cast<std::chrono::seconds>(
SystemClock::now().time_since_epoch()).count();
@@ -128,7 +118,7 @@ void DefaultFileSource::Impl::startCacheRequest(const Resource& resource) {
if (!response || expired()) {
// No response or stale cache. Run the real request.
startRealRequest(resource, std::move(response));
startRealRequest(request, std::move(response));
} else {
// The response is fresh. We're good to notify the caller.
notify(request, std::move(response), FileCache::Hint::No);
@@ -136,17 +126,15 @@ void DefaultFileSource::Impl::startCacheRequest(const Resource& resource) {
});
}
void DefaultFileSource::Impl::startRealRequest(const Resource& resource, std::shared_ptr<const Response> response) {
DefaultFileRequest* request = find(resource);
void DefaultFileSource::Impl::startRealRequest(DefaultFileRequest* request, std::shared_ptr<const Response> response) {
auto callback = [request, this] (std::shared_ptr<const Response> res, FileCache::Hint hint) {
notify(request, res, hint);
};
if (algo::starts_with(resource.url, "asset://")) {
request->request = assetContext->createRequest(resource, callback, loop, assetRoot);
if (algo::starts_with(request->resource.url, "asset://")) {
request->realRequest = assetContext->createRequest(request->resource, callback, loop, assetRoot);
} else {
request->request = httpContext->createRequest(resource, callback, loop, response);
request->realRequest = httpContext->createRequest(request->resource, callback, loop, response);
}
}
@@ -158,8 +146,11 @@ void DefaultFileSource::Impl::cancel(Request* req) {
// cancel the request and remove it from the pending list.
request->observers.erase(req);
if (request->observers.empty()) {
if (request->request) {
request->request->cancel();
if (request->cacheRequest) {
request->cacheRequest.reset();
}
if (request->realRequest) {
request->realRequest->cancel();
}
pending.erase(request->resource);
}

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

@@ -17,7 +17,9 @@ class RequestBase;
struct DefaultFileRequest {
const Resource resource;
std::set<Request*> observers;
RequestBase* request = nullptr;
std::unique_ptr<WorkRequest> cacheRequest;
RequestBase* realRequest = nullptr;
inline DefaultFileRequest(const Resource& resource_)
: resource(resource_) {}
@@ -39,8 +41,8 @@ public:
private:
DefaultFileRequest* find(const Resource&);
void startCacheRequest(const Resource&);
void startRealRequest(const Resource&, std::shared_ptr<const Response> = nullptr);
void startCacheRequest(DefaultFileRequest*);
void startRealRequest(DefaultFileRequest*, std::shared_ptr<const Response> = nullptr);
void notify(DefaultFileRequest*, std::shared_ptr<const Response>, FileCache::Hint);
std::unordered_map<Resource, DefaultFileRequest, Resource::Hash> pending;

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

@@ -2,6 +2,8 @@
#define MBGL_UTIL_RUN_LOOP
#include <mbgl/util/noncopyable.hpp>
#include <mbgl/util/work_task.hpp>
#include <mbgl/util/work_request.hpp>
#include <mbgl/util/uv_detail.hpp>
#include <functional>
@@ -23,11 +25,48 @@ public:
template <class Fn, class... Args>
void invoke(Fn&& fn, Args&&... args) {
auto tuple = std::make_tuple(std::move(args)...);
auto invokable = std::make_unique<Invoker<Fn, decltype(tuple), Args...>>(std::move(fn), std::move(tuple));
withMutex([&] { queue.push(std::move(invokable)); });
auto task = std::make_shared<Invoker<Fn, decltype(tuple), Args...>>(
std::move(fn),
std::move(tuple));
withMutex([&] { queue.push(task); });
async.send();
}
// Invoke fn(args...) on this RunLoop, then invoke callback() on the current RunLoop.
template <class Fn, class... Args>
std::unique_ptr<WorkRequest>
invokeWithResult(Fn&& fn, std::function<void ()> callback, Args&&... args) {
auto tuple = std::make_tuple(std::move(args)...);
auto task = std::make_shared<Invoker<Fn, decltype(tuple), Args...>>(
std::move(fn),
std::move(tuple));
task->bind(callback);
withMutex([&] { queue.push(task); });
async.send();
return std::make_unique<WorkRequest>(task);
}
// Invoke fn(args...) on this RunLoop, then invoke callback(result) on the current RunLoop.
template <class R, class Fn, class... Args>
std::unique_ptr<WorkRequest>
invokeWithResult(Fn&& fn, std::function<void (R)> callback, Args&&... args) {
auto tuple = std::make_tuple(std::move(args)...);
auto task = std::make_shared<InvokerWithResult<Fn, decltype(tuple), R, Args...>>(
std::move(fn),
std::move(tuple));
task->bind(callback);
withMutex([&] { queue.push(task); });
async.send();
return std::make_unique<WorkRequest>(task);
}
// Return a function that invokes the given function on this RunLoop.
template <class... Args>
auto bind(std::function<void (Args...)> fn) {
@@ -36,57 +75,119 @@ public:
};
}
// Invoke fn(args...) on this RunLoop, then invoke callback(result) on the current RunLoop.
template <class R, class Fn, class... Args>
void invokeWithResult(Fn&& fn, std::function<void (R)> callback, Args&&... args) {
invoke([fn = std::move(fn), callback = current.get()->bind(callback)] (Args&&... a) mutable {
callback(fn(std::forward<Args>(a)...));
}, std::forward<Args>(args)...);
}
// Invoke fn(args...) on this RunLoop, then invoke callback() on the current RunLoop.
template <class Fn, class... Args>
void invokeWithResult(Fn&& fn, std::function<void ()> callback, Args&&... args) {
invoke([fn = std::move(fn), callback = current.get()->bind(callback)] (Args&&... a) mutable {
fn(std::forward<Args>(a)...);
callback();
}, std::forward<Args>(args)...);
}
uv_loop_t* get() { return async.get()->loop; }
static uv::tls<RunLoop> current;
private:
// A movable type-erasing invokable entity wrapper. This allows to store arbitrary invokable
// things (like std::function<>, or the result of a movable-only std::bind()) in the queue.
// Source: http://stackoverflow.com/a/29642072/331379
struct Message {
virtual void operator()() = 0;
virtual ~Message() = default;
};
template <class F, class P, class... Args>
struct Invoker : Message {
class Invoker : public WorkTask, public std::enable_shared_from_this<Invoker<F, P, Args...>> {
public:
Invoker(F&& f, P&& p)
: func(std::move(f)),
params(std::move(p)) {
}
void operator()() override {
invoke(std::index_sequence_for<Args...>{});
using C = std::function<void ()>;
void bind(C after) {
auto task = this->shared_from_this();
callback = RunLoop::current.get()->bind(C([task, after] {
if (!task->canceled) {
after();
}
}));
}
void operator()() override {
// We are only running the task when there's an after callback to call. This means that an
// empty after callback will be treated as a cancelled request. The mutex will be locked while
// processing so that the cancel() callback will block.
std::lock_guard<std::mutex> lock(mutex);
if (!canceled) {
invoke(std::index_sequence_for<Args...>{});
}
}
void cancel() override {
// Remove the after callback to indicate that this callback has been canceled. The mutex will
// block when the task is currently in progres. When the task has not begun yet, the runTask()
// method will not do anything. When the task has been completed already, and the after callback
// was run as well, this will also do nothing.
std::lock_guard<std::mutex> lock(mutex);
canceled = true;
}
private:
template <std::size_t... I>
void invoke(std::index_sequence<I...>) {
func(std::forward<Args>(std::get<I>(params))...);
func(std::forward<Args>(std::get<I>(params))...);
if (callback) {
callback();
}
}
std::mutex mutex;
bool canceled = false;
F func;
P params;
C callback;
};
using Queue = std::queue<std::unique_ptr<Message>>;
template <class F, class P, class R, class... Args>
class InvokerWithResult : public WorkTask, public std::enable_shared_from_this<InvokerWithResult<F, P, R, Args...>> {
public:
InvokerWithResult(F&& f, P&& p)
: func(std::move(f)),
params(std::move(p)) {
}
using C = std::function<void (R)>;
void bind(C after) {
auto task = this->shared_from_this();
callback = RunLoop::current.get()->bind(C([task, after] (R result) {
if (!task->canceled) {
after(std::move(result));
}
}));
}
void operator()() override {
// We are only running the task when there's an after callback to call. This means that an
// empty after callback will be treated as a cancelled request. The mutex will be locked while
// processing so that the cancel() callback will block.
std::lock_guard<std::mutex> lock(mutex);
if (!canceled) {
invoke(std::index_sequence_for<Args...>{});
}
}
void cancel() override {
// Remove the after callback to indicate that this callback has been canceled. The mutex will
// block when the task is currently in progres. When the task has not begun yet, the runTask()
// method will not do anything. When the task has been completed already, and the after callback
// was run as well, this will also do nothing.
std::lock_guard<std::mutex> lock(mutex);
canceled = true;
}
private:
template <std::size_t... I>
void invoke(std::index_sequence<I...>) {
callback(func(std::forward<Args>(std::get<I>(params))...));
}
std::mutex mutex;
bool canceled = false;
F func;
P params;
C callback;
};
using Queue = std::queue<std::shared_ptr<WorkTask>>;
void withMutex(std::function<void()>&&);
void process();

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

@@ -36,15 +36,17 @@ public:
}
// Invoke object->fn(args...) in the runloop thread, then invoke callback(result) in the current thread.
template <class R, typename Fn, class... Args>
void invokeWithResult(Fn fn, std::function<void (R)> callback, Args&&... args) {
loop->invokeWithResult(bind<Fn, Args...>(fn), callback, std::forward<Args>(args)...);
template <typename Fn, class... Args>
std::unique_ptr<WorkRequest>
invokeWithResult(Fn fn, std::function<void ()> callback, Args&&... args) {
return loop->invokeWithResult(bind<Fn, Args...>(fn), callback, std::forward<Args>(args)...);
}
// 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(bind<Fn, Args...>(fn), callback, std::forward<Args>(args)...);
// Invoke object->fn(args...) in the runloop thread, then invoke callback(result) in the current thread.
template <class R, typename Fn, class... Args>
std::unique_ptr<WorkRequest>
invokeWithResult(Fn fn, std::function<void (R)> callback, Args&&... args) {
return loop->invokeWithResult(bind<Fn, Args...>(fn), callback, std::forward<Args>(args)...);
}
// Invoke object->fn(args...) in the runloop thread, and wait for the result.

37
src/mbgl/util/work_task.cpp
View File

@@ -1,37 +0,0 @@
#include <mbgl/util/work_task.hpp>
#include <cassert>
namespace mbgl {
WorkTask::WorkTask(std::function<void()> task_, std::function<void()> after_)
: task(task_), after(after_) {
assert(after);
}
void WorkTask::runTask() {
// We are only running the task when there's an after callback to call. This means that an
// empty after callback will be treated as a cancelled request. The mutex will be locked while
// processing so that the cancel() callback will block.
std::lock_guard<std::mutex> lock(mutex);
if (after) {
task();
}
}
void WorkTask::runAfter() {
if (after) {
after();
}
}
void WorkTask::cancel() {
// Remove the after callback to indicate that this callback has been canceled. The mutex will
// block when the task is currently in progres. When the task has not begun yet, the runTask()
// method will not do anything. When the task has been completed already, and the after callback
// was run as well, this will also do nothing.
std::lock_guard<std::mutex> lock(mutex);
after = {};
}
} // end namespace mbgl

20
src/mbgl/util/work_task.hpp
View File

@@ -3,25 +3,19 @@
#include <mbgl/util/noncopyable.hpp>
#include <functional>
#include <mutex>
namespace mbgl {
// A movable type-erasing function wrapper. This allows to store arbitrary invokable
// things (like std::function<>, or the result of a movable-only std::bind()) in the queue.
// Source: http://stackoverflow.com/a/29642072/331379
class WorkTask : private util::noncopyable {
public:
WorkTask(std::function<void()> task, std::function<void()> after);
virtual ~WorkTask() = default;
void runTask();
void runAfter();
void cancel();
private:
const std::function<void()> task;
std::function<void()> after;
std::mutex mutex;
virtual void operator()() = 0;
virtual void cancel() = 0;
};
} // end namespace mbgl
}
#endif

30
src/mbgl/util/worker.cpp
View File

@@ -10,41 +10,29 @@ namespace mbgl {
class Worker::Impl {
public:
Impl(uv_loop_t*) {}
Impl(uv_loop_t*, FileSource* fs) {
// FIXME: Workers should not access the FileSource but it
// is currently needed because of GlyphsPBF. See #1664.
util::ThreadContext::setFileSource(fs);
}
void doWork(std::shared_ptr<WorkTask>& task) {
task->runTask();
void doWork(Fn work) {
work();
}
};
Worker::Worker(std::size_t count) {
util::ThreadContext context = {"Worker", util::ThreadType::Worker, util::ThreadPriority::Low};
for (std::size_t i = 0; i < count; i++) {
std::unique_ptr<util::Thread<Impl>> worker(new util::Thread<Impl>(context));
// FIXME: Workers should not access the FileSource but it
// is currently needed because of GlyphsPBF. See #1664.
auto task = std::make_shared<WorkTask>([fs = util::ThreadContext::getFileSource()]{
util::ThreadContext::setFileSource(fs);
}, []{});
worker->invoke(&Worker::Impl::doWork, task);
threads.emplace_back(std::move(worker));
threads.emplace_back(std::make_unique<util::Thread<Impl>>(context, util::ThreadContext::getFileSource()));
}
}
Worker::~Worker() = default;
std::unique_ptr<WorkRequest> Worker::send(Fn work, Fn after) {
auto task = std::make_shared<WorkTask>(work, after);
auto request = std::make_unique<WorkRequest>(task);
threads[current]->invokeWithResult(&Worker::Impl::doWork, [task] {
task->runAfter();
}, task);
current = (current + 1) % threads.size();
return request;
return threads[current]->invokeWithResult(&Worker::Impl::doWork, after, work);
}
} // end namespace mbgl

31
test/miscellaneous/thread.cpp
View File

@@ -67,40 +67,46 @@ TEST(Thread, invoke) {
const std::thread::id tid = std::this_thread::get_id();
RunLoop loop(uv_default_loop());
std::vector<std::unique_ptr<mbgl::WorkRequest>> requests;
loop.invoke([&] {
EXPECT_EQ(tid, std::this_thread::get_id());
Thread<TestObject> thread({"Test", ThreadType::Map, ThreadPriority::Regular}, tid);
thread.invoke(&TestObject::fn1, 1);
thread.invokeWithResult<int>(&TestObject::fn2, [&] (int result) {
requests.push_back(thread.invokeWithResult<int>(&TestObject::fn2, [&] (int result) {
EXPECT_EQ(tid, std::this_thread::get_id());
EXPECT_EQ(result, 1);
});
}));
thread.invoke(&TestObject::transferIn, std::make_unique<int>(1));
thread.invokeWithResult<std::unique_ptr<int>>(&TestObject::transferOut, [&] (std::unique_ptr<int> result) {
requests.push_back(thread.invokeWithResult<std::unique_ptr<int>>(&TestObject::transferOut, [&] (std::unique_ptr<int> result) {
EXPECT_EQ(tid, std::this_thread::get_id());
EXPECT_EQ(*result, 1);
});
}));
thread.invokeWithResult<std::unique_ptr<int>>(&TestObject::transferInOut, [&] (std::unique_ptr<int> result) {
requests.push_back(thread.invokeWithResult<std::unique_ptr<int>>(&TestObject::transferInOut, [&] (std::unique_ptr<int> result) {
EXPECT_EQ(tid, std::this_thread::get_id());
EXPECT_EQ(*result, 1);
}, std::make_unique<int>(1));
}, std::make_unique<int>(1)));
thread.invoke(&TestObject::transferInShared, std::make_shared<int>(1));
thread.invokeWithResult<std::shared_ptr<int>>(&TestObject::transferOutShared, [&] (std::shared_ptr<int> result) {
requests.push_back(thread.invokeWithResult<std::shared_ptr<int>>(&TestObject::transferOutShared, [&] (std::shared_ptr<int> result) {
EXPECT_EQ(tid, std::this_thread::get_id());
EXPECT_EQ(*result, 1);
});
}));
// Cancelled request
thread.invokeWithResult(&TestObject::fn1, [&] {
ADD_FAILURE();
}, 1);
std::string test("test");
thread.invokeWithResult<std::string>(&TestObject::transferString, [&] (std::string result){
requests.push_back(thread.invokeWithResult<std::string>(&TestObject::transferString, [&] (std::string result){
EXPECT_EQ(tid, std::this_thread::get_id());
EXPECT_EQ(result, "test");
loop.stop();
}, test);
}, test));
test.clear();
});
@@ -117,14 +123,15 @@ TEST(Thread, context) {
const std::thread::id tid = std::this_thread::get_id();
RunLoop loop(uv_default_loop());
std::vector<std::unique_ptr<mbgl::WorkRequest>> requests;
loop.invoke([&] {
Thread<TestObject> thread({"Test", ThreadType::Worker, ThreadPriority::Low}, tid);
thread.invokeWithResult<bool>(&TestObject::checkContext, [&] (bool inTestThreadContext) {
requests.push_back(thread.invokeWithResult<bool>(&TestObject::checkContext, [&] (bool inTestThreadContext) {
EXPECT_EQ(inTestThreadContext, true);
loop.stop();
});
}));
});
uv_run(uv_default_loop(), UV_RUN_DEFAULT);