[core] Use an actor model for tile worker concurrency
This commit is contained in:
139
test/actor/actor.cpp
Normal file
139
test/actor/actor.cpp
Normal file
@@ -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();
|
||||
}
|
||||
42
test/actor/actor_ref.cpp
Normal file
42
test/actor/actor_ref.cpp
Normal file
@@ -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);
|
||||
}
|
||||
@@ -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;
|
||||
};
|
||||
|
||||
@@ -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();
|
||||
}
|
||||
|
||||
@@ -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;
|
||||
|
||||
Reference in New Issue
Block a user