Add cross-scheme slow-start warm-up ramp for newly added servers

docs/cn/client.md

@@ -290,6 +290,14 @@ locality-aware，优先选择延时低的下游，直到其延时高于其他机
 channel.Init("http://...", "random:min_working_instances=6 hold_seconds=10", &options);
 ```
 
+### 慢启动（预热）
+
+新加入集群或刚重启的server往往是“冷”的（缓存未命中、JIT未编译、连接池未建立），立即承担全量流量会推高其延时甚至过载。设置-lb_warmup_ms大于0（默认为0，即关闭）后，新加入负载均衡器的server先获得一小部分正常流量份额（-lb_warmup_min_weight，默认0.1），并在该时间窗口内线性爬升到100%。该机制对rr、wrr、random、la、p2c和一致性哈希均生效：la和p2c把爬升系数乘入权重，与延时评分自然叠加而不会互相干扰；其余算法按该系数概率性地把请求转给其他server（一致性哈希转给环上的下一个节点，预热期间会有部分请求偏离原有的哈希亲和性）。
+
+-lb_warmup_curve（默认1.0）控制爬升曲线：流量份额为max(lb_warmup_min_weight, progress^lb_warmup_curve)，progress在窗口内从0线性升到1。大于1的值让新server冷得更久，小于1则更激进。
+
+说明：预热的起点是server被加入负载均衡器的时刻。server被命名服务摘除后重新加入会重新预热；而短暂断连或健康检查失败不改变负载均衡器成员，不会重新预热。Channel初始化时所有server同时加入、一起爬升，相对流量比例不变，因此首次启动无需特殊处理。
+
 ## 健康检查
 
 连接断开的server会被暂时隔离而不会被负载均衡算法选中，brpc会定期连接被隔离的server，以检查他们是否恢复正常，间隔由参数-health_check_interval控制:

docs/en/client.md

@@ -291,6 +291,14 @@ This recovery mechanism requires the capabilities of downstream servers to be si
 channel.Init("http://...", "random:min_working_instances=6 hold_seconds=10", &options);
 ```
 
+### Slow start (warm-up)
+
+A server that just joined the cluster or restarted is often "cold" (empty caches, uncompiled JIT, unestablished connection pools); sending it a full traffic share immediately raises its latency or even overloads it. When -lb_warmup_ms is positive (default 0, disabled), a server newly added to a LoadBalancer gets a fraction of its normal traffic share at first (-lb_warmup_min_weight, default 0.1) and ramps up to 100% over the window. The mechanism works across rr, wrr, random, la, p2c and consistent hashing: la and p2c multiply the ramp into the weight so it composes with their latency scoring instead of fighting it; the other policies divert requests probabilistically to other servers (consistent hashing moves to the next node on the ring, so part of the hash affinity is temporarily diverted during warm-up).
+
+-lb_warmup_curve (default 1.0) shapes the ramp: the traffic share is max(lb_warmup_min_weight, progress^lb_warmup_curve) where progress rises linearly from 0 to 1 over the window. Values above 1 keep a new server colder for longer, values below 1 ramp more aggressively.
+
+Note: warm-up starts when the server is added to the LoadBalancer. A server removed by the naming service and added back restarts its ramp, while a transient disconnection or health-check failure does not change LB membership and keeps the ramp. At channel initialization all servers join and ramp together with unchanged relative shares, so initial startup needs no special casing.
+
 ## Health checking
 
 Servers whose connections are lost are isolated temporarily to prevent them from being selected by LoadBalancer. brpc connects isolated servers periodically to test if they're healthy again. The interval is controlled by gflag -health_check_interval:

src/brpc/load_balancer.cpp

@@ -16,7 +16,10 @@
 // under the License.
 
 
+#include <cmath>                                  // std::pow
 #include <gflags/gflags.h>
+#include "butil/fast_rand.h"                      // fast_rand_double
+#include "butil/time.h"                           // gettimeofday_us
 #include "brpc/reloadable_flags.h"
 #include "brpc/load_balancer.h"
 #include "brpc/socket.h"
@@ -30,7 +33,60 @@ DEFINE_int32(default_weight_of_wlb, 0, "Default weight value of Weighted LoadBal
              "problems when user is using wlb but forgot to set the weights of some of their "
              "downstream instances. Then these instances will be set default_weight_of_wlb as "
              "their weights. wlb policy degradation is not enabled by default.");
+DEFINE_int64(lb_warmup_ms, 0,
+             "When positive, a server newly added to a LoadBalancer gets "
+             "lb_warmup_min_weight of its normal traffic share at first and "
+             "ramps up to 100% over this period(ms). 0 disables the warm-up");
+DEFINE_double(lb_warmup_curve, 1.0,
+              "Shape of the warm-up ramp: the weight multiplier is "
+              "max(lb_warmup_min_weight, progress^lb_warmup_curve) where progress rises "
+              "linearly from 0 to 1 over lb_warmup_ms. Must be positive: 1 ramps "
+              "linearly, larger values keep a new server colder for longer");
 BRPC_VALIDATE_GFLAG(show_lb_in_vars, PassValidate);
+BRPC_VALIDATE_GFLAG(lb_warmup_ms, PassValidate);
+DEFINE_double(lb_warmup_min_weight, 0.1,
+              "Floor of the warm-up multiplier, in (0, 1]: the share of "
+              "normal traffic a server gets right after joining, so that "
+              "it still receives a trickle and latency-based policies keep "
+              "observing it");
+static bool ValidateWarmupCurve(const char*, double v) {
+    return v > 0.0;
+}
+static bool ValidateWarmupMinWeight(const char*, double v) {
+    return v > 0.0 && v <= 1.0;
+}
+BRPC_VALIDATE_GFLAG(lb_warmup_curve, ValidateWarmupCurve);
+BRPC_VALIDATE_GFLAG(lb_warmup_min_weight, ValidateWarmupMinWeight);
+
+
+double WarmupMultiplierImpl(int64_t join_time_us, int64_t now_us) {
+    const int64_t warmup_us = FLAGS_lb_warmup_ms * 1000L;
+    if (warmup_us <= 0 || join_time_us <= 0) {
+        return 1.0;
+    }
+    if (now_us <= 0) {
+        now_us = butil::gettimeofday_us();
+    }
+    const int64_t elapsed_us = now_us - join_time_us;
+    if (elapsed_us >= warmup_us) {
+        return 1.0;
+    }
+    const double min_weight = std::min(std::max(FLAGS_lb_warmup_min_weight, 1e-9), 1.0);
+    if (elapsed_us <= 0) {
+        // The clock went backwards, be conservative.
+        return min_weight;
+    }
+    double progress = (double)elapsed_us / (double)warmup_us;
+    if (FLAGS_lb_warmup_curve > 0 && FLAGS_lb_warmup_curve != 1.0) {
+        progress = std::pow(progress, FLAGS_lb_warmup_curve);
+    }
+    return std::max(progress, min_weight);
+}
+
+bool WarmupAcceptImpl(int64_t join_time_us, int64_t now_us) {
+    const double m = WarmupMultiplierImpl(join_time_us, now_us);
+    return m >= 1.0 || butil::fast_rand_double() < m;
+}
 
 // For assigning unique names for lb.
 static butil::static_atomic<int> g_lb_counter = BUTIL_STATIC_ATOMIC_INIT(0);

src/brpc/load_balancer.h

