[core] Use an actor model for tile worker concurrency

5 files changed, 201 insertions, 25 deletions

diff --git a/test/actor/actor.cpp b/test/actor/actor.cpp
new file mode 100644
index 0000000000..7bb76784d6
--- /dev/null
+++ b/test/actor/actor.cpp
@@ -0,0 +1,139 @@
+#include <mbgl/actor/actor.hpp>
+#include <mbgl/actor/thread_pool.hpp>
+
+#include <mbgl/test/util.hpp>
+
+#include <chrono>
+#include <functional>
+#include <future>
+
+using namespace mbgl;
+using namespace std::chrono_literals;
+
+TEST(Actor, Construction) {
+    // Construction is currently synchronous. It may become asynchronous in the future.
+
+    struct Test {
+        Test(ActorRef<Test>, bool& constructed) {
+            constructed = true;
+        };
+    };
+
+    ThreadPool pool { 1 };
+    bool constructed = false;
+    Actor<Test> test(pool, std::ref(constructed));
+
+    EXPECT_TRUE(constructed);
+}
+
+TEST(Actor, DestructionClosesMailbox) {
+    // Destruction blocks until the actor is not receiving.
+
+    struct Test {
+        std::promise<void> promise;
+        std::future<void> future;
+        std::atomic<bool> waited;
+
+        Test(ActorRef<Test>, std::promise<void> promise_, std::future<void> future_)
+            : promise(std::move(promise_)),
+              future(std::move(future_)),
+              waited(false) {
+        }
+
+        ~Test() {
+            EXPECT_TRUE(waited.load());
+        }
+
+        void wait() {
+            promise.set_value();
+            future.wait();
+            std::this_thread::sleep_for(1ms);
+            waited = true;
+        }
+    };
+
+    ThreadPool pool { 1 };
+
+    std::promise<void> enteredPromise;
+    std::future<void> enteredFuture = enteredPromise.get_future();
+
+    std::promise<void> exitingPromise;
+    std::future<void> exitingFuture = exitingPromise.get_future();
+
+    Actor<Test> test(pool, std::move(enteredPromise), std::move(exitingFuture));
+
+    test.invoke(&Test::wait);
+    enteredFuture.wait();
+    exitingPromise.set_value();
+}
+
+TEST(Actor, OrderedMailbox) {
+    // Messages are processed in order.
+
+    struct Test {
+        int last = 0;
+        std::promise<void> promise;
+
+        Test(ActorRef<Test>, std::promise<void> promise_)
+            : promise(std::move(promise_))  {
+        }
+
+        void receive(int i) {
+            EXPECT_EQ(i, last + 1);
+            last = i;
+        }
+
+        void end() {
+            promise.set_value();
+        }
+    };
+
+    ThreadPool pool { 1 };
+
+    std::promise<void> endedPromise;
+    std::future<void> endedFuture = endedPromise.get_future();
+    Actor<Test> test(pool, std::move(endedPromise));
+
+    for (auto i = 1; i <= 10; ++i) {
+        test.invoke(&Test::receive, i);
+    }
+
+    test.invoke(&Test::end);
+    endedFuture.wait();
+}
+
+TEST(Actor, NonConcurrentMailbox) {
+    // An individual actor is never itself concurrent.
+
+    struct Test {
+        int last = 0;
+        std::promise<void> promise;
+
+        Test(ActorRef<Test>, std::promise<void> promise_)
+            : promise(std::move(promise_))  {
+        }
+
+        void receive(int i) {
+            EXPECT_EQ(i, last + 1);
+            last = i;
+            std::this_thread::sleep_for(1ms);
+        }
+
+        void end() {
+            promise.set_value();
+        }
+    };
+
+    ThreadPool pool { 10 };
+
+    std::promise<void> endedPromise;
+    std::future<void> endedFuture = endedPromise.get_future();
+    Actor<Test> test(pool, std::move(endedPromise));
+
+    for (auto i = 1; i <= 10; ++i) {
+        test.invoke(&Test::receive, i);
+    }
+
+    test.invoke(&Test::end);
+    endedFuture.wait();
+}
diff --git a/test/actor/actor_ref.cpp b/test/actor/actor_ref.cpp
new file mode 100644
index 0000000000..655529035f
--- /dev/null
+++ b/test/actor/actor_ref.cpp
@@ -0,0 +1,42 @@
+#include <mbgl/actor/actor.hpp>
+#include <mbgl/actor/thread_pool.hpp>
+
+#include <mbgl/test/util.hpp>
+
+#include <chrono>
+#include <functional>
+#include <future>
+
+using namespace mbgl;
+using namespace std::chrono_literals;
+
+TEST(ActorRef, CanOutliveActor) {
+    // An ActorRef can outlive its actor. Doing does not extend the actor's lifetime.
+    // Sending a message to an ActorRef whose actor has died is a no-op.
+
+    struct Test {
+        bool& died;
+
+        Test(ActorRef<Test>, bool& died_)
+            : died(died_) {
+        }
+
+        ~Test() {
+            died = true;
+        }
+
+        void receive() {
+            FAIL();
+        }
+    };
+
+    ThreadPool pool { 1 };
+    bool died = false;
+
+    ActorRef<Test> test = [&] () {
+        return Actor<Test>(pool, std::ref(died)).self();
+    }();
+
+    EXPECT_TRUE(died);
+    test.invoke(&Test::receive);
+}
diff --git a/test/src/mbgl/test/stub_style_observer.hpp b/test/src/mbgl/test/stub_style_observer.hpp
index 4de3430fcc..aa780121f5 100644
--- a/test/src/mbgl/test/stub_style_observer.hpp
+++ b/test/src/mbgl/test/stub_style_observer.hpp
@@ -34,19 +34,15 @@ public:
         if (sourceError) sourceError(source, error);
     }
 
