Commit f5649fc5 authored by Dominik Charousset's avatar Dominik Charousset

Shutdown actors when removing final stream manager

parent 4a7db979
...@@ -214,7 +214,6 @@ public: ...@@ -214,7 +214,6 @@ public:
/// Consumes messages from the mailbox. /// Consumes messages from the mailbox.
struct mailbox_visitor { struct mailbox_visitor {
scheduled_actor* self; scheduled_actor* self;
resume_result& result;
size_t& handled_msgs; size_t& handled_msgs;
size_t max_throughput; size_t max_throughput;
......
...@@ -341,7 +341,6 @@ scheduled_actor::mailbox_visitor::operator()(mailbox_element& x) { ...@@ -341,7 +341,6 @@ scheduled_actor::mailbox_visitor::operator()(mailbox_element& x) {
CAF_LOG_TRACE(CAF_ARG(x) << CAF_ARG(handled_msgs)); CAF_LOG_TRACE(CAF_ARG(x) << CAF_ARG(handled_msgs));
switch (self->reactivate(x)) { switch (self->reactivate(x)) {
case activation_result::terminated: case activation_result::terminated:
result = resume_result::done;
return intrusive::task_result::stop; return intrusive::task_result::stop;
case activation_result::success: case activation_result::success:
return ++handled_msgs < max_throughput return ++handled_msgs < max_throughput
...@@ -359,7 +358,7 @@ scheduled_actor::resume(execution_unit* ctx, size_t max_throughput) { ...@@ -359,7 +358,7 @@ scheduled_actor::resume(execution_unit* ctx, size_t max_throughput) {
CAF_PUSH_AID(id()); CAF_PUSH_AID(id());
CAF_LOG_TRACE(CAF_ARG(max_throughput)); CAF_LOG_TRACE(CAF_ARG(max_throughput));
if (!activate(ctx)) if (!activate(ctx))
return resume_result::done; return resumable::done;
size_t handled_msgs = 0; size_t handled_msgs = 0;
actor_clock::time_point tout{actor_clock::duration_type{0}}; actor_clock::time_point tout{actor_clock::duration_type{0}};
auto reset_timeouts_if_needed = [&] { auto reset_timeouts_if_needed = [&] {
...@@ -374,8 +373,7 @@ scheduled_actor::resume(execution_unit* ctx, size_t max_throughput) { ...@@ -374,8 +373,7 @@ scheduled_actor::resume(execution_unit* ctx, size_t max_throughput) {
set_stream_timeout(tout); set_stream_timeout(tout);
} }
}; };
auto result = resume_result::awaiting_message; mailbox_visitor f{this, handled_msgs, max_throughput};
mailbox_visitor f{this, result, handled_msgs, max_throughput};
mailbox_element_ptr ptr; mailbox_element_ptr ptr;
// Timeout for calling `advance_streams`. // Timeout for calling `advance_streams`.
while (handled_msgs < max_throughput) { while (handled_msgs < max_throughput) {
...@@ -387,9 +385,11 @@ scheduled_actor::resume(execution_unit* ctx, size_t max_throughput) { ...@@ -387,9 +385,11 @@ scheduled_actor::resume(execution_unit* ctx, size_t max_throughput) {
if (mailbox().try_block()) if (mailbox().try_block())
return resumable::awaiting_message; return resumable::awaiting_message;
} }
// Immediately stop if the visitor reports an error. // Check whether the visitor left the actor without behavior.
if (result != awaiting_message) if (finalize()) {
return result; return resumable::done;
}
// Advance streams, i.e., try to generating credit or to emit batches.
auto now = clock().now(); auto now = clock().now();
if (now >= tout) if (now >= tout)
tout = advance_streams(now); tout = advance_streams(now);
...@@ -771,6 +771,12 @@ void scheduled_actor::do_become(behavior bhvr, bool discard_old) { ...@@ -771,6 +771,12 @@ void scheduled_actor::do_become(behavior bhvr, bool discard_old) {
} }
bool scheduled_actor::finalize() { bool scheduled_actor::finalize() {
CAF_LOG_TRACE("");
// Repeated calls always return `true` but have no side effects.
if (getf(is_cleaned_up_flag))
return true;
// An actor is considered alive as long as it has a behavior and didn't set
// the terminated flag.
if (has_behavior() && !getf(is_terminated_flag)) if (has_behavior() && !getf(is_terminated_flag))
return false; return false;
CAF_LOG_DEBUG("actor either has no behavior or has set an exit reason"); CAF_LOG_DEBUG("actor either has no behavior or has set an exit reason");
...@@ -778,6 +784,7 @@ bool scheduled_actor::finalize() { ...@@ -778,6 +784,7 @@ bool scheduled_actor::finalize() {
bhvr_stack_.clear(); bhvr_stack_.clear();
bhvr_stack_.cleanup(); bhvr_stack_.cleanup();
cleanup(std::move(fail_state_), context()); cleanup(std::move(fail_state_), context());
CAF_ASSERT(getf(is_cleaned_up_flag));
return true; return true;
} }
......
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