rajvarun77 commented on code in PR #3526:
URL: https://github.com/apache/brpc/pull/3526#discussion_r3962829939
##########
src/brpc/policy/consistent_hashing_load_balancer.h:
##########
@@ -47,6 +47,12 @@ class ConsistentHashingLoadBalancer : public LoadBalancer {
uint32_t hash;
ServerId server_sock;
butil::EndPoint server_addr; // To make sorting stable among all
clients
+ // Time when the server was added, for the warm-up ramp. Not part
+ // of ordering/equality: AddBatch merges with std::set_union, which
+ // keeps the existing node when a server is added again without
+ // having been removed, so its stamp is preserved. RemoveServer
+ // followed by AddServer rebuilds the nodes with a fresh stamp.
+ int64_t join_time_us;
Review Comment:
Done in d152ee57: `join_time_us = 0` default (0 = unstamped, never ramped).
##########
test/brpc_lb_warmup_unittest.cpp:
##########
@@ -0,0 +1,302 @@
+// 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 <unistd.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/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,
+ with_request_code ? butil::fast_rand() % UINT_MAX : 0u, nullptr };
+ brpc::SocketUniquePtr ptr;
+ brpc::LoadBalancer::SelectOut out(&ptr);
+ if (lb->SelectServer(in, &out) != 0) {
+ continue;
+ }
+ ++shares[ptr->id()];
+ if (out.need_feedback) {
+ brpc::LoadBalancer::CallInfo info;
+ info.begin_time_us = now_us;
+ info.server_id = ptr->id();
+ info.error_code = 0;
+ info.controller = nullptr;
+ lb->Feedback(info);
+ }
+ }
+ return shares;
+}
+
+class LbWarmupTest : public ::testing::Test {
+protected:
+ // Restores every flag the tests touch when the fixture is destroyed.
+ GFLAGS_NAMESPACE::FlagSaver _flag_saver;
+};
+
+TEST_F(LbWarmupTest, disabled_by_default) {
+ ASSERT_EQ(0, brpc::FLAGS_lb_warmup_ms);
+ // Any stamp maps to full weight when disabled.
+ ASSERT_DOUBLE_EQ(1.0, brpc::WarmupMultiplier(butil::gettimeofday_us(), 0));
+
+ // A just-added server gets its full share right away.
+ brpc::policy::RoundRobinLoadBalancer lb;
+ const brpc::ServerId a = CreateServer("127.0.0.1:8101");
+ const brpc::ServerId b = CreateServer("127.0.0.1:8102");
+ ASSERT_TRUE(lb.AddServer(a));
+ ASSERT_TRUE(lb.AddServer(b));
+ std::map<brpc::SocketId, int> shares =
+ CountShares(&lb, 2000, butil::gettimeofday_us());
+ ASSERT_GT(shares[a.id], 600);
+ ASSERT_GT(shares[b.id], 600);
+}
+
+TEST_F(LbWarmupTest, multiplier_math) {
+ brpc::FLAGS_lb_warmup_ms = 10000;
+ const int64_t join_us = 1000000;
+
+ // Unstamped server is never ramped.
+ ASSERT_DOUBLE_EQ(1.0, brpc::WarmupMultiplier(0, join_us));
+ // Ramp floor right after joining and on backward clock jumps.
+ ASSERT_DOUBLE_EQ(0.1, brpc::WarmupMultiplier(join_us, join_us));
+ ASSERT_DOUBLE_EQ(0.1, brpc::WarmupMultiplier(join_us + 5000000, join_us));
+ // Linear ramp.
+ ASSERT_DOUBLE_EQ(0.1, brpc::WarmupMultiplier(join_us, join_us + 500000));
+ ASSERT_DOUBLE_EQ(0.3, brpc::WarmupMultiplier(join_us, join_us + 3000000));
+ ASSERT_DOUBLE_EQ(0.5, brpc::WarmupMultiplier(join_us, join_us + 5000000));
+ ASSERT_DOUBLE_EQ(1.0, brpc::WarmupMultiplier(join_us, join_us + 10000000));
+ ASSERT_DOUBLE_EQ(1.0, brpc::WarmupMultiplier(join_us, join_us + 60000000));
+
+ // Curve shaping: >1 is more conservative early, <1 more aggressive.
+ brpc::FLAGS_lb_warmup_curve = 2.0;
+ ASSERT_DOUBLE_EQ(0.25, brpc::WarmupMultiplier(join_us, join_us + 5000000));
+ brpc::FLAGS_lb_warmup_curve = 0.5;
+ ASSERT_DOUBLE_EQ(0.5, brpc::WarmupMultiplier(join_us, join_us + 2500000));
Review Comment:
Done in d152ee57: the two curve cases use `ASSERT_NEAR(…, 1e-9)`.
##########
test/brpc_lb_warmup_unittest.cpp:
##########
@@ -0,0 +1,302 @@
+// 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 <unistd.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/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,
+ with_request_code ? butil::fast_rand() % UINT_MAX : 0u, nullptr };
+ brpc::SocketUniquePtr ptr;
+ brpc::LoadBalancer::SelectOut out(&ptr);
+ if (lb->SelectServer(in, &out) != 0) {
+ continue;
+ }
+ ++shares[ptr->id()];
+ if (out.need_feedback) {
+ brpc::LoadBalancer::CallInfo info;
+ info.begin_time_us = now_us;
+ info.server_id = ptr->id();
+ info.error_code = 0;
+ info.controller = nullptr;
+ lb->Feedback(info);
+ }
+ }
+ return shares;
+}
+
+class LbWarmupTest : public ::testing::Test {
+protected:
+ // Restores every flag the tests touch when the fixture is destroyed.
+ GFLAGS_NAMESPACE::FlagSaver _flag_saver;
+};
+
+TEST_F(LbWarmupTest, disabled_by_default) {
+ ASSERT_EQ(0, brpc::FLAGS_lb_warmup_ms);
+ // Any stamp maps to full weight when disabled.
+ ASSERT_DOUBLE_EQ(1.0, brpc::WarmupMultiplier(butil::gettimeofday_us(), 0));
+
+ // A just-added server gets its full share right away.
+ brpc::policy::RoundRobinLoadBalancer lb;
+ const brpc::ServerId a = CreateServer("127.0.0.1:8101");
+ const brpc::ServerId b = CreateServer("127.0.0.1:8102");
+ ASSERT_TRUE(lb.AddServer(a));
+ ASSERT_TRUE(lb.AddServer(b));
+ std::map<brpc::SocketId, int> shares =
+ CountShares(&lb, 2000, butil::gettimeofday_us());
+ ASSERT_GT(shares[a.id], 600);
+ ASSERT_GT(shares[b.id], 600);
+}
+
+TEST_F(LbWarmupTest, multiplier_math) {
+ brpc::FLAGS_lb_warmup_ms = 10000;
+ const int64_t join_us = 1000000;
+
+ // Unstamped server is never ramped.
+ ASSERT_DOUBLE_EQ(1.0, brpc::WarmupMultiplier(0, join_us));
+ // Ramp floor right after joining and on backward clock jumps.
+ ASSERT_DOUBLE_EQ(0.1, brpc::WarmupMultiplier(join_us, join_us));
+ ASSERT_DOUBLE_EQ(0.1, brpc::WarmupMultiplier(join_us + 5000000, join_us));
+ // Linear ramp.
+ ASSERT_DOUBLE_EQ(0.1, brpc::WarmupMultiplier(join_us, join_us + 500000));
+ ASSERT_DOUBLE_EQ(0.3, brpc::WarmupMultiplier(join_us, join_us + 3000000));
+ ASSERT_DOUBLE_EQ(0.5, brpc::WarmupMultiplier(join_us, join_us + 5000000));
+ ASSERT_DOUBLE_EQ(1.0, brpc::WarmupMultiplier(join_us, join_us + 10000000));
+ ASSERT_DOUBLE_EQ(1.0, brpc::WarmupMultiplier(join_us, join_us + 60000000));
+
+ // Curve shaping: >1 is more conservative early, <1 more aggressive.
+ brpc::FLAGS_lb_warmup_curve = 2.0;
+ ASSERT_DOUBLE_EQ(0.25, brpc::WarmupMultiplier(join_us, join_us + 5000000));
+ brpc::FLAGS_lb_warmup_curve = 0.5;
+ ASSERT_DOUBLE_EQ(0.5, brpc::WarmupMultiplier(join_us, join_us + 2500000));
+ brpc::FLAGS_lb_warmup_curve = 1.0;
+
+ // The floor is configurable.
+ brpc::FLAGS_lb_warmup_min_weight = 0.3;
+ ASSERT_DOUBLE_EQ(0.3, brpc::WarmupMultiplier(join_us, join_us));
+ ASSERT_DOUBLE_EQ(0.3, brpc::WarmupMultiplier(join_us, join_us + 1000000));
+ ASSERT_DOUBLE_EQ(0.5, brpc::WarmupMultiplier(join_us, join_us + 5000000));
+ brpc::FLAGS_lb_warmup_min_weight = 1.0;
+ ASSERT_DOUBLE_EQ(1.0, brpc::WarmupMultiplier(join_us, join_us));
+
+ brpc::FLAGS_lb_warmup_curve = 1.0;
+ brpc::FLAGS_lb_warmup_ms = 0;
+ ASSERT_DOUBLE_EQ(1.0, brpc::WarmupMultiplier(join_us, join_us));
+}
+
+TEST_F(LbWarmupTest, flag_validation) {
+ // Window must be non-negative and convertible to microseconds.
+ ASSERT_FALSE(GFLAGS_NAMESPACE::SetCommandLineOption("lb_warmup_ms",
"30000").empty());
+ ASSERT_TRUE(GFLAGS_NAMESPACE::SetCommandLineOption("lb_warmup_ms",
"-1").empty());
+ ASSERT_TRUE(GFLAGS_NAMESPACE::SetCommandLineOption(
+ "lb_warmup_ms", "9223372036854775807").empty());
+ ASSERT_EQ(30000, brpc::FLAGS_lb_warmup_ms);
+ // Curve must be positive; the floor must be in (0, 1].
+ ASSERT_FALSE(GFLAGS_NAMESPACE::SetCommandLineOption("lb_warmup_curve",
"2").empty());
+ ASSERT_TRUE(GFLAGS_NAMESPACE::SetCommandLineOption("lb_warmup_curve",
"0").empty());
+ ASSERT_TRUE(GFLAGS_NAMESPACE::SetCommandLineOption("lb_warmup_curve",
"-1").empty());
+ ASSERT_DOUBLE_EQ(2.0, brpc::FLAGS_lb_warmup_curve);
+
ASSERT_FALSE(GFLAGS_NAMESPACE::SetCommandLineOption("lb_warmup_min_weight",
"0.5").empty());
+ ASSERT_TRUE(GFLAGS_NAMESPACE::SetCommandLineOption("lb_warmup_min_weight",
"0").empty());
+ ASSERT_TRUE(GFLAGS_NAMESPACE::SetCommandLineOption("lb_warmup_min_weight",
"1.5").empty());
+ ASSERT_DOUBLE_EQ(0.5, brpc::FLAGS_lb_warmup_min_weight);
+}
+
+TEST_F(LbWarmupTest, accept_probability_follows_multiplier) {
+ brpc::FLAGS_lb_warmup_ms = 10000;
+ const int64_t join_us = butil::gettimeofday_us();
+ int accepted = 0;
+ const int N = 10000;
+ for (int i = 0; i < N; ++i) {
+ accepted += brpc::WarmupAccept(join_us, join_us + 5000000);
+ }
+ // ~N/2 accepts at multiplier 0.5.
+ ASSERT_GT(accepted, N * 4 / 10);
+ ASSERT_LT(accepted, N * 6 / 10);
+}
+
+TEST_F(LbWarmupTest, rr_ramp_and_rejoin) {
+ brpc::FLAGS_lb_warmup_ms = 300;
+ brpc::policy::RoundRobinLoadBalancer lb;
+ const brpc::ServerId a = CreateServer("127.0.0.1:8111");
+ ASSERT_TRUE(lb.AddServer(a));
+ usleep(400 * 1000);
+ const brpc::ServerId b = CreateServer("127.0.0.1:8112");
+ ASSERT_TRUE(lb.AddServer(b));
+
+ const int N = 4000;
+ // Server b is still cold, its share stays well below the even 50%.
+ std::map<brpc::SocketId, int> shares =
+ CountShares(&lb, N, butil::gettimeofday_us());
+ ASSERT_LT(shares[b.id], N / 4) << shares[b.id];
+ ASSERT_GT(shares[b.id], 0);
+
+ // Past the window(simulated by a future timestamp) shares even out.
+ shares = CountShares(&lb, N, butil::gettimeofday_us() + 1000000);
+ ASSERT_GT(shares[b.id], N * 35 / 100);
+ ASSERT_LT(shares[b.id], N * 65 / 100);
+
+ // Removing and re-adding restarts the ramp.
+ ASSERT_TRUE(lb.RemoveServer(b));
+ ASSERT_TRUE(lb.AddServer(b));
+ shares = CountShares(&lb, N, butil::gettimeofday_us());
Review Comment:
Done in d152ee57: join stamps now come from `LoadBalancerJoinTimeUs()`,
replaceable via `SetLoadBalancerClockForTesting()`. The tests install a fake
clock, advance it between `AddServer` calls and pass explicit `begin_time_us`;
no sleeps left, suite runs in ~3ms.
--
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]
