From b9fa1c9cda0d8f0418185da25ce4a34f95c01706 Mon Sep 17 00:00:00 2001 From: rajvarun77 <287367605+rajvarun77@users.noreply.github.com> Date: Wed, 15 Jul 2026 01:32:14 -0400 Subject: [PATCH] Add bounded-load consistent hashing load balancer (c_murmurhash_bl) 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. --- docs/cn/client.md | 6 + docs/en/client.md | 6 + src/brpc/global.cpp | 3 + .../consistent_hashing_load_balancer.cpp | 238 +++++++++++- .../policy/consistent_hashing_load_balancer.h | 54 ++- ...brpc_ch_bounded_load_balancer_unittest.cpp | 347 ++++++++++++++++++ 6 files changed, 641 insertions(+), 13 deletions(-) create mode 100644 test/brpc_ch_bounded_load_balancer_unittest.cpp diff --git a/docs/cn/client.md b/docs/cn/client.md index e659403c24..c1c011c8dc 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 266eb90289..2178be4cb1 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 successors, which keeps caches effective. The default factor comes from -chash_bounded_load_factor(default 1.25, must be > 1) and can be overridden per channel: `c_murmurhash_bl:load_factor=1.5`. The `replicas` parameter is supported as in `c_murmurhash`. + ### 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 mechanism for recovery after cluster downtime. When no server is available in the cluster, the cluster enters recovery state. Assuming that the minimum number of servers that can serve all requests is min_working_instances, current number of servers available in the cluster is q, then in recovery state, the probability of client accepting the request is q/min_working_instances, otherwise it is discarded. If q remains unchanged for a period of time(hold_seconds), the traffic is resent to all available servers and leaves recovery state. Whether the request is rejected in recovery state is indicated by whether controller.ErrorCode() is equal to brpc::ERJECT, and the rejected request will not be retried by the framework. diff --git a/src/brpc/global.cpp b/src/brpc/global.cpp index 812a79a183..0a0837e096 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 5ff1558f8b..31d1ad744a 100644 --- a/src/brpc/policy/consistent_hashing_load_balancer.cpp +++ b/src/brpc/policy/consistent_hashing_load_balancer.cpp @@ -18,23 +18,36 @@ #include // std::set_union #include +#include // std::ceil +#include // numeric_limits #include #include #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& ids) { + bg.clear(); + for (size_t i = 0; i < ids.size(); ++i) { + const std::shared_ptr* fg_load = fg.seek(ids[i]); + bg[ids[i]] = (fg_load != nullptr) + ? *fg_load : std::make_shared(); + } + // Non-zero so that both buffers are always rebuilt. + return 1; +} + +void ConsistentHashingBoundedLoadBalancer::SyncLoadMap() { + std::vector ids; + { + butil::DoublyBufferedData >::ScopedPtr s; + if (_db_hash_ring.Read(&s) != 0) { + return; + } + butil::FlatSet 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& servers) { + const size_t n = ConsistentHashingLoadBalancer::AddServersInBatch(servers); + if (n != 0) { + SyncLoadMap(); + } + return n; +} + +size_t ConsistentHashingBoundedLoadBalancer::RemoveServersInBatch( + const std::vector& 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 >::ScopedPtr s; + if (_db_hash_ring.Read(&s) != 0) { + return ENOMEM; + } + if (s->empty()) { + return ENODATA; + } + butil::DoublyBufferedData::ScopedPtr lm; + if (_db_load_map.Read(&lm) != 0) { + return ENOMEM; + } + int64_t capacity = std::numeric_limits::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::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* 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::ScopedPtr lm; + if (_db_load_map.Read(&lm) != 0) { + return; + } + const std::shared_ptr* 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 a4808c1a70..812cfa1581 100644 --- a/src/brpc/policy/consistent_hashing_load_balancer.h +++ b/src/brpc/policy/consistent_hashing_load_balancer.h @@ -21,9 +21,11 @@ #include // uint32_t #include +#include // std::shared_ptr #include // 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 @@ class ConsistentHashingLoadBalancer : public LoadBalancer { 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 > _db_hash_ring; + +private: void GetLoads(std::map *load_map); static size_t AddBatch(std::vector &bg, const std::vector &fg, const std::vector &servers, bool *executed); @@ -78,9 +88,45 @@ class ConsistentHashingLoadBalancer : public LoadBalancer { const std::vector &servers, bool *executed); static size_t Remove(std::vector &bg, const std::vector &fg, const ServerId& server, bool *executed); - size_t _num_replicas; - ConsistentHashingLoadBalancerType _type; - butil::DoublyBufferedData > _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& servers) override; + size_t RemoveServersInBatch(const std::vector& 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 inflight; + }; + // Counters are shared by both buffers like p2c's NodeStat. + typedef butil::FlatMap > 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& ids); + + double _load_factor; + butil::atomic _total_inflight; + butil::DoublyBufferedData _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 0000000000..e3098b9e9d --- /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 +#include +#include +#include +#include +#include +#include +#include +#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& 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 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 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 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::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 ids; + brpc::policy::ConsistentHashingBoundedLoadBalancer lb( + brpc::policy::CONS_HASH_LB_MURMUR3); + brpc::policy::ConsistentHashingLoadBalancer classic_lb( + brpc::policy::CONS_HASH_LB_MURMUR3); + std::map 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 bounded_order; + std::map 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 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 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 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 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