[core] Mailbox is handling Scheduler destructionupstream/mikhail_weak_ptr_for_scheduler

3 files changed, 40 insertions, 10 deletions

diff --git a/include/mbgl/actor/mailbox.hpp b/include/mbgl/actor/mailbox.hpp
index 2b9838ef29..dac2947df5 100644
--- a/include/mbgl/actor/mailbox.hpp
+++ b/include/mbgl/actor/mailbox.hpp
@@ -1,5 +1,6 @@
 #pragma once
 
+#include <mapbox/weak.hpp>
 #include <mbgl/util/optional.hpp>
 
 #include <functional>
@@ -38,7 +39,8 @@ public:
     static std::function<void()> makeClosure(std::weak_ptr<Mailbox>);
 
 private:
-    optional<Scheduler*> scheduler;
+    void scheduleThis();
+    mapbox::base::WeakPtr<Scheduler> scheduler;
 
     std::recursive_mutex receivingMutex;
     std::mutex pushingMutex;
diff --git a/src/mbgl/actor/mailbox.cpp b/src/mbgl/actor/mailbox.cpp
index 070e14bdb0..4d62ec8aba 100644
--- a/src/mbgl/actor/mailbox.cpp
+++ b/src/mbgl/actor/mailbox.cpp
@@ -8,9 +8,7 @@ namespace mbgl {
 
 Mailbox::Mailbox() = default;
 
-Mailbox::Mailbox(Scheduler& scheduler_)
-    : scheduler(&scheduler_) {
-}
+Mailbox::Mailbox(Scheduler& scheduler_) : scheduler(scheduler_.makeWeakPtr()) {}
 
 void Mailbox::open(Scheduler& scheduler_) {
     assert(!scheduler);
@@ -19,15 +17,15 @@ void Mailbox::open(Scheduler& scheduler_) {
     // mutexes in the same order.
     std::lock_guard<std::recursive_mutex> receivingLock(receivingMutex);
     std::lock_guard<std::mutex> pushingLock(pushingMutex);
-    
-    scheduler = &scheduler_;
+
+    scheduler = scheduler_.makeWeakPtr();
 
     if (closed) {
         return;
     }
     
     if (!queue.empty()) {
-        (*scheduler)->schedule(makeClosure(shared_from_this()));
+        scheduleThis();
     }
 }
 
@@ -56,8 +54,8 @@ void Mailbox::push(std::unique_ptr<Message> message) {
     std::lock_guard<std::mutex> queueLock(queueMutex);
     bool wasEmpty = queue.empty();
     queue.push(std::move(message));
-    if (wasEmpty && scheduler) {
-        (*scheduler)->schedule(makeClosure(shared_from_this()));
+    if (wasEmpty) {
+        scheduleThis();
     }
 }
 
@@ -84,10 +82,15 @@ void Mailbox::receive() {
     (*message)();
 
     if (!wasEmpty) {
-        (*scheduler)->schedule(makeClosure(shared_from_this()));
+        scheduleThis();
     }
 }
 
+void Mailbox::scheduleThis() {
+    auto guard = scheduler.lock();
+    if (scheduler) scheduler->schedule(makeClosure(shared_from_this()));
+}
+
 // static
 void Mailbox::maybeReceive(std::weak_ptr<Mailbox> mailbox) {
     if (auto locked = mailbox.lock()) {
diff --git a/test/actor/actor.test.cpp b/test/actor/actor.test.cpp
index 4b152f471a..832e5b98b6 100644
--- a/test/actor/actor.test.cpp
+++ b/test/actor/actor.test.cpp
@@ -90,6 +90,7 @@ TEST(Actor, DestructionBlocksOnSend) {
         std::promise<void> promise;
         std::future<void> future;
         std::atomic<bool> waited;
+        mapbox::base::WeakPtrFactory<Scheduler> weakFactory{this};
 
         TestScheduler(std::promise<void> promise_, std::future<void> future_)
             : promise(std::move(promise_)),
@@ -107,6 +108,7 @@ TEST(Actor, DestructionBlocksOnSend) {
             std::this_thread::sleep_for(1ms);
             waited = true;
         }
+        mapbox::base::WeakPtr<Scheduler> makeWeakPtr() override { return weakFactory.makeWeakPtr(); }
     };
 
     struct Test {
@@ -134,7 +136,30 @@ TEST(Actor, DestructionBlocksOnSend) {
     exitingPromise.set_value();
 
     actor.reset();
+
+    thread.join();
+}
+
+TEST(Actor, DestructionScheduler) {
+    // Destruction of the sheduler is handled in Mailbox::push().
+    // Passes if does not crash.
+
+    struct TestScheduler : public Scheduler {
+        mapbox::base::WeakPtrFactory<Scheduler> weakFactory{this};
+
+        void schedule(std::function<void()>) final { std::this_thread::sleep_for(1ms); }
+        mapbox::base::WeakPtr<Scheduler> makeWeakPtr() override { return weakFactory.makeWeakPtr(); }
+    };
+
+    struct Test {
+        Test(ActorRef<Test>) {}
+        void message() {}
+    };
+
+    auto scheduler = std::make_unique<TestScheduler>();
+    auto actor = std::make_unique<Actor<Test>>(*scheduler);
     scheduler.reset();
+    std::thread thread{[](ActorRef<Test> ref) { ref.invoke(&Test::message); }, actor->self()};
 
     thread.join();
 }