Unverified Commit cec81950 authored by Noir's avatar Noir Committed by GitHub

Merge pull request #1381

Limit how many actions actors may run in one go
parents b3164d5b 0e5a90e1
...@@ -41,6 +41,10 @@ is based on [Keep a Changelog](https://keepachangelog.com). ...@@ -41,6 +41,10 @@ is based on [Keep a Changelog](https://keepachangelog.com).
- The `mcast` and `ucast` operators now stop calling `on_next` immediately when - The `mcast` and `ucast` operators now stop calling `on_next` immediately when
disposed. disposed.
- Actors no longer terminate despite having open streams (#1377). - Actors no longer terminate despite having open streams (#1377).
- Actors reading from external sources such as SPSC buffers via a local flow
could end up in a long-running read loop. To avoid potentially starving other
actors or activities, scheduled actors now limit the amount of actions that
may run in one iteration (#1364).
## [0.19.0-rc.1] - 2022-10-31 ## [0.19.0-rc.1] - 2022-10-31
......
...@@ -32,6 +32,11 @@ constexpr parameter<T> make_parameter(std::string_view name, T fallback) { ...@@ -32,6 +32,11 @@ constexpr parameter<T> make_parameter(std::string_view name, T fallback) {
return {name, fallback}; return {name, fallback};
} }
/// Configures how many actions scheduled_actor::delay may add to the internal
/// queue for scheduled_actor::run_actions before being forced to push them to
/// the mailbox instead.
constexpr auto max_inline_actions_per_run = size_t{10};
} // namespace caf::defaults } // namespace caf::defaults
namespace caf::defaults::stream { namespace caf::defaults::stream {
......
...@@ -829,6 +829,11 @@ private: ...@@ -829,6 +829,11 @@ private:
/// message. /// message.
std::vector<action> actions_; std::vector<action> actions_;
/// Counter for scheduled_actor::delay to make sure
/// scheduled_actor::run_actions does not end up in a busy loop that might
/// starve other activities.
size_t delayed_actions_this_run_ = 0;
/// Stores ongoing activities such as flows that block the actor from /// Stores ongoing activities such as flows that block the actor from
/// terminating. /// terminating.
std::vector<disposable> watched_disposables_; std::vector<disposable> watched_disposables_;
...@@ -844,6 +849,12 @@ private: ...@@ -844,6 +849,12 @@ private:
/// Maps the ID of incoming stream batches to local state that allows the /// Maps the ID of incoming stream batches to local state that allows the
/// actor to push received batches into the local flow. /// actor to push received batches into the local flow.
std::unordered_map<uint64_t, detail::stream_bridge_sub_ptr> stream_bridges_; 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 } // namespace caf
...@@ -491,7 +491,22 @@ void scheduled_actor::schedule(action what) { ...@@ -491,7 +491,22 @@ void scheduled_actor::schedule(action what) {
} }
void scheduled_actor::delay(action what) { void scheduled_actor::delay(action what) {
// 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)); actions_.emplace_back(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, disposable scheduled_actor::delay_until(steady_time_point abs_time,
...@@ -971,6 +986,7 @@ void scheduled_actor::deregister_stream(uint64_t stream_id) { ...@@ -971,6 +986,7 @@ void scheduled_actor::deregister_stream(uint64_t stream_id) {
} }
void scheduled_actor::run_actions() { void scheduled_actor::run_actions() {
delayed_actions_this_run_ = 0;
if (!actions_.empty()) { if (!actions_.empty()) {
// Note: can't use iterators here since actions may add to the vector. // Note: can't use iterators here since actions may add to the vector.
for (auto index = size_t{0}; index < actions_.size(); ++index) { for (auto index = size_t{0}; index < actions_.size(); ++index) {
......
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