Copilot commented on code in PR #3229: URL: https://github.com/apache/brpc/pull/3229#discussion_r2869161173
########## src/brpc/backup_request_policy.cpp: ########## @@ -0,0 +1,174 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +#include "brpc/backup_request_policy.h" + +#include "butil/logging.h" +#include "bvar/reducer.h" +#include "bvar/window.h" +#include "butil/atomicops.h" +#include "butil/time.h" + +namespace brpc { + +// Standalone statistics module for tracking backup/total request ratio +// within a sliding time window. Each instance schedules two bvar::Window +// sampler tasks; keep this in mind for high channel-count deployments. +class BackupRateLimiter { +public: + BackupRateLimiter(double max_backup_ratio, + int window_size_seconds, + int update_interval_seconds) + : _max_backup_ratio(max_backup_ratio) + , _update_interval_us(update_interval_seconds * 1000000LL) + , _total_count() + , _backup_count() + , _total_window(&_total_count, window_size_seconds) + , _backup_window(&_backup_count, window_size_seconds) + , _cached_ratio(0.0) + , _last_update_us(0) { + } + + // All atomic operations use relaxed ordering intentionally. + // This is best-effort rate limiting: a slightly stale ratio is + // acceptable for approximate throttling. Within a single update interval, + // the cached ratio is not updated, so bursts up to update_interval_seconds + // in duration can exceed the configured max_backup_ratio transiently. + bool ShouldAllow() const { + const int64_t now_us = butil::cpuwide_time_us(); + int64_t last_us = _last_update_us.load(butil::memory_order_relaxed); + double ratio = _cached_ratio.load(butil::memory_order_relaxed); + + if (now_us - last_us >= _update_interval_us) { + if (_last_update_us.compare_exchange_strong( + last_us, now_us, butil::memory_order_relaxed)) { + int64_t total = _total_window.get_value(); + int64_t backup = _backup_window.get_value(); + // Fall back to cumulative counts when the window has no + // sampled data yet (cold-start within the first few seconds). + if (total <= 0) { + total = _total_count.get_value(); + backup = _backup_count.get_value(); + } + if (total > 0) { + ratio = static_cast<double>(backup) / total; + } else if (backup > 0) { + // Backups issued but no completions in window yet (latency spike). + // Be conservative to prevent backup storms. + ratio = 1.0; + } else { + // True cold-start: no traffic yet. Allow freely. + ratio = 0.0; + } + _cached_ratio.store(ratio, butil::memory_order_relaxed); + } + } + + bool allow = ratio < _max_backup_ratio; + if (allow) { + // Count backup decisions immediately for faster feedback + // during latency spikes (before RPCs complete). + _backup_count << 1; + } + return allow; + } Review Comment: The new rate limiter logic (cached ratio refresh, and transitions once total_count starts increasing via OnRPCEnd) is only partially covered by the added unit tests: current tests assert cold-start allow and suppression when backup>0/total==0, but don’t validate that calling OnRPCEnd() (and/or accumulating totals) brings the ratio back below the threshold and re-allows backups, nor that the computed ratio tracks backup/total as intended. Adding a small unit test that simulates completions by calling OnRPCEnd() and checks allow/suppress behavior across a few cycles would help prevent regressions. ########## src/brpc/controller.cpp: ########## @@ -351,8 +351,16 @@ void Controller::set_backup_request_ms(int64_t timeout_ms) { } int64_t Controller::backup_request_ms() const { - int timeout_ms = NULL != _backup_request_policy ? - _backup_request_policy->GetBackupRequestMs(this) : _backup_request_ms; + int timeout_ms = _backup_request_ms; + if (NULL != _backup_request_policy) { + const int32_t policy_ms = _backup_request_policy->GetBackupRequestMs(this); + // Any negative value means the policy defers to the channel-level + // backup_request_ms (set from ChannelOptions). The canonical sentinel + // is -1, but all negative values are treated the same way. + if (policy_ms >= 0) { + timeout_ms = policy_ms; + } Review Comment: Controller::backup_request_ms() now treats any negative value returned by BackupRequestPolicy::GetBackupRequestMs() as “defer to _backup_request_ms”. Previously, a policy could effectively disable backup requests by returning -1 (since Channel::CallMethod only arms the backup timer when backup_request_ms() >= 0). This is a behavior change for existing policies and can unexpectedly re-enable backups when ChannelOptions.backup_request_ms is set. Consider using a dedicated sentinel (e.g., UNSET_MAGIC_NUM) to mean “inherit”, and keep -1 as “disabled”, or otherwise clearly document the new contract for negative return values. ########## src/brpc/backup_request_policy.cpp: ########## @@ -0,0 +1,174 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +#include "brpc/backup_request_policy.h" + +#include "butil/logging.h" +#include "bvar/reducer.h" +#include "bvar/window.h" +#include "butil/atomicops.h" +#include "butil/time.h" + +namespace brpc { + +// Standalone statistics module for tracking backup/total request ratio +// within a sliding time window. Each instance schedules two bvar::Window +// sampler tasks; keep this in mind for high channel-count deployments. +class BackupRateLimiter { +public: + BackupRateLimiter(double max_backup_ratio, + int window_size_seconds, + int update_interval_seconds) + : _max_backup_ratio(max_backup_ratio) + , _update_interval_us(update_interval_seconds * 1000000LL) + , _total_count() + , _backup_count() + , _total_window(&_total_count, window_size_seconds) + , _backup_window(&_backup_count, window_size_seconds) + , _cached_ratio(0.0) + , _last_update_us(0) { + } + + // All atomic operations use relaxed ordering intentionally. + // This is best-effort rate limiting: a slightly stale ratio is + // acceptable for approximate throttling. Within a single update interval, + // the cached ratio is not updated, so bursts up to update_interval_seconds + // in duration can exceed the configured max_backup_ratio transiently. + bool ShouldAllow() const { + const int64_t now_us = butil::cpuwide_time_us(); + int64_t last_us = _last_update_us.load(butil::memory_order_relaxed); + double ratio = _cached_ratio.load(butil::memory_order_relaxed); + + if (now_us - last_us >= _update_interval_us) { + if (_last_update_us.compare_exchange_strong( + last_us, now_us, butil::memory_order_relaxed)) { + int64_t total = _total_window.get_value(); + int64_t backup = _backup_window.get_value(); + // Fall back to cumulative counts when the window has no + // sampled data yet (cold-start within the first few seconds). + if (total <= 0) { + total = _total_count.get_value(); + backup = _backup_count.get_value(); + } + if (total > 0) { + ratio = static_cast<double>(backup) / total; + } else if (backup > 0) { + // Backups issued but no completions in window yet (latency spike). + // Be conservative to prevent backup storms. + ratio = 1.0; + } else { + // True cold-start: no traffic yet. Allow freely. + ratio = 0.0; + } + _cached_ratio.store(ratio, butil::memory_order_relaxed); + } + } + + bool allow = ratio < _max_backup_ratio; + if (allow) { + // Count backup decisions immediately for faster feedback + // during latency spikes (before RPCs complete). + _backup_count << 1; + } + return allow; + } + + void OnRPCEnd(const Controller* /*controller*/) { + // Count all completed RPC legs (both original and backup RPCs). + // Backup decisions are counted in ShouldAllow() at decision time for + // faster feedback. As a result, the effective suppression threshold is + // (backup_count / total_legs), where total_legs includes both original + // and backup completions. + _total_count << 1; Review Comment: BackupRateLimiter::OnRPCEnd() currently increments _total_count by 1 per RPC, but the surrounding comments/PR description state the denominator is “all completed RPC legs (original + backup)”. Controller::OnRPCEnd is called once per user RPC, so this implementation does not count backup legs and will enforce a different ratio than documented. If the intended denominator is legs, consider incrementing by 1 + controller->has_backup_request() (or otherwise accounting for a backup leg) and update the comments accordingly. ```suggestion void OnRPCEnd(const Controller* controller) { // Count all completed RPC legs (both original and backup RPCs). // Backup decisions are counted in ShouldAllow() at decision time for // faster feedback. As a result, the effective suppression threshold is // (backup_count / total_legs), where total_legs includes both original // and backup completions. Each user RPC contributes 1 leg for the // original, plus 1 additional leg if a backup request was sent. int64_t legs = 1; if (controller != NULL && controller->has_backup_request()) { ++legs; } _total_count << legs; ``` ########## src/brpc/backup_request_policy.cpp: ########## @@ -0,0 +1,174 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +#include "brpc/backup_request_policy.h" + +#include "butil/logging.h" +#include "bvar/reducer.h" +#include "bvar/window.h" +#include "butil/atomicops.h" +#include "butil/time.h" + +namespace brpc { + +// Standalone statistics module for tracking backup/total request ratio +// within a sliding time window. Each instance schedules two bvar::Window +// sampler tasks; keep this in mind for high channel-count deployments. +class BackupRateLimiter { +public: + BackupRateLimiter(double max_backup_ratio, + int window_size_seconds, + int update_interval_seconds) + : _max_backup_ratio(max_backup_ratio) + , _update_interval_us(update_interval_seconds * 1000000LL) + , _total_count() + , _backup_count() + , _total_window(&_total_count, window_size_seconds) + , _backup_window(&_backup_count, window_size_seconds) + , _cached_ratio(0.0) + , _last_update_us(0) { + } + + // All atomic operations use relaxed ordering intentionally. + // This is best-effort rate limiting: a slightly stale ratio is + // acceptable for approximate throttling. Within a single update interval, + // the cached ratio is not updated, so bursts up to update_interval_seconds + // in duration can exceed the configured max_backup_ratio transiently. + bool ShouldAllow() const { + const int64_t now_us = butil::cpuwide_time_us(); + int64_t last_us = _last_update_us.load(butil::memory_order_relaxed); + double ratio = _cached_ratio.load(butil::memory_order_relaxed); + + if (now_us - last_us >= _update_interval_us) { + if (_last_update_us.compare_exchange_strong( + last_us, now_us, butil::memory_order_relaxed)) { + int64_t total = _total_window.get_value(); + int64_t backup = _backup_window.get_value(); + // Fall back to cumulative counts when the window has no + // sampled data yet (cold-start within the first few seconds). + if (total <= 0) { + total = _total_count.get_value(); + backup = _backup_count.get_value(); + } + if (total > 0) { + ratio = static_cast<double>(backup) / total; + } else if (backup > 0) { + // Backups issued but no completions in window yet (latency spike). + // Be conservative to prevent backup storms. + ratio = 1.0; + } else { + // True cold-start: no traffic yet. Allow freely. + ratio = 0.0; + } + _cached_ratio.store(ratio, butil::memory_order_relaxed); + } + } + + bool allow = ratio < _max_backup_ratio; + if (allow) { + // Count backup decisions immediately for faster feedback + // during latency spikes (before RPCs complete). + _backup_count << 1; + } + return allow; + } + + void OnRPCEnd(const Controller* /*controller*/) { + // Count all completed RPC legs (both original and backup RPCs). + // Backup decisions are counted in ShouldAllow() at decision time for + // faster feedback. As a result, the effective suppression threshold is + // (backup_count / total_legs), where total_legs includes both original + // and backup completions. + _total_count << 1; + } + +private: + double _max_backup_ratio; + int64_t _update_interval_us; + + bvar::Adder<int64_t> _total_count; + mutable bvar::Adder<int64_t> _backup_count; + bvar::Window<bvar::Adder<int64_t>> _total_window; + bvar::Window<bvar::Adder<int64_t>> _backup_window; + + mutable butil::atomic<double> _cached_ratio; + mutable butil::atomic<int64_t> _last_update_us; +}; + +// Internal BackupRequestPolicy that composes a BackupRateLimiter +// for ratio-based suppression. +class RateLimitedBackupPolicy : public BackupRequestPolicy { +public: + RateLimitedBackupPolicy(int32_t backup_request_ms, + double max_backup_ratio, + int window_size_seconds, + int update_interval_seconds) + : _backup_request_ms(backup_request_ms) + , _rate_limiter(max_backup_ratio, window_size_seconds, + update_interval_seconds) { + } + + int32_t GetBackupRequestMs(const Controller* /*controller*/) const override { + return _backup_request_ms; + } + + bool DoBackup(const Controller* /*controller*/) const override { + return _rate_limiter.ShouldAllow(); + } + + void OnRPCEnd(const Controller* controller) override { + _rate_limiter.OnRPCEnd(controller); + } + +private: + int32_t _backup_request_ms; + BackupRateLimiter _rate_limiter; +}; + +BackupRequestPolicy* CreateRateLimitedBackupPolicy( + const RateLimitedBackupPolicyOptions& options) { + if (options.backup_request_ms < -1) { + LOG(ERROR) << "Invalid backup_request_ms=" << options.backup_request_ms + << ", must be >= -1 (-1 means inherit from ChannelOptions)"; + return NULL; + } + if (options.max_backup_ratio <= 0 || options.max_backup_ratio > 1.0) { + LOG(ERROR) << "Invalid max_backup_ratio=" << options.max_backup_ratio + << ", must be in (0, 1]"; + return NULL; + } + if (options.window_size_seconds < 1 || options.window_size_seconds > 3600) { + LOG(ERROR) << "Invalid window_size_seconds=" << options.window_size_seconds + << ", must be in [1, 3600]"; + return NULL; + } + if (options.update_interval_seconds < 1) { + LOG(ERROR) << "Invalid update_interval_seconds=" + << options.update_interval_seconds << ", must be >= 1"; + return NULL; + } + if (options.update_interval_seconds > options.window_size_seconds) { + LOG(WARNING) << "update_interval_seconds=" << options.update_interval_seconds + << " exceeds window_size_seconds=" << options.window_size_seconds + << "; the ratio window will rarely refresh within its own period"; + } + return new RateLimitedBackupPolicy( + options.backup_request_ms, options.max_backup_ratio, + options.window_size_seconds, options.update_interval_seconds); Review Comment: CreateRateLimitedBackupPolicy() returns NULL for invalid parameters, but allocates with plain `new`, which may throw (or terminate if exceptions are disabled) on OOM. Elsewhere in brpc, similar factory-style functions typically use `new (std::nothrow)` when returning raw pointers (e.g. policy/auto_concurrency_limiter.cpp). Consider switching to `new (std::nothrow)` and returning NULL (optionally with LOG(ERROR)) if allocation fails, to keep failure semantics predictable. ########## docs/cn/backup_request.md: ########## @@ -39,6 +39,80 @@ my_func_latency << tm.u_elapsed(); // u代表微秒，还有s_elapsed(), m_elap // 好了，在/vars中会显示my_func_qps, my_func_latency, my_func_latency_cdf等很多计数器。 ``` +## Backup Request 限流 + +如需限制 backup request 的发送比例，可使用内置工厂函数创建限流策略，也可自行实现 `BackupRequestPolicy` 接口。 + +优先级顺序：`backup_request_policy` > `backup_request_ms`。 + +### 使用内置限流策略 + +调用 `CreateRateLimitedBackupPolicy` 创建限流策略，并将其设置到 `ChannelOptions.backup_request_policy`： + +```c++ +#include "brpc/backup_request_policy.h" +#include <memory> + +brpc::RateLimitedBackupPolicyOptions opts; +opts.backup_request_ms = 10; // 超过10ms未返回时发送backup请求 +opts.max_backup_ratio = 0.3; // backup请求比例上限30% +opts.window_size_seconds = 10; // 滑动窗口宽度（秒） +opts.update_interval_seconds = 5; // 缓存比例的刷新间隔（秒） + +// CreateRateLimitedBackupPolicy返回的指针由调用方负责释放。 +// 推荐使用unique_ptr管理生命周期，确保policy在Channel销毁后才释放。 +std::unique_ptr<brpc::BackupRequestPolicy> policy( + brpc::CreateRateLimitedBackupPolicy(opts)); + +brpc::ChannelOptions options; +options.backup_request_policy = policy.get(); // Channel不拥有该对象 +channel.Init(..., &options); +// policy在unique_ptr析构时自动释放，确保其生命周期长于channel即可。 +``` + +参数说明（`RateLimitedBackupPolicyOptions`）： + +| 字段 | 默认值 | 说明 | +|------|--------|------| +| `backup_request_ms` | -1 | 超时阈值（毫秒）；-1 表示继承 `ChannelOptions.backup_request_ms`，必须 >= -1。**仅在通过 `ChannelOptions.backup_request_policy` 设置策略时有效；通过 Controller 注入时必须显式指定 >= 0 的值。** | Review Comment: 参数表中写到“-1 … 仅在通过 ChannelOptions.backup_request_policy 设置时有效；通过 Controller 注入时必须显式指定 >= 0 的值”。但按当前 Controller::backup_request_ms() 的实现，如果调用方先通过 Controller::set_backup_request_ms() 设置了值，策略返回/配置为“继承”也可以工作。建议把这段表述改成：Controller 注入时需要某处显式提供 backup_request_ms（策略返回 >=0，或先设置 Controller.backup_request_ms 再让策略 defer），避免误导只能通过策略返回 >=0。 ```suggestion | `backup_request_ms` | -1 | 超时阈值（毫秒）；-1 表示继承 `ChannelOptions.backup_request_ms`，必须 >= -1。**通过 `ChannelOptions.backup_request_policy` 设置策略时，`-1` 表示继承 `ChannelOptions.backup_request_ms`；通过 Controller 注入时，需要某处显式提供 `backup_request_ms`（例如策略返回 `>= 0`，或先调用 `Controller::set_backup_request_ms()` 再让策略返回 `-1` 继承）。** | ``` ########## docs/en/backup_request.md: ########## @@ -39,6 +39,81 @@ my_func_latency << tm.u_elapsed(); // u represents for microsecond, and s_elaps // All work is done here. My_func_qps, my_func_latency, my_func_latency_cdf and many other counters would be shown in /vars. ``` +## Rate-limited backup requests + +To limit the ratio of backup requests sent, use the built-in factory function or implement the `BackupRequestPolicy` interface yourself. + +Priority order: `backup_request_policy` > `backup_request_ms`. + +### Using the built-in rate-limiting policy + +Call `CreateRateLimitedBackupPolicy` and set the result on `ChannelOptions.backup_request_policy`: + +```c++ +#include "brpc/backup_request_policy.h" +#include <memory> + +brpc::RateLimitedBackupPolicyOptions opts; +opts.backup_request_ms = 10; // send backup if RPC does not complete within 10ms +opts.max_backup_ratio = 0.3; // cap backup requests at 30% of total +opts.window_size_seconds = 10; // sliding window width in seconds +opts.update_interval_seconds = 5; // how often the cached ratio is refreshed + +// The caller owns the returned pointer. +// Use unique_ptr to manage the lifetime; ensure the policy outlives the channel. +std::unique_ptr<brpc::BackupRequestPolicy> policy( + brpc::CreateRateLimitedBackupPolicy(opts)); + +brpc::ChannelOptions options; +options.backup_request_policy = policy.get(); // NOT owned by channel +channel.Init(..., &options); +// policy is released automatically when unique_ptr goes out of scope, +// as long as it outlives the channel. +``` + +`RateLimitedBackupPolicyOptions` fields: + +| Field | Default | Description | +|-------|---------|-------------| +| `backup_request_ms` | -1 | Timeout threshold in ms; -1 means inherit from `ChannelOptions.backup_request_ms`; must be >= -1. **Only effective when the policy is set via `ChannelOptions.backup_request_policy`; controller-level injection always requires an explicit >= 0 value.** | Review Comment: The note in the options table says “-1 … Only effective when the policy is set via ChannelOptions.backup_request_policy; controller-level injection always requires an explicit >= 0 value.” With the current Controller::backup_request_ms() behavior, a controller-injected policy can still ‘inherit’ if the caller sets Controller::set_backup_request_ms() (or if _backup_request_ms is otherwise initialized). Consider rewording this to: controller-level injection requires an explicit backup_request_ms somewhere (either return >=0 from the policy, or set Controller.backup_request_ms and have the policy defer), rather than implying it must be returned by the policy. ```suggestion | `backup_request_ms` | -1 | Timeout threshold in ms; -1 means inherit from `ChannelOptions.backup_request_ms`; must be >= -1. **Inheritance is only applicable when the policy is set via `ChannelOptions.backup_request_policy`. For controller-level injection, there must be an explicit `backup_request_ms` defined somewhere (either the policy returns a value >= 0, or `Controller::backup_request_ms()` is set and the policy returns -1 to defer).** | ``` -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
