Commit c7eaa95b authored by Dominik Charousset's avatar Dominik Charousset

Fix synchronization issue on mailbox

Calling `reset_timeout_if_needed` *after* `mailbox().try_block()` can cause an
actor to run in two threads at once.
parent b5d28778
...@@ -495,12 +495,7 @@ public: ...@@ -495,12 +495,7 @@ public:
return ! bhvr_stack_.empty() || ! pending_responses_.empty(); return ! bhvr_stack_.empty() || ! pending_responses_.empty();
} }
behavior& get_behavior() { behavior& get_behavior();
if (! pending_responses_.empty()) {
return pending_responses_.front().second;
}
return bhvr_stack_.back();
}
virtual void initialize() = 0; virtual void initialize() = 0;
......
...@@ -600,22 +600,18 @@ resumable::resume_result local_actor::resume(execution_unit* eu, ...@@ -600,22 +600,18 @@ resumable::resume_result local_actor::resume(execution_unit* eu,
} }
// actor is cooperatively scheduled // actor is cooperatively scheduled
host(eu); host(eu);
auto done_cb = [&] { auto actor_done = [&]() -> bool {
CAF_LOG_TRACE(""); if (! has_behavior() || planned_exit_reason() != exit_reason::not_exited) {
CAF_LOG_DEBUG("actor either has no behavior or has set an exit reason");
on_exit();
bhvr_stack().clear(); bhvr_stack().clear();
bhvr_stack().cleanup(); bhvr_stack().cleanup();
on_exit();
CAF_LOG_DEBUG("on_exit did set a new behavior: ignored");
auto rsn = planned_exit_reason(); auto rsn = planned_exit_reason();
if (rsn == exit_reason::not_exited) { if (rsn == exit_reason::not_exited) {
rsn = exit_reason::normal; rsn = exit_reason::normal;
planned_exit_reason(rsn); planned_exit_reason(rsn);
} }
cleanup(rsn); cleanup(rsn);
};
auto actor_done = [&]() -> bool {
if (! has_behavior() || planned_exit_reason() != exit_reason::not_exited) {
done_cb();
return true; return true;
} }
return false; return false;
...@@ -638,7 +634,7 @@ resumable::resume_result local_actor::resume(execution_unit* eu, ...@@ -638,7 +634,7 @@ resumable::resume_result local_actor::resume(execution_unit* eu,
} }
int handled_msgs = 0; int handled_msgs = 0;
auto reset_timeout_if_needed = [&] { auto reset_timeout_if_needed = [&] {
if (handled_msgs > 0 && has_behavior()) { if (handled_msgs > 0) {
request_timeout(get_behavior().timeout()); request_timeout(get_behavior().timeout());
} }
}; };
...@@ -677,17 +673,16 @@ resumable::resume_result local_actor::resume(execution_unit* eu, ...@@ -677,17 +673,16 @@ resumable::resume_result local_actor::resume(execution_unit* eu,
} }
} else { } else {
CAF_LOG_DEBUG("no more element in mailbox; going to block"); CAF_LOG_DEBUG("no more element in mailbox; going to block");
if (mailbox().try_block()) {
reset_timeout_if_needed(); reset_timeout_if_needed();
if (mailbox().try_block()) {
return resumable::awaiting_message; return resumable::awaiting_message;
} }
CAF_LOG_DEBUG("try_block() interrupted by new message"); CAF_LOG_DEBUG("try_block() interrupted by new message");
} }
} }
if (! has_next_message() && mailbox().try_block()) {
reset_timeout_if_needed(); reset_timeout_if_needed();
if (! has_next_message() && mailbox().try_block())
return resumable::awaiting_message; return resumable::awaiting_message;
}
// time's up // time's up
return resumable::resume_later; return resumable::resume_later;
} }
...@@ -840,6 +835,11 @@ response_promise local_actor::make_response_promise() { ...@@ -840,6 +835,11 @@ response_promise local_actor::make_response_promise() {
return result; return result;
} }
behavior& local_actor::get_behavior() {
return pending_responses_.empty() ? bhvr_stack_.back()
: pending_responses_.front().second;
}
void local_actor::cleanup(uint32_t reason) { void local_actor::cleanup(uint32_t reason) {
CAF_LOG_TRACE(CAF_ARG(reason)); CAF_LOG_TRACE(CAF_ARG(reason));
detail::sync_request_bouncer f{reason}; detail::sync_request_bouncer f{reason};
......
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment