diff --git a/src/mbgl/map/tile_data.cpp b/src/mbgl/map/tile_data.cpp index 4025bacfc..c6d87c7bd 100644 --- a/src/mbgl/map/tile_data.cpp +++ b/src/mbgl/map/tile_data.cpp @@ -4,6 +4,7 @@ #include #include +#include #include using namespace mbgl; @@ -56,9 +57,7 @@ void TileData::cancel() { env.cancelRequest(req); req = nullptr; } - if (workRequest) { - workRequest.join(); - } + workRequest.reset(); } bool TileData::mayStartParsing() { diff --git a/src/mbgl/map/tile_data.hpp b/src/mbgl/map/tile_data.hpp index 3be10dc03..ecc53a6b5 100644 --- a/src/mbgl/map/tile_data.hpp +++ b/src/mbgl/map/tile_data.hpp @@ -7,7 +7,6 @@ #include #include -#include #include #include @@ -21,6 +20,7 @@ class SourceInfo; class StyleLayer; class Request; class Worker; +class WorkRequest; class TileData : private util::noncopyable { public: @@ -79,7 +79,7 @@ protected: Request *req = nullptr; std::string data; - WorkRequest workRequest; + std::unique_ptr workRequest; // Contains the tile ID string for painting debug information. DebugFontBuffer debugFontBuffer; diff --git a/src/mbgl/util/work_request.cpp b/src/mbgl/util/work_request.cpp index a7d4b9c02..69e939728 100644 --- a/src/mbgl/util/work_request.cpp +++ b/src/mbgl/util/work_request.cpp @@ -1,38 +1,16 @@ #include +#include + +#include namespace mbgl { -WorkRequest::WorkRequest() = default; - -WorkRequest::WorkRequest(Future&& future, JoinedFlag flag) - : complete(std::move(future)), - joinedFlag(flag) { -} - -WorkRequest::WorkRequest(WorkRequest&& o) - : complete(std::move(o.complete)), - joinedFlag(std::move(o.joinedFlag)) { +WorkRequest::WorkRequest(Task task_) : task(task_) { + assert(task); } WorkRequest::~WorkRequest() { - if (complete.valid()) { - join(); - } -} - -WorkRequest& WorkRequest::operator=(WorkRequest&& o) { - complete = std::move(o.complete); - joinedFlag = std::move(o.joinedFlag); - return *this; -} - -WorkRequest::operator bool() const { - return complete.valid(); -} - -void WorkRequest::join() { - *joinedFlag = true; - complete.get(); + task->cancel(); } } diff --git a/src/mbgl/util/work_request.hpp b/src/mbgl/util/work_request.hpp index f0f461b5d..f2aa2bbac 100644 --- a/src/mbgl/util/work_request.hpp +++ b/src/mbgl/util/work_request.hpp @@ -3,29 +3,20 @@ #include -#include +#include namespace mbgl { +class WorkTask; + class WorkRequest : public util::noncopyable { public: - using Future = std::future; - using JoinedFlag = std::shared_ptr>; - - WorkRequest(); - WorkRequest(Future&&, JoinedFlag); - WorkRequest(WorkRequest&&); + using Task = std::shared_ptr; + WorkRequest(Task); ~WorkRequest(); - WorkRequest& operator=(WorkRequest&&); - operator bool() const; - - // Wait for the worker task to complete. - void join(); - private: - Future complete; - JoinedFlag joinedFlag; + std::shared_ptr task; }; } diff --git a/src/mbgl/util/work_task.cpp b/src/mbgl/util/work_task.cpp new file mode 100644 index 000000000..ebec420de --- /dev/null +++ b/src/mbgl/util/work_task.cpp @@ -0,0 +1,37 @@ +#include + +#include + +namespace mbgl { + +WorkTask::WorkTask(std::function task_, std::function 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 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 lock(mutex); + after = {}; +} + +} // end namespace mbgl diff --git a/src/mbgl/util/work_task.hpp b/src/mbgl/util/work_task.hpp new file mode 100644 index 000000000..f730a31c5 --- /dev/null +++ b/src/mbgl/util/work_task.hpp @@ -0,0 +1,27 @@ +#ifndef MBGL_UTIL_WORK_TASK +#define MBGL_UTIL_WORK_TASK + +#include + +#include +#include + +namespace mbgl { + +class WorkTask : private util::noncopyable { +public: + WorkTask(std::function task, std::function after); + + void runTask(); + void runAfter(); + void cancel(); + +private: + const std::function task; + std::function after; + std::mutex mutex; +}; + +} // end namespace mbgl + +#endif diff --git a/src/mbgl/util/worker.cpp b/src/mbgl/util/worker.cpp index ad26ea5b1..984c55c95 100644 --- a/src/mbgl/util/worker.cpp +++ b/src/mbgl/util/worker.cpp @@ -1,4 +1,6 @@ #include +#include +#include #include #include @@ -10,8 +12,8 @@ class Worker::Impl { public: Impl(uv_loop_t*) {} - void doWork(std::packaged_task& task) { - task(); + void doWork(std::shared_ptr& task) { + task->runTask(); } }; @@ -23,21 +25,16 @@ Worker::Worker(std::size_t count) { Worker::~Worker() = default; -WorkRequest Worker::send(Fn work, Fn after) { - std::packaged_task task(work); - std::future future = task.get_future(); +std::unique_ptr Worker::send(Fn work, Fn after) { + auto task = std::make_shared(work, after); + auto request = util::make_unique(task); - std::shared_ptr> joined = std::make_shared>(); - *joined = false; - - threads[current]->invokeWithResult(&Worker::Impl::doWork, [joined, after] { - if (!*joined) { - after(); - } + threads[current]->invokeWithResult(&Worker::Impl::doWork, [task] { + task->runAfter(); }, task); current = (current + 1) % threads.size(); - return WorkRequest(std::move(future), joined); + return request; } -} +} // end namespace mbgl diff --git a/src/mbgl/util/worker.hpp b/src/mbgl/util/worker.hpp index 461ba36e5..0c001e419 100644 --- a/src/mbgl/util/worker.hpp +++ b/src/mbgl/util/worker.hpp @@ -2,13 +2,15 @@ #define MBGL_UTIL_WORKER #include -#include #include #include +#include namespace mbgl { +class WorkRequest; + class Worker : public mbgl::util::noncopyable { public: using Fn = std::function; @@ -26,7 +28,7 @@ public: // Together, this means that an object may make a work request with lambdas which // bind references to itself, and if and when those lambdas execute, the references // will still be valid. - WorkRequest send(Fn work, Fn after); + std::unique_ptr send(Fn work, Fn after); private: class Impl; diff --git a/test/miscellaneous/worker.cpp b/test/miscellaneous/worker.cpp index 50fabcebc..3d3c781b8 100644 --- a/test/miscellaneous/worker.cpp +++ b/test/miscellaneous/worker.cpp @@ -1,6 +1,7 @@ #include "../fixtures/util.hpp" #include +#include #include using namespace mbgl; @@ -10,7 +11,7 @@ TEST(Worker, ExecutesWorkAndAfter) { RunLoop loop(uv_default_loop()); Worker worker(1); - WorkRequest request; + std::unique_ptr request; bool didWork = false; bool didAfter = false; @@ -29,19 +30,22 @@ TEST(Worker, ExecutesWorkAndAfter) { EXPECT_TRUE(didAfter); } -TEST(Worker, WorkRequestJoinWaitsForWorkToComplete) { +TEST(Worker, WorkRequestDeletionWaitsForWorkToComplete) { RunLoop loop(uv_default_loop()); Worker worker(1); + std::promise started; bool didWork = false; loop.invoke([&] { - WorkRequest request = worker.send([&] { + auto request = worker.send([&] { + started.set_value(); usleep(10000); didWork = true; }, [&] {}); + started.get_future().get(); - request.join(); + request.reset(); EXPECT_TRUE(didWork); loop.stop(); }); @@ -53,18 +57,48 @@ TEST(Worker, WorkRequestJoinCancelsAfter) { RunLoop loop(uv_default_loop()); Worker worker(1); + std::promise started; bool didAfter = false; loop.invoke([&] { - WorkRequest request = worker.send([&] { + auto request = worker.send([&] { + started.set_value(); }, [&] { didAfter = true; }); + started.get_future().get(); - request.join(); + request.reset(); loop.stop(); }); uv_run(uv_default_loop(), UV_RUN_DEFAULT); EXPECT_FALSE(didAfter); } + +TEST(Worker, WorkRequestCancelsImmediately) { + RunLoop loop(uv_default_loop()); + + Worker worker(1); + + loop.invoke([&] { + std::promise started; + // First worker item. + auto request1 = worker.send([&] { + usleep(10000); + started.set_value(); + }, [&] {}); + + auto request2 = worker.send([&] { + ADD_FAILURE() << "Second work item should not be invoked"; + }, [&] {}); + request2.reset(); + + started.get_future().get(); + request1.reset(); + + loop.stop(); + }); + + uv_run(uv_default_loop(), UV_RUN_DEFAULT); +}