chenBright commented on code in PR #3367: URL: https://github.com/apache/brpc/pull/3367#discussion_r3534534840
########## src/brpc/policy/p2c_ewma_load_balancer.cpp: ########## @@ -0,0 +1,369 @@ +// 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 "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 { + +namespace { +const uint32_t DEFAULT_CHOICES = 2; +const uint32_t MAX_CHOICES = 64; +// Finagle's PeakEwma default decay time. +const int64_t DEFAULT_TAU_US = 10 * 1000000L; +// 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; Review Comment: These variables should be made configurable using gflag. ########## src/brpc/policy/p2c_ewma_load_balancer.cpp: ########## @@ -0,0 +1,369 @@ +// 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 "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 { + +namespace { +const uint32_t DEFAULT_CHOICES = 2; +const uint32_t MAX_CHOICES = 64; +// Finagle's PeakEwma default decay time. +const int64_t DEFAULT_TAU_US = 10 * 1000000L; +// 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() + : _choices(DEFAULT_CHOICES) + , _tau_us(DEFAULT_TAU_US) {} + +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( Review Comment: It seems that persistent RPC errors will cause ewma_us to keep growing. Could this lead to slow recovery? ########## src/brpc/policy/p2c_ewma_load_balancer.h: ########## @@ -0,0 +1,100 @@ +// 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; + + Servers() { + if (server_map.init(1024, 70) != 0) { + LOG(WARNING) << "Fail to init server_map"; + } + } Review Comment: No need to initialize the map. ########## src/brpc/policy/p2c_ewma_load_balancer.cpp: ########## @@ -0,0 +1,369 @@ +// 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 "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 { + +namespace { +const uint32_t DEFAULT_CHOICES = 2; +const uint32_t MAX_CHOICES = 64; +// Finagle's PeakEwma default decay time. +const int64_t DEFAULT_TAU_US = 10 * 1000000L; +// 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() + : _choices(DEFAULT_CHOICES) + , _tau_us(DEFAULT_TAU_US) {} + +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)); + } + + BAIDU_SCOPED_LOCK(stat->update_mutex); Review Comment: Please test the overhead of the lock. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
