Copilot commented on code in PR #3526:
URL: https://github.com/apache/brpc/pull/3526#discussion_r3958368674
##########
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:
`Node::join_time_us` is a new field but has no default initialization. If
any code path default-constructs a `Node` (or forgets to assign `join_time_us`
before use), this becomes undefined behavior and can break warm-up decisions.
Safer: give it a default member initializer (e.g., `int64_t join_time_us = 0;`)
so unstamped nodes deterministically behave as 'not 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:
These assertions rely on exact floating-point equality (`ASSERT_DOUBLE_EQ`)
involving `std::pow`, which can be sensitive to platform/libm differences. To
reduce test flakiness across toolchains, prefer `ASSERT_NEAR(expected, actual,
epsilon)` with a small tolerance for the curve-shaped cases.
##########
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:
Several tests use real `usleep(...)` to create join-time separation. This
slows down the test suite and can be flaky on busy CI machines due to
scheduling jitter. Consider avoiding wall-clock sleeps by driving time purely
via `SelectIn::begin_time_us` (and/or exposing a test hook/clock injection for
join timestamps) so warm-up phase transitions can be tested deterministically
without sleeping.
--
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]
