Join worker tasks before destroying TileData

Fixes #1309
This commit is contained in:
John Firebaugh
2015-05-02 13:31:56 -04:00
parent e476fcbf3b
commit 5c7867f435
11 changed files with 111 additions and 40 deletions

View File

@@ -24,7 +24,11 @@ LiveTileData::LiveTileData(const TileID& id_,
state = State::loaded;
}
LiveTileData::~LiveTileData() {}
LiveTileData::~LiveTileData() {
// Cancel in most derived class destructor so that worker tasks are joined before
// any member data goes away.
cancel();
}
void LiveTileData::parse() {
if (state != State::loaded) {

View File

@@ -9,6 +9,9 @@ RasterTileData::RasterTileData(const TileID& id_, TexturePool &texturePool,
}
RasterTileData::~RasterTileData() {
// Cancel in most derived class destructor so that worker tasks are joined before
// any member data goes away.
cancel();
}
void RasterTileData::parse() {

View File

@@ -55,21 +55,11 @@ void TileData::cancel() {
env.cancelRequest(req);
req = nullptr;
}
if (workRequest) {
workRequest.join();
}
}
void TileData::reparse(Worker& worker, std::function<void()> callback) {
util::ptr<TileData> tile = shared_from_this();
worker.send(
[this]() {
EnvironmentScope scope(env, ThreadType::TileWorker, "TileWorker_" + name);
parse();
},
[tile, callback]() {
// `tile` is bound in this lambda to ensure that if it's the last owning pointer,
// destruction happens on the map thread, not the worker thread. It is _not_ bound
// in the above lambda, because we do not want the possibility to arise that the
// after callback could execute and release the penultimate reference before the
// work callback has been destructed.
callback();
});
workRequest = worker.send([this] { parse(); }, callback);
}

View File

@@ -7,6 +7,7 @@
#include <mbgl/util/noncopyable.hpp>
#include <mbgl/util/ptr.hpp>
#include <mbgl/util/work_request.hpp>
#include <atomic>
#include <string>
@@ -21,8 +22,7 @@ class StyleLayer;
class Request;
class Worker;
class TileData : public std::enable_shared_from_this<TileData>,
private util::noncopyable {
class TileData : private util::noncopyable {
public:
enum class State {
invalid,
@@ -60,6 +60,8 @@ protected:
Request *req = nullptr;
std::string data;
WorkRequest workRequest;
// Contains the tile ID string for painting debug information.
DebugFontBuffer debugFontBuffer;

View File

@@ -28,6 +28,9 @@ VectorTileData::VectorTileData(const TileID& id_,
}
VectorTileData::~VectorTileData() {
// Cancel in most derived class destructor so that worker tasks are joined before
// any member data goes away.
cancel();
glyphAtlas.removeGlyphs(reinterpret_cast<uintptr_t>(this));
}

View File

@@ -33,21 +33,10 @@ public:
RunLoop* outer = current.get();
assert(outer);
invoke([fn, callback, outer] {
/*
With C++14, we could write:
outer->invoke([cb = std::move(callback), result = std::move(fn())] () mutable {
cb(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([] (std::function<void (R)>& cb, R& result) {
cb(std::move(result));
}, std::move(callback), std::move(fn())));
invoke([fn = std::move(fn), callback = std::move(callback), outer] () mutable {
outer->invoke([callback = std::move(callback), result = std::move(fn())] () mutable {
callback(std::move(result));
});
});
}
@@ -57,7 +46,7 @@ public:
RunLoop* outer = current.get();
assert(outer);
invoke([fn, callback, outer] {
invoke([fn = std::move(fn), callback = std::move(callback), outer] () mutable {
fn();
outer->invoke(std::move(callback));
});

View File

@@ -56,13 +56,13 @@ public:
// 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...), std::move(callback));
loop->invokeWithResult(std::bind(fn, object, std::move(args)...), std::move(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...), std::move(callback));
loop->invokeWithResult(std::bind(fn, object, std::move(args)...), std::move(callback));
}
// Invoke object->fn(args...) in the runloop thread, and wait for the result.

View File

@@ -0,0 +1,34 @@
#include <mbgl/util/work_request.hpp>
namespace mbgl {
WorkRequest::WorkRequest() = default;
WorkRequest::WorkRequest(Future&& future)
: complete(std::move(future)) {
}
WorkRequest::WorkRequest(WorkRequest&& o)
: complete(std::move(o.complete)) {
}
WorkRequest::~WorkRequest() {
if (complete.valid()) {
complete.get();
}
}
WorkRequest& WorkRequest::operator=(WorkRequest&& o) {
complete = std::move(o.complete);
return *this;
}
WorkRequest::operator bool() const {
return complete.valid();
}
void WorkRequest::join() {
complete.get();
}
}

View File

@@ -0,0 +1,31 @@
#ifndef MBGL_UTIL_WORK_REQUEST
#define MBGL_UTIL_WORK_REQUEST
#include <mbgl/util/noncopyable.hpp>
#include <future>
namespace mbgl {
class WorkRequest : public util::noncopyable {
public:
using Future = std::future<void>;
WorkRequest();
WorkRequest(Future&&);
WorkRequest(WorkRequest&&);
~WorkRequest();
WorkRequest& operator=(WorkRequest&&);
operator bool() const;
// Wait for the worker task to complete.
void join();
private:
Future complete;
};
}
#endif

View File

@@ -2,6 +2,7 @@
#include <mbgl/platform/platform.hpp>
#include <cassert>
#include <future>
namespace mbgl {
@@ -9,8 +10,8 @@ class Worker::Impl {
public:
Impl(uv_loop_t*) {}
void doWork(Fn work) {
work();
void doWork(std::packaged_task<void ()>& task) {
task();
}
};
@@ -22,9 +23,14 @@ Worker::Worker(std::size_t count) {
Worker::~Worker() = default;
void Worker::send(Fn&& work, Fn&& after) {
threads[current]->invokeWithResult(&Worker::Impl::doWork, std::move(after), std::move(work));
WorkRequest Worker::send(Fn work, Fn after) {
std::packaged_task<void ()> task(work);
std::future<void> future = task.get_future();
threads[current]->invokeWithResult(&Worker::Impl::doWork, std::move(after), task);
current = (current + 1) % threads.size();
return WorkRequest(std::move(future));
}
}

View File

@@ -2,6 +2,7 @@
#define MBGL_UTIL_WORKER
#include <mbgl/util/noncopyable.hpp>
#include <mbgl/util/work_request.hpp>
#include <mbgl/util/thread.hpp>
#include <functional>
@@ -15,7 +16,15 @@ public:
Worker(std::size_t count);
~Worker();
void send(Fn&& work, Fn&& after);
// Request work be done on a thread pool. The optional after callback is
// executed on the invoking thread, which must have a run loop, after the
// work is complete.
//
// The return value represents the request to perform the work asynchronously.
// Its destructor guarantees that the work is either cancelled and will never
// execute, or has finished executing. In other words, the WorkRequest is
// guaranteed to outlive any references held by the work function.
WorkRequest send(Fn work, Fn after);
private:
class Impl;