diff options
Diffstat (limited to 'vespalib/src')
| -rw-r--r-- | vespalib/src/vespa/vespalib/util/shared_operation_throttler.cpp | 51 | ||||
| -rw-r--r-- | vespalib/src/vespa/vespalib/util/shared_operation_throttler.h | 22 |
2 files changed, 60 insertions, 13 deletions
diff --git a/vespalib/src/vespa/vespalib/util/shared_operation_throttler.cpp b/vespalib/src/vespa/vespalib/util/shared_operation_throttler.cpp index dd790bcaa0a..1df5f85bb6b 100644 --- a/vespalib/src/vespa/vespalib/util/shared_operation_throttler.cpp +++ b/vespalib/src/vespa/vespalib/util/shared_operation_throttler.cpp @@ -23,6 +23,7 @@ public: } uint32_t current_window_size() const noexcept override { return 0; } uint32_t waiting_threads() const noexcept override { return 0; } + void reconfigure_dynamic_throttling(const DynamicThrottleParams&) noexcept override { /* no-op */ } private: void release_one() noexcept override { /* no-op */ } }; @@ -38,6 +39,7 @@ private: * messagebus/src/main/java/com/yahoo/messagebus/DynamicThrottlePolicy.java */ class DynamicThrottlePolicy { + SharedOperationThrottler::DynamicThrottleParams _active_params; std::function<steady_time()> _time_provider; uint32_t _num_sent; uint32_t _num_ok; @@ -66,6 +68,8 @@ public: void set_min_window_size(double min_size) noexcept; void set_window_size_decrement_factor(double decrement_factor) noexcept; + void configure(const SharedOperationThrottler::DynamicThrottleParams& params) noexcept; + [[nodiscard]] uint32_t current_window_size() const noexcept { return static_cast<uint32_t>(_window_size); } @@ -74,6 +78,8 @@ public: void process_response(bool success) noexcept; private: + void internal_unconditional_configure(const SharedOperationThrottler::DynamicThrottleParams& params) noexcept; + [[nodiscard]] uint64_t current_time_as_millis() noexcept { return count_ms(_time_provider().time_since_epoch()); } @@ -81,7 +87,8 @@ private: DynamicThrottlePolicy::DynamicThrottlePolicy(const SharedOperationThrottler::DynamicThrottleParams& params, std::function<steady_time()> time_provider) - : _time_provider(std::move(time_provider)), + : _active_params(params), + _time_provider(std::move(time_provider)), _num_sent(0), _num_ok(0), _resize_rate(3.0), @@ -98,14 +105,7 @@ DynamicThrottlePolicy::DynamicThrottlePolicy(const SharedOperationThrottler::Dyn _weight(1), _local_max_throughput(0) { - // We use setters for convenience, since setting one parameter may imply setting others, - // based on it, and there's frequently min/max capping of values. - set_window_size_increment(params.window_size_increment); - set_min_window_size(params.min_window_size); - set_max_window_size(params.max_window_size); - set_resize_rate(params.resize_rate); - set_window_size_decrement_factor(params.window_size_decrement_factor); - set_window_size_backoff(params.window_size_backoff); + internal_unconditional_configure(_active_params); } void @@ -146,6 +146,31 @@ DynamicThrottlePolicy::set_window_size_decrement_factor(double decrement_factor) _decrement_factor = decrement_factor; } +void +DynamicThrottlePolicy::internal_unconditional_configure(const SharedOperationThrottler::DynamicThrottleParams& params) noexcept +{ + // We use setters for convenience, since setting one parameter may imply setting others, + // based on it, and there's frequently min/max capping of values. + set_window_size_increment(params.window_size_increment); + set_min_window_size(params.min_window_size); + set_max_window_size(params.max_window_size); + set_resize_rate(params.resize_rate); + set_window_size_decrement_factor(params.window_size_decrement_factor); + set_window_size_backoff(params.window_size_backoff); +} + +void +DynamicThrottlePolicy::configure(const SharedOperationThrottler::DynamicThrottleParams& params) noexcept +{ + // To avoid any noise where setting parameters on the throttler may implicitly reduce the + // current window size (even though this isn't _currently_ the case), don't invoke any internal + // reconfiguration code unless the parameters have actually changed. + if (params != _active_params) { + internal_unconditional_configure(params); + _active_params = params; + } +} + bool DynamicThrottlePolicy::has_spare_capacity(uint32_t pending_count) noexcept { @@ -225,6 +250,7 @@ public: Token try_acquire_one() noexcept override; uint32_t current_window_size() const noexcept override; uint32_t waiting_threads() const noexcept override; + void reconfigure_dynamic_throttling(const DynamicThrottleParams& params) noexcept override; private: void release_one() noexcept override; // Non-const since actually checking the send window of a dynamic throttler might change @@ -340,6 +366,13 @@ DynamicOperationThrottler::waiting_threads() const noexcept return _waiting_threads; } +void +DynamicOperationThrottler::reconfigure_dynamic_throttling(const DynamicThrottleParams& params) noexcept +{ + std::unique_lock lock(_mutex); + _throttle_policy.configure(params); +} + } // anonymous namespace std::unique_ptr<SharedOperationThrottler> diff --git a/vespalib/src/vespa/vespalib/util/shared_operation_throttler.h b/vespalib/src/vespa/vespalib/util/shared_operation_throttler.h index 2a8a42dd1ba..030935339ed 100644 --- a/vespalib/src/vespa/vespalib/util/shared_operation_throttler.h +++ b/vespalib/src/vespa/vespalib/util/shared_operation_throttler.h @@ -51,9 +51,15 @@ public: virtual ~SharedOperationThrottler() = default; - // All methods are thread safe + // Acquire a valid throttling token, uninterruptedly blocking until one can be obtained. [[nodiscard]] virtual Token blocking_acquire_one() noexcept = 0; + // Attempt to acquire a valid throttling token, waiting up to `timeout` for one to be + // available. If the timeout is exceeded without any tokens becoming available, an + // invalid token will be returned. [[nodiscard]] virtual Token blocking_acquire_one(vespalib::duration timeout) noexcept = 0; + // Attempt to acquire a valid throttling token if one is immediately available. + // An invalid token will be returned if none is available. Never blocks (other than + // when contending for the internal throttler mutex). [[nodiscard]] virtual Token try_acquire_one() noexcept = 0; // May return 0, in which case the window size is unlimited. @@ -62,9 +68,6 @@ public: // Exposed for unit testing only. [[nodiscard]] virtual uint32_t waiting_threads() const noexcept = 0; - // Creates a throttler that does exactly zero throttling (but also has zero overhead and locking) - static std::unique_ptr<SharedOperationThrottler> make_unlimited_throttler(); - struct DynamicThrottleParams { uint32_t window_size_increment = 20; uint32_t min_window_size = 20; @@ -72,8 +75,19 @@ public: double resize_rate = 3.0; double window_size_decrement_factor = 1.2; double window_size_backoff = 0.95; + + bool operator==(const DynamicThrottleParams&) const noexcept = default; + bool operator!=(const DynamicThrottleParams&) const noexcept = default; }; + // No-op if underlying throttler does not use a dynamic policy, or if the supplied + // parameters are equal to the current configuration. + // FIXME leaky abstraction alert! + virtual void reconfigure_dynamic_throttling(const DynamicThrottleParams& params) noexcept = 0; + + // Creates a throttler that does exactly zero throttling (but also has zero overhead and locking) + static std::unique_ptr<SharedOperationThrottler> make_unlimited_throttler(); + // Creates a throttler that uses a DynamicThrottlePolicy under the hood static std::unique_ptr<SharedOperationThrottler> make_dynamic_throttler(const DynamicThrottleParams& params); static std::unique_ptr<SharedOperationThrottler> make_dynamic_throttler(const DynamicThrottleParams& params, |