rajvarun77 commented on code in PR #3526:
URL: https://github.com/apache/brpc/pull/3526#discussion_r3964197720
##########
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:
Done in c8504917: the clock pointer is a `butil::atomic` now (relaxed
load/store); header comment says it is test-only and not meant to change while
LoadBalancers are in use.
##########
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:
Done in c8504917: the `now_us <= 0` path reads the same injectable clock;
hook renamed `LoadBalancerNowUs()` since it serves both purposes.
##########
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:
Done in c8504917: `size_t& index`.
##########
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:
Done in c8504917: `void BeforeRecycle(brpc::Socket*) override`.
##########
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:
Done in c8504917.
--
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]
