This is an automated email from the ASF dual-hosted git repository.
chenBright pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/brpc.git
The following commit(s) were added to refs/heads/master by this push:
new c7b18016 Add bounded-load consistent hashing load balancer
(c_murmurhash_bl) (#3502)
c7b18016 is described below
commit c7b180164cbf641cc15c3d510e5e5493a8371b92
Author: rajvarun77 <[email protected]>
AuthorDate: Sun Sep 6 03:06:23 2026 -0400
Add bounded-load consistent hashing load balancer (c_murmurhash_bl) (#3502)
Classic consistent hashing routes a hot key to one server with no
relief valve: that server saturates while its ring neighbors idle.
c_murmurhash_bl implements "Consistent Hashing with Bounded Loads"
(Mirrokni et al., CACM 2017) on the same murmurhash ring: each server
accepts at most ceil(load_factor * average in-flight) requests, and an
at-capacity server overflows clockwise to the next server with spare
capacity, so spilled requests always land on the same ring successors.
load_factor defaults to -chash_bounded_load_factor(1.25, validated > 1)
and is overridable per channel: c_murmurhash_bl:load_factor=1.5.
In-flight accounting uses relaxed per-server counters shared by both
DoublyBufferedData buffers, incremented at selection and decremented in
Feedback(); the total is restored even if the server was removed in
between, and the counter map is resynced from the ring on membership
changes. The ring code is reused by subclassing
ConsistentHashingLoadBalancer via a per-key SetParameter() hook. The
existing `replicas' parameter is now documented for all CH schemes,
settling the old "TODO: or 160?" on the replica count.
Includes unit tests (cap under hot-key load, overflow to ring
successor, factor validation, feedback decrement, removal consistency,
replicas parameter) and docs in cn/en client.md.
Co-authored-by: rajvarun77 <[email protected]>
---
docs/cn/client.md | 6 +
docs/en/client.md | 6 +
src/brpc/global.cpp | 3 +
.../policy/consistent_hashing_load_balancer.cpp | 238 +++++++++++++-
src/brpc/policy/consistent_hashing_load_balancer.h | 54 +++-
test/brpc_ch_bounded_load_balancer_unittest.cpp | 347 +++++++++++++++++++++
6 files changed, 641 insertions(+), 13 deletions(-)
diff --git a/docs/cn/client.md b/docs/cn/client.md
index e659403c..c1c011c8 100755
--- a/docs/cn/client.md
+++ b/docs/cn/client.md
@@ -277,10 +277,16 @@ locality-aware，优先选择延时低的下游，直到其延时高于其他机
注意甄别请求中的“主键”部分和“属性”部分，不要为了偷懒或通用，就把请求的所有内容一股脑儿计算出哈希值，属性的变化会使请求的目的地发生剧烈的变化。另外也要注意padding问题，比如struct
Foo { int32_t a; int64_t b; }在64位机器上a和b之间有4个字节的空隙，内容未定义，如果像hash(&foo,
sizeof(foo))这样计算哈希值，结果就是未定义的，得把内容紧密排列或序列化后再算。
+每台服务器的虚拟节点数默认由-chash_num_replicas控制（默认100），可按channel覆盖：`c_murmurhash:replicas=300`。
+
实现原理请查看[Consistent Hashing](consistent_hashing.md)。
其他lb不需要设置Controller.set_request_code()，如果调用了request_code也不会被lb使用，例如：lb=rr调用了Controller.set_request_code()，即使所有RPC的request_code都相同，也依然是rr。
+### c_murmurhash_bl
+
+即带负载上限的一致性哈希（"Consistent Hashing with Bounded Loads"，Mirrokni等，CACM
2017）。哈希环与`c_murmurhash`完全相同，但每台服务器额外有容量上限`ceil(load_factor *
平均在途请求数)`。当哈希命中的服务器已达上限时，请求沿哈希环顺时针溢出到下一台有余量的服务器，因此热点key不再压垮单台服务器，且溢出请求总是落到环上固定的后继节点，对cache仍然友好。系数默认来自-chash_bounded_load_factor（默认1.25，必须大于1），可按channel覆盖：`c_murmurhash_bl:load_factor=1.5`。`replicas`参数与`c_murmurhash`相同。
+
### 从集群宕机后恢复时的客户端限流
集群宕机指的是集群中所有server都处于不可用的状态。由于健康检查机制，当集群恢复正常后，server会间隔性地上线。当某一个server上线后，所有的流量会发送过去，可能导致服务再次过载。若熔断开启，则可能导致其它server上线前该server再次熔断，集群永远无法恢复。作为解决方案，brpc提供了在集群宕机后恢复时的限流机制：当集群中没有可用server时，集群进入恢复状态，假设正好能服务所有请求的server数量为min_working_instances，当前集群可用的server数量为q，则在恢复状态时，client接受请求的概率为q/min_working_instances，否则丢弃；若一段时间hold_seconds内q保持不变，则把流量重新发送全部可用的server上，并离开恢复状态。在恢复阶段时，可以通过判断controller.ErrorCode()是否等于brpc::ERJECT来判断该次请求是否被拒绝，被拒绝的请求不会被框架重试。
diff --git a/docs/en/client.md b/docs/en/client.md
index 266eb902..2178be4c 100644
--- a/docs/en/client.md
+++ b/docs/en/client.md
@@ -278,10 +278,16 @@ Need to set Controller.set_request_code() before RPC
otherwise the RPC will fail
Do distinguish "key" and "attributes" of the request. Don't compute
request_code by full content of the request just for quick. Minor change in
attributes may result in totally different hash code and change destination
dramatically. Another cause is padding, for example: `struct Foo { int32_t a;
int64_t b; }` has a 4-byte undefined gap between `a` and `b` on 64-bit
machines, result of `hash(&foo, sizeof(foo))` is undefined. Fields need to be
packed or serialized before hashing.
+Number of virtual nodes per server defaults to -chash_num_replicas(default
100) and can be overridden per channel: `c_murmurhash:replicas=300`.
+
Check out [Consistent Hashing](consistent_hashing.md) for more details.
Other kind of lb does not need to set Controller.set_request_code(). If
request code is set, it will not be used by lb. For example, lb=rr, and call
Controller.set_request_code(), even if request_code is the same for every
request, lb will balance the requests using the rr policy.
+### c_murmurhash_bl
+
+which is consistent hashing with bounded loads("Consistent Hashing with
Bounded Loads", Mirrokni et al., CACM 2017). The hash ring is identical to
`c_murmurhash`, but each server additionally has a capacity of
`ceil(load_factor * average in-flight requests)`. When the hashed-to server is
at capacity, the request overflows clockwise to the next server on the ring
with spare capacity, so a hot key no longer saturates a single server while
overflowed requests always land on the same ring su [...]
+
### Client-side throttling for recovery from cluster downtime
Cluster downtime refers to the state in which all servers in the cluster are
unavailable. Due to the health check mechanism, when the cluster returns to
normal, server will go online one by one. When a server is online, all traffic
will be sent to it, which may cause the service to be overloaded again. If
circuit breaker is enabled, server may be offline again before the other
servers go online, and the cluster can never be recovered. As a solution, brpc
provides a client-side throttling [...]
diff --git a/src/brpc/global.cpp b/src/brpc/global.cpp
index 812a79a1..0a0837e0 100644
--- a/src/brpc/global.cpp
+++ b/src/brpc/global.cpp
@@ -138,6 +138,7 @@ struct GlobalExtensions {
, ch_mh_lb(CONS_HASH_LB_MURMUR3)
, ch_md5_lb(CONS_HASH_LB_MD5)
, ch_ketama_lb(CONS_HASH_LB_KETAMA)
+ , ch_mh_bl_lb(CONS_HASH_LB_MURMUR3)
, constant_cl(0) {
}
@@ -163,6 +164,7 @@ struct GlobalExtensions {
ConsistentHashingLoadBalancer ch_mh_lb;
ConsistentHashingLoadBalancer ch_md5_lb;
ConsistentHashingLoadBalancer ch_ketama_lb;
+ ConsistentHashingBoundedLoadBalancer ch_mh_bl_lb;
DynPartLoadBalancer dynpart_lb;
AutoConcurrencyLimiter auto_cl;
@@ -411,6 +413,7 @@ static void GlobalInitializeOrDieImpl() {
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);
+ LoadBalancerExtension()->RegisterOrDie("c_murmurhash_bl",
&g_ext->ch_mh_bl_lb);
LoadBalancerExtension()->RegisterOrDie("_dynpart", &g_ext->dynpart_lb);
// Compress Handlers
diff --git a/src/brpc/policy/consistent_hashing_load_balancer.cpp
b/src/brpc/policy/consistent_hashing_load_balancer.cpp
index 5ff1558f..31d1ad74 100644
--- a/src/brpc/policy/consistent_hashing_load_balancer.cpp
+++ b/src/brpc/policy/consistent_hashing_load_balancer.cpp
@@ -18,23 +18,36 @@
#include <algorithm> //
std::set_union
#include <array>
+#include <cmath> // std::ceil
+#include <limits> //
numeric_limits
#include <gflags/gflags.h>
#include <openssl/md5.h>
#include "butil/containers/flat_map.h"
#include "butil/errno.h"
#include "butil/strings/string_number_conversions.h"
#include "brpc/socket.h"
+#include "brpc/reloadable_flags.h"
#include "brpc/policy/consistent_hashing_load_balancer.h"
#include "brpc/policy/hasher.h"
namespace brpc {
namespace policy {
-// TODO: or 160?
-DEFINE_int32(chash_num_replicas, 100,
- "default number of replicas per server in chash");
-DEFINE_bool(consistent_hashing_enable_server_tag, false,
+DEFINE_int32(chash_num_replicas, 100,
+ "default number of replicas per server in chash, "
+ "overridable per channel with the `replicas' parameter");
+DEFINE_bool(consistent_hashing_enable_server_tag, false,
"if consistent hashing enable server with tag");
+DEFINE_double(chash_bounded_load_factor, 1.25,
+ "default capacity factor of bounded-load consistent hashing"
+ "(c_*_bl): a server takes at most ceil(factor * average "
+ "in-flight) requests before overflowing to its ring successor, "
+ "overridable per channel with the `load_factor' parameter");
+
+static bool ValidateLoadFactor(const char*, double factor) {
+ return factor > 1.0;
+}
+BRPC_VALIDATE_GFLAG(chash_bounded_load_factor, ValidateLoadFactor);
// Defined in hasher.cpp.
const char* GetHashName(HashFunc hasher);
@@ -395,16 +408,223 @@ bool ConsistentHashingLoadBalancer::SetParameters(const
butil::StringPiece& para
LOG(ERROR) << "Empty value for " << sp.key() << " in lb parameter";
return false;
}
- if (sp.key() == "replicas") {
- if (!butil::StringToSizeT(sp.value(), &_num_replicas)) {
- return false;
+ if (!SetParameter(sp.key(), sp.value())) {
+ return false;
+ }
+ }
+ return true;
+}
+
+bool ConsistentHashingLoadBalancer::SetParameter(
+ const butil::StringPiece& key, const butil::StringPiece& value) {
+ if (key == "replicas") {
+ return butil::StringToSizeT(value, &_num_replicas);
+ }
+ LOG(ERROR) << "Failed to set this unknown parameters " << key << '=' <<
value;
+ return true;
+}
+
+ConsistentHashingBoundedLoadBalancer::ConsistentHashingBoundedLoadBalancer(
+ ConsistentHashingLoadBalancerType type)
+ : ConsistentHashingLoadBalancer(type)
+ , _load_factor(FLAGS_chash_bounded_load_factor)
+ , _total_inflight(0) {}
+
+size_t ConsistentHashingBoundedLoadBalancer::ResetLoads(
+ LoadMap& bg, const LoadMap& fg, const std::vector<SocketId>& ids) {
+ bg.clear();
+ for (size_t i = 0; i < ids.size(); ++i) {
+ const std::shared_ptr<ServerLoad>* fg_load = fg.seek(ids[i]);
+ bg[ids[i]] = (fg_load != nullptr)
+ ? *fg_load : std::make_shared<ServerLoad>();
+ }
+ // Non-zero so that both buffers are always rebuilt.
+ return 1;
+}
+
+void ConsistentHashingBoundedLoadBalancer::SyncLoadMap() {
+ std::vector<SocketId> ids;
+ {
+ butil::DoublyBufferedData<std::vector<Node> >::ScopedPtr s;
+ if (_db_hash_ring.Read(&s) != 0) {
+ return;
+ }
+ butil::FlatSet<SocketId> id_set;
+ ids.reserve(s->size() / std::max(_num_replicas, (size_t)1));
+ for (size_t i = 0; i < s->size(); ++i) {
+ const SocketId id = (*s)[i].server_sock.id;
+ if (id_set.seek(id) == nullptr && id_set.insert(id) != nullptr) {
+ ids.push_back(id);
}
- continue;
}
- LOG(ERROR) << "Failed to set this unknown parameters " <<
sp.key_and_value();
}
+ _db_load_map.ModifyWithForeground(ResetLoads, ids);
+}
+
+bool ConsistentHashingBoundedLoadBalancer::AddServer(const ServerId& server) {
+ if (!ConsistentHashingLoadBalancer::AddServer(server)) {
+ return false;
+ }
+ SyncLoadMap();
return true;
}
+bool ConsistentHashingBoundedLoadBalancer::RemoveServer(const ServerId&
server) {
+ if (!ConsistentHashingLoadBalancer::RemoveServer(server)) {
+ return false;
+ }
+ SyncLoadMap();
+ return true;
+}
+
+size_t ConsistentHashingBoundedLoadBalancer::AddServersInBatch(
+ const std::vector<ServerId>& servers) {
+ const size_t n = ConsistentHashingLoadBalancer::AddServersInBatch(servers);
+ if (n != 0) {
+ SyncLoadMap();
+ }
+ return n;
+}
+
+size_t ConsistentHashingBoundedLoadBalancer::RemoveServersInBatch(
+ const std::vector<ServerId>& servers) {
+ const size_t n =
ConsistentHashingLoadBalancer::RemoveServersInBatch(servers);
+ if (n != 0) {
+ SyncLoadMap();
+ }
+ return n;
+}
+
+LoadBalancer* ConsistentHashingBoundedLoadBalancer::New(
+ const butil::StringPiece& params) const {
+ ConsistentHashingBoundedLoadBalancer* lb =
+ new (std::nothrow) ConsistentHashingBoundedLoadBalancer(_type);
+ if (lb && !lb->SetParameters(params)) {
+ delete lb;
+ lb = nullptr;
+ }
+ return lb;
+}
+
+bool ConsistentHashingBoundedLoadBalancer::SetParameter(
+ const butil::StringPiece& key, const butil::StringPiece& value) {
+ if (key == "load_factor") {
+ double factor = 0.0;
+ if (!butil::StringToDouble(value.as_string(), &factor) ||
+ factor <= 1.0) {
+ LOG(ERROR) << "Invalid load_factor=`" << value
+ << "', must be a number > 1";
+ return false;
+ }
+ _load_factor = factor;
+ return true;
+ }
+ return ConsistentHashingLoadBalancer::SetParameter(key, value);
+}
+
+int ConsistentHashingBoundedLoadBalancer::SelectServer(
+ const SelectIn& in, SelectOut* out) {
+ if (!in.has_request_code) {
+ LOG(ERROR) << "Controller.set_request_code() is required";
+ return EINVAL;
+ }
+ if (in.request_code > UINT_MAX) {
+ LOG(ERROR) << "request_code must be 32-bit currently";
+ return EINVAL;
+ }
+ butil::DoublyBufferedData<std::vector<Node> >::ScopedPtr s;
+ if (_db_hash_ring.Read(&s) != 0) {
+ return ENOMEM;
+ }
+ if (s->empty()) {
+ return ENODATA;
+ }
+ butil::DoublyBufferedData<LoadMap>::ScopedPtr lm;
+ if (_db_load_map.Read(&lm) != 0) {
+ return ENOMEM;
+ }
+ int64_t capacity = std::numeric_limits<int64_t>::max();
+ if (!lm->empty()) {
+ const int64_t total =
_total_inflight.load(butil::memory_order_relaxed);
+ capacity = (int64_t)std::ceil(
+ _load_factor * (double)(total + 1) / (double)lm->size());
+ }
+ std::vector<Node>::const_iterator choice =
+ std::lower_bound(s->begin(), s->end(), (uint32_t)in.request_code);
+ if (choice == s->end()) {
+ choice = s->begin();
+ }
+ // Walk clockwise from the hashed-to node and take the first server under
+ // capacity. With load_factor > 1 at least one server is below the average
+ // whenever counters are consistent, so the walk finds one; the first
+ // acceptable server is kept as a fallback to guard against transient
+ // inconsistency of the relaxed counters.
+ SocketUniquePtr fallback_ptr;
+ ServerLoad* fallback_load = nullptr;
+ ServerLoad* selected_load = nullptr;
+ for (size_t i = 0; i < s->size(); ++i) {
+ SocketUniquePtr ptr;
+ if (((i + 1) == s->size() // always take last chance
+ || !ExcludedServers::IsExcluded(in.excluded,
choice->server_sock.id))
+ && IsServerAvailable(choice->server_sock.id, &ptr)) {
+ const std::shared_ptr<ServerLoad>* pload =
+ lm->seek(choice->server_sock.id);
+ ServerLoad* load = (pload != nullptr) ? pload->get() : nullptr;
+ const int32_t inflight = (load != nullptr)
+ ? load->inflight.load(butil::memory_order_relaxed) : 0;
+ if (inflight < capacity) {
+ selected_load = load;
+ out->ptr->swap(ptr);
+ break;
+ }
+ if (fallback_ptr.get() == nullptr) {
+ fallback_load = load;
+ fallback_ptr.swap(ptr);
+ }
+ }
+ if (++choice == s->end()) {
+ choice = s->begin();
+ }
+ }
+ if (out->ptr->get() == nullptr) {
+ if (fallback_ptr.get() == nullptr) {
+ return EHOSTDOWN;
+ }
+ selected_load = fallback_load;
+ out->ptr->swap(fallback_ptr);
+ }
+ if (in.changable_weights && selected_load != nullptr) {
+ selected_load->inflight.fetch_add(1, butil::memory_order_relaxed);
+ _total_inflight.fetch_add(1, butil::memory_order_relaxed);
+ out->need_feedback = true;
+ }
+ return 0;
+}
+
+void ConsistentHashingBoundedLoadBalancer::Feedback(const CallInfo& info) {
+ _total_inflight.fetch_sub(1, butil::memory_order_relaxed);
+ butil::DoublyBufferedData<LoadMap>::ScopedPtr lm;
+ if (_db_load_map.Read(&lm) != 0) {
+ return;
+ }
+ const std::shared_ptr<ServerLoad>* pload = lm->seek(info.server_id);
+ if (pload != nullptr) {
+ // If the server was removed after selection, its counter is already
+ // gone and only the total needs restoring.
+ (*pload)->inflight.fetch_sub(1, butil::memory_order_relaxed);
+ }
+}
+
+void ConsistentHashingBoundedLoadBalancer::Describe(
+ std::ostream& os, const DescribeOptions& options) {
+ if (!options.verbose) {
+ os << "c_hash_bl";
+ return;
+ }
+ os << "BoundedLoad{load_factor=" << _load_factor << " total_inflight="
+ << _total_inflight.load(butil::memory_order_relaxed) << "}\n";
+ ConsistentHashingLoadBalancer::Describe(os, options);
+}
+
} // namespace policy
} // namespace brpc
diff --git a/src/brpc/policy/consistent_hashing_load_balancer.h
b/src/brpc/policy/consistent_hashing_load_balancer.h
index a4808c1a..812cfa15 100644
--- a/src/brpc/policy/consistent_hashing_load_balancer.h
+++ b/src/brpc/policy/consistent_hashing_load_balancer.h
@@ -21,9 +21,11 @@
#include <stdint.h> // uint32_t
#include <functional>
+#include <memory> // std::shared_ptr
#include <vector> // std::vector
#include "butil/endpoint.h" // butil::EndPoint
#include "butil/containers/doubly_buffered_data.h"
+#include "butil/containers/flat_map.h" // FlatMap
#include "brpc/load_balancer.h"
@@ -69,8 +71,16 @@ public:
int SelectServer(const SelectIn &in, SelectOut *out);
void Describe(std::ostream &os, const DescribeOptions& options);
-private:
+protected:
bool SetParameters(const butil::StringPiece& params);
+ virtual bool SetParameter(const butil::StringPiece& key,
+ const butil::StringPiece& value);
+
+ size_t _num_replicas;
+ ConsistentHashingLoadBalancerType _type;
+ butil::DoublyBufferedData<std::vector<Node> > _db_hash_ring;
+
+private:
void GetLoads(std::map<butil::EndPoint, double> *load_map);
static size_t AddBatch(std::vector<Node> &bg, const std::vector<Node> &fg,
const std::vector<Node> &servers, bool *executed);
@@ -78,9 +88,45 @@ private:
const std::vector<ServerId> &servers, bool
*executed);
static size_t Remove(std::vector<Node> &bg, const std::vector<Node> &fg,
const ServerId& server, bool *executed);
- size_t _num_replicas;
- ConsistentHashingLoadBalancerType _type;
- butil::DoublyBufferedData<std::vector<Node> > _db_hash_ring;
+};
+
+// "Consistent Hashing with Bounded Loads" (Mirrokni et al., CACM 2017) on
+// top of the parent's hash ring: a server whose in-flight count reached
+// ceil(load_factor * average_inflight) overflows to the next server
+// clockwise on the ring with spare capacity, so a hot key spills to its
+// ring successors instead of saturating a single server.
+class ConsistentHashingBoundedLoadBalancer : public
ConsistentHashingLoadBalancer {
+public:
+ explicit ConsistentHashingBoundedLoadBalancer(
+ ConsistentHashingLoadBalancerType type);
+ bool AddServer(const ServerId& server) override;
+ bool RemoveServer(const ServerId& server) override;
+ size_t AddServersInBatch(const std::vector<ServerId>& servers) override;
+ size_t RemoveServersInBatch(const std::vector<ServerId>& servers) override;
+ LoadBalancer* New(const butil::StringPiece& params) const override;
+ int SelectServer(const SelectIn& in, SelectOut* out) override;
+ void Feedback(const CallInfo& info) override;
+ void Describe(std::ostream& os, const DescribeOptions& options) override;
+
+private:
+ struct ServerLoad {
+ ServerLoad() : inflight(0) {}
+ butil::atomic<int32_t> inflight;
+ };
+ // Counters are shared by both buffers like p2c's NodeStat.
+ typedef butil::FlatMap<SocketId, std::shared_ptr<ServerLoad> > LoadMap;
+
+ bool SetParameter(const butil::StringPiece& key,
+ const butil::StringPiece& value) override;
+ // Rebuild the load map from the sockets currently on the ring, keeping
+ // the counters of servers that stay.
+ void SyncLoadMap();
+ static size_t ResetLoads(LoadMap& bg, const LoadMap& fg,
+ const std::vector<SocketId>& ids);
+
+ double _load_factor;
+ butil::atomic<int64_t> _total_inflight;
+ butil::DoublyBufferedData<LoadMap> _db_load_map;
};
} // namespace policy
diff --git a/test/brpc_ch_bounded_load_balancer_unittest.cpp
b/test/brpc_ch_bounded_load_balancer_unittest.cpp
new file mode 100644
index 00000000..e3098b9e
--- /dev/null
+++ b/test/brpc_ch_bounded_load_balancer_unittest.cpp
@@ -0,0 +1,347 @@
+// 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 <cmath>
+#include <cstdlib>
+#include <cstring>
+#include <map>
+#include <sstream>
+#include <vector>
+#include <gflags/gflags.h>
+#include <gtest/gtest.h>
+#include "butil/macros.h"
+#include "brpc/socket.h"
+#include "brpc/excluded_servers.h"
+#include "brpc/policy/consistent_hashing_load_balancer.h"
+#include "brpc/policy/hasher.h"
+
+namespace {
+
+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;
+ EXPECT_EQ(0, brpc::Socket::Create(options, &id.id));
+ id.tag = tag;
+ return id;
+}
+
+void DestroyServers(const std::vector<brpc::ServerId>& ids) {
+ for (size_t i = 0; i < ids.size(); ++i) {
+ brpc::Socket::SetFailed(ids[i].id);
+ }
+}
+
+brpc::LoadBalancer::SelectIn MakeInput(uint64_t code,
+ bool changable_weights = true) {
+ brpc::LoadBalancer::SelectIn in = { 0, changable_weights, true, code,
nullptr };
+ return in;
+}
+
+int64_t TotalInflightOf(brpc::LoadBalancer* lb) {
+ std::ostringstream os;
+ brpc::DescribeOptions opt;
+ opt.verbose = true;
+ lb->Describe(os, opt);
+ const std::string desc = os.str();
+ const size_t pos = desc.find("total_inflight=");
+ EXPECT_NE(std::string::npos, pos) << desc;
+ return strtoll(desc.c_str() + pos + strlen("total_inflight="), nullptr,
10);
+}
+
+class CHBoundedLoadTest : public testing::Test {};
+
+TEST_F(CHBoundedLoadTest, load_factor_validation) {
+ brpc::policy::ConsistentHashingBoundedLoadBalancer lb(
+ brpc::policy::CONS_HASH_LB_MURMUR3);
+ ASSERT_EQ(nullptr, lb.New("load_factor=1.0"));
+ ASSERT_EQ(nullptr, lb.New("load_factor=0.5"));
+ ASSERT_EQ(nullptr, lb.New("load_factor=abc"));
+ brpc::LoadBalancer* valid = lb.New("load_factor=1.5");
+ ASSERT_TRUE(valid != nullptr);
+ valid->Destroy();
+
+ ASSERT_EQ("", GFLAGS_NAMESPACE::SetCommandLineOption(
+ "chash_bounded_load_factor", "0.9"));
+ ASSERT_EQ("", GFLAGS_NAMESPACE::SetCommandLineOption(
+ "chash_bounded_load_factor", "1.0"));
+ ASSERT_NE("", GFLAGS_NAMESPACE::SetCommandLineOption(
+ "chash_bounded_load_factor", "1.25"));
+}
+
+TEST_F(CHBoundedLoadTest, replicas_parameter) {
+ brpc::policy::ConsistentHashingLoadBalancer classic_lb(
+ brpc::policy::CONS_HASH_LB_MURMUR3);
+ brpc::policy::ConsistentHashingBoundedLoadBalancer bounded_lb(
+ brpc::policy::CONS_HASH_LB_MURMUR3);
+ const brpc::LoadBalancer* lbs[] = { &classic_lb, &bounded_lb };
+ for (size_t i = 0; i < arraysize(lbs); ++i) {
+ ASSERT_EQ(nullptr, lbs[i]->New("replicas=abc"));
+ brpc::LoadBalancer* lb = lbs[i]->New("replicas=300");
+ ASSERT_TRUE(lb != nullptr);
+ std::ostringstream os;
+ brpc::DescribeOptions opt;
+ opt.verbose = true;
+ lb->Describe(os, opt);
+ ASSERT_NE(std::string::npos,
+ os.str().find("replica per host: 300")) << os.str();
+ lb->Destroy();
+ }
+ brpc::LoadBalancer* lb = bounded_lb.New("replicas=200 load_factor=2");
+ ASSERT_TRUE(lb != nullptr);
+ lb->Destroy();
+}
+
+TEST_F(CHBoundedLoadTest, hot_key_is_capped) {
+ const size_t N = 8;
+ const size_t K = 200;
+ const double FACTOR = 1.25;
+ std::vector<brpc::ServerId> ids;
+ brpc::policy::ConsistentHashingBoundedLoadBalancer lb(
+ brpc::policy::CONS_HASH_LB_MURMUR3);
+ brpc::policy::ConsistentHashingLoadBalancer classic_lb(
+ brpc::policy::CONS_HASH_LB_MURMUR3);
+ for (size_t i = 0; i < N; ++i) {
+ char addr[32];
+ snprintf(addr, sizeof(addr), "192.168.1.%d:8080", (int)i);
+ ids.push_back(CreateServer(addr));
+ }
+ ASSERT_EQ(N, lb.AddServersInBatch(ids));
+ ASSERT_EQ(N, classic_lb.AddServersInBatch(ids));
+
+ const std::string hot_key = "hot_key";
+ brpc::LoadBalancer::SelectIn in =
+ MakeInput(brpc::policy::MurmurHash32(hot_key.data(), hot_key.size()));
+
+ // Classic CH sends every request for the key to one server.
+ std::map<brpc::SocketId, size_t> classic_counts;
+ for (size_t i = 0; i < K; ++i) {
+ brpc::SocketUniquePtr ptr;
+ brpc::LoadBalancer::SelectOut out(&ptr);
+ ASSERT_EQ(0, classic_lb.SelectServer(in, &out));
+ ++classic_counts[ptr->id()];
+ }
+ ASSERT_EQ(1UL, classic_counts.size());
+ const brpc::SocketId primary = classic_counts.begin()->first;
+
+ // Bounded-load CH spreads the same hot key once the primary server hits
+ // its capacity: no server exceeds ceil(FACTOR * K / N) outstanding
+ // requests(the cap of the last selection, when total load is largest).
+ std::map<brpc::SocketId, size_t> counts;
+ for (size_t i = 0; i < K; ++i) {
+ brpc::SocketUniquePtr ptr;
+ brpc::LoadBalancer::SelectOut out(&ptr);
+ ASSERT_EQ(0, lb.SelectServer(in, &out));
+ ASSERT_TRUE(out.need_feedback);
+ ++counts[ptr->id()];
+ }
+ const size_t cap = (size_t)std::ceil(FACTOR * K / N);
+ size_t max_count = 0;
+ for (std::map<brpc::SocketId, size_t>::iterator it = counts.begin();
+ it != counts.end(); ++it) {
+ max_count = std::max(max_count, it->second);
+ }
+ ASSERT_LE(max_count, cap);
+ ASSERT_GT(counts.size(), 1UL);
+ // The first selection matches classic CH: locality is unchanged until
+ // the primary server is at capacity.
+ ASSERT_GT(counts[primary], 0UL);
+ ASSERT_EQ((int64_t)K, TotalInflightOf(&lb));
+
+ DestroyServers(ids);
+}
+
+TEST_F(CHBoundedLoadTest, overflow_walks_to_ring_successor) {
+ const size_t N = 8;
+ const size_t K = 200;
+ std::vector<brpc::ServerId> ids;
+ brpc::policy::ConsistentHashingBoundedLoadBalancer lb(
+ brpc::policy::CONS_HASH_LB_MURMUR3);
+ brpc::policy::ConsistentHashingLoadBalancer classic_lb(
+ brpc::policy::CONS_HASH_LB_MURMUR3);
+ std::map<brpc::SocketId, brpc::ServerId> id_map;
+ for (size_t i = 0; i < N; ++i) {
+ char addr[32];
+ snprintf(addr, sizeof(addr), "192.168.1.%d:8080", (int)i);
+ ids.push_back(CreateServer(addr));
+ id_map[ids.back().id] = ids.back();
+ }
+ ASSERT_EQ(N, lb.AddServersInBatch(ids));
+ ASSERT_EQ(N, classic_lb.AddServersInBatch(ids));
+
+ const std::string hot_key = "another_hot_key";
+ brpc::LoadBalancer::SelectIn in =
+ MakeInput(brpc::policy::MurmurHash32(hot_key.data(), hot_key.size()));
+
+ // First-use order of servers under overflow.
+ std::vector<brpc::SocketId> bounded_order;
+ std::map<brpc::SocketId, size_t> seen;
+ for (size_t i = 0; i < K; ++i) {
+ brpc::SocketUniquePtr ptr;
+ brpc::LoadBalancer::SelectOut out(&ptr);
+ ASSERT_EQ(0, lb.SelectServer(in, &out));
+ if (++seen[ptr->id()] == 1) {
+ bounded_order.push_back(ptr->id());
+ }
+ }
+ ASSERT_GT(bounded_order.size(), 1UL);
+
+ // Expected overflow targets are the ring successors: what classic CH
+ // picks as servers are removed one by one.
+ for (size_t i = 0; i < bounded_order.size(); ++i) {
+ brpc::SocketUniquePtr ptr;
+ brpc::LoadBalancer::SelectOut out(&ptr);
+ ASSERT_EQ(0, classic_lb.SelectServer(in, &out));
+ ASSERT_EQ(bounded_order[i], ptr->id()) << "i=" << i;
+ ASSERT_TRUE(classic_lb.RemoveServer(id_map[ptr->id()]));
+ }
+
+ DestroyServers(ids);
+}
+
+TEST_F(CHBoundedLoadTest, feedback_decrements_load) {
+ const size_t N = 3;
+ std::vector<brpc::ServerId> ids;
+ brpc::policy::ConsistentHashingBoundedLoadBalancer lb(
+ brpc::policy::CONS_HASH_LB_MURMUR3);
+ for (size_t i = 0; i < N; ++i) {
+ char addr[32];
+ snprintf(addr, sizeof(addr), "192.168.1.%d:8080", (int)i);
+ ids.push_back(CreateServer(addr));
+ }
+ ASSERT_EQ(N, lb.AddServersInBatch(ids));
+
+ const std::string key = "some_key";
+ brpc::LoadBalancer::SelectIn in =
+ MakeInput(brpc::policy::MurmurHash32(key.data(), key.size()));
+
+ // With capacity ceil(1.25 * 1 / 3) = 1 an idle ring always routes the
+ // key to its primary server, so select+feedback staying on one server
+ // for many rounds proves the counters return to zero every round.
+ brpc::SocketId primary = 0;
+ for (size_t i = 0; i < 100; ++i) {
+ brpc::SocketUniquePtr ptr;
+ brpc::LoadBalancer::SelectOut out(&ptr);
+ ASSERT_EQ(0, lb.SelectServer(in, &out));
+ ASSERT_TRUE(out.need_feedback);
+ if (i == 0) {
+ primary = ptr->id();
+ } else {
+ ASSERT_EQ(primary, ptr->id()) << "i=" << i;
+ }
+ const brpc::LoadBalancer::CallInfo info = { 0, ptr->id(), 0, nullptr };
+ lb.Feedback(info);
+ }
+ ASSERT_EQ(0, TotalInflightOf(&lb));
+
+ // Saturate the primary without feedback: the key overflows...
+ std::vector<brpc::SocketId> outstanding;
+ bool overflowed = false;
+ for (size_t i = 0; i < 20; ++i) {
+ brpc::SocketUniquePtr ptr;
+ brpc::LoadBalancer::SelectOut out(&ptr);
+ ASSERT_EQ(0, lb.SelectServer(in, &out));
+ outstanding.push_back(ptr->id());
+ overflowed |= (ptr->id() != primary);
+ }
+ ASSERT_TRUE(overflowed);
+ // ...and returns to the primary once the load drains.
+ for (size_t i = 0; i < outstanding.size(); ++i) {
+ const brpc::LoadBalancer::CallInfo info = { 0, outstanding[i], 0,
nullptr };
+ lb.Feedback(info);
+ }
+ ASSERT_EQ(0, TotalInflightOf(&lb));
+ brpc::SocketUniquePtr ptr;
+ brpc::LoadBalancer::SelectOut out(&ptr);
+ ASSERT_EQ(0, lb.SelectServer(in, &out));
+ ASSERT_EQ(primary, ptr->id());
+ const brpc::LoadBalancer::CallInfo info = { 0, ptr->id(), 0, nullptr };
+ lb.Feedback(info);
+
+ DestroyServers(ids);
+}
+
+TEST_F(CHBoundedLoadTest,
feedback_after_server_removed_keeps_total_consistent) {
+ const size_t N = 3;
+ std::vector<brpc::ServerId> ids;
+ brpc::policy::ConsistentHashingBoundedLoadBalancer lb(
+ brpc::policy::CONS_HASH_LB_MURMUR3);
+ for (size_t i = 0; i < N; ++i) {
+ char addr[32];
+ snprintf(addr, sizeof(addr), "192.168.1.%d:8080", (int)i);
+ ids.push_back(CreateServer(addr));
+ }
+ ASSERT_EQ(N, lb.AddServersInBatch(ids));
+
+ const std::string key = "some_key";
+ brpc::LoadBalancer::SelectIn in =
+ MakeInput(brpc::policy::MurmurHash32(key.data(), key.size()));
+ brpc::SocketUniquePtr ptr;
+ brpc::LoadBalancer::SelectOut out(&ptr);
+ ASSERT_EQ(0, lb.SelectServer(in, &out));
+ ASSERT_EQ(1, TotalInflightOf(&lb));
+
+ brpc::ServerId selected;
+ for (size_t i = 0; i < ids.size(); ++i) {
+ if (ids[i].id == ptr->id()) {
+ selected = ids[i];
+ }
+ }
+ ASSERT_TRUE(lb.RemoveServer(selected));
+ const brpc::LoadBalancer::CallInfo info = { 0, ptr->id(), 0, nullptr };
+ lb.Feedback(info);
+ ASSERT_EQ(0, TotalInflightOf(&lb));
+ ASSERT_TRUE(lb.AddServer(selected));
+
+ DestroyServers(ids);
+}
+
+TEST_F(CHBoundedLoadTest, no_accounting_without_changable_weights) {
+ const size_t N = 4;
+ std::vector<brpc::ServerId> ids;
+ brpc::policy::ConsistentHashingBoundedLoadBalancer lb(
+ brpc::policy::CONS_HASH_LB_MURMUR3);
+ for (size_t i = 0; i < N; ++i) {
+ char addr[32];
+ snprintf(addr, sizeof(addr), "192.168.1.%d:8080", (int)i);
+ ids.push_back(CreateServer(addr));
+ }
+ ASSERT_EQ(N, lb.AddServersInBatch(ids));
+
+ const std::string key = "some_key";
+ brpc::LoadBalancer::SelectIn in = MakeInput(
+ brpc::policy::MurmurHash32(key.data(), key.size()), false);
+ brpc::SocketId first = 0;
+ for (size_t i = 0; i < 50; ++i) {
+ brpc::SocketUniquePtr ptr;
+ brpc::LoadBalancer::SelectOut out(&ptr);
+ ASSERT_EQ(0, lb.SelectServer(in, &out));
+ ASSERT_FALSE(out.need_feedback);
+ if (i == 0) {
+ first = ptr->id();
+ } else {
+ ASSERT_EQ(first, ptr->id());
+ }
+ }
+ ASSERT_EQ(0, TotalInflightOf(&lb));
+
+ DestroyServers(ids);
+}
+
+} // namespace
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