-    void onTileLoaded(Source& source, const OverscaledTileID& tileID, TileLoadState loadState) override {
-        if (tileLoaded) tileLoaded(source, tileID, loadState);
-    }
+    void onTileChanged(Source& source, const OverscaledTileID& tileID) override {
+        if (tileChanged) tileChanged(source, tileID);
+    };
 
     void
     onTileError(Source& source, const OverscaledTileID& tileID, std::exception_ptr error) override {
         if (tileError) tileError(source, tileID, error);
     }
 
-    void onTileUpdated(Source& source, const OverscaledTileID& tileID) override {
-        if (tileUpdated) tileUpdated(source, tileID);
-    };
-
     void onResourceError(std::exception_ptr error) override {
         if (resourceError) resourceError(error);
     };
@@ -57,8 +53,7 @@ public:
     std::function<void (std::exception_ptr)> spriteError;
     std::function<void (Source&)> sourceLoaded;
     std::function<void (Source&, std::exception_ptr)> sourceError;
-    std::function<void (Source&, const OverscaledTileID&, TileLoadState)> tileLoaded;
+    std::function<void (Source&, const OverscaledTileID&)> tileChanged;
     std::function<void (Source&, const OverscaledTileID&, std::exception_ptr)> tileError;
-    std::function<void (Source&, const OverscaledTileID&)> tileUpdated;
     std::function<void (std::exception_ptr)> resourceError;
 };
diff --git a/test/style/source.cpp b/test/style/source.cpp
index a33467dedb..519ca9288e 100644
--- a/test/style/source.cpp
+++ b/test/style/source.cpp
@@ -11,10 +11,10 @@
 #include <mbgl/util/string.hpp>
 #include <mbgl/util/io.hpp>
 #include <mbgl/util/tileset.hpp>
+#include <mbgl/actor/thread_pool.hpp>
 #include <mbgl/platform/log.hpp>
 
 #include <mbgl/map/transform.hpp>
-#include <mbgl/util/worker.hpp>
 #include <mbgl/style/style.hpp>
 #include <mbgl/style/update_parameters.hpp>
 #include <mbgl/style/layers/line_layer.hpp>
@@ -31,7 +31,7 @@ public:
     StubStyleObserver observer;
     Transform transform;
     TransformState transformState;
-    Worker worker { 1 };
+    ThreadPool threadPool { 1 };
     AnnotationManager annotationManager { 1.0 };
     style::Style style { fileSource, 1.0 };
 
@@ -39,9 +39,8 @@ public:
         1.0,
         MapDebugOptions(),
         transformState,
-        worker,
+        threadPool,
         fileSource,
-        true,
         MapMode::Continuous,
         annotationManager,
         style
@@ -123,7 +122,7 @@ TEST(Source, RasterTileEmpty) {
         return response;
     };
 
