Merge pull request #1470 from mapbox/1470-cancelable-work-requests

Make WorkRequests cancelable

src/mbgl/map/tile_data.cpp

@@ -4,6 +4,7 @@
 
 #include <mbgl/storage/file_source.hpp>
 #include <mbgl/util/worker.hpp>
+#include <mbgl/util/work_request.hpp>
 #include <mbgl/platform/log.hpp>
 
 using namespace mbgl;
@@ -56,9 +57,7 @@ void TileData::cancel() {
         env.cancelRequest(req);
         req = nullptr;
     }
-    if (workRequest) {
-        workRequest.join();
-    }
+    workRequest.reset();
 }
 
 bool TileData::mayStartParsing() {

src/mbgl/map/tile_data.hpp

@@ -7,7 +7,6 @@
 
 #include <mbgl/util/noncopyable.hpp>
 #include <mbgl/util/ptr.hpp>
-#include <mbgl/util/work_request.hpp>
 
 #include <atomic>
 #include <string>
@@ -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> workRequest;
 
     // Contains the tile ID string for painting debug information.
     DebugFontBuffer debugFontBuffer;

src/mbgl/util/work_request.cpp

@@ -1,38 +1,16 @@
 #include <mbgl/util/work_request.hpp>
+#include <mbgl/util/work_task.hpp>
+
+#include <cassert>
 
 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();
 }
 
 }

src/mbgl/util/work_request.hpp

@@ -3,29 +3,20 @@
 
 #include <mbgl/util/noncopyable.hpp>
 
-#include <future>
+#include <memory>
 
 namespace mbgl {
 
+class WorkTask;
+
 class WorkRequest : public util::noncopyable {
 public:
-    using Future = std::future<void>;
-    using JoinedFlag = std::shared_ptr<std::atomic<bool>>;
-
-    WorkRequest();
-    WorkRequest(Future&&, JoinedFlag);
-    WorkRequest(WorkRequest&&);
+    using Task = std::shared_ptr<WorkTask>;
+    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<WorkTask> task;
 };
 
 }

src/mbgl/util/work_task.cpp

@@ -0,0 +1,37 @@
+#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

src/mbgl/util/work_task.hpp

@@ -0,0 +1,27 @@
+#ifndef MBGL_UTIL_WORK_TASK
+#define MBGL_UTIL_WORK_TASK
+
+#include <mbgl/util/noncopyable.hpp>
+
+#include <functional>
+#include <mutex>
+
+namespace mbgl {
+
+class WorkTask : private util::noncopyable {
+public:
+    WorkTask(std::function<void()> task, std::function<void()> after);
+
+    void runTask();
+    void runAfter();
+    void cancel();
+
+private:
+    const std::function<void()> task;
+    std::function<void()> after;
+    std::mutex mutex;
+};
+
+} // end namespace mbgl
+
+#endif

src/mbgl/util/worker.cpp

@@ -1,4 +1,6 @@
 #include <mbgl/util/worker.hpp>
+#include <mbgl/util/work_task.hpp>
+#include <mbgl/util/work_request.hpp>
 #include <mbgl/platform/platform.hpp>
 
 #include <cassert>
@@ -10,8 +12,8 @@ class Worker::Impl {
 public:
     Impl(uv_loop_t*) {}
 
-    void doWork(std::packaged_task<void ()>& task) {
-        task();
+    void doWork(std::shared_ptr<WorkTask>& 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<void ()> task(work);
-    std::future<void> future = task.get_future();
+std::unique_ptr<WorkRequest> Worker::send(Fn work, Fn after) {
+    auto task = std::make_shared<WorkTask>(work, after);
+    auto request = util::make_unique<WorkRequest>(task);
 
-    std::shared_ptr<std::atomic<bool>> joined = std::make_shared<std::atomic<bool>>();
-    *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

src/mbgl/util/worker.hpp

@@ -2,13 +2,15 @@
 #define MBGL_UTIL_WORKER
 
 #include <mbgl/util/noncopyable.hpp>
-#include <mbgl/util/work_request.hpp>
 #include <mbgl/util/thread.hpp>
 
 #include <functional>
+#include <memory>
 
 namespace mbgl {
 
+class WorkRequest;
+
 class Worker : public mbgl::util::noncopyable {
 public:
     using Fn = std::function<void ()>;
@@ -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<WorkRequest> send(Fn work, Fn after);
 
 private:
     class Impl;

test/miscellaneous/worker.cpp

@@ -1,6 +1,7 @@
 #include "../fixtures/util.hpp"
 
 #include <mbgl/util/worker.hpp>
+#include <mbgl/util/work_request.hpp>
 #include <mbgl/util/run_loop.hpp>
 
 using namespace mbgl;
@@ -10,7 +11,7 @@ TEST(Worker, ExecutesWorkAndAfter) {
     RunLoop loop(uv_default_loop());
 
     Worker worker(1);
-    WorkRequest request;
+    std::unique_ptr<WorkRequest> 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<void> 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<void> 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<void> 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);
+}