brpc加redis cluster支持

CMakeLists.txt

@@ -383,6 +383,7 @@ set(BUTIL_SOURCES
     ${PROJECT_SOURCE_DIR}/src/butil/time.cpp
     ${PROJECT_SOURCE_DIR}/src/butil/zero_copy_stream_as_streambuf.cpp
     ${PROJECT_SOURCE_DIR}/src/butil/crc32c.cc
+    ${PROJECT_SOURCE_DIR}/src/butil/redis_cluster_slot.cc
     ${PROJECT_SOURCE_DIR}/src/butil/containers/case_ignored_flat_map.cpp
     ${PROJECT_SOURCE_DIR}/src/butil/iobuf.cpp
     ${PROJECT_SOURCE_DIR}/src/butil/binary_printer.cpp

example/http_c++/http_client.cpp

@@ -56,7 +56,7 @@ int main(int argc, char* argv[]) {
 
     // Initialize the channel, NULL means using default options. 
     // options, see `brpc/channel.h'.
-    if (channel.Init(url, FLAGS_load_balancer.c_str(), &options) != 0) {
+    if (channel.Init(url, "", &options) != 0) {
         LOG(ERROR) << "Fail to initialize channel";
         return -1;
     }

example/redis_c++/redis_cli.cpp

@@ -23,18 +23,38 @@
 #include <readline/history.h>
 #include <gflags/gflags.h>
 #include <butil/logging.h>
+#include <butil/redis_cluster_slot.h>
 #include <brpc/channel.h>
 #include <brpc/redis.h>
+#include <brpc/policy/redis_authenticator.h>
 
 DEFINE_string(connection_type, "", "Connection type. Available values: single, pooled, short");
 DEFINE_string(server, "127.0.0.1:6379", "IP Address of server");
 DEFINE_int32(timeout_ms, 1000, "RPC timeout in milliseconds");
 DEFINE_int32(max_retry, 3, "Max retries(not including the first RPC)"); 
+DEFINE_string(auth, "", "auth...");
+DEFINE_uint64(request_code, 0, "request code");
 
 namespace brpc {
 const char* logo();
 }
 
+static std::string extractSecondWord(const std::string &str) {
+    std::istringstream iss(str);
+    std::vector<std::string> words;
+    std::string word;
+
+    while (iss >> word) {
+        words.push_back(word);
+    }
+
+    if (words.size() >= 2) {
+        return words[1];
+    } else {
+        return "";
+    }
+}
+
 // Send `command' to redis-server via `channel'
 static bool access_redis(brpc::Channel& channel, const char* command) {
     brpc::RedisRequest request;
@@ -44,6 +64,7 @@ static bool access_redis(brpc::Channel& channel, const char* command) {
     }
     brpc::RedisResponse response;
     brpc::Controller cntl;
+    cntl.set_request_code(butil::cal_slot_num(extractSecondWord(command)));
     channel.CallMethod(NULL, &cntl, &request, &response, NULL);
     if (cntl.Failed()) {
         LOG(ERROR) << "Fail to access redis, " << cntl.ErrorText();
@@ -89,7 +110,11 @@ int main(int argc, char* argv[]) {
     options.connection_type = FLAGS_connection_type;
     options.timeout_ms = FLAGS_timeout_ms/*milliseconds*/;
     options.max_retry = FLAGS_max_retry;
-    if (channel.Init(FLAGS_server.c_str(), &options) != 0) {
+    if (!FLAGS_auth.empty()) {
+        brpc::policy::RedisAuthenticator* auth = new brpc::policy::RedisAuthenticator(FLAGS_auth);
+        options.auth = auth;
+    }
+    if (channel.Init(FLAGS_server.c_str(), "c_redis_cluster", &options) != 0) {
         LOG(ERROR) << "Fail to initialize channel";
         return -1;
     }

src/brpc/channel.cpp

@@ -24,7 +24,6 @@
 #include "butil/third_party/murmurhash3/murmurhash3.h"
 #include "butil/strings/string_util.h"
 #include "bthread/unstable.h"                        // bthread_timer_add
-#include "brpc/socket_map.h"                         // SocketMapInsert
 #include "brpc/compress.h"
 #include "brpc/global.h"
 #include "brpc/span.h"

src/brpc/channel.h

@@ -32,6 +32,7 @@
 #include "brpc/adaptive_connection_type.h"  // AdaptiveConnectionType
 #include "brpc/socket_id.h"                 // SocketId
 #include "brpc/controller.h"                // brpc::Controller
+#include "brpc/socket_map.h"                         // SocketMapInsert
 #include "brpc/details/profiler_linker.h"
 #include "brpc/retry_policy.h"
 #include "brpc/naming_service_filter.h"

src/brpc/controller.cpp

@@ -29,7 +29,6 @@
 #include "bthread/unstable.h"
 #include "bvar/bvar.h"
 #include "brpc/socket.h"
-#include "brpc/socket_map.h"
 #include "brpc/channel.h"
 #include "brpc/load_balancer.h"
 #include "brpc/closure_guard.h"
@@ -265,6 +264,7 @@ void Controller::ResetPods() {
     _sender = NULL;
     _request_code = 0;
     _single_server_id = INVALID_SOCKET_ID;
+    _tmp_single_server_id = INVALID_SOCKET_ID;
     _unfinished_call = NULL;
     _stream_creator = NULL;
     _accessed = NULL;
@@ -587,6 +587,12 @@ void Controller::OnVersionedRPCReturned(const CompletionInfo& info,
         return;
     }
 
+    if (saved_error == EMOVED) {
+        //todo set real serverid....
+        _current_call.OnComplete(this, _error_code, info.responded, false);
+        return IssueRPC(butil::gettimeofday_us());
+    }
+
     if ((!_error_code && _retry_policy == NULL) ||
         _current_call.nretry >= _max_retry) {
         goto END_OF_RPC;
@@ -1020,14 +1026,23 @@ void Controller::IssueRPC(int64_t start_realtime_us) {
     if (SingleServer()) {
         // Don't use _current_call.peer_id which is set to -1 after construction
         // of the backup call.
-        const int rc = Socket::Address(_single_server_id, &tmp_sock);
+        int rc;
+        SocketId peer_id;
+        if (_single_server_id != INVALID_SOCKET_ID) {
+            rc = Socket::Address(_single_server_id, &tmp_sock);
+            peer_id = _single_server_id;
+        } else {
+            rc = Socket::Address(_tmp_single_server_id, &tmp_sock);
+            peer_id = _tmp_single_server_id;
+            _tmp_single_server_id = INVALID_SOCKET_ID;
+        }
         if (rc != 0 || (!is_health_check_call() && !tmp_sock->IsAvailable())) {
             SetFailed(EHOSTDOWN, "Not connected to %s yet, server_id=%" PRIu64,
                       endpoint2str(_remote_side).c_str(), _single_server_id);
             tmp_sock.reset();  // Release ref ASAP
             return HandleSendFailed();
         }
-        _current_call.peer_id = _single_server_id;
+        _current_call.peer_id = peer_id;
     } else {
         LoadBalancer::SelectIn sel_in =
             { start_realtime_us, true,

src/brpc/controller.h

@@ -43,6 +43,7 @@
 #include "brpc/callback.h"
 #include "brpc/progressive_attachment.h"       // ProgressiveAttachment
 #include "brpc/progressive_reader.h"           // ProgressiveReader
+#include "brpc/socket_map.h"
 #include "brpc/grpc.h"
 #include "brpc/kvmap.h"
 
@@ -564,6 +565,10 @@ friend void policy::ProcessThriftRequest(InputMessageBase*);
     // -1 means no deadline.
     int64_t deadline_us() const { return _deadline_us; }
 
+    void set_tmp_single_socket_id (const SocketId& id) {
+        _tmp_single_server_id = id;
+    }
+
 private:
     struct CompletionInfo {
         CallId id;           // call_id of the corresponding request
@@ -679,7 +684,7 @@ friend void policy::ProcessThriftRequest(InputMessageBase*);
 
     void HandleStreamConnection(Socket *host_socket);
 
-    bool SingleServer() const { return _single_server_id != INVALID_SOCKET_ID; }
+    bool SingleServer() const { return _single_server_id != INVALID_SOCKET_ID || _tmp_single_server_id != INVALID_SOCKET_ID; }
 
     void SubmitSpan();
 
@@ -774,6 +779,7 @@ friend void policy::ProcessThriftRequest(InputMessageBase*);
     RPCSender* _sender;
     uint64_t _request_code;
     SocketId _single_server_id;
+    SocketId _tmp_single_server_id;
     butil::intrusive_ptr<SharedLoadBalancer> _lb;
 
     // for passing parameters to created bthread, don't modify it otherwhere.

src/brpc/errno.proto

@@ -49,6 +49,7 @@ enum Errno {
     ELIMIT                  = 2004;  // Reached server's limit on resources
     ECLOSE                  = 2005;  // Close socket initiatively
     EITP                    = 2006;  // Failed Itp response
+    EMOVED                  = 2007;  // Failed Itp response
 
     // Errno related to RDMA (may happen at both sides)
     ERDMA                   = 3001;  // RDMA verbs error

src/brpc/global.cpp

@@ -39,6 +39,7 @@
 #include "brpc/policy/consul_naming_service.h"
 #include "brpc/policy/discovery_naming_service.h"
 #include "brpc/policy/nacos_naming_service.h"
+#include "brpc/policy/redis_cluster_naming_service.h"
 
 // Load Balancers
 #include "brpc/policy/round_robin_load_balancer.h"
@@ -123,6 +124,7 @@ struct GlobalExtensions {
         , ch_mh_lb(CONS_HASH_LB_MURMUR3)
         , ch_md5_lb(CONS_HASH_LB_MD5)
         , ch_ketama_lb(CONS_HASH_LB_KETAMA)
+        , ch_redis_cluster_lb(CONS_HASH_LB_REDIS_CLUSTER)
         , constant_cl(0) {
     }
 
@@ -138,6 +140,7 @@ struct GlobalExtensions {
     ConsulNamingService cns;
     DiscoveryNamingService dcns;
     NacosNamingService nns;
+    RedisClusterNamingService rcns;
 
     RoundRobinLoadBalancer rr_lb;
     WeightedRoundRobinLoadBalancer wrr_lb;
@@ -147,6 +150,7 @@ struct GlobalExtensions {
     ConsistentHashingLoadBalancer ch_mh_lb;
     ConsistentHashingLoadBalancer ch_md5_lb;
     ConsistentHashingLoadBalancer ch_ketama_lb;
+    ConsistentHashingLoadBalancer ch_redis_cluster_lb;
     DynPartLoadBalancer dynpart_lb;
 
     AutoConcurrencyLimiter auto_cl;
@@ -363,6 +367,7 @@ static void GlobalInitializeOrDieImpl() {
     NamingServiceExtension()->RegisterOrDie("consul", &g_ext->cns);
     NamingServiceExtension()->RegisterOrDie("discovery", &g_ext->dcns);
     NamingServiceExtension()->RegisterOrDie("nacos", &g_ext->nns);
+    NamingServiceExtension()->RegisterOrDie("redis_cluster", &g_ext->rcns);
 
     // Load Balancers
     LoadBalancerExtension()->RegisterOrDie("rr", &g_ext->rr_lb);
@@ -373,6 +378,7 @@ static void GlobalInitializeOrDieImpl() {
     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);
+    LoadBalancerExtension()->RegisterOrDie("c_redis_cluster", &g_ext->ch_redis_cluster_lb);
     LoadBalancerExtension()->RegisterOrDie("_dynpart", &g_ext->dynpart_lb);
 
     // Compress Handlers

src/brpc/policy/consistent_hashing_load_balancer.cpp

@@ -121,6 +121,38 @@ bool KetamaReplicaPolicy::Build(ServerId server,
     return true;
 }
 
+class RedisClusterPolicy : public ReplicaPolicy {
+public:
+
+    virtual bool Build(ServerId server,
+                       size_t num_replicas,
+                       std::vector<ConsistentHashingLoadBalancer::Node>* replicas) const;
+
+    virtual const char* name() const { return "RedisCluster"; }
+};
+
+bool RedisClusterPolicy::Build(ServerId server,
+                                 size_t,
+                                 std::vector<ConsistentHashingLoadBalancer::Node>* replicas) const {
+    SocketUniquePtr ptr;
+    if (Socket::AddressFailedAsWell(server.id, &ptr) == -1) {
+        return false;
+    }
+    replicas->clear();
+    std::stringstream ss(server.tag);
+    int slot_start, slot_end;
+    char delimiter;
+    ss >> slot_start >> delimiter >> slot_end;
+    for (size_t i = slot_start; i <= slot_end; ++i) {
+        ConsistentHashingLoadBalancer::Node node;
+        node.hash = i;
+        node.server_sock = server;
+        node.server_addr = ptr->remote_side();
+        replicas->push_back(node);
+    }
+    return true;
+}
+
 namespace {
 
 pthread_once_t s_replica_policy_once = PTHREAD_ONCE_INIT;
@@ -130,7 +162,8 @@ void InitReplicaPolicy() {
     g_replica_policy = new std::array<const ReplicaPolicy*, CONS_HASH_LB_LAST>({
         new DefaultReplicaPolicy(MurmurHash32),
         new DefaultReplicaPolicy(MD5Hash32),
-        new KetamaReplicaPolicy
+        new KetamaReplicaPolicy,
+        new RedisClusterPolicy
     });
 }
 
@@ -245,12 +278,12 @@ size_t ConsistentHashingLoadBalancer::AddServersInBatch(
     std::sort(add_nodes.begin(), add_nodes.end());
     bool executed = false;
     const size_t ret = _db_hash_ring.ModifyWithForeground(AddBatch, add_nodes, &executed);
-    CHECK(ret % _num_replicas == 0);
-    const size_t n = ret / _num_replicas;
-    LOG_IF(ERROR, n != servers.size())
-        << "Fail to AddServersInBatch, expected " << servers.size()
-        << " actually " << n;
-    return n;
+    // CHECK(ret % _num_replicas == 0);
+    // const size_t n = ret / _num_replicas;
+    // LOG_IF(ERROR, n != servers.size())
+    //     << "Fail to AddServersInBatch, expected " << servers.size()
+    //     << " actually " << n;
+    return add_nodes.size();
 }
 
 bool ConsistentHashingLoadBalancer::RemoveServer(const ServerId& server) {

src/brpc/policy/consistent_hashing_load_balancer.h

@@ -36,9 +36,10 @@ enum ConsistentHashingLoadBalancerType {
     CONS_HASH_LB_MURMUR3 = 0,
     CONS_HASH_LB_MD5 = 1,
     CONS_HASH_LB_KETAMA = 2,
+    CONS_HASH_LB_REDIS_CLUSTER = 3,
 
     // Identify the last one.
-    CONS_HASH_LB_LAST = 3
+    CONS_HASH_LB_LAST = 4
 };
 
 class ConsistentHashingLoadBalancer : public LoadBalancer {