-    test.observer.tileLoaded = [&] (Source& source, const OverscaledTileID&, TileLoadState) {
+    test.observer.tileChanged = [&] (Source& source, const OverscaledTileID&) {
         EXPECT_EQ("source", source.getID());
         test.end();
     };
@@ -138,7 +137,7 @@ TEST(Source, RasterTileEmpty) {
     RasterSource source("source", tileset, 512);
     source.baseImpl->setObserver(&test.observer);
     source.baseImpl->loadDescription(test.fileSource);
-    source.baseImpl->loadTiles(test.updateParameters);
+    source.baseImpl->updateTiles(test.updateParameters);
 
     test.run();
 }
@@ -152,7 +151,7 @@ TEST(Source, VectorTileEmpty) {
         return response;
     };
 
-    test.observer.tileLoaded = [&] (Source& source, const OverscaledTileID&, TileLoadState) {
+    test.observer.tileChanged = [&] (Source& source, const OverscaledTileID&) {
         EXPECT_EQ("source", source.getID());
         test.end();
     };
@@ -167,7 +166,7 @@ TEST(Source, VectorTileEmpty) {
     VectorSource source("source", tileset);
     source.baseImpl->setObserver(&test.observer);
     source.baseImpl->loadDescription(test.fileSource);
-    source.baseImpl->loadTiles(test.updateParameters);
+    source.baseImpl->updateTiles(test.updateParameters);
 
     test.run();
 }
@@ -196,7 +195,7 @@ TEST(Source, RasterTileFail) {
     RasterSource source("source", tileset, 512);
     source.baseImpl->setObserver(&test.observer);
     source.baseImpl->loadDescription(test.fileSource);
-    source.baseImpl->loadTiles(test.updateParameters);
+    source.baseImpl->updateTiles(test.updateParameters);
 
     test.run();
 }
@@ -225,7 +224,7 @@ TEST(Source, VectorTileFail) {
     VectorSource source("source", tileset);
     source.baseImpl->setObserver(&test.observer);
     source.baseImpl->loadDescription(test.fileSource);
-    source.baseImpl->loadTiles(test.updateParameters);
+    source.baseImpl->updateTiles(test.updateParameters);
 
     test.run();
 }
@@ -253,7 +252,7 @@ TEST(Source, RasterTileCorrupt) {
     RasterSource source("source", tileset, 512);
     source.baseImpl->setObserver(&test.observer);
     source.baseImpl->loadDescription(test.fileSource);
-    source.baseImpl->loadTiles(test.updateParameters);
+    source.baseImpl->updateTiles(test.updateParameters);
 
     test.run();
 }
@@ -285,7 +284,7 @@ TEST(Source, VectorTileCorrupt) {
     VectorSource source("source", tileset);
     source.baseImpl->setObserver(&test.observer);
     source.baseImpl->loadDescription(test.fileSource);
-    source.baseImpl->loadTiles(test.updateParameters);
+    source.baseImpl->updateTiles(test.updateParameters);
 
     test.run();
 }
@@ -298,7 +297,7 @@ TEST(Source, RasterTileCancel) {
         return optional<Response>();
     };
 
-    test.observer.tileLoaded = [&] (Source&, const OverscaledTileID&, TileLoadState) {
+    test.observer.tileChanged = [&] (Source&, const OverscaledTileID&) {
         FAIL() << "Should never be called";
     };
 
@@ -312,7 +311,7 @@ TEST(Source, RasterTileCancel) {
     RasterSource source("source", tileset, 512);
     source.baseImpl->setObserver(&test.observer);
     source.baseImpl->loadDescription(test.fileSource);
-    source.baseImpl->loadTiles(test.updateParameters);
+    source.baseImpl->updateTiles(test.updateParameters);
 
     test.run();
 }
@@ -325,7 +324,7 @@ TEST(Source, VectorTileCancel) {
         return optional<Response>();
     };
 
-    test.observer.tileLoaded = [&] (Source&, const OverscaledTileID&, TileLoadState) {
+    test.observer.tileChanged = [&] (Source&, const OverscaledTileID&) {
         FAIL() << "Should never be called";
     };
 
@@ -339,7 +338,7 @@ TEST(Source, VectorTileCancel) {
     VectorSource source("source", tileset);
     source.baseImpl->setObserver(&test.observer);
     source.baseImpl->loadDescription(test.fileSource);
-    source.baseImpl->loadTiles(test.updateParameters);
+    source.baseImpl->updateTiles(test.updateParameters);
 
     test.run();
 }
diff --git a/test/style/style.cpp b/test/style/style.cpp
index 0d04fa5834..ff03e3c2cc 100644
--- a/test/style/style.cpp
+++ b/test/style/style.cpp
@@ -4,6 +4,7 @@
 #include <mbgl/style/style.hpp>
 #include <mbgl/style/source_impl.hpp>
 #include <mbgl/util/io.hpp>
+#include <mbgl/util/run_loop.hpp>
 
 using namespace mbgl;
 using namespace mbgl::style;