cdjingit commented on a change in pull request #592: lb configurable &&
consistency hash lb refactor
URL: https://github.com/apache/incubator-brpc/pull/592#discussion_r273807198
##########
File path: src/brpc/policy/consistent_hashing_load_balancer.cpp
##########
@@ -29,16 +31,119 @@ 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) {
+// Defined in hasher.cpp.
+const char* GetHashName(HashFunc hasher);
+
+class ReplicaPolicy {
+public:
+ virtual ~ReplicaPolicy() = default;
+
+ virtual bool Build(ServerId server,
+ size_t num_replicas,
+ std::vector<ConsistentHashingLoadBalancer::Node>*
replicas) const = 0;
+ virtual const char* name() const = 0;
+};
+
+class DefaultReplicaPolicy : public ReplicaPolicy {
+public:
+ DefaultReplicaPolicy(HashFunc hash) : _hash_func(hash) {}
+
+ virtual bool Build(ServerId server,
+ size_t num_replicas,
+ std::vector<ConsistentHashingLoadBalancer::Node>*
replicas) const;
+
+ virtual const char* name() const { return GetHashName(_hash_func); }
+
+private:
+ HashFunc _hash_func;
+};
+
+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;
+
+ virtual const char* name() const { return "ketama"; }
+};
+
+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;
}
+namespace {
+
+pthread_once_t s_replica_policy_once = PTHREAD_ONCE_INIT;
+const std::array<const ReplicaPolicy*, CONS_HASH_LB_LAST>* g_replica_policy =
nullptr;
+
+void init_replica_policy() {
Review comment:
这个包裹在namespace {}中了
----------------------------------------------------------------
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
