Commit 61bf48f2 authored by Dominik Charousset's avatar Dominik Charousset

Make sure actors with pending actions stay alive

parent a225c0d8
......@@ -849,6 +849,12 @@ private:
/// Maps the ID of incoming stream batches to local state that allows the
/// actor to push received batches into the local flow.
std::unordered_map<uint64_t, detail::stream_bridge_sub_ptr> stream_bridges_;
/// Special-purpose behavior for scheduled_actor::delay. When pushing an
/// action to the mailbox, we register this behavior as the response handler.
/// This is to make sure that actor does not terminate because it thinks it's
/// done before processing the delayed action.
behavior delay_bhvr_;
};
} // namespace caf
......@@ -491,10 +491,22 @@ void scheduled_actor::schedule(action what) {
}
void scheduled_actor::delay(action what) {
if (delayed_actions_this_run_++ < defaults::max_inline_actions_per_run)
// Happy path: push it to `actions_`, where we run it from `run_actions`.
if (delayed_actions_this_run_++ < defaults::max_inline_actions_per_run) {
actions_.emplace_back(std::move(what));
else
schedule(std::move(what));
return;
}
// Slow path: we send a "request" with the action to ourselves. The pending
// request makes sure that the action keeps the actor alive until processed.
if (!delay_bhvr_) {
delay_bhvr_ = behavior{[](action& f) {
CAF_LOG_DEBUG("run delayed action");
f.run();
}};
}
auto res_id = new_request_id(message_priority::normal).response_id();
enqueue(nullptr, res_id, make_message(std::move(what)), context_);
add_multiplexed_response_handler(res_id, delay_bhvr_);
}
disposable scheduled_actor::delay_until(steady_time_point abs_time,
......
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