[core] Block in Mailbox::close() until neither receive nor push are in progress

Otherwise, an ActorRef that's in the process of sending a message could attempt to access an invalid Scheduler reference:

      Thread 1                     Thread 2
--------------------------------------------------
 Scheduler::Scheduler
     Actor::Actor
                              weakMailbox.lock()
     Actor::~Actor
 Scheduler::~Scheduler
                               mailbox->push()
                             scheduler.schedule() 💣
This commit is contained in:
John Firebaugh
2017-05-22 16:05:58 -07:00
parent 8b448786fe
commit 2c8343aa82
3 changed files with 79 additions and 12 deletions

View File

@@ -23,8 +23,10 @@ public:
private:
Scheduler& scheduler;
std::mutex closingMutex;
bool closing { false };
std::mutex receivingMutex;
std::mutex pushingMutex;
bool closed { false };
std::mutex queueMutex;
std::queue<std::unique_ptr<Message>> queue;

View File

@@ -10,8 +10,23 @@ Mailbox::Mailbox(Scheduler& scheduler_)
: scheduler(scheduler_) {
}
void Mailbox::close() {
// Block until neither receive() nor push() are in progress. Two mutexes are used because receive()
// 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);
std::lock_guard<std::mutex> pushingLock(pushingMutex);
closed = true;
}
void Mailbox::push(std::unique_ptr<Message> message) {
assert(!closing);
std::lock_guard<std::mutex> pushingLock(pushingMutex);
if (closed) {
return;
}
std::lock_guard<std::mutex> queueLock(queueMutex);
bool wasEmpty = queue.empty();
@@ -21,16 +36,10 @@ void Mailbox::push(std::unique_ptr<Message> message) {
}
}
void Mailbox::close() {
// Block until the scheduler is guaranteed not to be executing receive().
std::lock_guard<std::mutex> closingLock(closingMutex);
closing = true;
}
void Mailbox::receive() {
std::lock_guard<std::mutex> closingLock(closingMutex);
std::lock_guard<std::mutex> receivingLock(receivingMutex);
if (closing) {
if (closed) {
return;
}

View File

@@ -26,7 +26,7 @@ TEST(Actor, Construction) {
EXPECT_TRUE(constructed);
}
TEST(Actor, DestructionClosesMailbox) {
TEST(Actor, DestructionBlocksOnReceive) {
// Destruction blocks until the actor is not receiving.
struct Test {
@@ -67,6 +67,62 @@ TEST(Actor, DestructionClosesMailbox) {
exitingPromise.set_value();
}
TEST(Actor, DestructionBlocksOnSend) {
// Destruction blocks until the actor is not being sent a message.
struct TestScheduler : public Scheduler {
std::promise<void> promise;
std::future<void> future;
std::atomic<bool> waited;
TestScheduler(std::promise<void> promise_, std::future<void> future_)
: promise(std::move(promise_)),
future(std::move(future_)),
waited(false) {
}
~TestScheduler() {
EXPECT_TRUE(waited.load());
}
void schedule(std::weak_ptr<Mailbox>) final {
promise.set_value();
future.wait();
std::this_thread::sleep_for(1ms);
waited = true;
}
};
struct Test {
Test(ActorRef<Test>) {}
void message() {}
};
std::promise<void> enteredPromise;
std::future<void> enteredFuture = enteredPromise.get_future();
std::promise<void> exitingPromise;
std::future<void> exitingFuture = exitingPromise.get_future();
auto scheduler = std::make_unique<TestScheduler>(std::move(enteredPromise), std::move(exitingFuture));
auto actor = std::make_unique<Actor<Test>>(*scheduler);
std::thread thread {
[] (ActorRef<Test> ref) {
ref.invoke(&Test::message);
},
actor->self()
};
enteredFuture.wait();
exitingPromise.set_value();
actor.reset();
scheduler.reset();
thread.join();
}
TEST(Actor, OrderedMailbox) {
// Messages are processed in order.