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

Simplify RunLoop

Browse Source 
Instead of transforming between return value and callback, pass
the wrapped callback on to the invoked function as the last argument.

This eliminates the need for multiple invokeWithArgument overloads
and permits invoked functions to be asynchronous.
This commit is contained in:
John Firebaugh
2015-06-11 19:03:19 -04:00
parent 3ab682e627
commit fc64e97bce
8 changed files with 149 additions and 216 deletions
Download Patch File Download Diff File Expand all files Collapse all files

14
src/mbgl/map/source.cpp
View File

@@ -233,7 +233,9 @@ bool Source::handlePartialTile(const TileID& id, Worker&) {
return true;
}
util::ptr<TileData> data = it->second.lock();
// Note: this uses a raw pointer; we don't want the callback binding to have a
// shared pointer.
VectorTileData* data = static_cast<VectorTileData*>(it->second.lock().get());
if (!data) {
return true;
}
@@ -241,14 +243,12 @@ bool Source::handlePartialTile(const TileID& id, Worker&) {
// The signal is only emitted if there was an actual change on the tile. The
// tile can be in a "partial" state waiting for resources and get reparsed on
// the arrival of new resources that were needed by another tile.
size_t bucketCount = static_cast<VectorTileData*>(data.get())->countBuckets();
auto callback = [this, data, bucketCount]() {
if (static_cast<VectorTileData*>(data.get())->countBuckets() > bucketCount) {
size_t bucketCount = data->countBuckets();
return data->reparse([this, data, bucketCount]() {
if (data->countBuckets() > bucketCount) {
emitTileLoaded(false);
}
};
return data->reparse(callback);
});
}
TileData::State Source::addTile(MapData& data,

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

@@ -25,7 +25,7 @@ public:
template <class Fn, class... Args>
void invoke(Fn&& fn, Args&&... args) {
auto tuple = std::make_tuple(std::move(args)...);
auto task = std::make_shared<Invoker<Fn, decltype(tuple), Args...>>(
auto task = std::make_shared<Invoker<Fn, decltype(tuple)>>(
std::move(fn),
std::move(tuple));
@@ -33,44 +33,24 @@ public:
async.send();
}
// Invoke fn(args...) on this RunLoop, then invoke callback() on the current RunLoop.
template <class Fn, class... Args>
// Invoke fn(args...) on this RunLoop, then invoke callback(results...) on the current RunLoop.
template <class Fn, class Cb, 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...>>(
invokeWithCallback(Fn&& fn, Cb&& callback, Args&&... args) {
auto flag = std::make_shared<bool>();
*flag = false;
auto after = RunLoop::current.get()->bind([flag, callback] (auto&&... results) {
if (!*flag) {
callback(std::move(results)...);
}
});
auto tuple = std::make_tuple(std::move(args)..., after);
auto task = std::make_shared<Invoker<Fn, decltype(tuple)>>(
std::move(fn),
std::move(tuple));
// `task` is a shared pointer with ownership in the following three places:
// 1. In the `queue` of pending invocations.
// 2. In the `WorkRequest` result.
// 3. In the lambda binding of the callback to be executed on the invoking
// RunLoop. This last shared ownership is necessary in the case where
// callback execution has been scheduled (queued on the invoking RunLoop),
// but the other two places have released ownership -- i.e. the task was
// cancelled after the work is completed, but before the callback is
// executed. In this case, the lambda binding checks the cancellation flag
// and does not execute the original callback.
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);
std::move(tuple),
flag);
withMutex([&] { queue.push(task); });
async.send();
@@ -79,10 +59,11 @@ public:
}
// Return a function that invokes the given function on this RunLoop.
template <class... Args>
auto bind(std::function<void (Args...)> fn) {
return [this, fn = std::move(fn)] (Args&&... args) {
invoke(std::move(fn), std::move(args)...);
template <class Fn>
auto bind(Fn&& fn) {
return [this, fn = std::move(fn)] (auto&&... args) {
// `this->` is a workaround for https://gcc.gnu.org/bugzilla/show_bug.cgi?id=61636
this->invoke(std::move(fn), std::move(args)...);
};
}
@@ -91,111 +72,47 @@ public:
static uv::tls<RunLoop> current;
private:
template <class F, class P, class... Args>
class Invoker : public WorkTask, public std::enable_shared_from_this<Invoker<F, P, Args...>> {
template <class F, class P>
class Invoker : public WorkTask {
public:
Invoker(F&& f, P&& p)
: func(std::move(f)),
Invoker(F&& f, P&& p, std::shared_ptr<bool> canceled_ = nullptr)
: canceled(canceled_),
func(std::move(f)),
params(std::move(p)) {
}
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.
// Lock the mutex while processing so that cancel() will block.
std::lock_guard<std::mutex> lock(mutex);
if (!canceled) {
invoke(std::index_sequence_for<Args...>{});
if (!canceled || !*canceled) {
invoke(std::make_index_sequence<std::tuple_size<P>::value>{});
}
}
// If the task has not yet begun, this will cancel it.
// If the task is in progress, this will block until it completed. (Currently
// necessary because of shared state, but should be removed.) It will also
// cancel the after callback.
// If the task has completed, but the after callback has not executed, this
// will cancel the after callback.
// If the task has completed and the after callback has executed, this will
// do nothing.
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;
*canceled = true;
}
private:
template <std::size_t... I>
void invoke(std::index_sequence<I...>) {
func(std::forward<Args>(std::get<I>(params))...);
if (callback) {
callback();
}
func(std::get<I>(std::forward<P>(params))...);
}
std::mutex mutex;
bool canceled = false;
std::shared_ptr<bool> canceled;
F func;
P params;
C callback;
};
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>>;

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

@@ -32,21 +32,14 @@ public:
// Invoke object->fn(args...) in the runloop thread.
template <typename Fn, class... Args>
void invoke(Fn fn, Args&&... args) {
loop->invoke(bind<Fn, Args...>(fn), std::forward<Args>(args)...);
loop->invoke(bind(fn), std::forward<Args>(args)...);
}
// Invoke object->fn(args...) in the runloop thread, then invoke callback(result) in the current thread.
template <typename Fn, class... Args>
template <typename Fn, class Cb, 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(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)...);
invokeWithCallback(Fn fn, Cb&& callback, Args&&... args) {
return loop->invokeWithCallback(bind(fn), callback, std::forward<Args>(args)...);
}
// Invoke object->fn(args...) in the runloop thread, and wait for the result.
@@ -73,9 +66,11 @@ private:
Thread& operator=(const Thread&) = delete;
Thread& operator=(Thread&&) = delete;
template <typename Fn, class... Args>
template <typename Fn>
auto bind(Fn fn) {
return [fn, this] (Args&&... a) { return (object->*fn)(std::forward<Args>(a)...); };
return [fn, this] (auto &&... args) {
return (object->*fn)(std::forward<decltype(args)>(args)...);
};
}
template <typename P, std::size_t... I>

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

@@ -20,25 +20,30 @@ public:
util::ThreadContext::setFileSource(fs);
}
bool parseRasterTile(RasterBucket* bucket, std::string data) {
return bucket->setImage(data);
void parseRasterTile(RasterBucket* bucket, std::string data, std::function<void (bool)> callback) {
callback(bucket->setImage(data));
}
TileParseResult parseVectorTile(TileWorker* worker, std::string data) {
void parseVectorTile(TileWorker* worker, std::string data, std::function<void (TileParseResult)> callback) {
try {
pbf tilePBF(reinterpret_cast<const unsigned char *>(data.data()), data.size());
return worker->parse(VectorTile(tilePBF));
callback(worker->parse(VectorTile(tilePBF)));
} catch (const std::exception& ex) {
return TileParseResult(ex.what());
callback(TileParseResult(ex.what()));
}
}
TileParseResult parseLiveTile(TileWorker* worker, const LiveTile* tile) {
return worker->parse(*tile);
void parseLiveTile(TileWorker* worker, const LiveTile* tile, std::function<void (TileParseResult)> callback) {
try {
callback(worker->parse(*tile));
} catch (const std::exception& ex) {
callback(TileParseResult(ex.what()));
}
}
void redoPlacement(TileWorker* worker, float angle, bool collisionDebug) {
void redoPlacement(TileWorker* worker, float angle, bool collisionDebug, std::function<void ()> callback) {
worker->redoPlacement(angle, collisionDebug);
callback();
}
};
@@ -53,22 +58,22 @@ Worker::~Worker() = default;
std::unique_ptr<WorkRequest> Worker::parseRasterTile(RasterBucket& bucket, std::string data, std::function<void (bool)> callback) {
current = (current + 1) % threads.size();
return threads[current]->invokeWithResult(&Worker::Impl::parseRasterTile, callback, &bucket, data);
return threads[current]->invokeWithCallback(&Worker::Impl::parseRasterTile, callback, &bucket, data);
}
std::unique_ptr<WorkRequest> Worker::parseVectorTile(TileWorker& worker, std::string data, std::function<void (TileParseResult)> callback) {
current = (current + 1) % threads.size();
return threads[current]->invokeWithResult(&Worker::Impl::parseVectorTile, callback, &worker, data);
return threads[current]->invokeWithCallback(&Worker::Impl::parseVectorTile, callback, &worker, data);
}
std::unique_ptr<WorkRequest> Worker::parseLiveTile(TileWorker& worker, const LiveTile& tile, std::function<void (TileParseResult)> callback) {
current = (current + 1) % threads.size();
return threads[current]->invokeWithResult(&Worker::Impl::parseLiveTile, callback, &worker, &tile);
return threads[current]->invokeWithCallback(&Worker::Impl::parseLiveTile, callback, &worker, &tile);
}
std::unique_ptr<WorkRequest> Worker::redoPlacement(TileWorker& worker, float angle, bool collisionDebug, std::function<void ()> callback) {
current = (current + 1) % threads.size();
return threads[current]->invokeWithResult(&Worker::Impl::redoPlacement, callback, &worker, angle, collisionDebug);
return threads[current]->invokeWithCallback(&Worker::Impl::redoPlacement, callback, &worker, angle, collisionDebug);
}
} // end namespace mbgl