Copilot commented on code in PR #3526:
URL: https://github.com/apache/brpc/pull/3526#discussion_r3964178725
##########
src/brpc/load_balancer.cpp:
##########
@@ -30,7 +35,74 @@ DEFINE_int32(default_weight_of_wlb, 0, "Default weight value
of Weighted LoadBal
"problems when user is using wlb but forgot to set the weights of
some of their "
"downstream instances. Then these instances will be set
default_weight_of_wlb as "
"their weights. wlb policy degradation is not enabled by
default.");
+DEFINE_int64(lb_warmup_ms, 0,
+ "When positive, a server newly added to a LoadBalancer gets "
+ "lb_warmup_min_weight of its normal traffic share at first and "
+ "ramps up to 100% over this period(ms). 0 disables the warm-up");
+DEFINE_double(lb_warmup_curve, 1.0,
+ "Shape of the warm-up ramp: the weight multiplier is "
+ "max(lb_warmup_min_weight, progress^lb_warmup_curve) where
progress rises "
+ "linearly from 0 to 1 over lb_warmup_ms. Must be positive: 1
ramps "
+ "linearly, larger values keep a new server colder for longer");
+static bool ValidateWarmupMs(const char*, int64_t v) {
+ // Must survive the conversion to microseconds.
+ return v >= 0 && v <= INT64_MAX / 1000;
+}
BRPC_VALIDATE_GFLAG(show_lb_in_vars, PassValidate);
+BRPC_VALIDATE_GFLAG(lb_warmup_ms, ValidateWarmupMs);
+DEFINE_double(lb_warmup_min_weight, 0.1,
+ "Floor of the warm-up multiplier, in (0, 1]: the share of "
+ "normal traffic a server gets right after joining, so that "
+ "it still receives a trickle and latency-based policies keep "
+ "observing it");
+static bool ValidateWarmupCurve(const char*, double v) {
+ return v > 0.0;
+}
+static bool ValidateWarmupMinWeight(const char*, double v) {
+ return v > 0.0 && v <= 1.0;
+}
+BRPC_VALIDATE_GFLAG(lb_warmup_curve, ValidateWarmupCurve);
+BRPC_VALIDATE_GFLAG(lb_warmup_min_weight, ValidateWarmupMinWeight);
+
+static int64_t (*g_lb_clock_us)() = NULL;
+
+int64_t LoadBalancerJoinTimeUs() {
+ return g_lb_clock_us != NULL ? g_lb_clock_us() : butil::gettimeofday_us();
+}
+
+void SetLoadBalancerClockForTesting(int64_t (*clock_us)()) {
+ g_lb_clock_us = clock_us;
+}
+
+
+double WarmupMultiplierImpl(int64_t join_time_us, int64_t now_us) {
+ const int64_t warmup_us = FLAGS_lb_warmup_ms * 1000L;
+ if (warmup_us <= 0 || join_time_us <= 0) {
+ return 1.0;
+ }
+ if (now_us <= 0) {
+ now_us = butil::gettimeofday_us();
Review Comment:
The code introduces a configurable clock for stamping `join_time_us`, but
when `now_us <= 0` the ramp uses the real wall clock
(`butil::gettimeofday_us()`) rather than the same clock source used for
join-stamping. This can make the ramp inconsistent under tests that rely on the
injected clock and can be surprising for callers who pass `now_us == 0`
expecting a consistent time base. A concrete fix is to use the same clock
source as stamping (e.g., call `LoadBalancerJoinTimeUs()` or introduce a
separate “now” helper that respects the injected clock) when `now_us <= 0`.
##########
src/brpc/load_balancer.cpp:
##########
@@ -30,7 +35,74 @@ DEFINE_int32(default_weight_of_wlb, 0, "Default weight value
of Weighted LoadBal
"problems when user is using wlb but forgot to set the weights of
some of their "
"downstream instances. Then these instances will be set
default_weight_of_wlb as "
"their weights. wlb policy degradation is not enabled by
default.");
+DEFINE_int64(lb_warmup_ms, 0,
+ "When positive, a server newly added to a LoadBalancer gets "
+ "lb_warmup_min_weight of its normal traffic share at first and "
+ "ramps up to 100% over this period(ms). 0 disables the warm-up");
+DEFINE_double(lb_warmup_curve, 1.0,
+ "Shape of the warm-up ramp: the weight multiplier is "
+ "max(lb_warmup_min_weight, progress^lb_warmup_curve) where
progress rises "
+ "linearly from 0 to 1 over lb_warmup_ms. Must be positive: 1
ramps "
+ "linearly, larger values keep a new server colder for longer");
+static bool ValidateWarmupMs(const char*, int64_t v) {
+ // Must survive the conversion to microseconds.
+ return v >= 0 && v <= INT64_MAX / 1000;
+}
BRPC_VALIDATE_GFLAG(show_lb_in_vars, PassValidate);
+BRPC_VALIDATE_GFLAG(lb_warmup_ms, ValidateWarmupMs);
+DEFINE_double(lb_warmup_min_weight, 0.1,
+ "Floor of the warm-up multiplier, in (0, 1]: the share of "
+ "normal traffic a server gets right after joining, so that "
+ "it still receives a trickle and latency-based policies keep "
+ "observing it");
+static bool ValidateWarmupCurve(const char*, double v) {
+ return v > 0.0;
+}
+static bool ValidateWarmupMinWeight(const char*, double v) {
+ return v > 0.0 && v <= 1.0;
+}
+BRPC_VALIDATE_GFLAG(lb_warmup_curve, ValidateWarmupCurve);
+BRPC_VALIDATE_GFLAG(lb_warmup_min_weight, ValidateWarmupMinWeight);
+
+static int64_t (*g_lb_clock_us)() = NULL;
+
+int64_t LoadBalancerJoinTimeUs() {
+ return g_lb_clock_us != NULL ? g_lb_clock_us() : butil::gettimeofday_us();
+}
+
+void SetLoadBalancerClockForTesting(int64_t (*clock_us)()) {
+ g_lb_clock_us = clock_us;
+}
Review Comment:
`g_lb_clock_us` is a non-atomic global function pointer read/written without
synchronization. If `SetLoadBalancerClockForTesting` can run concurrently with
`LoadBalancerJoinTimeUs()` (or if tests run in parallel), this is a data race.
Consider either (a) making this hook strictly test-only (e.g., guarded by a
test macro / moved into an internal-only header) and documenting it must be set
before any threads start, or (b) making the pointer access thread-safe (e.g.,
`std::atomic` for the pointer or a mutex-protected getter/setter).
##########
test/brpc_lb_warmup_unittest.cpp:
##########
@@ -0,0 +1,354 @@
+// 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 <map>
+#include <sstream>
+#include <vector>
+#include <gflags/gflags.h>
+#include <gtest/gtest.h>
+#include "butil/fast_rand.h"
+#include "butil/time.h"
+#include "brpc/socket.h"
+#include "brpc/load_balancer.h"
+#include "brpc/policy/round_robin_load_balancer.h"
+#include "brpc/policy/randomized_load_balancer.h"
+#include "brpc/policy/weighted_round_robin_load_balancer.h"
+#include "brpc/policy/consistent_hashing_load_balancer.h"
+#include "brpc/policy/locality_aware_load_balancer.h"
+#include "brpc/policy/p2c_ewma_load_balancer.h"
+
+namespace brpc {
+DECLARE_double(lb_warmup_curve);
+DECLARE_double(lb_warmup_min_weight);
+}
+
+namespace {
+
+class SaveRecycle : public brpc::SocketUser {
+ void BeforeRecycle(brpc::Socket* s) { delete this; }
Review Comment:
`BeforeRecycle` should be marked `override` to ensure it correctly matches
the base virtual signature, and the unused parameter `s` may trigger warnings
under stricter build flags. Consider updating the method to `void
BeforeRecycle(brpc::Socket* /*s*/) override { delete this; }` (or equivalent
project convention) to make intent explicit and reduce warning risk.
##########
src/brpc/policy/weighted_round_robin_load_balancer.h:
##########
@@ -79,7 +82,8 @@ class WeightedRoundRobinLoadBalancer : public LoadBalancer {
static size_t BatchRemove(Servers& bg, const std::vector<ServerId>&
servers);
static SocketId GetServerInNextStride(const std::vector<Server>&
server_list,
const std::unordered_set<SocketId>&
filter,
- TLS& tls);
+ TLS& tls,
+ size_t* index);
Review Comment:
`GetServerInNextStride` now requires a `size_t* index` out-param and the
implementation unconditionally dereferences it. This is error-prone because the
signature permits passing `nullptr` (and newly introduces that failure mode).
Prefer making it a reference (`size_t& index`) or add an explicit runtime
check/contract (e.g., assert `index != NULL`) to prevent undefined behavior.
##########
test/brpc_lb_warmup_unittest.cpp:
##########
@@ -0,0 +1,354 @@
+// 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 <map>
+#include <sstream>
+#include <vector>
+#include <gflags/gflags.h>
+#include <gtest/gtest.h>
+#include "butil/fast_rand.h"
+#include "butil/time.h"
+#include "brpc/socket.h"
+#include "brpc/load_balancer.h"
+#include "brpc/policy/round_robin_load_balancer.h"
+#include "brpc/policy/randomized_load_balancer.h"
+#include "brpc/policy/weighted_round_robin_load_balancer.h"
+#include "brpc/policy/consistent_hashing_load_balancer.h"
+#include "brpc/policy/locality_aware_load_balancer.h"
+#include "brpc/policy/p2c_ewma_load_balancer.h"
+
+namespace brpc {
+DECLARE_double(lb_warmup_curve);
+DECLARE_double(lb_warmup_min_weight);
+}
+
+namespace {
+
+class SaveRecycle : public brpc::SocketUser {
+ void BeforeRecycle(brpc::Socket* s) { delete this; }
+};
+
+brpc::ServerId CreateServer(const char* addr, const char* tag = "") {
+ butil::EndPoint point;
+ EXPECT_EQ(0, str2endpoint(addr, &point));
+ brpc::ServerId id(8888);
+ brpc::SocketOptions options;
+ options.remote_side = point;
+ options.user = new SaveRecycle;
+ EXPECT_EQ(0, brpc::Socket::Create(options, &id.id));
+ id.tag = tag;
+ return id;
+}
+
+// Select `count' times at `now_us' and return times each server was chosen.
+// Feeds back immediately when the LB asks for it(la).
+std::map<brpc::SocketId, int> CountShares(
+ brpc::LoadBalancer* lb, int count, int64_t now_us,
+ bool changable_weights = false, bool with_request_code = false) {
+ std::map<brpc::SocketId, int> shares;
+ for (int i = 0; i < count; ++i) {
+ brpc::LoadBalancer::SelectIn in = {
+ now_us, changable_weights, with_request_code,
Review Comment:
Fix typo in parameter name: `changable_weights` → `changeable_weights`
(keeps intent clear and avoids propagating the misspelling into future test
helpers/call sites).
--
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]
