diff --git a/cmake/core-files.cmake b/cmake/core-files.cmake index 66f569abc..f4ee786df 100644 --- a/cmake/core-files.cmake +++ b/cmake/core-files.cmake @@ -4,6 +4,8 @@ set(MBGL_CORE_FILES # actor include/mbgl/actor/actor.hpp include/mbgl/actor/actor_ref.hpp + include/mbgl/actor/aspiring_actor.hpp + include/mbgl/actor/established_actor.hpp include/mbgl/actor/mailbox.hpp include/mbgl/actor/message.hpp include/mbgl/actor/scheduler.hpp diff --git a/include/mbgl/actor/actor.hpp b/include/mbgl/actor/actor.hpp index a0df19208..0052fad24 100644 --- a/include/mbgl/actor/actor.hpp +++ b/include/mbgl/actor/actor.hpp @@ -1,5 +1,7 @@ #pragma once +#include +#include #include #include #include @@ -8,6 +10,7 @@ #include #include #include +#include namespace mbgl { @@ -34,65 +37,33 @@ namespace mbgl { the lifetime of the owning Actor, and sending a message to a `Ref` whose `Actor` has died is a no-op. (In the future, a dead-letters queue or log may be implemented.) - Construction and destruction of an actor is currently synchronous: the corresponding `O` + Construction and destruction of an Actor is synchronous: the corresponding `O` object is constructed synchronously by the `Actor` constructor, and destructed synchronously by the `~Actor` destructor, after ensuring that the `O` is not currently receiving an - asynchronous message. (Construction and destruction may change to be asynchronous in the - future.) The constructor of `O` is passed an `ActorRef` referring to itself (which it - can use to self-send messages), followed by the forwarded arguments passed to `Actor`. + asynchronous message. The constructor of `O` is passed an `ActorRef` referring to itself + (which it can use to self-send messages), followed by the forwarded arguments passed to + `Actor`. Asynchronous object construction can be accomplished by directly using the + lower-level types, `AspiringActor` and `EstablishedActor`. Please don't send messages that contain shared pointers or references. That subverts the purpose of the actor model: prohibiting direct concurrent access to shared state. */ - template -class Actor : public util::noncopyable { +class Actor { public: + template + Actor(Scheduler& scheduler, Args&&... args) + : target(scheduler, parent, std::forward(args)...) {} - // Enabled for Objects with a constructor taking ActorRef as the first parameter - template , Args...>::value>::type * = nullptr> - Actor(Scheduler& scheduler, Args&&... args_) - : mailbox(std::make_shared(scheduler)), - object(self(), std::forward(args_)...) { - } - - // Enabled for plain Objects - template, Args...>::value>::type * = nullptr> - Actor(Scheduler& scheduler, Args&& ... args_) - : mailbox(std::make_shared(scheduler)), object(std::forward(args_)...) { - } - - ~Actor() { - mailbox->close(); - } - - template - void invoke(Fn fn, Args&&... args) { - mailbox->push(actor::makeMessage(object, fn, std::forward(args)...)); - } - - template - auto ask(Fn fn, Args&&... args) { - // Result type is deduced from the function's return type - using ResultType = typename std::result_of::type; - - std::promise promise; - auto future = promise.get_future(); - mailbox->push(actor::makeMessage(std::move(promise), object, fn, std::forward(args)...)); - return future; - } + Actor(const Actor&) = delete; ActorRef> self() { - return ActorRef>(object, mailbox); - } - - operator ActorRef>() { - return self(); + return parent.self(); } private: - std::shared_ptr mailbox; - Object object; + AspiringActor parent; + EstablishedActor target; }; } // namespace mbgl diff --git a/include/mbgl/actor/aspiring_actor.hpp b/include/mbgl/actor/aspiring_actor.hpp new file mode 100644 index 000000000..6c410cdfc --- /dev/null +++ b/include/mbgl/actor/aspiring_actor.hpp @@ -0,0 +1,60 @@ +#pragma once + +#include +#include +#include + +#include +#include +#include +#include + +namespace mbgl { + +template +class EstablishedActor; + +template +class Actor; + +/* + An `AspiringActor` is one half of the pair of types that comprise an actor (see `Actor`), + the other half being `EstablishedActor`. It is responsible for: + - ownership of the actor's `Mailbox` + - allocating the memory for (but *not* constructing) the target object `O` + + Using these two pieces--the mailbox and a stable address for `O`--an `AspiringActor` can + accept messages for the target object, or provide `ActorRef`s that do so, before the object + has actually been constructed by the corresponding `EstablishedActor`. (Such messages are + queued in the mailbox until after the object is constructed.) + + This allows for an `AspiringActor` to be created and safely used by a thread other than the + one on which the target object will (eventually) live. +*/ +template +class AspiringActor { +public: + AspiringActor() : mailbox(std::make_shared()) { + // mailbox starts closed because the `Object` hasn't yet been constructed + assert(!mailbox->isOpen()); + } + + AspiringActor(const AspiringActor&) = delete; + + ActorRef> self() { + return ActorRef>(object(), mailbox); + } + +private: + std::shared_ptr mailbox; + std::aligned_storage_t objectStorage; + + Object& object() { + return *reinterpret_cast(&objectStorage); + } + + friend class EstablishedActor; + friend class Actor; +}; + +} // namespace mbgl diff --git a/include/mbgl/actor/established_actor.hpp b/include/mbgl/actor/established_actor.hpp new file mode 100644 index 000000000..da0d8ac70 --- /dev/null +++ b/include/mbgl/actor/established_actor.hpp @@ -0,0 +1,80 @@ +#pragma once + +#include +#include +#include +#include + +#include +#include +#include +#include + +namespace mbgl { + +/* + An `EstablishedActor` is one half of the pair of types that comprise an actor (see `Actor`), + the other half being `AspiringActor`. It is responsible for managing the lifetime of the + target object `O` and the open/closed state of the parent's `mailbox`. + + The `O` object's lifetime is contained by that of its owning `EstablishedActor`: the + `EstablishedActor` constructor executes the `O` constructor via "placement new", constructing + it at the address provided by the parent `AspiringActor`, and the `~EstablishedActor` destructor + similarly executes the `~O` destructor (after closing the mailbox). `EstablishedActor` should + therefore live entirely on the thread intended to own `O`. +*/ + +template +class EstablishedActor { +public: + // Construct the Object from a parameter pack `args` (i.e. `Object(args...)`) + template ::value || + std::is_constructible, Args...>::value + >::type * = nullptr> + EstablishedActor(Scheduler& scheduler, AspiringActor& parent_, Args&& ... args) + : parent(parent_) { + emplaceObject(std::forward(args)...); + parent.mailbox->open(scheduler); + } + + // Construct the `Object` from a tuple containing the constructor arguments (i.e. + // `Object(std::get<0>(args), std::get<1>(args), ...)`) + template >::value> + EstablishedActor(Scheduler& scheduler, AspiringActor& parent_, ArgsTuple&& args) + : parent(parent_) { + emplaceObject(std::forward(args), std::make_index_sequence{}); + parent.mailbox->open(scheduler); + } + + EstablishedActor(const EstablishedActor&) = delete; + + ~EstablishedActor() { + parent.mailbox->close(); + parent.object().~Object(); + } + +private: + // Enabled for Objects with a constructor taking ActorRef as the first parameter + template , Args...>::value>::type * = nullptr> + void emplaceObject(Args&&... args_) { + new (&parent.objectStorage) Object(parent.self(), std::forward(args_)...); + } + + // Enabled for plain Objects + template ::value>::type * = nullptr> + void emplaceObject(Args&&... args_) { + new (&parent.objectStorage) Object(std::forward(args_)...); + } + + // Used to expand a tuple holding the constructor arguments + template + void emplaceObject(ArgsTuple&& args, std::index_sequence) { + emplaceObject(std::move(std::get(std::forward(args)))...); + (void) args; // mark args as used: if it's empty tuple, it's not actually used above. + } + + AspiringActor& parent; +}; + +} // namespace mbgl diff --git a/include/mbgl/actor/mailbox.hpp b/include/mbgl/actor/mailbox.hpp index 8ecf91701..23c579917 100644 --- a/include/mbgl/actor/mailbox.hpp +++ b/include/mbgl/actor/mailbox.hpp @@ -1,5 +1,7 @@ #pragma once +#include + #include #include #include @@ -11,17 +13,30 @@ class Message; class Mailbox : public std::enable_shared_from_this { public: + + // Create a "holding" mailbox, messages to which will remain queued, + // unconsumed, until the mailbox is associated with a Scheduler using + // start(). This allows a Mailbox object to be created on one thread and + // later transferred to a different target thread that may not yet exist. + Mailbox(); + Mailbox(Scheduler&); - void push(std::unique_ptr); - + // Attach the given scheduler to this mailbox and begin processing messages + // sent to it. The mailbox must be a "holding" mailbox, as created by the + // default constructor Mailbox(). + void open(Scheduler& scheduler_); void close(); + + bool isOpen() const; + + void push(std::unique_ptr); void receive(); static void maybeReceive(std::weak_ptr); private: - Scheduler& scheduler; + optional scheduler; std::recursive_mutex receivingMutex; std::mutex pushingMutex; diff --git a/include/mbgl/actor/scheduler.hpp b/include/mbgl/actor/scheduler.hpp index d8a26ebea..75ead29f0 100644 --- a/include/mbgl/actor/scheduler.hpp +++ b/include/mbgl/actor/scheduler.hpp @@ -31,6 +31,11 @@ class Mailbox; class Scheduler { public: virtual ~Scheduler() = default; + + // Used by a Mailbox when it has a message in its queue to request that it + // be scheduled. Specifically, the scheduler is expected to asynchronously + // call `receive() on the given mailbox, provided it still exists at that + // time. virtual void schedule(std::weak_ptr) = 0; // Set/Get the current Scheduler for this thread diff --git a/include/mbgl/util/thread.hpp b/include/mbgl/util/thread.hpp index 74e722b02..bc5842734 100644 --- a/include/mbgl/util/thread.hpp +++ b/include/mbgl/util/thread.hpp @@ -37,45 +37,55 @@ namespace util { // - `Object` can use `Timer` and do asynchronous I/O, like wait for sockets events. // template -class Thread : public Scheduler { +class Thread { public: template Thread(const std::string& name, Args&&... args) { - std::promise running; - thread = std::thread([&] { + std::promise running_; + running = running_.get_future(); + + auto capturedArgs = std::make_tuple(std::forward(args)...); + + thread = std::thread([ + this, + name, + capturedArgs = std::move(capturedArgs), + runningPromise = std::move(running_) + ] () mutable { platform::setCurrentThreadName(name); platform::makeThreadLowPriority(); util::RunLoop loop_(util::RunLoop::Type::New); loop = &loop_; + EstablishedActor establishedActor(loop_, object, std::move(capturedArgs)); - object = std::make_unique>(*this, std::forward(args)...); - running.set_value(); - + runningPromise.set_value(); + loop->run(); + + (void) establishedActor; + loop = nullptr; }); - - running.get_future().get(); } - ~Thread() override { + ~Thread() { if (paused) { resume(); } - std::promise joinable; + std::promise stoppable; + + running.wait(); - // Kill the actor, so we don't get more - // messages posted on this scheduler after - // we delete the RunLoop. + // Invoke a noop task on the run loop to ensure that we're executing + // run() before we call stop() loop->invoke([&] { - object.reset(); - joinable.set_value(); + stoppable.set_value(); }); - joinable.get_future().get(); + stoppable.get_future().get(); loop->stop(); thread.join(); @@ -85,8 +95,8 @@ public: // can be used to send messages to `Object`. It is safe // to the non-owning reference to outlive this object // and be used after the `Thread<>` gets destroyed. - ActorRef> actor() const { - return object->self(); + ActorRef> actor() { + return object.self(); } // Pauses the `Object` thread. It will prevent the object to wake @@ -103,6 +113,8 @@ public: auto pausing = paused->get_future(); + running.wait(); + loop->invoke(RunLoop::Priority::High, [this] { auto resuming = resumed->get_future(); paused->set_value(); @@ -127,13 +139,12 @@ public: private: MBGL_STORE_THREAD(tid); - void schedule(std::weak_ptr mailbox) override { - loop->schedule(mailbox); - } + AspiringActor object; std::thread thread; - std::unique_ptr> object; + std::future running; + std::unique_ptr> paused; std::unique_ptr> resumed; diff --git a/platform/android/src/snapshotter/map_snapshotter.cpp b/platform/android/src/snapshotter/map_snapshotter.cpp index 155fdf81f..e8fcc6177 100644 --- a/platform/android/src/snapshotter/map_snapshotter.cpp +++ b/platform/android/src/snapshotter/map_snapshotter.cpp @@ -58,8 +58,8 @@ MapSnapshotter::MapSnapshotter(jni::JNIEnv& _env, showLogo = _showLogo; // Create the core snapshotter - snapshotter = std::make_unique(fileSource, - *threadPool, + snapshotter = std::make_unique(&fileSource, + threadPool, style, size, pixelRatio, @@ -173,4 +173,4 @@ void MapSnapshotter::registerNative(jni::JNIEnv& env) { } } // namespace android -} // namespace mbgl \ No newline at end of file +} // namespace mbgl diff --git a/platform/android/src/snapshotter/map_snapshotter.hpp b/platform/android/src/snapshotter/map_snapshotter.hpp index 3be2cb4f6..7b72452c4 100644 --- a/platform/android/src/snapshotter/map_snapshotter.hpp +++ b/platform/android/src/snapshotter/map_snapshotter.hpp @@ -76,4 +76,4 @@ private: }; } // namespace android -} // namespace mbgl \ No newline at end of file +} // namespace mbgl diff --git a/platform/darwin/src/MGLMapSnapshotter.mm b/platform/darwin/src/MGLMapSnapshotter.mm index 2a2bef8fb..d9fa04421 100644 --- a/platform/darwin/src/MGLMapSnapshotter.mm +++ b/platform/darwin/src/MGLMapSnapshotter.mm @@ -469,7 +469,7 @@ const CGFloat MGLSnapshotterMinimumPixelSize = 64; } // Create the snapshotter - _mbglMapSnapshotter = std::make_unique(*mbglFileSource, *_mbglThreadPool, style, size, pixelRatio, cameraOptions, coordinateBounds); + _mbglMapSnapshotter = std::make_unique(mbglFileSource, _mbglThreadPool, style, size, pixelRatio, cameraOptions, coordinateBounds); } @end diff --git a/platform/default/default_file_source.cpp b/platform/default/default_file_source.cpp index 89aabeb8d..f07012149 100644 --- a/platform/default/default_file_source.cpp +++ b/platform/default/default_file_source.cpp @@ -19,15 +19,10 @@ namespace mbgl { class DefaultFileSource::Impl { public: - Impl(ActorRef self, std::shared_ptr assetFileSource_, const std::string& cachePath, uint64_t maximumCacheSize) + Impl(std::shared_ptr assetFileSource_, std::string cachePath, uint64_t maximumCacheSize) : assetFileSource(assetFileSource_) - , localFileSource(std::make_unique()) { - // Initialize the Database asynchronously so as to not block Actor creation. - self.invoke(&Impl::initializeOfflineDatabase, cachePath, maximumCacheSize); - } - - void initializeOfflineDatabase(std::string cachePath, uint64_t maximumCacheSize) { - offlineDatabase = std::make_unique(cachePath, maximumCacheSize); + , localFileSource(std::make_unique()) + , offlineDatabase(std::make_unique(cachePath, maximumCacheSize)) { } void setAPIBaseURL(const std::string& url) { diff --git a/platform/default/mbgl/map/map_snapshotter.cpp b/platform/default/mbgl/map/map_snapshotter.cpp index a909e3fe9..149ef22e7 100644 --- a/platform/default/mbgl/map/map_snapshotter.cpp +++ b/platform/default/mbgl/map/map_snapshotter.cpp @@ -13,8 +13,8 @@ namespace mbgl { class MapSnapshotter::Impl { public: - Impl(FileSource&, - Scheduler&, + Impl(FileSource*, + std::shared_ptr, const std::pair style, const Size&, const float pixelRatio, @@ -40,20 +40,22 @@ public: void snapshot(ActorRef); private: + std::shared_ptr scheduler; HeadlessFrontend frontend; Map map; }; -MapSnapshotter::Impl::Impl(FileSource& fileSource, - Scheduler& scheduler, +MapSnapshotter::Impl::Impl(FileSource* fileSource, + std::shared_ptr scheduler_, const std::pair style, const Size& size, const float pixelRatio, const optional cameraOptions, const optional region, const optional programCacheDir) - : frontend(size, pixelRatio, fileSource, scheduler, programCacheDir) - , map(frontend, MapObserver::nullObserver(), size, pixelRatio, fileSource, scheduler, MapMode::Static) { + : scheduler(std::move(scheduler_)) + , frontend(size, pixelRatio, *fileSource, *scheduler, programCacheDir) + , map(frontend, MapObserver::nullObserver(), size, pixelRatio, *fileSource, *scheduler, MapMode::Static) { if (style.first) { map.getStyle().loadJSON(style.second); @@ -149,15 +151,15 @@ LatLngBounds MapSnapshotter::Impl::getRegion() const { return map.latLngBoundsForCamera(getCameraOptions()); } -MapSnapshotter::MapSnapshotter(FileSource& fileSource, - Scheduler& scheduler, +MapSnapshotter::MapSnapshotter(FileSource* fileSource, + std::shared_ptr scheduler, const std::pair style, const Size& size, const float pixelRatio, const optional cameraOptions, const optional region, const optional programCacheDir) - : impl(std::make_unique>("Map Snapshotter", fileSource, scheduler, style, size, pixelRatio, cameraOptions, region, programCacheDir)) { + : impl(std::make_unique>("Map Snapshotter", fileSource, std::move(scheduler), style, size, pixelRatio, cameraOptions, region, programCacheDir)) { } MapSnapshotter::~MapSnapshotter() = default; diff --git a/platform/default/mbgl/map/map_snapshotter.hpp b/platform/default/mbgl/map/map_snapshotter.hpp index b9e630766..f40d1e4b7 100644 --- a/platform/default/mbgl/map/map_snapshotter.hpp +++ b/platform/default/mbgl/map/map_snapshotter.hpp @@ -25,8 +25,8 @@ class Style; class MapSnapshotter { public: - MapSnapshotter(FileSource& fileSource, - Scheduler& scheduler, + MapSnapshotter(FileSource* fileSource, + std::shared_ptr scheduler, const std::pair style, const Size&, const float pixelRatio, diff --git a/src/mbgl/actor/mailbox.cpp b/src/mbgl/actor/mailbox.cpp index 373c24275..8ee8dca11 100644 --- a/src/mbgl/actor/mailbox.cpp +++ b/src/mbgl/actor/mailbox.cpp @@ -6,8 +6,30 @@ namespace mbgl { +Mailbox::Mailbox() { +} + Mailbox::Mailbox(Scheduler& scheduler_) - : scheduler(scheduler_) { + : scheduler(&scheduler_) { +} + +void Mailbox::open(Scheduler& scheduler_) { + assert(!scheduler); + + // As with close(), block until neither receive() nor push() are in progress, and acquire the two + // mutexes in the same order. + std::lock_guard receivingLock(receivingMutex); + std::lock_guard pushingLock(pushingMutex); + + scheduler = &scheduler_; + + if (closed) { + return; + } + + if (!queue.empty()) { + (*scheduler)->schedule(shared_from_this()); + } } void Mailbox::close() { @@ -22,6 +44,9 @@ void Mailbox::close() { closed = true; } +bool Mailbox::isOpen() const { return bool(scheduler); } + + void Mailbox::push(std::unique_ptr message) { std::lock_guard pushingLock(pushingMutex); @@ -32,13 +57,15 @@ void Mailbox::push(std::unique_ptr message) { std::lock_guard queueLock(queueMutex); bool wasEmpty = queue.empty(); queue.push(std::move(message)); - if (wasEmpty) { - scheduler.schedule(shared_from_this()); + if (wasEmpty && scheduler) { + (*scheduler)->schedule(shared_from_this()); } } void Mailbox::receive() { std::lock_guard receivingLock(receivingMutex); + + assert(scheduler); if (closed) { return; @@ -58,7 +85,7 @@ void Mailbox::receive() { (*message)(); if (!wasEmpty) { - scheduler.schedule(shared_from_this()); + (*scheduler)->schedule(shared_from_this()); } } diff --git a/src/mbgl/sprite/sprite_loader.cpp b/src/mbgl/sprite/sprite_loader.cpp index 93d6dfd9a..df4fe6e8d 100644 --- a/src/mbgl/sprite/sprite_loader.cpp +++ b/src/mbgl/sprite/sprite_loader.cpp @@ -86,7 +86,7 @@ void SpriteLoader::emitSpriteLoadedIfComplete() { return; } - loader->worker.invoke(&SpriteLoaderWorker::parse, loader->image, loader->json); + loader->worker.self().invoke(&SpriteLoaderWorker::parse, loader->image, loader->json); } void SpriteLoader::onParsed(std::vector>&& result) { diff --git a/src/mbgl/style/sources/custom_geometry_source.cpp b/src/mbgl/style/sources/custom_geometry_source.cpp index b37490a5c..6ce7c1be1 100644 --- a/src/mbgl/style/sources/custom_geometry_source.cpp +++ b/src/mbgl/style/sources/custom_geometry_source.cpp @@ -30,15 +30,15 @@ void CustomGeometrySource::loadDescription(FileSource&) { void CustomGeometrySource::setTileData(const CanonicalTileID& tileID, const GeoJSON& data) { - loader->invoke(&CustomTileLoader::setTileData, tileID, data); + loader->self().invoke(&CustomTileLoader::setTileData, tileID, data); } void CustomGeometrySource::invalidateTile(const CanonicalTileID& tileID) { - loader->invoke(&CustomTileLoader::invalidateTile, tileID); + loader->self().invoke(&CustomTileLoader::invalidateTile, tileID); } void CustomGeometrySource::invalidateRegion(const LatLngBounds& bounds) { - loader->invoke(&CustomTileLoader::invalidateRegion, bounds, impl().getZoomRange()); + loader->self().invoke(&CustomTileLoader::invalidateRegion, bounds, impl().getZoomRange()); } } // namespace style diff --git a/src/mbgl/tile/geometry_tile.cpp b/src/mbgl/tile/geometry_tile.cpp index af28fe396..d686d8440 100644 --- a/src/mbgl/tile/geometry_tile.cpp +++ b/src/mbgl/tile/geometry_tile.cpp @@ -86,7 +86,7 @@ void GeometryTile::setData(std::unique_ptr data_) { pending = true; ++correlationID; - worker.invoke(&GeometryTileWorker::setData, std::move(data_), correlationID); + worker.self().invoke(&GeometryTileWorker::setData, std::move(data_), correlationID); } @@ -112,14 +112,14 @@ void GeometryTile::setLayers(const std::vector>& layers) } ++correlationID; - worker.invoke(&GeometryTileWorker::setLayers, std::move(impls), correlationID); + worker.self().invoke(&GeometryTileWorker::setLayers, std::move(impls), correlationID); } void GeometryTile::setShowCollisionBoxes(const bool showCollisionBoxes_) { if (showCollisionBoxes != showCollisionBoxes_) { showCollisionBoxes = showCollisionBoxes_; ++correlationID; - worker.invoke(&GeometryTileWorker::setShowCollisionBoxes, showCollisionBoxes, correlationID); + worker.self().invoke(&GeometryTileWorker::setShowCollisionBoxes, showCollisionBoxes, correlationID); } } @@ -153,7 +153,7 @@ void GeometryTile::onError(std::exception_ptr err, const uint64_t resultCorrelat } void GeometryTile::onGlyphsAvailable(GlyphMap glyphs) { - worker.invoke(&GeometryTileWorker::onGlyphsAvailable, std::move(glyphs)); + worker.self().invoke(&GeometryTileWorker::onGlyphsAvailable, std::move(glyphs)); } void GeometryTile::getGlyphs(GlyphDependencies glyphDependencies) { @@ -161,7 +161,7 @@ void GeometryTile::getGlyphs(GlyphDependencies glyphDependencies) { } void GeometryTile::onImagesAvailable(ImageMap images, uint64_t imageCorrelationID) { - worker.invoke(&GeometryTileWorker::onImagesAvailable, std::move(images), imageCorrelationID); + worker.self().invoke(&GeometryTileWorker::onImagesAvailable, std::move(images), imageCorrelationID); } void GeometryTile::getImages(ImageRequestPair pair) { diff --git a/src/mbgl/tile/raster_dem_tile.cpp b/src/mbgl/tile/raster_dem_tile.cpp index 5db298cf4..f29861ee7 100644 --- a/src/mbgl/tile/raster_dem_tile.cpp +++ b/src/mbgl/tile/raster_dem_tile.cpp @@ -48,7 +48,7 @@ void RasterDEMTile::setMetadata(optional modified_, optional data) { pending = true; ++correlationID; - worker.invoke(&RasterDEMTileWorker::parse, data, correlationID, encoding); + worker.self().invoke(&RasterDEMTileWorker::parse, data, correlationID, encoding); } void RasterDEMTile::onParsed(std::unique_ptr result, const uint64_t resultCorrelationID) { diff --git a/src/mbgl/tile/raster_tile.cpp b/src/mbgl/tile/raster_tile.cpp index ff23d4493..cc71c04ba 100644 --- a/src/mbgl/tile/raster_tile.cpp +++ b/src/mbgl/tile/raster_tile.cpp @@ -37,7 +37,7 @@ void RasterTile::setMetadata(optional modified_, optional void RasterTile::setData(std::shared_ptr data) { pending = true; ++correlationID; - worker.invoke(&RasterTileWorker::parse, data, correlationID); + worker.self().invoke(&RasterTileWorker::parse, data, correlationID); } void RasterTile::onParsed(std::unique_ptr result, const uint64_t resultCorrelationID) { diff --git a/test/actor/actor.test.cpp b/test/actor/actor.test.cpp index 967dc152d..7493abe26 100644 --- a/test/actor/actor.test.cpp +++ b/test/actor/actor.test.cpp @@ -1,5 +1,6 @@ #include #include +#include #include @@ -7,13 +8,12 @@ #include #include #include +#include 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, bool& constructed) { constructed = true; @@ -27,6 +27,25 @@ TEST(Actor, Construction) { EXPECT_TRUE(constructed); } +TEST(Actor, Destruction) { + struct Test { + Test(ActorRef, bool& destructed_) : destructed(destructed_) {}; + ~Test() { + destructed = true; + } + + bool& destructed; + }; + + ThreadPool pool { 1 }; + bool destructed = false; + { + Actor test(pool, std::ref(destructed)); + } + + EXPECT_TRUE(destructed); +} + TEST(Actor, DestructionBlocksOnReceive) { // Destruction blocks until the actor is not receiving. @@ -63,7 +82,7 @@ TEST(Actor, DestructionBlocksOnReceive) { Actor test(pool, std::move(enteredPromise), std::move(exitingFuture)); - test.invoke(&Test::wait); + test.self().invoke(&Test::wait); enteredFuture.wait(); exitingPromise.set_value(); } @@ -145,7 +164,7 @@ TEST(Actor, DestructionAllowedInReceiveOnSameThread) { auto test = std::make_unique>(pool); // Callback (triggered while mutex is locked in Mailbox::receive()) - test->invoke(&Test::callMeBack, [&]() { + test->self().invoke(&Test::callMeBack, [&]() { // Destroy the Actor/Mailbox in the same thread test.reset(); callbackFiredPromise.set_value(); @@ -180,16 +199,16 @@ TEST(Actor, SelfDestructionDoesntCrashWaitingReceivingThreads) { std::atomic waitingMessageProcessed {false}; // Callback (triggered while mutex is locked in Mailbox::receive()) - closingActor->invoke(&Test::callMeBack, [&]() { + closingActor->self().invoke(&Test::callMeBack, [&]() { // Queue up another message from another thread std::promise messageQueuedPromise; - waitingActor->invoke(&Test::callMeBack, [&]() { + waitingActor->self().invoke(&Test::callMeBack, [&]() { // This will be waiting on the mutex in // Mailbox::receive(), holding a lock // on the weak_ptr so the mailbox is not // destroyed - closingActor->invoke(&Test::callMeBack, [&]() { + closingActor->self().invoke(&Test::callMeBack, [&]() { waitingMessageProcessed.store(true); }); messageQueuedPromise.set_value(); @@ -239,10 +258,10 @@ TEST(Actor, OrderedMailbox) { Actor test(pool, std::move(endedPromise)); for (auto i = 1; i <= 10; ++i) { - test.invoke(&Test::receive, i); + test.self().invoke(&Test::receive, i); } - test.invoke(&Test::end); + test.self().invoke(&Test::end); endedFuture.wait(); } @@ -275,10 +294,10 @@ TEST(Actor, NonConcurrentMailbox) { Actor test(pool, std::move(endedPromise)); for (auto i = 1; i <= 10; ++i) { - test.invoke(&Test::receive, i); + test.self().invoke(&Test::receive, i); } - test.invoke(&Test::end); + test.self().invoke(&Test::end); endedFuture.wait(); } @@ -297,7 +316,7 @@ TEST(Actor, Ask) { ThreadPool pool { 2 }; Actor test(pool); - auto result = test.ask(&Test::doubleIt, 1); + auto result = test.self().ask(&Test::doubleIt, 1); ASSERT_TRUE(result.valid()); @@ -324,7 +343,7 @@ TEST(Actor, AskVoid) { bool executed = false; Actor actor(pool, executed); - actor.ask(&Test::doIt).get(); + actor.self().ask(&Test::doIt).get(); EXPECT_TRUE(executed); } @@ -355,6 +374,58 @@ TEST(Actor, NoSelfActorRef) { auto future = promise.get_future(); Actor withArguments(pool, std::move(promise)); - withArguments.invoke(&WithArguments::receive); + withArguments.self().invoke(&WithArguments::receive); future.wait(); } + +TEST(Actor, TwoPhaseConstruction) { + // This test mimics, in simplified form, the approach used by the Thread to construct + // its actor in two parts so that the Thread instance can be created without waiting + // for the target thread to be up and running. + + struct Test { + Test(ActorRef, std::shared_ptr destroyed_) + : destroyed(std::move(destroyed_)) {}; + + ~Test() { + *destroyed = true; + } + + void callMe(std::promise p) { + p.set_value(); + } + + void stop() { + util::RunLoop::Get()->stop(); + } + + std::shared_ptr destroyed; + }; + + AspiringActor parent; + + auto destroyed = std::make_shared(false); + + std::promise queueExecuted; + auto queueExecutedFuture = queueExecuted.get_future(); + + parent.self().invoke(&Test::callMe, std::move(queueExecuted)); + parent.self().invoke(&Test::stop); + + auto thread = std::thread([ + capturedArgs = std::make_tuple(destroyed), + &parent + ] () mutable { + util::RunLoop loop(util::RunLoop::Type::New); + EstablishedActor test(loop, parent, capturedArgs); + loop.run(); + }); + + // should not hang + queueExecutedFuture.get(); + thread.join(); + + EXPECT_TRUE(*destroyed); +} + + diff --git a/test/util/thread.test.cpp b/test/util/thread.test.cpp index 76fb5ce3f..2bcb9d895 100644 --- a/test/util/thread.test.cpp +++ b/test/util/thread.test.cpp @@ -15,11 +15,11 @@ class TestObject { public: TestObject(ActorRef, std::thread::id otherTid) : tid(std::this_thread::get_id()) { - EXPECT_NE(tid, otherTid); + EXPECT_NE(tid, otherTid); // Object is created on child thread } ~TestObject() { - EXPECT_EQ(tid, std::this_thread::get_id()); + EXPECT_EQ(tid, std::this_thread::get_id()); // Object is destroyed on child thread } void fn1(int val) const { @@ -275,3 +275,60 @@ TEST(Thread, PauseResume) { thread.actor().invoke(&TestWorker::send, [&] { loop.stop(); }); loop.run(); } + + +class TestWorkerDelayedConstruction { +public: + TestWorkerDelayedConstruction(ActorRef, std::future start) { + start.get(); + } + + void send(std::function cb) { + cb(); + } + +private: + Timer timer; +}; + +TEST(Thread, InvokeBeforeChildStarts) { + RunLoop loop; + + std::promise start; + Thread thread("Test", start.get_future()); + + std::atomic count { 0 }; + + for (unsigned i = 0; i < 100; ++i) { + thread.actor().invoke(&TestWorkerDelayedConstruction::send, [&] { ++count; }); + } + + thread.actor().invoke(&TestWorkerDelayedConstruction::send, [&] { loop.stop(); }); + + // This test will be flaky if messages are consumed before the target object is constructed. + ASSERT_EQ(count, 0); + + start.set_value(); + + loop.run(); + + ASSERT_EQ(count, 100); +} + +TEST(Thread, DeleteBeforeChildStarts) { + std::atomic_bool flag(false); + std::promise start; + + Thread control("Control"); + auto thread = std::make_unique>("Test", start.get_future()); + + thread->actor().invoke(&TestWorkerDelayedConstruction::send, [&] { flag = true; }); + + control.actor().invoke(&TestWorker::sendDelayed, [&] { start.set_value(); }); + + // Should not hang. + thread.reset(); + + // Should process the queue before destruction. + ASSERT_TRUE(flag); +}