Merge pull request #20935 from vespa-engine/vekterli/support-live-reconfig-of-dynamic-throttle-params

Support live reconfig of dynamic persistence throttling [run-systemtest]

4 files changed, 100 insertions, 20 deletions

diff --git a/configdefinitions/src/vespa/stor-filestor.def b/configdefinitions/src/vespa/stor-filestor.def
index 99f79a92c07..f1fc667ca9d 100644
--- a/configdefinitions/src/vespa/stor-filestor.def
+++ b/configdefinitions/src/vespa/stor-filestor.def
@@ -80,6 +80,22 @@ resource_usage_reporter_noise_level double default=0.001
 ##  - DYNAMIC uses DynamicThrottlePolicy under the hood and will block if the window
 ##    is full (if a blocking throttler API call is invoked).
 ##
+async_operation_throttler.type enum { UNLIMITED, DYNAMIC } default=UNLIMITED restart
+## Internal throttler tuning parameters that only apply when type == DYNAMIC:
+async_operation_throttler.window_size_increment int default=20
+async_operation_throttler.window_size_decrement_factor double default=1.2
+async_operation_throttler.window_size_backoff double default=0.95
+
+## Specify throttling used for async persistence operations. This throttling takes place
+## before operations are dispatched to Proton and serves as a limiter for how many
+## operations may be in flight in Proton's internal queues.
+##
+##  - UNLIMITED is, as it says on the tin, unlimited. Offers no actual throttling, but
+##    has near zero overhead and never blocks.
+##  - DYNAMIC uses DynamicThrottlePolicy under the hood and will block if the window
+##    is full (if a blocking throttler API call is invoked).
+##
+## TODO deprecate in favor of the async_operation_throttler struct instead.
 async_operation_throttler_type enum { UNLIMITED, DYNAMIC } default=UNLIMITED restart
 
 ## Specifies the extent the throttling window is increased by when the async throttle
@@ -88,4 +104,5 @@ async_operation_throttler_type enum { UNLIMITED, DYNAMIC } default=UNLIMITED res
 ## value, number of threads).
 ##
 ## Only applies if async_operation_throttler_type == DYNAMIC.
+## DEPRECATED! use the async_operation_throttler struct instead
 async_operation_dynamic_throttling_window_increment int default=20 restart
diff --git a/storage/src/vespa/storage/persistence/filestorage/filestormanager.cpp b/storage/src/vespa/storage/persistence/filestorage/filestormanager.cpp
index 03b16e75297..ec7793cdd8d 100644
--- a/storage/src/vespa/storage/persistence/filestorage/filestormanager.cpp
+++ b/storage/src/vespa/storage/persistence/filestorage/filestormanager.cpp
@@ -143,17 +143,29 @@ selectSequencer(StorFilestorConfig::ResponseSequencerType sequencerType) {
     }
 }
 
+vespalib::SharedOperationThrottler::DynamicThrottleParams
+dynamic_throttle_params_from_config(const StorFilestorConfig& config, size_t num_threads)
+{
+    const auto& cfg_params = config.asyncOperationThrottler;
+    auto win_size_incr = std::max(static_cast<size_t>(std::max(cfg_params.windowSizeIncrement, 1)), num_threads);
+
+    vespalib::SharedOperationThrottler::DynamicThrottleParams params;
+    params.window_size_increment        = win_size_incr;
+    params.min_window_size              = win_size_incr;
+    params.window_size_decrement_factor = cfg_params.windowSizeDecrementFactor;
+    params.window_size_backoff          = cfg_params.windowSizeBackoff;
+    return params;
+}
+
 std::unique_ptr<vespalib::SharedOperationThrottler>
 make_operation_throttler_from_config(const StorFilestorConfig& config, size_t num_threads)
 {
-    const bool use_dynamic_throttling = (config.asyncOperationThrottlerType == StorFilestorConfig::AsyncOperationThrottlerType::DYNAMIC);
+    // TODO only use struct config field instead once config model is updated
+    const bool use_dynamic_throttling = ((config.asyncOperationThrottlerType  == StorFilestorConfig::AsyncOperationThrottlerType::DYNAMIC) ||
+                                         (config.asyncOperationThrottler.type == StorFilestorConfig::AsyncOperationThrottler::Type::DYNAMIC));
     if (use_dynamic_throttling) {
-        auto config_win_size_incr = std::max(config.asyncOperationDynamicThrottlingWindowIncrement, 1);
-        auto win_size_increment = std::max(static_cast<size_t>(config_win_size_incr), num_threads);
-        vespalib::SharedOperationThrottler::DynamicThrottleParams params;
-        params.window_size_increment = win_size_increment;
-        params.min_window_size = win_size_increment;
-        return vespalib::SharedOperationThrottler::make_dynamic_throttler(params);
+        auto dyn_params = dynamic_throttle_params_from_config(config, num_threads);
+        return vespalib::SharedOperationThrottler::make_dynamic_throttler(dyn_params);
     } else {
         return vespalib::SharedOperationThrottler::make_unlimited_throttler();
     }
@@ -229,6 +241,10 @@ FileStorManager::configure(std::unique_ptr<StorFilestorConfig> config)
                                                                    *_filestorHandler, i % numStripes, _component));
         }
         _bucketExecutorRegistration = _provider->register_executor(std::make_shared<BucketExecutorWrapper>(*this));
+    } else {
+        assert(_filestorHandler);
+        auto updated_dyn_throttle_params = dynamic_throttle_params_from_config(*config, _threads.size());
+        _filestorHandler->operation_throttler().reconfigure_dynamic_throttling(updated_dyn_throttle_params);
     }
 }
 
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,