From e83309fe1f98a5d8eb348be47c212ef3da41527e Mon Sep 17 00:00:00 2001 From: rajvarun77 <287367605+rajvarun77@users.noreply.github.com> Date: Thu, 2 Jul 2026 02:58:47 -0400 Subject: [PATCH 1/2] Add Power-of-Two-Choices Peak-EWMA load balancer (p2c) Implements the p2c load balancing policy proposed in #3340: each selection samples two random servers (configurable via choices=N) and routes to the lower peak-EWMA latency * (inflight+1) / weight score. Upward latency spikes take effect immediately while recovery decays over tau_ms (default 10s), so a degraded server is shed within one observation at O(1) selection cost. Uses only existing LoadBalancer hooks (SelectServer/Feedback, need_feedback) with DoublyBufferedData membership like rr/la; per-node stats are shared_ptr-owned by both buffers. Registered as "p2c" in global.cpp. Includes unit tests (functional, weighted, exclusion, error punishment, concurrency churn) and docs in cn/en client.md. Co-Authored-By: Claude Fable 5 --- docs/cn/client.md | 4 + docs/cn/rpc_press.md | 2 +- docs/en/client.md | 4 + src/brpc/global.cpp | 3 + src/brpc/policy/p2c_ewma_load_balancer.cpp | 366 ++++++++++++++++++ src/brpc/policy/p2c_ewma_load_balancer.h | 100 +++++ test/brpc_p2c_ewma_load_balancer_unittest.cpp | 346 +++++++++++++++++ tools/rpc_press/rpc_press.cpp | 2 +- 8 files changed, 825 insertions(+), 2 deletions(-) create mode 100644 src/brpc/policy/p2c_ewma_load_balancer.cpp create mode 100644 src/brpc/policy/p2c_ewma_load_balancer.h create mode 100644 test/brpc_p2c_ewma_load_balancer_unittest.cpp diff --git a/docs/cn/client.md b/docs/cn/client.md index 609becc086..bb16eefb68 100755 --- a/docs/cn/client.md +++ b/docs/cn/client.md @@ -260,6 +260,10 @@ int main() { locality-aware，优先选择延时低的下游，直到其延时高于其他机器，无需其他设置。实现原理请查看[Locality-aware load balancing](lalb.md)。 +### p2c + +即power-of-two-choices加peak-EWMA延时评分。每次选择随机采样两台服务器，把请求发给`延时 * (inflight + 1) / 权重`得分较低的那台。其中延时是对尖峰敏感的滑动平均：延时上升立即生效，恢复则按`tau_ms`（默认10秒）衰减。变慢或出错的服务器在一次观察内即被避开，且选择开销与集群规模无关，为O(1)。权重取自实例tag（同wrr，默认为1）。可选参数：`p2c:choices=4`（每次比较4台采样服务器，适合多台机器同时劣化的场景）、`p2c:tau_ms=5000`。 + ### c_murmurhash or c_md5 一致性哈希，与简单hash的不同之处在于增加或删除机器时不会使分桶结果剧烈变化，特别适合cache类服务。 diff --git a/docs/cn/rpc_press.md b/docs/cn/rpc_press.md index 325728463e..357340ae97 100644 --- a/docs/cn/rpc_press.md +++ b/docs/cn/rpc_press.md @@ -33,7 +33,7 @@ json也可以写在文件中，假如./input.json包含了上述两个请求，- 可选参数: - -inc: 包含被import的proto文件的路径。rpc_press默认支持import目录下的其他proto文件，但如果proto文件在其他目录，就要通过这个参数指定，多个路径用分号(;)分隔。 -- -lb_policy: 指定负载均衡算法，默认为空，可选项为: rr random la c_murmurhash c_md5，具体见[负载均衡](client.md#负载均衡)。 +- -lb_policy: 指定负载均衡算法，默认为空，可选项为: rr random la p2c c_murmurhash c_md5，具体见[负载均衡](client.md#负载均衡)。 - -timeout_ms: 设定超时,单位是毫秒(milliseconds),默认是1000(1秒) - -max_retry: 最大的重试次数,默认是3, 一般无需修改. brpc的重试行为具体请见[这里](client.md#重试). - -protocol: 连接server使用的协议，可选项见[协议](client.md#协议), 默认是baidu_std diff --git a/docs/en/client.md b/docs/en/client.md index 60c458b62c..8da0c6b156 100644 --- a/docs/en/client.md +++ b/docs/en/client.md @@ -258,6 +258,10 @@ Requirements of instance tag is the same as wrr. which is locality-aware. Perfer servers with lower latencies, until the latency is higher than others, no other settings. Check out [Locality-aware load balancing](lalb.md) for more details. +### p2c + +which is power-of-two-choices with peak-EWMA latency scoring. Each selection samples two random servers and routes to the one with the lower `latency * (inflight + 1) / weight` score, where the latency is a peak-sensitive moving average: an upward spike takes effect immediately while recovery decays over `tau_ms`(default 10s). A slow or failing server is shed within one observation and selection cost is O(1) regardless of cluster size. Weight comes from the instance tag as in wrr(default 1). Optional parameters: `p2c:choices=4`(compare 4 sampled servers instead of 2, useful when many servers degrade at once), `p2c:tau_ms=5000`. + ### c_murmurhash or c_md5 which is consistent hashing. Adding or removing servers does not make destinations of requests change as dramatically as in simple hashing. It's especially suitable for caching services. diff --git a/src/brpc/global.cpp b/src/brpc/global.cpp index 90f19cd5bc..70a4666abe 100644 --- a/src/brpc/global.cpp +++ b/src/brpc/global.cpp @@ -46,6 +46,7 @@ #include "brpc/policy/randomized_load_balancer.h" #include "brpc/policy/weighted_randomized_load_balancer.h" #include "brpc/policy/locality_aware_load_balancer.h" +#include "brpc/policy/p2c_ewma_load_balancer.h" #include "brpc/policy/consistent_hashing_load_balancer.h" #include "brpc/policy/hasher.h" #include "brpc/policy/dynpart_load_balancer.h" @@ -155,6 +156,7 @@ struct GlobalExtensions { RandomizedLoadBalancer randomized_lb; WeightedRandomizedLoadBalancer wr_lb; LocalityAwareLoadBalancer la_lb; + P2CEwmaLoadBalancer p2c_ewma_lb; ConsistentHashingLoadBalancer ch_mh_lb; ConsistentHashingLoadBalancer ch_md5_lb; ConsistentHashingLoadBalancer ch_ketama_lb; @@ -391,6 +393,7 @@ static void GlobalInitializeOrDieImpl() { LoadBalancerExtension()->RegisterOrDie("random", &g_ext->randomized_lb); LoadBalancerExtension()->RegisterOrDie("wr", &g_ext->wr_lb); LoadBalancerExtension()->RegisterOrDie("la", &g_ext->la_lb); + LoadBalancerExtension()->RegisterOrDie("p2c", &g_ext->p2c_ewma_lb); LoadBalancerExtension()->RegisterOrDie("c_murmurhash", &g_ext->ch_mh_lb); LoadBalancerExtension()->RegisterOrDie("c_md5", &g_ext->ch_md5_lb); LoadBalancerExtension()->RegisterOrDie("c_ketama", &g_ext->ch_ketama_lb); diff --git a/src/brpc/policy/p2c_ewma_load_balancer.cpp b/src/brpc/policy/p2c_ewma_load_balancer.cpp new file mode 100644 index 0000000000..d72c348328 --- /dev/null +++ b/src/brpc/policy/p2c_ewma_load_balancer.cpp @@ -0,0 +1,366 @@ +// 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 // std::exp +#include "butil/fast_rand.h" // fast_rand_less_than +#include "butil/time.h" // gettimeofday_us +#include "butil/string_splitter.h" // KeyValuePairsSplitter +#include "butil/strings/string_number_conversions.h" // StringToUint +#include "brpc/socket.h" +#include "brpc/controller.h" +#include "brpc/policy/p2c_ewma_load_balancer.h" + +namespace brpc { +namespace policy { + +namespace { +const uint32_t DEFAULT_CHOICES = 2; +const uint32_t MAX_CHOICES = 64; +// Finagle's PeakEwma default decay time. +const int64_t DEFAULT_TAU_US = 10 * 1000000L; +// Floor of the latency term so that in-flight counts break ties between +// servers that have no latency observation yet. +const double MIN_LATENCY_TERM_US = 1.0; + +uint32_t WeightOfTag(const std::string& tag) { + if (tag.empty()) { + return 1; + } + uint32_t weight = 0; + if (!butil::StringToUint(tag, &weight) || weight == 0) { + LOG(WARNING) << "Invalid weight tag=`" << tag << "', use weight=1"; + return 1; + } + return weight; +} +} // namespace + +P2CEwmaLoadBalancer::P2CEwmaLoadBalancer() + : _choices(DEFAULT_CHOICES) + , _tau_us(DEFAULT_TAU_US) {} + +bool P2CEwmaLoadBalancer::Add(Servers& bg, const Servers& fg, + const ServerId& id) { + if (bg.server_list.capacity() < 128) { + bg.server_list.reserve(128); + } + if (bg.server_map.seek(id.id) != NULL) { + return false; + } + ServerInfo info = { id.id, WeightOfTag(id.tag), NULL }; + const size_t* pindex = fg.server_map.seek(id.id); + if (pindex == NULL) { + // Both buffers do not have the server. Create the stat structure + // which will be shared by both buffers. + info.stat = std::make_shared(); + } else { + // Already added to the other buffer, share its stat. + info.stat = fg.server_list[*pindex].stat; + } + bg.server_map[id.id] = bg.server_list.size(); + bg.server_list.push_back(info); + return true; +} + +bool P2CEwmaLoadBalancer::Remove(Servers& bg, const ServerId& id) { + size_t* pindex = bg.server_map.seek(id.id); + if (pindex == NULL) { + return false; + } + const size_t index = *pindex; + bg.server_list[index] = bg.server_list.back(); + bg.server_map[bg.server_list[index].id] = index; + bg.server_list.pop_back(); + bg.server_map.erase(id.id); + return true; +} + +size_t P2CEwmaLoadBalancer::BatchAdd( + Servers& bg, const Servers& fg, const std::vector& servers) { + size_t count = 0; + for (size_t i = 0; i < servers.size(); ++i) { + count += !!Add(bg, fg, servers[i]); + } + return count; +} + +size_t P2CEwmaLoadBalancer::BatchRemove( + Servers& bg, const std::vector& servers) { + size_t count = 0; + for (size_t i = 0; i < servers.size(); ++i) { + count += !!Remove(bg, servers[i]); + } + return count; +} + +bool P2CEwmaLoadBalancer::AddServer(const ServerId& id) { + return _db_servers.ModifyWithForeground(Add, id); +} + +bool P2CEwmaLoadBalancer::RemoveServer(const ServerId& id) { + return _db_servers.Modify(Remove, id); +} + +size_t P2CEwmaLoadBalancer::AddServersInBatch( + const std::vector& servers) { + const size_t n = _db_servers.ModifyWithForeground(BatchAdd, servers); + LOG_IF(ERROR, n != servers.size()) + << "Fail to AddServersInBatch, expected " << servers.size() + << " actually " << n; + return n; +} + +size_t P2CEwmaLoadBalancer::RemoveServersInBatch( + const std::vector& servers) { + return _db_servers.Modify(BatchRemove, servers); +} + +double P2CEwmaLoadBalancer::Score( + const ServerInfo& info, int64_t now_us) const { + const int64_t ewma_us = + info.stat->ewma_us.load(butil::memory_order_relaxed); + const int32_t inflight = + info.stat->inflight.load(butil::memory_order_relaxed); + double latency_term = MIN_LATENCY_TERM_US; + if (ewma_us > 0) { + // Decay the (possibly stale) EWMA at read time so that a server + // penalized long ago regains traffic and gets re-observed. + const int64_t stamp_us = + info.stat->stamp_us.load(butil::memory_order_relaxed); + const int64_t elapsed_us = now_us - stamp_us; + double decayed = (double)ewma_us; + if (elapsed_us > 0) { + decayed *= std::exp(-(double)elapsed_us / (double)_tau_us); + } + latency_term = std::max(decayed, MIN_LATENCY_TERM_US); + } + // Clamp so that a transiently negative counter can not invert routing. + const int32_t load = std::max(inflight + 1, 1); + return latency_term * (double)load / (double)info.weight; +} + +int P2CEwmaLoadBalancer::SelectServer(const SelectIn& in, SelectOut* out) { + butil::DoublyBufferedData::ScopedPtr s; + if (_db_servers.Read(&s) != 0) { + return ENOMEM; + } + const size_t n = s->server_list.size(); + if (n == 0) { + return ENODATA; + } + const int64_t now_us = butil::gettimeofday_us(); + + const ServerInfo* best = NULL; + double best_score = 0; + SocketUniquePtr best_ptr; + // Score the server at `index' and keep it if it beats the current best. + // Excluded and unavailable servers are skipped. + auto consider = [&](size_t index) { + const ServerInfo& info = s->server_list[index]; + if (ExcludedServers::IsExcluded(in.excluded, info.id)) { + return; + } + SocketUniquePtr ptr; + if (!IsServerAvailable(info.id, &ptr)) { + return; + } + const double score = Score(info, now_us); + if (best == NULL || score < best_score) { + best = &info; + best_score = score; + best_ptr.swap(ptr); + } + }; + + const size_t choices = std::min((size_t)_choices, n); + if (choices >= n) { + // Scan from a random offset so that equal scores do not herd all + // clients onto the lowest-indexed server. + const size_t start = butil::fast_rand_less_than(n); + for (size_t i = 0; i < n; ++i) { + consider((start + i) % n); + } + } else { + // Sample `choices' distinct random servers. Attempts are bounded so + // that duplicated draws never loop for long. + size_t chosen[MAX_CHOICES]; + size_t nchosen = 0; + const size_t max_attempts = 4 * choices + 8; + for (size_t attempt = 0; + attempt < max_attempts && nchosen < choices; ++attempt) { + const size_t index = butil::fast_rand_less_than(n); + bool duplicated = false; + for (size_t i = 0; i < nchosen; ++i) { + if (chosen[i] == index) { + duplicated = true; + break; + } + } + if (duplicated) { + continue; + } + chosen[nchosen++] = index; + consider(index); + } + if (best == NULL) { + // All sampled servers were excluded or unavailable, fall back + // to scoring the whole list before violating exclusion below. + for (size_t i = 0; i < n; ++i) { + consider(i); + } + } + } + + if (best == NULL) { + // Always take last chance: all servers are excluded, send to any + // available one as rr/random do. + for (size_t i = 0; i < n; ++i) { + if (IsServerAvailable(s->server_list[i].id, &best_ptr)) { + best = &s->server_list[i]; + break; + } + } + if (best == NULL) { + return EHOSTDOWN; + } + } + if (in.changable_weights) { + best->stat->inflight.fetch_add(1, butil::memory_order_relaxed); + out->need_feedback = true; + } + out->ptr->swap(best_ptr); + return 0; +} + +void P2CEwmaLoadBalancer::Feedback(const CallInfo& info) { + butil::DoublyBufferedData::ScopedPtr s; + if (_db_servers.Read(&s) != 0) { + return; + } + const size_t* pindex = s->server_map.seek(info.server_id); + if (pindex == NULL) { + // The server was removed after selection, its stat is gone with it. + return; + } + NodeStat* stat = s->server_list[*pindex].stat.get(); + stat->inflight.fetch_sub(1, butil::memory_order_relaxed); + + const int64_t now_us = butil::gettimeofday_us(); + int64_t latency_us = now_us - info.begin_time_us; + if (latency_us <= 0) { + // time skews, ignore the sample. + return; + } + if (info.error_code != 0) { + // Punish failures with at least the timeout(if any) and twice the + // current average, so that a failing server keeps losing comparisons + // even when it fails fast. + const int64_t timeout_us = info.controller->timeout_ms() * 1000L; + latency_us = std::max(latency_us, timeout_us); + latency_us = std::max( + latency_us, 2 * stat->ewma_us.load(butil::memory_order_relaxed)); + } + + BAIDU_SCOPED_LOCK(stat->update_mutex); + const int64_t ewma_us = stat->ewma_us.load(butil::memory_order_relaxed); + int64_t new_ewma_us = latency_us; + if (ewma_us > 0 && latency_us < ewma_us) { + // Downward samples decay the average while upward spikes(handled by + // the branch above) replace it immediately. + const int64_t elapsed_us = + now_us - stat->stamp_us.load(butil::memory_order_relaxed); + const double w = elapsed_us > 0 + ? std::exp(-(double)elapsed_us / (double)_tau_us) : 1.0; + new_ewma_us = (int64_t)(ewma_us * w + latency_us * (1.0 - w)); + } + stat->ewma_us.store(new_ewma_us, butil::memory_order_relaxed); + stat->stamp_us.store(now_us, butil::memory_order_relaxed); +} + +P2CEwmaLoadBalancer* P2CEwmaLoadBalancer::New( + const butil::StringPiece& params) const { + P2CEwmaLoadBalancer* lb = new (std::nothrow) P2CEwmaLoadBalancer; + if (lb != NULL && !lb->SetParameters(params)) { + delete lb; + lb = NULL; + } + return lb; +} + +bool P2CEwmaLoadBalancer::SetParameters(const butil::StringPiece& params) { + for (butil::KeyValuePairsSplitter sp(params.begin(), params.end(), ' ', '='); + sp; ++sp) { + if (sp.value().empty()) { + LOG(ERROR) << "Empty value for " << sp.key() << " in lb parameter"; + return false; + } + if (sp.key() == "choices") { + unsigned choices = 0; + if (!butil::StringToUint(sp.value().as_string(), &choices) || + choices < 2 || choices > MAX_CHOICES) { + LOG(ERROR) << "Invalid choices=`" << sp.value() << "'"; + return false; + } + _choices = choices; + } else if (sp.key() == "tau_ms") { + int64_t tau_ms = 0; + if (!butil::StringToInt64(sp.value(), &tau_ms) || tau_ms <= 0) { + LOG(ERROR) << "Invalid tau_ms=`" << sp.value() << "'"; + return false; + } + _tau_us = tau_ms * 1000L; + } else { + LOG(ERROR) << "Unknown parameter " << sp.key_and_value(); + return false; + } + } + return true; +} + +void P2CEwmaLoadBalancer::Destroy() { + delete this; +} + +void P2CEwmaLoadBalancer::Describe( + std::ostream& os, const DescribeOptions& options) { + if (!options.verbose) { + os << "p2c"; + return; + } + os << "P2CEwma{choices=" << _choices << " tau_ms=" << _tau_us / 1000; + butil::DoublyBufferedData::ScopedPtr s; + if (_db_servers.Read(&s) != 0) { + os << " fail to read _db_servers"; + } else { + const int64_t now_us = butil::gettimeofday_us(); + os << " n=" << s->server_list.size() << ':'; + for (size_t i = 0; i < s->server_list.size(); ++i) { + const ServerInfo& info = s->server_list[i]; + os << ' ' << info.id << '(' << "w=" << info.weight + << " ewma_us=" + << info.stat->ewma_us.load(butil::memory_order_relaxed) + << " inflight=" + << info.stat->inflight.load(butil::memory_order_relaxed) + << " score=" << Score(info, now_us) << ')'; + } + } + os << '}'; +} + +} // namespace policy +} // namespace brpc diff --git a/src/brpc/policy/p2c_ewma_load_balancer.h b/src/brpc/policy/p2c_ewma_load_balancer.h new file mode 100644 index 0000000000..38902859c0 --- /dev/null +++ b/src/brpc/policy/p2c_ewma_load_balancer.h @@ -0,0 +1,100 @@ +// 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. + + +#ifndef BRPC_POLICY_P2C_EWMA_LOAD_BALANCER_H +#define BRPC_POLICY_P2C_EWMA_LOAD_BALANCER_H + +#include // std::shared_ptr +#include // std::vector +#include "butil/containers/flat_map.h" // FlatMap +#include "butil/containers/doubly_buffered_data.h" // DoublyBufferedData +#include "brpc/load_balancer.h" + +namespace brpc { +namespace policy { + +// Power-of-Two-Choices with Peak-EWMA latency scoring ("p2c"). +// Each selection samples `choices'(default 2) distinct servers and routes +// to the one with the lower load score: +// score = peak_ewma_latency_us * (inflight + 1) / weight +// The latency EWMA is peak-sensitive: an upward spike replaces the average +// immediately while recovery decays with time constant `tau_ms'(default 10s), +// so a degraded server is shed within one observation. Selection is O(1) +// regardless of fleet size. Weight is got from tag of ServerId(default 1). +class P2CEwmaLoadBalancer : public LoadBalancer { +public: + P2CEwmaLoadBalancer(); + bool AddServer(const ServerId& id) override; + bool RemoveServer(const ServerId& id) override; + size_t AddServersInBatch(const std::vector& servers) override; + size_t RemoveServersInBatch(const std::vector& servers) override; + int SelectServer(const SelectIn& in, SelectOut* out) override; + void Feedback(const CallInfo& info) override; + P2CEwmaLoadBalancer* New(const butil::StringPiece& params) const override; + void Destroy() override; + void Describe(std::ostream& os, const DescribeOptions&) override; + +private: + // Mutable per-server load state. Allocated once when the server is first + // added and shared by both buffers of _db_servers, so a stable pointer + // can be used from SelectServer()/Feedback() without copying. + struct NodeStat { + NodeStat() : inflight(0), ewma_us(0), stamp_us(0) {} + butil::atomic inflight; + // Peak-sensitive EWMA of latency in us. 0 means no observation yet. + butil::atomic ewma_us; + // Time of the last EWMA update. + butil::atomic stamp_us; + butil::Mutex update_mutex; + }; + struct ServerInfo { + SocketId id; + uint32_t weight; + std::shared_ptr stat; + }; + struct Servers { + std::vector server_list; + // Maps SocketId to index in server_list. + butil::FlatMap server_map; + + Servers() { + if (server_map.init(1024, 70) != 0) { + LOG(WARNING) << "Fail to init server_map"; + } + } + }; + + static bool Add(Servers& bg, const Servers& fg, const ServerId& id); + static bool Remove(Servers& bg, const ServerId& id); + static size_t BatchAdd(Servers& bg, const Servers& fg, + const std::vector& servers); + static size_t BatchRemove(Servers& bg, const std::vector& servers); + + bool SetParameters(const butil::StringPiece& params); + // Load score of the server at `now_us'. Smaller is better. + double Score(const ServerInfo& info, int64_t now_us) const; + + butil::DoublyBufferedData _db_servers; + uint32_t _choices; + int64_t _tau_us; +}; + +} // namespace policy +} // namespace brpc + +#endif // BRPC_POLICY_P2C_EWMA_LOAD_BALANCER_H diff --git a/test/brpc_p2c_ewma_load_balancer_unittest.cpp b/test/brpc_p2c_ewma_load_balancer_unittest.cpp new file mode 100644 index 0000000000..5df4e91638 --- /dev/null +++ b/test/brpc_p2c_ewma_load_balancer_unittest.cpp @@ -0,0 +1,346 @@ +// 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 +#include +#include +#include +#include "butil/macros.h" +#include "butil/time.h" +#include "brpc/socket.h" +#include "brpc/controller.h" +#include "brpc/excluded_servers.h" +#include "brpc/policy/p2c_ewma_load_balancer.h" + +namespace { + +butil::atomic nrecycle(0); +class SaveRecycle : public brpc::SocketUser { + void BeforeRecycle(brpc::Socket* s) { + nrecycle.fetch_add(1, butil::memory_order_relaxed); + 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; +} + +// Report a call that took `latency_us' back to the load balancer. +void FeedbackLatency(brpc::LoadBalancer* lb, brpc::SocketId server_id, + int64_t latency_us, int error_code = 0, + const brpc::Controller* cntl = NULL) { + brpc::LoadBalancer::CallInfo info; + info.begin_time_us = butil::gettimeofday_us() - latency_us; + info.server_id = server_id; + info.error_code = error_code; + info.controller = cntl; + lb->Feedback(info); +} + +class P2CEwmaLoadBalancerTest : public ::testing::Test { +protected: + void SetUp() override { + _lb = new brpc::policy::P2CEwmaLoadBalancer; + } + void TearDown() override { + _lb->Destroy(); + } + + int Select(brpc::SocketUniquePtr* ptr, bool* need_feedback = NULL, + const brpc::ExcludedServers* excluded = NULL) { + brpc::LoadBalancer::SelectIn in = { + butil::gettimeofday_us(), true, false, 0u, excluded }; + brpc::LoadBalancer::SelectOut out(ptr); + const int rc = _lb->SelectServer(in, &out); + if (need_feedback != NULL) { + *need_feedback = out.need_feedback; + } + return rc; + } + + brpc::policy::P2CEwmaLoadBalancer* _lb; +}; + +TEST_F(P2CEwmaLoadBalancerTest, add_remove_servers) { + brpc::SocketUniquePtr ptr; + ASSERT_EQ(ENODATA, Select(&ptr)); + + std::vector ids; + ids.push_back(CreateServer("127.0.0.1:7777")); + ids.push_back(CreateServer("127.0.0.1:7778")); + ids.push_back(CreateServer("127.0.0.1:7779")); + + ASSERT_TRUE(_lb->AddServer(ids[0])); + // Duplicated server is rejected. + ASSERT_FALSE(_lb->AddServer(ids[0])); + ASSERT_EQ(2u, _lb->AddServersInBatch( + std::vector(ids.begin() + 1, ids.end()))); + + ASSERT_EQ(0, Select(&ptr)); + + ASSERT_TRUE(_lb->RemoveServer(ids[0])); + ASSERT_FALSE(_lb->RemoveServer(ids[0])); + ASSERT_EQ(2u, _lb->RemoveServersInBatch( + std::vector(ids.begin() + 1, ids.end()))); + ASSERT_EQ(ENODATA, Select(&ptr)); +} + +TEST_F(P2CEwmaLoadBalancerTest, single_server) { + const brpc::ServerId id = CreateServer("127.0.0.1:7780"); + ASSERT_TRUE(_lb->AddServer(id)); + for (int i = 0; i < 10; ++i) { + brpc::SocketUniquePtr ptr; + bool need_feedback = false; + ASSERT_EQ(0, Select(&ptr, &need_feedback)); + ASSERT_EQ(id.id, ptr->id()); + ASSERT_TRUE(need_feedback); + FeedbackLatency(_lb, id.id, 1000); + } +} + +TEST_F(P2CEwmaLoadBalancerTest, prefers_lower_latency) { + const brpc::ServerId fast = CreateServer("127.0.0.1:7781"); + const brpc::ServerId slow = CreateServer("127.0.0.1:7782"); + ASSERT_TRUE(_lb->AddServer(fast)); + ASSERT_TRUE(_lb->AddServer(slow)); + + // Servers respond with their characteristic latency; both get observed + // within the first rounds because unobserved servers score best. + std::map counts; + const int kRounds = 1000; + for (int i = 0; i < kRounds; ++i) { + brpc::SocketUniquePtr ptr; + ASSERT_EQ(0, Select(&ptr)); + ++counts[ptr->id()]; + // Selected server responds with its own characteristic latency. + FeedbackLatency(_lb, ptr->id(), ptr->id() == fast.id ? 1000 : 50000); + } + LOG(INFO) << "fast=" << counts[fast.id] << " slow=" << counts[slow.id]; + // The fast server should receive almost all traffic. + ASSERT_GE(counts[fast.id], (size_t)(0.9 * kRounds)); +} + +TEST_F(P2CEwmaLoadBalancerTest, sheds_degraded_server) { + const brpc::ServerId a = CreateServer("127.0.0.1:7783"); + const brpc::ServerId b = CreateServer("127.0.0.1:7784"); + ASSERT_TRUE(_lb->AddServer(a)); + ASSERT_TRUE(_lb->AddServer(b)); + // Warm up with `b' slightly faster so it is the preferred server. + for (int i = 0; i < 20; ++i) { + brpc::SocketUniquePtr ptr; + ASSERT_EQ(0, Select(&ptr)); + FeedbackLatency(_lb, ptr->id(), ptr->id() == a.id ? 1000 : 500); + } + + // `b' degrades: its first slow response must shift traffic to `a' + // immediately(peak-sensitivity), long before an averaging window would. + std::map counts; + for (int i = 0; i < 100; ++i) { + brpc::SocketUniquePtr ptr; + ASSERT_EQ(0, Select(&ptr)); + ++counts[ptr->id()]; + FeedbackLatency(_lb, ptr->id(), ptr->id() == a.id ? 1000 : 100000); + } + ASSERT_GE(counts[a.id], 90u); +} + +TEST_F(P2CEwmaLoadBalancerTest, inflight_breaks_ties) { + const brpc::ServerId a = CreateServer("127.0.0.1:7785"); + const brpc::ServerId b = CreateServer("127.0.0.1:7786"); + ASSERT_TRUE(_lb->AddServer(a)); + ASSERT_TRUE(_lb->AddServer(b)); + + // With no latency observations, scores only differ by in-flight count, + // so two selections without feedback must go to different servers. + brpc::SocketUniquePtr ptr1; + brpc::SocketUniquePtr ptr2; + ASSERT_EQ(0, Select(&ptr1)); + ASSERT_EQ(0, Select(&ptr2)); + ASSERT_NE(ptr1->id(), ptr2->id()); + + // After feedback(decrementing in-flight), both are tied again and the + // third selection must not crash or fail. + FeedbackLatency(_lb, ptr1->id(), 1000); + FeedbackLatency(_lb, ptr2->id(), 1000); + brpc::SocketUniquePtr ptr3; + ASSERT_EQ(0, Select(&ptr3)); +} + +TEST_F(P2CEwmaLoadBalancerTest, weighted_split_by_inflight) { + // With equal latency, steady-state in-flight counts converge to the + // 1:2:4 weight ratio because the score is (inflight+1)/weight. + const brpc::ServerId w1 = CreateServer("127.0.0.1:7787", "1"); + const brpc::ServerId w2 = CreateServer("127.0.0.1:7788", "2"); + const brpc::ServerId w4 = CreateServer("127.0.0.1:7789", "4"); + brpc::policy::P2CEwmaLoadBalancer* lb = + _lb->New(butil::StringPiece("choices=3")); + ASSERT_TRUE(lb != NULL); + ASSERT_TRUE(lb->AddServer(w1)); + ASSERT_TRUE(lb->AddServer(w2)); + ASSERT_TRUE(lb->AddServer(w4)); + + std::map counts; + const int kRounds = 700; + for (int i = 0; i < kRounds; ++i) { + brpc::SocketUniquePtr ptr; + brpc::LoadBalancer::SelectIn in = { + butil::gettimeofday_us(), true, false, 0u, NULL }; + brpc::LoadBalancer::SelectOut out(&ptr); + ASSERT_EQ(0, lb->SelectServer(in, &out)); + ++counts[ptr->id()]; + // No feedback: requests stay in flight. + } + LOG(INFO) << "w1=" << counts[w1.id] << " w2=" << counts[w2.id] + << " w4=" << counts[w4.id]; + const double share1 = counts[w1.id] / (double)kRounds; + const double share2 = counts[w2.id] / (double)kRounds; + const double share4 = counts[w4.id] / (double)kRounds; + ASSERT_NEAR(share1, 1.0 / 7, 0.05); + ASSERT_NEAR(share2, 2.0 / 7, 0.05); + ASSERT_NEAR(share4, 4.0 / 7, 0.05); + lb->Destroy(); +} + +TEST_F(P2CEwmaLoadBalancerTest, error_feedback_punishes_server) { + const brpc::ServerId good = CreateServer("127.0.0.1:7790"); + const brpc::ServerId bad = CreateServer("127.0.0.1:7791"); + ASSERT_TRUE(_lb->AddServer(good)); + ASSERT_TRUE(_lb->AddServer(bad)); + // Warm up with `bad' slightly faster so it is the preferred server. + for (int i = 0; i < 20; ++i) { + brpc::SocketUniquePtr ptr; + ASSERT_EQ(0, Select(&ptr)); + FeedbackLatency(_lb, ptr->id(), ptr->id() == good.id ? 1000 : 500); + } + + // `bad' starts failing fast(1ms), but failures are punished with at + // least the RPC timeout so it keeps losing selections. + brpc::Controller cntl; + cntl.set_timeout_ms(100); + std::map counts; + for (int i = 0; i < 100; ++i) { + brpc::SocketUniquePtr ptr; + ASSERT_EQ(0, Select(&ptr)); + ++counts[ptr->id()]; + if (ptr->id() == good.id) { + FeedbackLatency(_lb, good.id, 1000); + } else { + FeedbackLatency(_lb, bad.id, 1000, ETIMEDOUT, &cntl); + } + } + ASSERT_GE(counts[good.id], 90u); +} + +TEST_F(P2CEwmaLoadBalancerTest, excluded_servers) { + const brpc::ServerId a = CreateServer("127.0.0.1:7792"); + const brpc::ServerId b = CreateServer("127.0.0.1:7793"); + ASSERT_TRUE(_lb->AddServer(a)); + ASSERT_TRUE(_lb->AddServer(b)); + + brpc::ExcludedServers* excluded = brpc::ExcludedServers::Create(2); + excluded->Add(a.id); + for (int i = 0; i < 20; ++i) { + brpc::SocketUniquePtr ptr; + ASSERT_EQ(0, Select(&ptr, NULL, excluded)); + ASSERT_EQ(b.id, ptr->id()); + FeedbackLatency(_lb, b.id, 1000); + } + // All servers excluded: still take the last chance instead of failing. + excluded->Add(b.id); + brpc::SocketUniquePtr ptr; + ASSERT_EQ(0, Select(&ptr, NULL, excluded)); + brpc::ExcludedServers::Destroy(excluded); +} + +TEST_F(P2CEwmaLoadBalancerTest, invalid_parameters) { + brpc::LoadBalancer* lb = _lb->New(butil::StringPiece("")); + ASSERT_TRUE(lb != NULL); + lb->Destroy(); + lb = _lb->New(butil::StringPiece("choices=4 tau_ms=5000")); + ASSERT_TRUE(lb != NULL); + lb->Destroy(); + ASSERT_TRUE(_lb->New(butil::StringPiece("choices=1")) == NULL); + ASSERT_TRUE(_lb->New(butil::StringPiece("choices=abc")) == NULL); + ASSERT_TRUE(_lb->New(butil::StringPiece("tau_ms=0")) == NULL); + ASSERT_TRUE(_lb->New(butil::StringPiece("unknown=1")) == NULL); +} + +struct ChurnArg { + brpc::policy::P2CEwmaLoadBalancer* lb; + butil::atomic stop; + butil::atomic nselected; +}; + +void* SelectAndFeedback(void* void_arg) { + ChurnArg* arg = static_cast(void_arg); + while (!arg->stop.load(butil::memory_order_relaxed)) { + brpc::SocketUniquePtr ptr; + brpc::LoadBalancer::SelectIn in = { + butil::gettimeofday_us(), true, false, 0u, NULL }; + brpc::LoadBalancer::SelectOut out(&ptr); + if (arg->lb->SelectServer(in, &out) == 0) { + arg->nselected.fetch_add(1, butil::memory_order_relaxed); + if (out.need_feedback) { + FeedbackLatency(arg->lb, ptr->id(), 1000); + } + } + } + return NULL; +} + +TEST_F(P2CEwmaLoadBalancerTest, concurrent_select_with_churn) { + std::vector ids; + for (int i = 0; i < 8; ++i) { + char addr[32]; + snprintf(addr, sizeof(addr), "127.0.0.1:%d", 7800 + i); + ids.push_back(CreateServer(addr)); + ASSERT_TRUE(_lb->AddServer(ids.back())); + } + + ChurnArg arg; + arg.lb = _lb; + arg.stop.store(false); + arg.nselected.store(0); + pthread_t threads[4]; + for (size_t i = 0; i < ARRAY_SIZE(threads); ++i) { + ASSERT_EQ(0, pthread_create( + &threads[i], NULL, SelectAndFeedback, &arg)); + } + // Churn membership while selections are running. + const int64_t stop_at_us = butil::gettimeofday_us() + 1000000L; + while (butil::gettimeofday_us() < stop_at_us) { + ASSERT_TRUE(_lb->RemoveServer(ids[0])); + ASSERT_TRUE(_lb->AddServer(ids[0])); + } + arg.stop.store(true); + for (size_t i = 0; i < ARRAY_SIZE(threads); ++i) { + ASSERT_EQ(0, pthread_join(threads[i], NULL)); + } + LOG(INFO) << "selected " << arg.nselected.load() << " times"; + ASSERT_GT(arg.nselected.load(), 0u); +} + +} // namespace diff --git a/tools/rpc_press/rpc_press.cpp b/tools/rpc_press/rpc_press.cpp index cb69ccfc16..a9a98d60c7 100644 --- a/tools/rpc_press/rpc_press.cpp +++ b/tools/rpc_press/rpc_press.cpp @@ -31,7 +31,7 @@ DEFINE_string(method, "example.EchoService.Echo", "The full method name"); DEFINE_string(server, "0.0.0.0:8002", "ip:port of the server when -load_balancer is empty, the naming service otherwise"); DEFINE_string(input, "", "The file containing requests in json format"); DEFINE_string(output, "", "The file containing responses in json format"); -DEFINE_string(lb_policy, "", "The load balancer algorithm: rr, random, la, c_murmurhash, c_md5"); +DEFINE_string(lb_policy, "", "The load balancer algorithm: rr, random, la, p2c, c_murmurhash, c_md5"); DEFINE_int32(thread_num, 0, "Number of threads to send requests. 0: automatically chosen according to -qps"); DEFINE_string(protocol, "baidu_std", "baidu_std hulu_pbrpc sofa_pbrpc http public_pbrpc nova_pbrpc ubrpc_compack..."); DEFINE_string(connection_type, "", "Type of connections: single, pooled, short"); From aeb33d5e56e2f78df3e47118d4cbd7a9b7d037e6 Mon Sep 17 00:00:00 2001 From: rajvarun77 <287367605+rajvarun77@users.noreply.github.com> Date: Tue, 7 Jul 2026 00:51:14 -0400 Subject: [PATCH 2/2] Reuse SelectIn.begin_time_us to avoid extra clock read per selection Co-Authored-By: Claude Fable 5 --- src/brpc/policy/p2c_ewma_load_balancer.cpp | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/src/brpc/policy/p2c_ewma_load_balancer.cpp b/src/brpc/policy/p2c_ewma_load_balancer.cpp index d72c348328..9283aeb99a 100644 --- a/src/brpc/policy/p2c_ewma_load_balancer.cpp +++ b/src/brpc/policy/p2c_ewma_load_balancer.cpp @@ -163,7 +163,10 @@ int P2CEwmaLoadBalancer::SelectServer(const SelectIn& in, SelectOut* out) { if (n == 0) { return ENODATA; } - const int64_t now_us = butil::gettimeofday_us(); + // Reuse the caller-provided timestamp to avoid an extra clock read per + // selection; only Channel::CheckHealth passes 0. + const int64_t now_us = in.begin_time_us > 0 + ? in.begin_time_us : butil::gettimeofday_us(); const ServerInfo* best = NULL; double best_score = 0;