[core] allow self closing mailbox/actor

This commit is contained in:
Ivo van Dongen
2017-05-25 09:42:48 +03:00
committed by Ivo van Dongen
parent f0d0e7423b
commit b608d85db7
3 changed files with 36 additions and 3 deletions

View File

@@ -23,7 +23,7 @@ public:
private:
Scheduler& scheduler;
std::mutex receivingMutex;
std::recursive_mutex receivingMutex;
std::mutex pushingMutex;
bool closed { false };

View File

@@ -15,7 +15,8 @@ void Mailbox::close() {
// must not block send(). Of the two, the receiving mutex must be acquired first, because that is
// the order that an actor will obtain them when it self-sends a message, and consistent lock
// acquisition order prevents deadlocks.
std::lock_guard<std::mutex> receivingLock(receivingMutex);
// The receiving mutex is recursive to allow a mailbox (and thus the actor) to close itself.
std::lock_guard<std::recursive_mutex> receivingLock(receivingMutex);
std::lock_guard<std::mutex> pushingLock(pushingMutex);
closed = true;
@@ -37,7 +38,7 @@ void Mailbox::push(std::unique_ptr<Message> message) {
}
void Mailbox::receive() {
std::lock_guard<std::mutex> receivingLock(receivingMutex);
std::lock_guard<std::recursive_mutex> receivingLock(receivingMutex);
if (closed) {
return;

View File

@@ -6,6 +6,7 @@
#include <chrono>
#include <functional>
#include <future>
#include <memory>
using namespace mbgl;
using namespace std::chrono_literals;
@@ -123,6 +124,37 @@ TEST(Actor, DestructionBlocksOnSend) {
thread.join();
}
TEST(Actor, DestructionAllowedInReceiveOnSameThread) {
// Destruction doesn't block if occurring on the same
// thread as receive(). This prevents deadlocks and
// allows for self-closing actors
struct Test {
Test(ActorRef<Test>){};
void callMeBack(std::function<void ()> callback) {
callback();
}
};
ThreadPool pool { 1 };
std::promise<void> callbackFiredPromise;
auto test = std::make_unique<Actor<Test>>(pool);
// Callback (triggered while mutex is locked in Mailbox::receive())
test->invoke(&Test::callMeBack, [&]() {
// Destroy the Actor/Mailbox in the same thread
test.reset();
callbackFiredPromise.set_value();
});
auto status = callbackFiredPromise.get_future().wait_for(std::chrono::seconds(1));
ASSERT_EQ(std::future_status::ready, status);
}
TEST(Actor, OrderedMailbox) {
// Messages are processed in order.