@@ -113,6 +113,31 @@ class LoadBalancer : public NonConstDescribable, public Destroyable {
 
 DECLARE_bool(show_lb_in_vars);
 DECLARE_int32(default_weight_of_wlb);
+DECLARE_int64(lb_warmup_ms);
+
+double WarmupMultiplierImpl(int64_t join_time_us, int64_t now_us);
+bool WarmupAcceptImpl(int64_t join_time_us, int64_t now_us);
+
+// Slow start: while -lb_warmup_ms is positive, a server newly added to a
+// LoadBalancer serves a ramping fraction of its normal traffic share, from
+// about 10% right after joining to 100% at the end of the window. The ramp
+// restarts when a removed server is added back(naming service flap); a
+// transiently disconnected server does not change LB membership and keeps
+// its ramp. Servers added together(e.g. at channel init) ramp together and
+// keep their relative shares.
+// Returns the weight multiplier in (0, 1] for a server that joined the
+// LoadBalancer at `join_time_us'(gettimeofday_us). `now_us' <= 0 makes the
+// function read the clock itself.
+inline double WarmupMultiplier(int64_t join_time_us, int64_t now_us) {
+    return FLAGS_lb_warmup_ms <= 0 ?
+        1.0 : WarmupMultiplierImpl(join_time_us, now_us);
+}
+
+// Probabilistic form of WarmupMultiplier for policies without changable
+// weights: returns true with probability WarmupMultiplier(...).
+inline bool WarmupAccept(int64_t join_time_us, int64_t now_us) {
+    return FLAGS_lb_warmup_ms <= 0 || WarmupAcceptImpl(join_time_us, now_us);
+}
 
 // A intrusively shareable load balancer created from name.
 class SharedLoadBalancer : public SharedObject, public NonConstDescribable {

src/brpc/policy/consistent_hashing_load_balancer.cpp

@@ -22,6 +22,7 @@
 #include <openssl/md5.h>
 #include "butil/containers/flat_map.h"
 #include "butil/errno.h"
+#include "butil/time.h"
 #include "butil/strings/string_number_conversions.h"
 #include "brpc/socket.h"
 #include "brpc/policy/consistent_hashing_load_balancer.h"
@@ -71,6 +72,7 @@ bool DefaultReplicaPolicy::Build(ServerId server,
         return false;
     }
     replicas->clear();
+    const int64_t join_time_us = butil::gettimeofday_us();
     for (size_t i = 0; i < num_replicas; ++i) {
         char host[256];
         int len = 0;
@@ -85,6 +87,7 @@ bool DefaultReplicaPolicy::Build(ServerId server,
         node.hash = _hash_func(host, len);
         node.server_sock = server;
         node.server_addr = ptr->remote_side();
+        node.join_time_us = join_time_us;
         replicas->push_back(node);
     }
     return true;
@@ -107,6 +110,7 @@ bool KetamaReplicaPolicy::Build(ServerId server,
         return false;
     }
     replicas->clear();
+    const int64_t join_time_us = butil::gettimeofday_us();
     const size_t points_per_hash = 4;
     CHECK(num_replicas % points_per_hash == 0)
         << "Ketam hash replicas number(" << num_replicas << ") should be n*4";
@@ -126,6 +130,7 @@ bool KetamaReplicaPolicy::Build(ServerId server,
             ConsistentHashingLoadBalancer::Node node;
             node.server_sock = server;
             node.server_addr = ptr->remote_side();
+            node.join_time_us = join_time_us;
             node.hash = ((uint32_t) (digest[3 + j * 4] & 0xFF) << 24)
                       | ((uint32_t) (digest[2 + j * 4] & 0xFF) << 16)
                       | ((uint32_t) (digest[1 + j * 4] & 0xFF) << 8)
@@ -321,7 +326,8 @@ int ConsistentHashingLoadBalancer::SelectServer(
     }
     for (size_t i = 0; i < s->size(); ++i) {
         if (((i + 1) == s->size() // always take last chance
-             || !ExcludedServers::IsExcluded(in.excluded, choice->server_sock.id))
+             || (!ExcludedServers::IsExcluded(in.excluded, choice->server_sock.id)
+                 && WarmupAccept(choice->join_time_us, in.begin_time_us)))
             && IsServerAvailable(choice->server_sock.id, out->ptr)) {
             return 0;
         } else {

src/brpc/policy/consistent_hashing_load_balancer.h

@@ -47,6 +47,10 @@ class ConsistentHashingLoadBalancer : public LoadBalancer {
         uint32_t hash;
         ServerId server_sock;
         butil::EndPoint server_addr;  // To make sorting stable among all clients
+        // Time when the server was added, for the warm-up ramp. Not part
+        // of ordering/equality so that re-adding an existing server keeps
+        // its original stamp.
+        int64_t join_time_us;
         bool operator<(const Node &rhs) const {
             if (hash < rhs.hash) { return true; }
             if (hash > rhs.hash) { return false; }

src/brpc/policy/locality_aware_load_balancer.cpp

@@ -555,6 +555,7 @@ LocalityAwareLoadBalancer::Weight::Weight(int64_t initial_weight)
     , _old_index((size_t)-1L)
     , _old_weight(0)
     , _avg_latency(0)
+    , _join_time_us(butil::gettimeofday_us())
     , _time_q(_time_q_items, sizeof(_time_q_items), butil::NOT_OWN_STORAGE) {
 }
 

src/brpc/policy/locality_aware_load_balancer.h

@@ -101,6 +101,7 @@ class LocalityAwareLoadBalancer : public LoadBalancer {
         size_t _old_index;
         int64_t _old_weight;
         int64_t _avg_latency;
+        int64_t _join_time_us;
         butil::BoundedQueue<TimeInfo> _time_q;
         // content of _time_q
         TimeInfo _time_q_items[RECV_QUEUE_SIZE];
@@ -176,6 +177,10 @@ inline int64_t LocalityAwareLoadBalancer::Weight::ResetWeight(
             new_weight = new_weight * punish_latency / inflight_delay;
         }
     }
+    const double wm = WarmupMultiplier(_join_time_us, now_us);
+    if (wm < 1.0) {
+        new_weight = (int64_t)(new_weight * wm);
+    }
     if (new_weight < FLAGS_min_weight) {
         new_weight = FLAGS_min_weight;
     }

src/brpc/policy/p2c_ewma_load_balancer.cpp

@@ -81,6 +81,7 @@ bool P2CEwmaLoadBalancer::Add(Servers& bg, const Servers& fg,
         // Both buffers do not have the server. Create the stat structure
         // which will be shared by both buffers.
         info.stat = std::make_shared<NodeStat>();
+        info.stat->join_time_us = butil::gettimeofday_us();
     } else {
         // Already added to the other buffer, share its stat.
         info.stat = fg.server_list[*pindex].stat;
@@ -164,7 +165,10 @@ double P2CEwmaLoadBalancer::Score(
     }
     // 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;
+    // The warm-up multiplier discounts the effective weight, composing with
+    // (instead of fighting) the latency score of a cold server.
+    return latency_term * (double)load /
+        ((double)info.weight * WarmupMultiplier(info.stat->join_time_us, now_us));
 }
 
 int P2CEwmaLoadBalancer::SelectServer(const SelectIn& in, SelectOut* out) {

src/brpc/policy/p2c_ewma_load_balancer.h

@@ -54,7 +54,9 @@ class P2CEwmaLoadBalancer : public LoadBalancer {
     // 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) {}
+        NodeStat() : join_time_us(0), inflight(0), ewma_us(0), stamp_us(0) {}
+        // Time when the server was added, for the warm-up ramp.
+        int64_t join_time_us;
         butil::atomic<int32_t> inflight;
         // Peak-sensitive EWMA of latency in us. 0 means no observation yet.
         butil::atomic<int64_t> ewma_us;

src/brpc/policy/randomized_load_balancer.cpp

@@ -18,6 +18,7 @@
 
 #include "butil/macros.h"
 #include "butil/fast_rand.h"
+#include "butil/time.h"
 #include "bthread/prime_offset.h"
 #include "brpc/socket.h"
 #include "brpc/policy/randomized_load_balancer.h"
@@ -36,6 +37,7 @@ bool RandomizedLoadBalancer::Add(Servers& bg, const ServerId& id) {
     }
     bg.server_map[id] = bg.server_list.size();
     bg.server_list.push_back(id);
+    bg.join_times.push_back(butil::gettimeofday_us());
     return true;
 }
 
@@ -44,8 +46,10 @@ bool RandomizedLoadBalancer::Remove(Servers& bg, const ServerId& id) {
     if (it != bg.server_map.end()) {
         size_t index = it->second;
         bg.server_list[index] = bg.server_list.back();
+        bg.join_times[index] = bg.join_times.back();
         bg.server_map[bg.server_list[index]] = index;
         bg.server_list.pop_back();
+        bg.join_times.pop_back();
         bg.server_map.erase(it);
         return true;
     }
@@ -112,7 +116,8 @@ int RandomizedLoadBalancer::SelectServer(const SelectIn& in, SelectOut* out) {
     for (size_t i = 0; i < n; ++i) {
         const SocketId id = s->server_list[offset].id;
         if (((i + 1) == n  // always take last chance
-             || !ExcludedServers::IsExcluded(in.excluded, id))
+             || (!ExcludedServers::IsExcluded(in.excluded, id)
+                 && WarmupAccept(s->join_times[offset], in.begin_time_us)))
             && IsServerAvailable(id, out->ptr)) {
             // We found an available server
             return 0;

src/brpc/policy/randomized_load_balancer.h

@@ -45,6 +45,8 @@ class RandomizedLoadBalancer : public LoadBalancer {
 private:
     struct Servers {
         std::vector<ServerId> server_list;
+        // Time when server_list[i] was added, for the warm-up ramp.
+        std::vector<int64_t> join_times;
         std::map<ServerId, size_t> server_map;
     };
     bool SetParameters(const butil::StringPiece& params);

src/brpc/policy/round_robin_load_balancer.cpp

@@ -18,6 +18,7 @@
 
 #include "butil/macros.h"
 #include "butil/fast_rand.h"
+#include "butil/time.h"
 #include "bthread/prime_offset.h"
 #include "brpc/socket.h"
 #include "brpc/policy/round_robin_load_balancer.h"
@@ -36,6 +37,7 @@ bool RoundRobinLoadBalancer::Add(Servers& bg, const ServerId& id) {
     }
     bg.server_map[id] = bg.server_list.size();
     bg.server_list.push_back(id);
+    bg.join_times.push_back(butil::gettimeofday_us());
     return true;
 }
 
@@ -44,8 +46,10 @@ bool RoundRobinLoadBalancer::Remove(Servers& bg, const ServerId& id) {
     if (it != bg.server_map.end()) {
         const size_t index = it->second;
         bg.server_list[index] = bg.server_list.back();
+        bg.join_times[index] = bg.join_times.back();
         bg.server_map[bg.server_list[index]] = index;
         bg.server_list.pop_back();
+        bg.join_times.pop_back();
         bg.server_map.erase(it);
         return true;
     }
@@ -119,7 +123,8 @@ int RoundRobinLoadBalancer::SelectServer(const SelectIn& in, SelectOut* out) {
         tls.offset = (tls.offset + tls.stride) % n;
         const SocketId id = s->server_list[tls.offset].id;
         if (((i + 1) == n  // always take last chance
-             || !ExcludedServers::IsExcluded(in.excluded, id))
+             || (!ExcludedServers::IsExcluded(in.excluded, id)
+                 && WarmupAccept(s->join_times[tls.offset], in.begin_time_us)))
             && IsServerAvailable(id, out->ptr)) {
             s.tls() = tls;
             return 0;

src/brpc/policy/round_robin_load_balancer.h

@@ -44,6 +44,8 @@ class RoundRobinLoadBalancer : public LoadBalancer {
 private:
     struct Servers {
         std::vector<ServerId> server_list;
+        // Time when server_list[i] was added, for the warm-up ramp.
+        std::vector<int64_t> join_times;
         std::map<ServerId, size_t> server_map;
     };
     struct TLS {

src/brpc/policy/weighted_round_robin_load_balancer.cpp

@@ -19,6 +19,7 @@
 #include <algorithm>
 
 #include "butil/fast_rand.h"
+#include "butil/time.h"
 #include "brpc/socket.h"
 #include "brpc/policy/weighted_round_robin_load_balancer.h"
 #include "butil/strings/string_number_conversions.h"
@@ -91,7 +92,7 @@ bool WeightedRoundRobinLoadBalancer::Add(Servers& bg, const ServerId& id) {
     bool insert_server =
              bg.server_map.emplace(id.id, bg.server_list.size()).second;
     if (insert_server) {
-        bg.server_list.emplace_back(id.id, weight);
+        bg.server_list.emplace_back(id.id, weight, butil::gettimeofday_us());
         bg.weight_sum += weight;
         return true;
     }
@@ -182,8 +183,15 @@ int WeightedRoundRobinLoadBalancer::SelectServer(const SelectIn& in, SelectOut*
     size_t remain_servers = s->server_list.size();
     while (remain_servers > 0) {
         SocketId server_id = GetServerInNextStride(s->server_list, filter, tls_temp);
+        bool warmup_pass = true;
+        if (remain_servers > 1 && FLAGS_lb_warmup_ms > 0) {
+            warmup_pass = WarmupAccept(
+                s->server_list[s->server_map.at(server_id)].join_time_us,
+                in.begin_time_us);
+        }
         if ((remain_servers == 1 // always take last chance
-                || !ExcludedServers::IsExcluded(in.excluded, server_id))
+                || (!ExcludedServers::IsExcluded(in.excluded, server_id)
+                    && warmup_pass))
             && Socket::Address(server_id, out->ptr) == 0
             && (*out->ptr)->IsAvailable()) {
             // update tls.

src/brpc/policy/weighted_round_robin_load_balancer.h

@@ -43,9 +43,12 @@ class WeightedRoundRobinLoadBalancer : public LoadBalancer {
 
 private:
     struct Server {
-        Server(SocketId s_id = 0, uint32_t s_w = 0): id(s_id), weight(s_w) {}
+        Server(SocketId s_id = 0, uint32_t s_w = 0, int64_t s_jt = 0)
+            : id(s_id), weight(s_w), join_time_us(s_jt) {}
         SocketId id;
         uint32_t weight;
+        // Time when the server was added, for the warm-up ramp.
+        int64_t join_time_us;
     };
     struct Servers {
         // The value is configured weight for each server.