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.

Author: rajvarun77

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来判断该次请求是否被拒绝，被拒绝的请求不会被框架重试。

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.

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

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