jamesge commented on a change in pull request #592: lb configurable &&
consistency hash lb refactor
URL: https://github.com/apache/incubator-brpc/pull/592#discussion_r272427635
##########
File path: src/brpc/policy/consistent_hashing_load_balancer.cpp
##########
@@ -29,16 +31,111 @@ namespace policy {
DEFINE_int32(chash_num_replicas, 100,
"default number of replicas per server in chash");
-ConsistentHashingLoadBalancer::ConsistentHashingLoadBalancer(HashFunc hash)
- : _hash(hash)
- , _num_replicas(FLAGS_chash_num_replicas) {
+class ReplicaPolicy {
+public:
+ ReplicaPolicy() : _hash_func(nullptr) {}
+ ReplicaPolicy(HashFunc hash) : _hash_func(hash) {}
+ virtual ~ReplicaPolicy() = default;
+
+ virtual bool Build(ServerId server,
+ size_t num_replicas,
+ std::vector<ConsistentHashingLoadBalancer::Node>*
replicas) const = 0;
+
+protected:
+ HashFunc _hash_func;
+};
+
+class DefaultReplicaPolicy : public ReplicaPolicy {
+public:
+ DefaultReplicaPolicy(HashFunc hash) : ReplicaPolicy(hash) {}
+
+ virtual bool Build(ServerId server,
+ size_t num_replicas,
+ std::vector<ConsistentHashingLoadBalancer::Node>*
replicas) const;
+};
+
+bool DefaultReplicaPolicy::Build(ServerId server,
+ size_t num_replicas,
+
std::vector<ConsistentHashingLoadBalancer::Node>* replicas) const {
+ SocketUniquePtr ptr;
+ if (Socket::AddressFailedAsWell(server.id, &ptr) == -1) {
+ return false;
+ }
+ replicas->clear();
+ for (size_t i = 0; i < num_replicas; ++i) {
+ char host[32];
+ int len = snprintf(host, sizeof(host), "%s-%lu",
+ endpoint2str(ptr->remote_side()).c_str(), i);
+ ConsistentHashingLoadBalancer::Node node;
+ node.hash = _hash_func(host, len);
+ node.server_sock = server;
+ node.server_addr = ptr->remote_side();
+ replicas->push_back(node);
+ }
+ return true;
+}
+
+class KetamaReplicaPolicy : public ReplicaPolicy {
+public:
+ virtual bool Build(ServerId server,
+ size_t num_replicas,
+ std::vector<ConsistentHashingLoadBalancer::Node>*
replicas) const;
+};
+
+bool KetamaReplicaPolicy::Build(ServerId server,
+ size_t num_replicas,
+
std::vector<ConsistentHashingLoadBalancer::Node>* replicas) const {
+ SocketUniquePtr ptr;
+ if (Socket::AddressFailedAsWell(server.id, &ptr) == -1) {
+ return false;
+ }
+ replicas->clear();
+ const size_t points_per_hash = 4;
+ CHECK(num_replicas % points_per_hash == 0)
+ << "Ketam hash replicas number(" << num_replicas << ") should be n*4";
+ for (size_t i = 0; i < num_replicas / points_per_hash; ++i) {
+ char host[32];
+ int len = snprintf(host, sizeof(host), "%s-%lu",
+ endpoint2str(ptr->remote_side()).c_str(), i);
+ unsigned char digest[16];
+ MD5HashSignature(host, len, digest);
+ for (size_t j = 0; j < points_per_hash; ++j) {
+ ConsistentHashingLoadBalancer::Node node;
+ node.server_sock = server;
+ node.server_addr = ptr->remote_side();
+ 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)
+ | (digest[0 + j * 4] & 0xFF);
+ replicas->push_back(node);
+ }
+ }
+ return true;
}
-ConsistentHashingLoadBalancer::ConsistentHashingLoadBalancer(
- HashFunc hash,
- size_t num_replicas)
- : _hash(hash)
- , _num_replicas(num_replicas) {
+namespace {
+
+const std::map<std::string, const ReplicaPolicy*> g_replica_policy_map = {
Review comment:
如果有一个线程在main函数之后还发了个rpc（不少见），这里可能就挂了把。建议搞成once初始化+数组(而不是map)的形式
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscr...@brpc.apache.org
For additional commands, e-mail: dev-h...@brpc.apache.org
