Commit a225c0d8 authored by Dominik Charousset's avatar Dominik Charousset

Limit how many actions actors may run in one go

Limit the maximum number of actions that `scheduled_actor::delay` may
add to the internal queue before being forced to push them to the
mailbox instead. The new upper bound is 10. This should be small enough
to break long-running activity chains early but still allow for small
bursts to run without interruption. When reaching the limit, `delay`
schedules the action instead of adding it to the internal queue. This
avoids the risk of `scheduled_actor::run_actions` entering a busy loop
that may starve other activities.
parent ea9fb5d4
...@@ -32,6 +32,10 @@ is based on [Keep a Changelog](https://keepachangelog.com). ...@@ -32,6 +32,10 @@ is based on [Keep a Changelog](https://keepachangelog.com).
operators that use the `ucast` operator internally. operators that use the `ucast` operator internally.
- 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 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_;
......
...@@ -491,7 +491,10 @@ void scheduled_actor::schedule(action what) { ...@@ -491,7 +491,10 @@ void scheduled_actor::schedule(action what) {
} }
void scheduled_actor::delay(action what) { void scheduled_actor::delay(action what) {
if (delayed_actions_this_run_++ < defaults::max_inline_actions_per_run)
actions_.emplace_back(std::move(what)); actions_.emplace_back(std::move(what));
else
schedule(std::move(what));
} }
disposable scheduled_actor::delay_until(steady_time_point abs_time, disposable scheduled_actor::delay_until(steady_time_point abs_time,
...@@ -946,6 +949,7 @@ void scheduled_actor::deregister_stream(uint64_t stream_id) { ...@@ -946,6 +949,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