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 32c95225 Add Power-of-Two-Choices Peak-EWMA load balancer (p2c) (#3367)
32c95225 is described below
commit 32c95225e83b88d8037560ba33c43977b4641508
Author: rajvarun77 <[email protected]>
AuthorDate: Sun Jul 12 10:41:54 2026 -0400
Add Power-of-Two-Choices Peak-EWMA load balancer (p2c) (#3367)
* Add Power-of-Two-Choices Peak-EWMA load balancer (p2c)
Implements the p2c load balancing policy proposed in #3340: each
selection samples two random servers (configurable via choices=N) and
routes to the lower peak-EWMA latency * (inflight+1) / weight score.
Upward latency spikes take effect immediately while recovery decays
over tau_ms (default 10s), so a degraded server is shed within one
observation at O(1) selection cost.
Uses only existing LoadBalancer hooks (SelectServer/Feedback,
need_feedback) with DoublyBufferedData membership like rr/la; per-node
stats are shared_ptr-owned by both buffers. Registered as "p2c" in
global.cpp. Includes unit tests (functional, weighted, exclusion,
error punishment, concurrency churn) and docs in cn/en client.md.
Co-Authored-By: Claude Fable 5 <[email protected]>
* Reuse SelectIn.begin_time_us to avoid extra clock read per selection
Co-Authored-By: Claude Fable 5 <[email protected]>
* Address review: gflags for tunables, cap error punishment, drop FlatMap
init
- p2c_default_choices/p2c_default_tau_ms gflags as channel-overridable
defaults
- p2c_max_punish_ms(default 30s) caps the doubled-EWMA error punishment so a
persistently failing server recovers in bounded time after turning healthy
- Servers::server_map relies on FlatMap small-map auto-init
- tests: error_punish_is_capped + feedback_lock_overhead benchmark
Co-Authored-By: Claude Fable 5 <[email protected]>
---------
Co-authored-by: rajvarun77 <[email protected]>
Co-authored-by: Claude Fable 5 <[email protected]>
---
docs/cn/client.md | 4 +
docs/cn/rpc_press.md | 2 +-
docs/en/client.md | 4 +
src/brpc/global.cpp | 3 +
src/brpc/policy/p2c_ewma_load_balancer.cpp | 389 +++++++++++++++++++++++
src/brpc/policy/p2c_ewma_load_balancer.h | 94 ++++++
test/brpc_p2c_ewma_load_balancer_unittest.cpp | 428 ++++++++++++++++++++++++++
tools/rpc_press/rpc_press.cpp | 2 +-
8 files changed, 924 insertions(+), 2 deletions(-)
diff --git a/docs/cn/client.md b/docs/cn/client.md
index 609becc0..bb16eefb 100755
--- a/docs/cn/client.md
+++ b/docs/cn/client.md
@@ -260,6 +260,10 @@ int main() {
locality-aware，优先选择延时低的下游，直到其延时高于其他机器，无需其他设置。实现原理请查看[Locality-aware load
balancing](lalb.md)。
+### p2c
+
+即power-of-two-choices加peak-EWMA延时评分。每次选择随机采样两台服务器，把请求发给`延时 * (inflight + 1) /
权重`得分较低的那台。其中延时是对尖峰敏感的滑动平均：延时上升立即生效，恢复则按`tau_ms`（默认10秒）衰减。变慢或出错的服务器在一次观察内即被避开，且选择开销与集群规模无关，为O(1)。权重取自实例tag（同wrr，默认为1）。可选参数：`p2c:choices=4`（每次比较4台采样服务器，适合多台机器同时劣化的场景）、`p2c:tau_ms=5000`。
+
### c_murmurhash or c_md5
一致性哈希，与简单hash的不同之处在于增加或删除机器时不会使分桶结果剧烈变化，特别适合cache类服务。
diff --git a/docs/cn/rpc_press.md b/docs/cn/rpc_press.md
index 32572846..357340ae 100644
--- a/docs/cn/rpc_press.md
+++ b/docs/cn/rpc_press.md
@@ -33,7 +33,7 @@ json也可以写在文件中，假如./input.json包含了上述两个请求，-
可选参数:
- -inc:
包含被import的proto文件的路径。rpc_press默认支持import目录下的其他proto文件，但如果proto文件在其他目录，就要通过这个参数指定，多个路径用分号(;)分隔。
-- -lb_policy: 指定负载均衡算法，默认为空，可选项为: rr random la c_murmurhash
c_md5，具体见[负载均衡](client.md#负载均衡)。
+- -lb_policy: 指定负载均衡算法，默认为空，可选项为: rr random la p2c c_murmurhash
c_md5，具体见[负载均衡](client.md#负载均衡)。
- -timeout_ms: 设定超时,单位是毫秒(milliseconds),默认是1000(1秒)
- -max_retry: 最大的重试次数,默认是3, 一般无需修改. brpc的重试行为具体请见[这里](client.md#重试).
- -protocol: 连接server使用的协议，可选项见[协议](client.md#协议), 默认是baidu_std
diff --git a/docs/en/client.md b/docs/en/client.md
index 60c458b6..8da0c6b1 100644
--- a/docs/en/client.md
+++ b/docs/en/client.md
@@ -258,6 +258,10 @@ Requirements of instance tag is the same as wrr.
which is locality-aware. Perfer servers with lower latencies, until the
latency is higher than others, no other settings. Check out [Locality-aware
load balancing](lalb.md) for more details.
+### p2c
+
+which is power-of-two-choices with peak-EWMA latency scoring. Each selection
samples two random servers and routes to the one with the lower `latency *
(inflight + 1) / weight` score, where the latency is a peak-sensitive moving
average: an upward spike takes effect immediately while recovery decays over
`tau_ms`(default 10s). A slow or failing server is shed within one observation
and selection cost is O(1) regardless of cluster size. Weight comes from the
instance tag as in wrr(default [...]
+
### c_murmurhash or c_md5
which is consistent hashing. Adding or removing servers does not make
destinations of requests change as dramatically as in simple hashing. It's
especially suitable for caching services.
diff --git a/src/brpc/global.cpp b/src/brpc/global.cpp
index c1bfbcdd..e9402512 100644
--- a/src/brpc/global.cpp
+++ b/src/brpc/global.cpp
@@ -47,6 +47,7 @@
#include "brpc/policy/randomized_load_balancer.h"
#include "brpc/policy/weighted_randomized_load_balancer.h"
#include "brpc/policy/locality_aware_load_balancer.h"
+#include "brpc/policy/p2c_ewma_load_balancer.h"
#include "brpc/policy/consistent_hashing_load_balancer.h"
#include "brpc/policy/hasher.h"
#include "brpc/policy/dynpart_load_balancer.h"
@@ -157,6 +158,7 @@ struct GlobalExtensions {
RandomizedLoadBalancer randomized_lb;
WeightedRandomizedLoadBalancer wr_lb;
LocalityAwareLoadBalancer la_lb;
+ P2CEwmaLoadBalancer p2c_ewma_lb;
ConsistentHashingLoadBalancer ch_mh_lb;
ConsistentHashingLoadBalancer ch_md5_lb;
ConsistentHashingLoadBalancer ch_ketama_lb;
@@ -407,6 +409,7 @@ static void GlobalInitializeOrDieImpl() {
LoadBalancerExtension()->RegisterOrDie("random", &g_ext->randomized_lb);
LoadBalancerExtension()->RegisterOrDie("wr", &g_ext->wr_lb);
LoadBalancerExtension()->RegisterOrDie("la", &g_ext->la_lb);
+ LoadBalancerExtension()->RegisterOrDie("p2c", &g_ext->p2c_ewma_lb);
LoadBalancerExtension()->RegisterOrDie("c_murmurhash", &g_ext->ch_mh_lb);
LoadBalancerExtension()->RegisterOrDie("c_md5", &g_ext->ch_md5_lb);
LoadBalancerExtension()->RegisterOrDie("c_ketama", &g_ext->ch_ketama_lb);
diff --git a/src/brpc/policy/p2c_ewma_load_balancer.cpp
b/src/brpc/policy/p2c_ewma_load_balancer.cpp
new file mode 100644
index 00000000..2b3f5bb5
--- /dev/null
+++ b/src/brpc/policy/p2c_ewma_load_balancer.cpp
@@ -0,0 +1,389 @@
+// 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> // std::exp
+#include <gflags/gflags.h>
+#include "butil/fast_rand.h" // fast_rand_less_than
+#include "butil/time.h" // gettimeofday_us
+#include "butil/string_splitter.h" // KeyValuePairsSplitter
+#include "butil/strings/string_number_conversions.h" // StringToUint
+#include "brpc/socket.h"
+#include "brpc/controller.h"
+#include "brpc/policy/p2c_ewma_load_balancer.h"
+
+namespace brpc {
+namespace policy {
+
+DEFINE_uint32(p2c_default_choices, 2,
+ "Default number of servers sampled per selection in p2c, "
+ "overridable per channel with the `choices' parameter");
+DEFINE_int64(p2c_default_tau_ms, 10000,
+ "Default decay time(ms) of the peak-EWMA latency in p2c "
+ "(Finagle's PeakEwma default), overridable per channel with "
+ "the `tau_ms' parameter");
+DEFINE_int64(p2c_max_punish_ms, 30000,
+ "Cap(ms) on the punished latency recorded for a failed call in "
+ "p2c, bounding how long a persistently failing server takes to "
+ "recover after it turns healthy. 0 means no cap");
+
+namespace {
+// Upper bound of the `choices' parameter. Compile-time because it sizes the
+// sampling array on SelectServer's stack.
+const uint32_t MAX_CHOICES = 64;
+// Floor of the latency term so that in-flight counts break ties between
+// servers that have no latency observation yet.
+const double MIN_LATENCY_TERM_US = 1.0;
+
+uint32_t WeightOfTag(const std::string& tag) {
+ if (tag.empty()) {
+ return 1;
+ }
+ uint32_t weight = 0;
+ if (!butil::StringToUint(tag, &weight) || weight == 0) {
+ LOG(WARNING) << "Invalid weight tag=`" << tag << "', use weight=1";
+ return 1;
+ }
+ return weight;
+}
+} // namespace
+
+P2CEwmaLoadBalancer::P2CEwmaLoadBalancer()
+ // Clamp so that broken flag values can not disable comparison or decay.
+ : _choices(std::min(std::max(FLAGS_p2c_default_choices, 2u), MAX_CHOICES))
+ , _tau_us(std::max<int64_t>(FLAGS_p2c_default_tau_ms, 1) * 1000L) {}
+
+bool P2CEwmaLoadBalancer::Add(Servers& bg, const Servers& fg,
+ const ServerId& id) {
+ if (bg.server_list.capacity() < 128) {
+ bg.server_list.reserve(128);
+ }
+ if (bg.server_map.seek(id.id) != NULL) {
+ return false;
+ }
+ ServerInfo info = { id.id, WeightOfTag(id.tag), NULL };
+ const size_t* pindex = fg.server_map.seek(id.id);
+ if (pindex == NULL) {
+ // Both buffers do not have the server. Create the stat structure
+ // which will be shared by both buffers.
+ info.stat = std::make_shared<NodeStat>();
+ } else {
+ // Already added to the other buffer, share its stat.
+ info.stat = fg.server_list[*pindex].stat;
+ }
+ bg.server_map[id.id] = bg.server_list.size();
+ bg.server_list.push_back(info);
+ return true;
+}
+
+bool P2CEwmaLoadBalancer::Remove(Servers& bg, const ServerId& id) {
+ size_t* pindex = bg.server_map.seek(id.id);
+ if (pindex == NULL) {
+ return false;
+ }
+ const size_t index = *pindex;
+ bg.server_list[index] = bg.server_list.back();
+ bg.server_map[bg.server_list[index].id] = index;
+ bg.server_list.pop_back();
+ bg.server_map.erase(id.id);
+ return true;
+}
+
+size_t P2CEwmaLoadBalancer::BatchAdd(
+ Servers& bg, const Servers& fg, const std::vector<ServerId>& servers) {
+ size_t count = 0;
+ for (size_t i = 0; i < servers.size(); ++i) {
+ count += !!Add(bg, fg, servers[i]);
+ }
+ return count;
+}
+
+size_t P2CEwmaLoadBalancer::BatchRemove(
+ Servers& bg, const std::vector<ServerId>& servers) {
+ size_t count = 0;
+ for (size_t i = 0; i < servers.size(); ++i) {
+ count += !!Remove(bg, servers[i]);
+ }
+ return count;
+}
+
+bool P2CEwmaLoadBalancer::AddServer(const ServerId& id) {
+ return _db_servers.ModifyWithForeground(Add, id);
+}
+
+bool P2CEwmaLoadBalancer::RemoveServer(const ServerId& id) {
+ return _db_servers.Modify(Remove, id);
+}
+
+size_t P2CEwmaLoadBalancer::AddServersInBatch(
+ const std::vector<ServerId>& servers) {
+ const size_t n = _db_servers.ModifyWithForeground(BatchAdd, servers);
+ LOG_IF(ERROR, n != servers.size())
+ << "Fail to AddServersInBatch, expected " << servers.size()
+ << " actually " << n;
+ return n;
+}
+
+size_t P2CEwmaLoadBalancer::RemoveServersInBatch(
+ const std::vector<ServerId>& servers) {
+ return _db_servers.Modify(BatchRemove, servers);
+}
+
+double P2CEwmaLoadBalancer::Score(
+ const ServerInfo& info, int64_t now_us) const {
+ const int64_t ewma_us =
+ info.stat->ewma_us.load(butil::memory_order_relaxed);
+ const int32_t inflight =
+ info.stat->inflight.load(butil::memory_order_relaxed);
+ double latency_term = MIN_LATENCY_TERM_US;
+ if (ewma_us > 0) {
+ // Decay the (possibly stale) EWMA at read time so that a server
+ // penalized long ago regains traffic and gets re-observed.
+ const int64_t stamp_us =
+ info.stat->stamp_us.load(butil::memory_order_relaxed);
+ const int64_t elapsed_us = now_us - stamp_us;
+ double decayed = (double)ewma_us;
+ if (elapsed_us > 0) {
+ decayed *= std::exp(-(double)elapsed_us / (double)_tau_us);
+ }
+ latency_term = std::max(decayed, MIN_LATENCY_TERM_US);
+ }
+ // Clamp so that a transiently negative counter can not invert routing.
+ const int32_t load = std::max(inflight + 1, 1);
+ return latency_term * (double)load / (double)info.weight;
+}
+
+int P2CEwmaLoadBalancer::SelectServer(const SelectIn& in, SelectOut* out) {
+ butil::DoublyBufferedData<Servers>::ScopedPtr s;
+ if (_db_servers.Read(&s) != 0) {
+ return ENOMEM;
+ }
+ const size_t n = s->server_list.size();
+ if (n == 0) {
+ return ENODATA;
+ }
+ // Reuse the caller-provided timestamp to avoid an extra clock read per
+ // selection; only Channel::CheckHealth passes 0.
+ const int64_t now_us = in.begin_time_us > 0
+ ? in.begin_time_us : butil::gettimeofday_us();
+
+ const ServerInfo* best = NULL;
+ double best_score = 0;
+ SocketUniquePtr best_ptr;
+ // Score the server at `index' and keep it if it beats the current best.
+ // Excluded and unavailable servers are skipped.
+ auto consider = [&](size_t index) {
+ const ServerInfo& info = s->server_list[index];
+ if (ExcludedServers::IsExcluded(in.excluded, info.id)) {
+ return;
+ }
+ SocketUniquePtr ptr;
+ if (!IsServerAvailable(info.id, &ptr)) {
+ return;
+ }
+ const double score = Score(info, now_us);
+ if (best == NULL || score < best_score) {
+ best = &info;
+ best_score = score;
+ best_ptr.swap(ptr);
+ }
+ };
+
+ const size_t choices = std::min((size_t)_choices, n);
+ if (choices >= n) {
+ // Scan from a random offset so that equal scores do not herd all
+ // clients onto the lowest-indexed server.
+ const size_t start = butil::fast_rand_less_than(n);
+ for (size_t i = 0; i < n; ++i) {
+ consider((start + i) % n);
+ }
+ } else {
+ // Sample `choices' distinct random servers. Attempts are bounded so
+ // that duplicated draws never loop for long.
+ size_t chosen[MAX_CHOICES];
+ size_t nchosen = 0;
+ const size_t max_attempts = 4 * choices + 8;
+ for (size_t attempt = 0;
+ attempt < max_attempts && nchosen < choices; ++attempt) {
+ const size_t index = butil::fast_rand_less_than(n);
+ bool duplicated = false;
+ for (size_t i = 0; i < nchosen; ++i) {
+ if (chosen[i] == index) {
+ duplicated = true;
+ break;
+ }
+ }
+ if (duplicated) {
+ continue;
+ }
+ chosen[nchosen++] = index;
+ consider(index);
+ }
+ if (best == NULL) {
+ // All sampled servers were excluded or unavailable, fall back
+ // to scoring the whole list before violating exclusion below.
+ for (size_t i = 0; i < n; ++i) {
+ consider(i);
+ }
+ }
+ }
+
+ if (best == NULL) {
+ // Always take last chance: all servers are excluded, send to any
+ // available one as rr/random do.
+ for (size_t i = 0; i < n; ++i) {
+ if (IsServerAvailable(s->server_list[i].id, &best_ptr)) {
+ best = &s->server_list[i];
+ break;
+ }
+ }
+ if (best == NULL) {
+ return EHOSTDOWN;
+ }
+ }
+ if (in.changable_weights) {
+ best->stat->inflight.fetch_add(1, butil::memory_order_relaxed);
+ out->need_feedback = true;
+ }
+ out->ptr->swap(best_ptr);
+ return 0;
+}
+
+void P2CEwmaLoadBalancer::Feedback(const CallInfo& info) {
+ butil::DoublyBufferedData<Servers>::ScopedPtr s;
+ if (_db_servers.Read(&s) != 0) {
+ return;
+ }
+ const size_t* pindex = s->server_map.seek(info.server_id);
+ if (pindex == NULL) {
+ // The server was removed after selection, its stat is gone with it.
+ return;
+ }
+ NodeStat* stat = s->server_list[*pindex].stat.get();
+ stat->inflight.fetch_sub(1, butil::memory_order_relaxed);
+
+ const int64_t now_us = butil::gettimeofday_us();
+ int64_t latency_us = now_us - info.begin_time_us;
+ if (latency_us <= 0) {
+ // time skews, ignore the sample.
+ return;
+ }
+ if (info.error_code != 0) {
+ // Punish failures with at least the timeout(if any) and twice the
+ // current average, so that a failing server keeps losing comparisons
+ // even when it fails fast.
+ const int64_t timeout_us = info.controller->timeout_ms() * 1000L;
+ latency_us = std::max(latency_us, timeout_us);
+ latency_us = std::max(
+ latency_us, 2 * stat->ewma_us.load(butil::memory_order_relaxed));
+ // Cap the punishment: consecutive failures double the EWMA each
+ // time, which otherwise grows without bound and delays recovery
+ // after the server turns healthy.
+ const int64_t max_punish_us = FLAGS_p2c_max_punish_ms * 1000L;
+ if (max_punish_us > 0 && latency_us > max_punish_us) {
+ latency_us = max_punish_us;
+ }
+ }
+
+ BAIDU_SCOPED_LOCK(stat->update_mutex);
+ const int64_t ewma_us = stat->ewma_us.load(butil::memory_order_relaxed);
+ int64_t new_ewma_us = latency_us;
+ if (ewma_us > 0 && latency_us < ewma_us) {
+ // Downward samples decay the average while upward spikes(handled by
+ // the branch above) replace it immediately.
+ const int64_t elapsed_us =
+ now_us - stat->stamp_us.load(butil::memory_order_relaxed);
+ const double w = elapsed_us > 0
+ ? std::exp(-(double)elapsed_us / (double)_tau_us) : 1.0;
+ new_ewma_us = (int64_t)(ewma_us * w + latency_us * (1.0 - w));
+ }
+ stat->ewma_us.store(new_ewma_us, butil::memory_order_relaxed);
+ stat->stamp_us.store(now_us, butil::memory_order_relaxed);
+}
+
+P2CEwmaLoadBalancer* P2CEwmaLoadBalancer::New(
+ const butil::StringPiece& params) const {
+ P2CEwmaLoadBalancer* lb = new (std::nothrow) P2CEwmaLoadBalancer;
+ if (lb != NULL && !lb->SetParameters(params)) {
+ delete lb;
+ lb = NULL;
+ }
+ return lb;
+}
+
+bool P2CEwmaLoadBalancer::SetParameters(const butil::StringPiece& params) {
+ for (butil::KeyValuePairsSplitter sp(params.begin(), params.end(), ' ',
'=');
+ sp; ++sp) {
+ if (sp.value().empty()) {
+ LOG(ERROR) << "Empty value for " << sp.key() << " in lb parameter";
+ return false;
+ }
+ if (sp.key() == "choices") {
+ unsigned choices = 0;
+ if (!butil::StringToUint(sp.value().as_string(), &choices) ||
+ choices < 2 || choices > MAX_CHOICES) {
+ LOG(ERROR) << "Invalid choices=`" << sp.value() << "'";
+ return false;
+ }
+ _choices = choices;
+ } else if (sp.key() == "tau_ms") {
+ int64_t tau_ms = 0;
+ if (!butil::StringToInt64(sp.value(), &tau_ms) || tau_ms <= 0) {
+ LOG(ERROR) << "Invalid tau_ms=`" << sp.value() << "'";
+ return false;
+ }
+ _tau_us = tau_ms * 1000L;
+ } else {
+ LOG(ERROR) << "Unknown parameter " << sp.key_and_value();
+ return false;
+ }
+ }
+ return true;
+}
+
+void P2CEwmaLoadBalancer::Destroy() {
+ delete this;
+}
+
+void P2CEwmaLoadBalancer::Describe(
+ std::ostream& os, const DescribeOptions& options) {
+ if (!options.verbose) {
+ os << "p2c";
+ return;
+ }
+ os << "P2CEwma{choices=" << _choices << " tau_ms=" << _tau_us / 1000;
+ butil::DoublyBufferedData<Servers>::ScopedPtr s;
+ if (_db_servers.Read(&s) != 0) {
+ os << " fail to read _db_servers";
+ } else {
+ const int64_t now_us = butil::gettimeofday_us();
+ os << " n=" << s->server_list.size() << ':';
+ for (size_t i = 0; i < s->server_list.size(); ++i) {
+ const ServerInfo& info = s->server_list[i];
+ os << ' ' << info.id << '(' << "w=" << info.weight
+ << " ewma_us="
+ << info.stat->ewma_us.load(butil::memory_order_relaxed)
+ << " inflight="
+ << info.stat->inflight.load(butil::memory_order_relaxed)
+ << " score=" << Score(info, now_us) << ')';
+ }
+ }
+ os << '}';
+}
+
+} // namespace policy
+} // namespace brpc
diff --git a/src/brpc/policy/p2c_ewma_load_balancer.h
b/src/brpc/policy/p2c_ewma_load_balancer.h
new file mode 100644
index 00000000..1594f159
--- /dev/null
+++ b/src/brpc/policy/p2c_ewma_load_balancer.h
@@ -0,0 +1,94 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+
+#ifndef BRPC_POLICY_P2C_EWMA_LOAD_BALANCER_H
+#define BRPC_POLICY_P2C_EWMA_LOAD_BALANCER_H
+
+#include <memory> // std::shared_ptr
+#include <vector> // std::vector
+#include "butil/containers/flat_map.h" // FlatMap
+#include "butil/containers/doubly_buffered_data.h" // DoublyBufferedData
+#include "brpc/load_balancer.h"
+
+namespace brpc {
+namespace policy {
+
+// Power-of-Two-Choices with Peak-EWMA latency scoring ("p2c").
+// Each selection samples `choices'(default 2) distinct servers and routes
+// to the one with the lower load score:
+// score = peak_ewma_latency_us * (inflight + 1) / weight
+// The latency EWMA is peak-sensitive: an upward spike replaces the average
+// immediately while recovery decays with time constant `tau_ms'(default 10s),
+// so a degraded server is shed within one observation. Selection is O(1)
+// regardless of fleet size. Weight is got from tag of ServerId(default 1).
+class P2CEwmaLoadBalancer : public LoadBalancer {
+public:
+ P2CEwmaLoadBalancer();
+ bool AddServer(const ServerId& id) override;
+ bool RemoveServer(const ServerId& id) override;
+ size_t AddServersInBatch(const std::vector<ServerId>& servers) override;
+ size_t RemoveServersInBatch(const std::vector<ServerId>& servers) override;
+ int SelectServer(const SelectIn& in, SelectOut* out) override;
+ void Feedback(const CallInfo& info) override;
+ P2CEwmaLoadBalancer* New(const butil::StringPiece& params) const override;
+ void Destroy() override;
+ void Describe(std::ostream& os, const DescribeOptions&) override;
+
+private:
+ // Mutable per-server load state. Allocated once when the server is first
+ // added and shared by both buffers of _db_servers, so a stable pointer
+ // can be used from SelectServer()/Feedback() without copying.
+ struct NodeStat {
+ NodeStat() : inflight(0), ewma_us(0), stamp_us(0) {}
+ butil::atomic<int32_t> inflight;
+ // Peak-sensitive EWMA of latency in us. 0 means no observation yet.
+ butil::atomic<int64_t> ewma_us;
+ // Time of the last EWMA update.
+ butil::atomic<int64_t> stamp_us;
+ butil::Mutex update_mutex;
+ };
+ struct ServerInfo {
+ SocketId id;
+ uint32_t weight;
+ std::shared_ptr<NodeStat> stat;
+ };
+ struct Servers {
+ std::vector<ServerInfo> server_list;
+ // Maps SocketId to index in server_list.
+ butil::FlatMap<SocketId, size_t> server_map;
+ };
+
+ static bool Add(Servers& bg, const Servers& fg, const ServerId& id);
+ static bool Remove(Servers& bg, const ServerId& id);
+ static size_t BatchAdd(Servers& bg, const Servers& fg,
+ const std::vector<ServerId>& servers);
+ static size_t BatchRemove(Servers& bg, const std::vector<ServerId>&
servers);
+
+ bool SetParameters(const butil::StringPiece& params);
+ // Load score of the server at `now_us'. Smaller is better.
+ double Score(const ServerInfo& info, int64_t now_us) const;
+
+ butil::DoublyBufferedData<Servers> _db_servers;
+ uint32_t _choices;
+ int64_t _tau_us;
+};
+
+} // namespace policy
+} // namespace brpc
+
+#endif // BRPC_POLICY_P2C_EWMA_LOAD_BALANCER_H
diff --git a/test/brpc_p2c_ewma_load_balancer_unittest.cpp
b/test/brpc_p2c_ewma_load_balancer_unittest.cpp
new file mode 100644
index 00000000..922bfc39
--- /dev/null
+++ b/test/brpc_p2c_ewma_load_balancer_unittest.cpp
@@ -0,0 +1,428 @@
+// 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 <cstdlib>
+#include <map>
+#include <sstream>
+#include <vector>
+#include <gflags/gflags.h>
+#include <gtest/gtest.h>
+#include <pthread.h>
+#include <unistd.h>
+#include "butil/macros.h"
+#include "butil/time.h"
+#include "brpc/socket.h"
+#include "brpc/controller.h"
+#include "brpc/excluded_servers.h"
+#include "brpc/policy/p2c_ewma_load_balancer.h"
+
+namespace brpc {
+namespace policy {
+DECLARE_int64(p2c_max_punish_ms);
+}
+}
+
+namespace {
+
+butil::atomic<size_t> nrecycle(0);
+class SaveRecycle : public brpc::SocketUser {
+ void BeforeRecycle(brpc::Socket* s) {
+ nrecycle.fetch_add(1, butil::memory_order_relaxed);
+ delete this;
+ }
+};
+
+brpc::ServerId CreateServer(const char* addr, const char* tag = "") {
+ butil::EndPoint point;
+ EXPECT_EQ(0, str2endpoint(addr, &point));
+ brpc::ServerId id(8888);
+ brpc::SocketOptions options;
+ options.remote_side = point;
+ options.user = new SaveRecycle;
+ EXPECT_EQ(0, brpc::Socket::Create(options, &id.id));
+ id.tag = tag;
+ return id;
+}
+
+// Report a call that took `latency_us' back to the load balancer.
+void FeedbackLatency(brpc::LoadBalancer* lb, brpc::SocketId server_id,
+ int64_t latency_us, int error_code = 0,
+ const brpc::Controller* cntl = NULL) {
+ brpc::LoadBalancer::CallInfo info;
+ info.begin_time_us = butil::gettimeofday_us() - latency_us;
+ info.server_id = server_id;
+ info.error_code = error_code;
+ info.controller = cntl;
+ lb->Feedback(info);
+}
+
+class P2CEwmaLoadBalancerTest : public ::testing::Test {
+protected:
+ void SetUp() override {
+ _lb = new brpc::policy::P2CEwmaLoadBalancer;
+ }
+ void TearDown() override {
+ _lb->Destroy();
+ }
+
+ int Select(brpc::SocketUniquePtr* ptr, bool* need_feedback = NULL,
+ const brpc::ExcludedServers* excluded = NULL) {
+ brpc::LoadBalancer::SelectIn in = {
+ butil::gettimeofday_us(), true, false, 0u, excluded };
+ brpc::LoadBalancer::SelectOut out(ptr);
+ const int rc = _lb->SelectServer(in, &out);
+ if (need_feedback != NULL) {
+ *need_feedback = out.need_feedback;
+ }
+ return rc;
+ }
+
+ brpc::policy::P2CEwmaLoadBalancer* _lb;
+};
+
+TEST_F(P2CEwmaLoadBalancerTest, add_remove_servers) {
+ brpc::SocketUniquePtr ptr;
+ ASSERT_EQ(ENODATA, Select(&ptr));
+
+ std::vector<brpc::ServerId> ids;
+ ids.push_back(CreateServer("127.0.0.1:7777"));
+ ids.push_back(CreateServer("127.0.0.1:7778"));
+ ids.push_back(CreateServer("127.0.0.1:7779"));
+
+ ASSERT_TRUE(_lb->AddServer(ids[0]));
+ // Duplicated server is rejected.
+ ASSERT_FALSE(_lb->AddServer(ids[0]));
+ ASSERT_EQ(2u, _lb->AddServersInBatch(
+ std::vector<brpc::ServerId>(ids.begin() + 1, ids.end())));
+
+ ASSERT_EQ(0, Select(&ptr));
+
+ ASSERT_TRUE(_lb->RemoveServer(ids[0]));
+ ASSERT_FALSE(_lb->RemoveServer(ids[0]));
+ ASSERT_EQ(2u, _lb->RemoveServersInBatch(
+ std::vector<brpc::ServerId>(ids.begin() + 1, ids.end())));
+ ASSERT_EQ(ENODATA, Select(&ptr));
+}
+
+TEST_F(P2CEwmaLoadBalancerTest, single_server) {
+ const brpc::ServerId id = CreateServer("127.0.0.1:7780");
+ ASSERT_TRUE(_lb->AddServer(id));
+ for (int i = 0; i < 10; ++i) {
+ brpc::SocketUniquePtr ptr;
+ bool need_feedback = false;
+ ASSERT_EQ(0, Select(&ptr, &need_feedback));
+ ASSERT_EQ(id.id, ptr->id());
+ ASSERT_TRUE(need_feedback);
+ FeedbackLatency(_lb, id.id, 1000);
+ }
+}
+
+TEST_F(P2CEwmaLoadBalancerTest, prefers_lower_latency) {
+ const brpc::ServerId fast = CreateServer("127.0.0.1:7781");
+ const brpc::ServerId slow = CreateServer("127.0.0.1:7782");
+ ASSERT_TRUE(_lb->AddServer(fast));
+ ASSERT_TRUE(_lb->AddServer(slow));
+
+ // Servers respond with their characteristic latency; both get observed
+ // within the first rounds because unobserved servers score best.
+ std::map<brpc::SocketId, size_t> counts;
+ const int kRounds = 1000;
+ for (int i = 0; i < kRounds; ++i) {
+ brpc::SocketUniquePtr ptr;
+ ASSERT_EQ(0, Select(&ptr));
+ ++counts[ptr->id()];
+ // Selected server responds with its own characteristic latency.
+ FeedbackLatency(_lb, ptr->id(), ptr->id() == fast.id ? 1000 : 50000);
+ }
+ LOG(INFO) << "fast=" << counts[fast.id] << " slow=" << counts[slow.id];
+ // The fast server should receive almost all traffic.
+ ASSERT_GE(counts[fast.id], (size_t)(0.9 * kRounds));
+}
+
+TEST_F(P2CEwmaLoadBalancerTest, sheds_degraded_server) {
+ const brpc::ServerId a = CreateServer("127.0.0.1:7783");
+ const brpc::ServerId b = CreateServer("127.0.0.1:7784");
+ ASSERT_TRUE(_lb->AddServer(a));
+ ASSERT_TRUE(_lb->AddServer(b));
+ // Warm up with `b' slightly faster so it is the preferred server.
+ for (int i = 0; i < 20; ++i) {
+ brpc::SocketUniquePtr ptr;
+ ASSERT_EQ(0, Select(&ptr));
+ FeedbackLatency(_lb, ptr->id(), ptr->id() == a.id ? 1000 : 500);
+ }
+
+ // `b' degrades: its first slow response must shift traffic to `a'
+ // immediately(peak-sensitivity), long before an averaging window would.
+ std::map<brpc::SocketId, size_t> counts;
+ for (int i = 0; i < 100; ++i) {
+ brpc::SocketUniquePtr ptr;
+ ASSERT_EQ(0, Select(&ptr));
+ ++counts[ptr->id()];
+ FeedbackLatency(_lb, ptr->id(), ptr->id() == a.id ? 1000 : 100000);
+ }
+ ASSERT_GE(counts[a.id], 90u);
+}
+
+TEST_F(P2CEwmaLoadBalancerTest, inflight_breaks_ties) {
+ const brpc::ServerId a = CreateServer("127.0.0.1:7785");
+ const brpc::ServerId b = CreateServer("127.0.0.1:7786");
+ ASSERT_TRUE(_lb->AddServer(a));
+ ASSERT_TRUE(_lb->AddServer(b));
+
+ // With no latency observations, scores only differ by in-flight count,
+ // so two selections without feedback must go to different servers.
+ brpc::SocketUniquePtr ptr1;
+ brpc::SocketUniquePtr ptr2;
+ ASSERT_EQ(0, Select(&ptr1));
+ ASSERT_EQ(0, Select(&ptr2));
+ ASSERT_NE(ptr1->id(), ptr2->id());
+
+ // After feedback(decrementing in-flight), both are tied again and the
+ // third selection must not crash or fail.
+ FeedbackLatency(_lb, ptr1->id(), 1000);
+ FeedbackLatency(_lb, ptr2->id(), 1000);
+ brpc::SocketUniquePtr ptr3;
+ ASSERT_EQ(0, Select(&ptr3));
+}
+
+TEST_F(P2CEwmaLoadBalancerTest, weighted_split_by_inflight) {
+ // With equal latency, steady-state in-flight counts converge to the
+ // 1:2:4 weight ratio because the score is (inflight+1)/weight.
+ const brpc::ServerId w1 = CreateServer("127.0.0.1:7787", "1");
+ const brpc::ServerId w2 = CreateServer("127.0.0.1:7788", "2");
+ const brpc::ServerId w4 = CreateServer("127.0.0.1:7789", "4");
+ brpc::policy::P2CEwmaLoadBalancer* lb =
+ _lb->New(butil::StringPiece("choices=3"));
+ ASSERT_TRUE(lb != NULL);
+ ASSERT_TRUE(lb->AddServer(w1));
+ ASSERT_TRUE(lb->AddServer(w2));
+ ASSERT_TRUE(lb->AddServer(w4));
+
+ std::map<brpc::SocketId, size_t> counts;
+ const int kRounds = 700;
+ for (int i = 0; i < kRounds; ++i) {
+ brpc::SocketUniquePtr ptr;
+ brpc::LoadBalancer::SelectIn in = {
+ butil::gettimeofday_us(), true, false, 0u, NULL };
+ brpc::LoadBalancer::SelectOut out(&ptr);
+ ASSERT_EQ(0, lb->SelectServer(in, &out));
+ ++counts[ptr->id()];
+ // No feedback: requests stay in flight.
+ }
+ LOG(INFO) << "w1=" << counts[w1.id] << " w2=" << counts[w2.id]
+ << " w4=" << counts[w4.id];
+ const double share1 = counts[w1.id] / (double)kRounds;
+ const double share2 = counts[w2.id] / (double)kRounds;
+ const double share4 = counts[w4.id] / (double)kRounds;
+ ASSERT_NEAR(share1, 1.0 / 7, 0.05);
+ ASSERT_NEAR(share2, 2.0 / 7, 0.05);
+ ASSERT_NEAR(share4, 4.0 / 7, 0.05);
+ lb->Destroy();
+}
+
+TEST_F(P2CEwmaLoadBalancerTest, error_feedback_punishes_server) {
+ const brpc::ServerId good = CreateServer("127.0.0.1:7790");
+ const brpc::ServerId bad = CreateServer("127.0.0.1:7791");
+ ASSERT_TRUE(_lb->AddServer(good));
+ ASSERT_TRUE(_lb->AddServer(bad));
+ // Warm up with `bad' slightly faster so it is the preferred server.
+ for (int i = 0; i < 20; ++i) {
+ brpc::SocketUniquePtr ptr;
+ ASSERT_EQ(0, Select(&ptr));
+ FeedbackLatency(_lb, ptr->id(), ptr->id() == good.id ? 1000 : 500);
+ }
+
+ // `bad' starts failing fast(1ms), but failures are punished with at
+ // least the RPC timeout so it keeps losing selections.
+ brpc::Controller cntl;
+ cntl.set_timeout_ms(100);
+ std::map<brpc::SocketId, size_t> counts;
+ for (int i = 0; i < 100; ++i) {
+ brpc::SocketUniquePtr ptr;
+ ASSERT_EQ(0, Select(&ptr));
+ ++counts[ptr->id()];
+ if (ptr->id() == good.id) {
+ FeedbackLatency(_lb, good.id, 1000);
+ } else {
+ FeedbackLatency(_lb, bad.id, 1000, ETIMEDOUT, &cntl);
+ }
+ }
+ ASSERT_GE(counts[good.id], 90u);
+}
+
+TEST_F(P2CEwmaLoadBalancerTest, excluded_servers) {
+ const brpc::ServerId a = CreateServer("127.0.0.1:7792");
+ const brpc::ServerId b = CreateServer("127.0.0.1:7793");
+ ASSERT_TRUE(_lb->AddServer(a));
+ ASSERT_TRUE(_lb->AddServer(b));
+
+ brpc::ExcludedServers* excluded = brpc::ExcludedServers::Create(2);
+ excluded->Add(a.id);
+ for (int i = 0; i < 20; ++i) {
+ brpc::SocketUniquePtr ptr;
+ ASSERT_EQ(0, Select(&ptr, NULL, excluded));
+ ASSERT_EQ(b.id, ptr->id());
+ FeedbackLatency(_lb, b.id, 1000);
+ }
+ // All servers excluded: still take the last chance instead of failing.
+ excluded->Add(b.id);
+ brpc::SocketUniquePtr ptr;
+ ASSERT_EQ(0, Select(&ptr, NULL, excluded));
+ brpc::ExcludedServers::Destroy(excluded);
+}
+
+TEST_F(P2CEwmaLoadBalancerTest, invalid_parameters) {
+ brpc::LoadBalancer* lb = _lb->New(butil::StringPiece(""));
+ ASSERT_TRUE(lb != NULL);
+ lb->Destroy();
+ lb = _lb->New(butil::StringPiece("choices=4 tau_ms=5000"));
+ ASSERT_TRUE(lb != NULL);
+ lb->Destroy();
+ ASSERT_TRUE(_lb->New(butil::StringPiece("choices=1")) == NULL);
+ ASSERT_TRUE(_lb->New(butil::StringPiece("choices=abc")) == NULL);
+ ASSERT_TRUE(_lb->New(butil::StringPiece("tau_ms=0")) == NULL);
+ ASSERT_TRUE(_lb->New(butil::StringPiece("unknown=1")) == NULL);
+}
+
+struct ChurnArg {
+ brpc::policy::P2CEwmaLoadBalancer* lb;
+ butil::atomic<bool> stop;
+ butil::atomic<size_t> nselected;
+};
+
+void* SelectAndFeedback(void* void_arg) {
+ ChurnArg* arg = static_cast<ChurnArg*>(void_arg);
+ while (!arg->stop.load(butil::memory_order_relaxed)) {
+ brpc::SocketUniquePtr ptr;
+ brpc::LoadBalancer::SelectIn in = {
+ butil::gettimeofday_us(), true, false, 0u, NULL };
+ brpc::LoadBalancer::SelectOut out(&ptr);
+ if (arg->lb->SelectServer(in, &out) == 0) {
+ arg->nselected.fetch_add(1, butil::memory_order_relaxed);
+ if (out.need_feedback) {
+ FeedbackLatency(arg->lb, ptr->id(), 1000);
+ }
+ }
+ }
+ return NULL;
+}
+
+TEST_F(P2CEwmaLoadBalancerTest, concurrent_select_with_churn) {
+ std::vector<brpc::ServerId> ids;
+ for (int i = 0; i < 8; ++i) {
+ char addr[32];
+ snprintf(addr, sizeof(addr), "127.0.0.1:%d", 7800 + i);
+ ids.push_back(CreateServer(addr));
+ ASSERT_TRUE(_lb->AddServer(ids.back()));
+ }
+
+ ChurnArg arg;
+ arg.lb = _lb;
+ arg.stop.store(false);
+ arg.nselected.store(0);
+ pthread_t threads[4];
+ for (size_t i = 0; i < ARRAY_SIZE(threads); ++i) {
+ ASSERT_EQ(0, pthread_create(
+ &threads[i], NULL, SelectAndFeedback, &arg));
+ }
+ // Churn membership while selections are running.
+ const int64_t stop_at_us = butil::gettimeofday_us() + 1000000L;
+ while (butil::gettimeofday_us() < stop_at_us) {
+ ASSERT_TRUE(_lb->RemoveServer(ids[0]));
+ ASSERT_TRUE(_lb->AddServer(ids[0]));
+ }
+ arg.stop.store(true);
+ for (size_t i = 0; i < ARRAY_SIZE(threads); ++i) {
+ ASSERT_EQ(0, pthread_join(threads[i], NULL));
+ }
+ LOG(INFO) << "selected " << arg.nselected.load() << " times";
+ ASSERT_GT(arg.nselected.load(), 0u);
+}
+
+TEST_F(P2CEwmaLoadBalancerTest, error_punish_is_capped) {
+ const brpc::ServerId id = CreateServer("127.0.0.1:7900");
+ ASSERT_TRUE(_lb->AddServer(id));
+ brpc::Controller cntl;
+ cntl.set_timeout_ms(100);
+ // Persistent failures double the punished EWMA per sample; without the
+ // cap 64 rounds would push it past 2^64.
+ for (int i = 0; i < 64; ++i) {
+ brpc::SocketUniquePtr ptr;
+ ASSERT_EQ(0, Select(&ptr));
+ FeedbackLatency(_lb, id.id, 1000, ETIMEDOUT, &cntl);
+ }
+ std::ostringstream os;
+ brpc::DescribeOptions options;
+ options.verbose = true;
+ _lb->Describe(os, options);
+ const std::string desc = os.str();
+ const size_t pos = desc.find("ewma_us=");
+ ASSERT_NE(std::string::npos, pos) << desc;
+ const int64_t ewma_us = strtoll(desc.c_str() + pos + 8, NULL, 10);
+ ASSERT_LE(ewma_us, brpc::policy::FLAGS_p2c_max_punish_ms * 1000L) << desc;
+}
+
+struct FeedbackBenchArg {
+ brpc::policy::P2CEwmaLoadBalancer* lb;
+ brpc::SocketId server_id;
+ butil::atomic<bool> stop;
+ butil::atomic<size_t> nfeedback;
+};
+
+void* FeedbackHammer(void* void_arg) {
+ FeedbackBenchArg* arg = static_cast<FeedbackBenchArg*>(void_arg);
+ size_t n = 0;
+ while (!arg->stop.load(butil::memory_order_relaxed)) {
+ FeedbackLatency(arg->lb, arg->server_id, 1000);
+ ++n;
+ }
+ arg->nfeedback.fetch_add(n, butil::memory_order_relaxed);
+ return NULL;
+}
+
+TEST_F(P2CEwmaLoadBalancerTest, feedback_lock_overhead) {
+ // All threads feed back to a single server so update_mutex sees
+ // worst-case contention.
+ const brpc::ServerId id = CreateServer("127.0.0.1:7901");
+ ASSERT_TRUE(_lb->AddServer(id));
+ for (size_t nthread = 1; nthread <= 4; nthread *= 2) {
+ FeedbackBenchArg arg;
+ arg.lb = _lb;
+ arg.server_id = id.id;
+ arg.stop.store(false);
+ arg.nfeedback.store(0);
+ pthread_t threads[4];
+ const int64_t begin_us = butil::gettimeofday_us();
+ for (size_t i = 0; i < nthread; ++i) {
+ ASSERT_EQ(0, pthread_create(
+ &threads[i], NULL, FeedbackHammer, &arg));
+ }
+ usleep(500 * 1000);
+ arg.stop.store(true);
+ for (size_t i = 0; i < nthread; ++i) {
+ ASSERT_EQ(0, pthread_join(threads[i], NULL));
+ }
+ const int64_t elapsed_us = butil::gettimeofday_us() - begin_us;
+ const size_t n = arg.nfeedback.load();
+ ASSERT_GT(n, 0u);
+ LOG(INFO) << "Feedback on 1 shared server, " << nthread
+ << " thread(s): " << (elapsed_us * 1000L * nthread) / n
+ << "ns/op (" << n << " ops in " << elapsed_us / 1000 <<
"ms)";
+ }
+}
+
+} // namespace
diff --git a/tools/rpc_press/rpc_press.cpp b/tools/rpc_press/rpc_press.cpp
index cb69ccfc..a9a98d60 100644
--- a/tools/rpc_press/rpc_press.cpp
+++ b/tools/rpc_press/rpc_press.cpp
@@ -31,7 +31,7 @@ DEFINE_string(method, "example.EchoService.Echo", "The full
method name");
DEFINE_string(server, "0.0.0.0:8002", "ip:port of the server when
-load_balancer is empty, the naming service otherwise");
DEFINE_string(input, "", "The file containing requests in json format");
DEFINE_string(output, "", "The file containing responses in json format");
-DEFINE_string(lb_policy, "", "The load balancer algorithm: rr, random, la,
c_murmurhash, c_md5");
+DEFINE_string(lb_policy, "", "The load balancer algorithm: rr, random, la,
p2c, c_murmurhash, c_md5");
DEFINE_int32(thread_num, 0, "Number of threads to send requests. 0:
automatically chosen according to -qps");
DEFINE_string(protocol, "baidu_std", "baidu_std hulu_pbrpc sofa_pbrpc http
public_pbrpc nova_pbrpc ubrpc_compack...");
DEFINE_string(connection_type, "", "Type of connections: single, pooled,
short");
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
