Copilot commented on code in PR #3367: URL: https://github.com/apache/brpc/pull/3367#discussion_r3525112707
########## src/brpc/policy/p2c_ewma_load_balancer.cpp: ########## @@ -0,0 +1,366 @@ +// 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; + } + const int64_t now_us = butil::gettimeofday_us(); Review Comment: SelectServer recomputes `now_us` via `butil::gettimeofday_us()` even though the caller already provides a timestamp in `SelectIn.begin_time_us` (set to `start_realtime_us` in `Controller::IssueRPC`). This adds an extra syscall/clock read on every selection, which can be noticeable at high QPS. Consider reusing `in.begin_time_us` when it is non-zero and only falling back to `gettimeofday_us()` for call sites (like `Channel::CheckHealth`) that pass 0. -- 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]
