support rebalance handling

BUILD

@@ -118,6 +118,7 @@ BUTIL_SRCS = [
     "src/butil/third_party/snappy/snappy-stubs-internal.cc",
     "src/butil/third_party/snappy/snappy.cc",
     "src/butil/third_party/murmurhash3/murmurhash3.cpp",
+    "src/butil/third_party/libvbucket/rfc1321/md5c.c",
     "src/butil/third_party/libvbucket/cJSON.c",
     "src/butil/third_party/libvbucket/crc32.c",
     "src/butil/third_party/libvbucket/ketama.c",

CMakeLists.txt

@@ -207,6 +207,7 @@ set(BUTIL_SOURCES
     ${PROJECT_SOURCE_DIR}/src/butil/third_party/snappy/snappy-stubs-internal.cc
     ${PROJECT_SOURCE_DIR}/src/butil/third_party/snappy/snappy.cc
     ${PROJECT_SOURCE_DIR}/src/butil/third_party/murmurhash3/murmurhash3.cpp
+    ${PROJECT_SOURCE_DIR}/src/butil/third_party/libvbucket/rfc1321/md5c.c
     ${PROJECT_SOURCE_DIR}/src/butil/third_party/libvbucket/cJSON.c
     ${PROJECT_SOURCE_DIR}/src/butil/third_party/libvbucket/crc32.c
     ${PROJECT_SOURCE_DIR}/src/butil/third_party/libvbucket/ketama.c

Makefile

@@ -47,6 +47,7 @@ BUTIL_SOURCES = \
     src/butil/third_party/snappy/snappy-stubs-internal.cc \
     src/butil/third_party/snappy/snappy.cc \
     src/butil/third_party/murmurhash3/murmurhash3.cpp \
+    src/butil/third_party/libvbucket/rfc1321/md5c.c \
     src/butil/third_party/libvbucket/cJSON.c \
     src/butil/third_party/libvbucket/crc32.c \
     src/butil/third_party/libvbucket/ketama.c \

src/brpc/channel.h

@@ -124,6 +124,8 @@ struct ChannelOptions {
 class Channel : public ChannelBase {
 friend class Controller;
 friend class SelectiveChannel;
+friend class CouchbaseChannel;
+friend class CouchbaseServerListener;
 public:
     Channel(ProfilerLinker = ProfilerLinker());
     ~Channel();

src/brpc/controller.h

@@ -62,8 +62,6 @@ class MongoContext;
 class RetryPolicy;
 class InputMessageBase;
 class ThriftStub;
-class CouchbaseChannel;
-class CouchbaseDone;
 namespace policy {
 class OnServerStreamCreated;
 void ProcessMongoRequest(InputMessageBase*);

src/brpc/couchbase.cpp

@@ -38,8 +38,8 @@ int CouchbaseRequest::ParseRequest(
     return 0;
 }
 
-bool CouchbaseRequest::BuildNewWithVBucketId(CouchbaseRequest* request, 
-                                             const size_t vbucket_id) const {
+bool CouchbaseRequest::BuildVBucketId(const size_t vbucket_id,
+                                      CouchbaseRequest* request) const {
     if (this == request) {
         return false;
     }
@@ -56,17 +56,19 @@ bool CouchbaseRequest::BuildNewWithVBucketId(CouchbaseRequest* request,
     }
     _buf.append_to(&request->_buf, n - sizeof(header), sizeof(header));
     request->_pipelined_count = _pipelined_count;
+    request->_read_replicas = _read_replicas;
     return true;
 }
 
-bool CouchbaseRequest::ReplicasGet(const butil::StringPiece& key) {
+bool CouchbaseRequest::ReplicasGet(const butil::StringPiece& key, 
+                                   const size_t vbucket_id) {
     const policy::MemcacheRequestHeader header = {
         policy::MC_MAGIC_REQUEST,
         0x83,
         butil::HostToNet16(key.size()),
         0,
         policy::MC_BINARY_RAW_BYTES,
-        0,
+        butil::HostToNet16(vbucket_id),
         butil::HostToNet32(key.size()),
         0,
         0
@@ -77,10 +79,33 @@ bool CouchbaseRequest::ReplicasGet(const butil::StringPiece& key) {
     if (_buf.append(key.data(), key.size())) {
         return false;
     }
+    _read_replicas = true;
     ++_pipelined_count;
     return true;
 }
 
+bool CouchbaseResponse::RecoverOptCodeForReplicasRead() {
+    const size_t n = _buf.size();
+    policy::MemcacheResponseHeader header;
+    if (n < sizeof(header)) {
+        butil::string_printf(&_err, "buffer is too small to contain a header");
+        return false;
+    }
+    _buf.copy_to(&header, sizeof(header));
+    if (header.command != (uint8_t)policy::MC_BINARY_REPLICAS_READ) {
+        butil::string_printf(&_err, "not a replicas get response");
+        return false;
+    }
+    header.command = (uint8_t)policy::MC_BINARY_GET;
+    CouchbaseResponse response;
+    if (response._buf.append(&header, sizeof(header))) {
+        return false;
+    }
+    _buf.append_to(&response._buf, n - sizeof(header), sizeof(header));
+    Swap(&response);
+    return true;
+}
+
 bool CouchbaseResponse::GetStatus(Status* st) {
     const size_t n = _buf.size();
     policy::MemcacheResponseHeader header;

src/brpc/couchbase.h

@@ -24,14 +24,18 @@ namespace brpc {
 
 // Request to couchbase.
 // Do not support pipeline multiple operations in one request and sent now.
+// Do not support Flush/Version
 class CouchbaseRequest : public MemcacheRequest {
+friend class CouchbaseChannel;
+friend class VBucketContext; 
 public:
     void Swap(CouchbaseRequest* other) {
         MemcacheRequest::Swap(other);
     }
 
-    bool Get(const butil::StringPiece& key) {
+    bool Get(const butil::StringPiece& key, bool read_replicas = false) {
         MemcacheRequest::Clear();
+        _read_replicas = read_replicas;
         return MemcacheRequest::Get(key);
     }
 
@@ -101,24 +105,28 @@ public:
 
     void CopyFrom(const CouchbaseRequest& from) {
         MemcacheRequest::CopyFrom(from);
+        _read_replicas = from._read_replicas;
     }
 
+private:
     int ParseRequest(std::string* key, 
                      policy::MemcacheBinaryCommand* command) const;
 
-    bool BuildNewWithVBucketId(CouchbaseRequest* request, 
-                               const size_t vbucket_id) const;
+    bool BuildVBucketId(const size_t vbucket_id,
+                        CouchbaseRequest* request) const;
 
-    bool ReplicasGet(const butil::StringPiece& key);
+    bool ReplicasGet(const butil::StringPiece& key, const size_t vbucket_id);
 
-private:
     void MergeFrom(const CouchbaseRequest& from);
 
     int pipelined_count();
+    
+    bool read_replicas() const { return _read_replicas; } 
+
+    bool _read_replicas = false;
 };
 
-// Request to couchbase.
-// Do not support pipeline multiple operations in one request and sent now.
+// Response from couchbase.
 class CouchbaseResponse : public MemcacheResponse {
 public:
     void Swap(CouchbaseResponse* other) {
@@ -133,6 +141,8 @@ public:
 
     bool GetStatus(Status* status);
 
+    bool RecoverOptCodeForReplicasRead();
+
 private:
     void MergeFrom(const CouchbaseResponse& from);
 

src/brpc/couchbase_channel.h

@@ -25,11 +25,59 @@
 #include "butil/containers/doubly_buffered_data.h"
 
 namespace brpc {
+// It is used to detect the new master server of vbuckets when the lastest 
+// vbucket mapping has not been received during rebalance.
+class DetectedMaster {
+public:
+    DetectedMaster() : _verified(false), _index(-1) {}
+    butil::atomic<bool> _verified;
+    butil::atomic<int> _index;
+
+private:
+    DetectedMaster(const DetectedMaster&) = delete;
+    DetectedMaster& operator=(const DetectedMaster&) = delete;
+};
 
 using CouchbaseChannelMap = 
     std::unordered_map<std::string, std::unique_ptr<Channel>>; 
+using DetectedVBucketMap = std::vector<DetectedMaster>;
+
+// Couchbase has two type of distribution used to map keys to servers.
+// One is vbucket distribution and other is ketama distribution.
+// This struct describes vbucket distribution of couchbase.
+// 'num_replicas': the number of copies that will be stored on servers of one
+//                 vbucket. Each vbucket must have this number of servers 
+//                 indexes plus one.
+// '_vbucket': A zero-based indexed by vBucketId. The entries in the _vbucket
+//            are arrays of integers, where each integer is a zero-based 
+//            index into the '_servers'. 
+// '_fvbucket': It is fast forward map with same struct as _vbucket. It is 
+//              used to provide the final vBubcket-to-server map during the
+//              statrt of the rebalance. 
+// '_servers': all servers of a bucket.                    
+// '_channel_map': the memcache channel for each server.
+// TODO: support ketama vbucket distribution
+struct VBucketServerMap {
+    uint64_t _version = 0;
+    int _num_replicas = 0;    
+    std::vector<std::vector<int>> _vbucket;
+    std::vector<std::vector<int>> _fvbucket;
+    std::vector<std::string> _servers;
+    CouchbaseChannelMap _channel_map;
+};
+
+enum VBucketStatus {
+    FORWARD_CREATE = 0x00,
+    FORWARD_FINISH = 0x01,
+    FORWARD_KEEPING = 0x02,
+    FORWARD_CHANGE = 0x03,
+    MASTER_CHANGE_WITHOUT_F = 0x04,
+    MASTER_KEEPING_WITHOUT_F = 0x05,
+    NO_CHANGE = 0x06,
+};
 
 class CouchbaseServerListener;
+class VBucketContext;
 
 // A couchbase channel maps different key to sub memcache channel according to 
 // current vbuckets mapping. It retrieves current vbuckets mapping by maintain
@@ -40,17 +88,27 @@ class CouchbaseServerListener;
 // For async rpc, Should not delete this channel until rpc done.
 class CouchbaseChannel : public ChannelBase/*non-copyable*/ {
 friend class CouchbaseServerListener;
+friend class VBucketContext;
+friend class CouchbaseDone;
 public:
     CouchbaseChannel();
     ~CouchbaseChannel();
 
     // You MUST initialize a couchbasechannel before using it. 
     // 'Server_addr': address list of couchbase servers. On these addresses, we 
-    //                can get vbucket map. 
+    //                can get vbucket map. Like following: "addr1:port1,addr2:port2" 
+    // 'bucket_name': the bucket name of couchbase server to access.
     // 'options': is used for each memcache channel of vbucket. The protocol 
     //            should be PROTOCOL_MEMCACHE. If 'options' is null, 
     //            use default options. 
-    int Init(const char* server_addr, const ChannelOptions* options);
+    int Init(const char* server_addr, const char* bucket_name, 
+             const ChannelOptions* options);
+
+    // 'listen_url': from this url, we can get vbucket map. Usually, it is 
+    // somthing like following:
+    // "http://host:port/pools/default/bucketsStreaming/bucket_name" or
+    // "http://host:port/pools/default/buckets/bucket_name"
+    int Init(const char* listen_url, const ChannelOptions* options);
 
     // TODO: Do not support pipeline mode now.
     // Send request to the mapped channel according to the key of request.
@@ -62,45 +120,38 @@ public:
 
     void Describe(std::ostream& os, const DescribeOptions& options);
 
-    // Couchbase has two type of distribution used to map keys to servers.
-    // One is vbucket distribution and other is ketama distribution.
-    // This struct describes vbucket distribution of couchbase.
-    // 'num_replicas': the number of copies that will be stored on servers of one
-    //                 vbucket. Each vbucket must have this number of servers 
-    //                 indexes plus one.
-    // '_vbucket': A zero-based indexed by vBucketId. The entries in the _vbucket
-    //            are arrays of integers, where each integer is a zero-based 
-    //            index into the '_servers'. 
-    // '_fvbucket': It is fast forward map with same struct as _vbucket. It is 
-    //              used to provide the final vBubcket-to-server map during the
-    //              statrt of the rebalance. 
-    // '_servers': all servers of a bucket.                    
-    // '_channel_map': the memcache channel for each server.
-    // TODO: support ketama vbucket distribution
-    struct VBucketServerMap {
-        uint64_t _version = 0;
-        int _num_replicas = 0;    
-        std::vector<std::vector<int>> _vbucket;
-        std::vector<std::vector<int>> _fvbucket;
-        std::vector<std::string> _servers;
-        CouchbaseChannelMap _channel_map;
-    };
-
 private:
     int CheckHealth();
 
-    Channel* SelectMasterChannel(const VBucketServerMap* vb_map, 
-                                 const size_t vb_index);
+    Channel* SelectBackupChannel(const VBucketServerMap* vb_map, 
+                                 const size_t vb_index, const int reason, 
+                                 VBucketContext* context);
 
     Channel* GetMappedChannel(const std::string* server,
                               const VBucketServerMap* vb_map);
 
-    const CouchbaseChannelMap& GetChannelMap();
+    const VBucketServerMap* vbucket_map();
+
+    bool IsNeedRetry(const Controller* cntl, const VBucketContext& context,
+                     CouchbaseResponse* response, int* reason, 
+                     std::string* error_text);
+
+    bool DoRetry(const int reason, Controller* cntl, 
+                 CouchbaseResponse* response, VBucketContext* vb_ct);
+
+    int GetDetectedMaster(const VBucketServerMap* vb_map, const size_t vb_index);
 
-    const std::string* GetMaster(const VBucketServerMap* vb_map, 
-                                 const size_t vb_index, int* index = nullptr);
+    void UpdateDetectedMasterIfNeeded(const int reason, 
+                                      const VBucketContext& context);
 
-    size_t Hash(const butil::StringPiece& key, const size_t vbuckets_num);
+    bool IsInRebalancing(const VBucketServerMap* vb_map) {
+        return !vb_map->_fvbucket.empty();
+    }
+ 
+    const std::string* GetNextRetryServer(
+        const VBucketStatus change, const int reason, 
+        const VBucketServerMap* vb_map, const size_t vb_index, 
+        VBucketContext* context);
 
     bool UpdateVBucketServerMap(
         const int num_replicas,
@@ -121,10 +172,13 @@ private:
 
     std::string GetAuthentication() const;
 
-    // Options for each memcache channel of vbucket.
+    // Options for each memcache channel of real servers.
     ChannelOptions _common_options; 
-    // Listener monitor and update vbucket map information.
+    // Listener monitor and update vbucket map.
     std::unique_ptr<CouchbaseServerListener> _listener;
+    // We need detect new vbucket map due to current vbucket map is invalid 
+    // during rebalance.
+    std::unique_ptr<DetectedVBucketMap> _detected_vbucket_map;
     butil::DoublyBufferedData<VBucketServerMap> _vbucket_map;
 };
 

src/brpc/global.cpp

@@ -30,6 +30,7 @@
 #include "brpc/policy/domain_naming_service.h"
 #include "brpc/policy/remote_file_naming_service.h"
 #include "brpc/policy/consul_naming_service.h"
+#include "brpc/policy/couchbase_naming_service.h"
 
 // Load Balancers
 #include "brpc/policy/round_robin_load_balancer.h"
@@ -119,6 +120,7 @@ struct GlobalExtensions {
     DomainNamingService dns;
     RemoteFileNamingService rfns;
     ConsulNamingService cns;
+    CouchbaseNamingService cblns;
 
     RoundRobinLoadBalancer rr_lb;
     WeightedRoundRobinLoadBalancer wrr_lb;
@@ -337,6 +339,7 @@ static void GlobalInitializeOrDieImpl() {
   NamingServiceExtension()->RegisterOrDie("redis", &g_ext->dns);
     NamingServiceExtension()->RegisterOrDie("remotefile", &g_ext->rfns);
     NamingServiceExtension()->RegisterOrDie("consul", &g_ext->cns);
+    NamingServiceExtension()->RegisterOrDie("couchbase_list", &g_ext->cblns);
 
     // Load Balancers
     LoadBalancerExtension()->RegisterOrDie("rr", &g_ext->rr_lb);

src/brpc/policy/couchbase_naming_service.cpp

+// Copyright (c) 2018 Iqiyi, Inc.
+// 
+// Licensed 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.
+
+// Authors: Cai,Daojin (Caidaojin@qiyi.com)
+
+#include <stdlib.h>                                   // strtol
+#include <string>                                     // std::string
+#include <set>                                        // std::set
+#include "butil/string_splitter.h"                     // StringSplitter
+#include "butil/strings/string_piece.h"
+#include "butil/strings/string_split.h"
+#include "butil/strings/string_number_conversions.h"
+#include "brpc/log.h"
+#include "brpc/policy/couchbase_naming_service.h"
+
+namespace brpc {
+namespace policy {
+
+// Defined in file_naming_service.cpp
+bool SplitIntoServerAndTag(const butil::StringPiece& line,
+                           butil::StringPiece* server_addr,
+                           butil::StringPiece* tag);
+
+butil::Mutex CouchbaseNamingService::_mutex; 
+std::unordered_map<std::string, std::string> CouchbaseNamingService::servers_map;
+
+bool CouchbaseNamingService::ParseListenUrl(
+    const butil::StringPiece listen_url, std::string* server, 
+    std::string* streaming_uri, std::string* init_uri) {
+    do {
+        const size_t pos = listen_url.find("//");
+        if (pos == listen_url.npos) {
+            break;
+        }
+        const size_t host_pos = listen_url.find('/', pos + 2);        
+        if (host_pos == listen_url.npos) {
+            break;
+        }
+        butil::StringPiece sub_str = listen_url.substr(pos + 2, host_pos - pos - 2);
+        server->clear();
+        server->append(sub_str.data(), sub_str.length());
+        butil::EndPoint point;
+        if (butil::str2endpoint(server->c_str(), &point) != 0) {
+            LOG(FATAL) << "Failed to get address and port \'" 
+                       << server << "\'.";
+            break; 
+        }
+        butil::StringPiece uri_sub = listen_url;
+        uri_sub.remove_prefix(host_pos);
+        size_t uri_pos = uri_sub.find("/bucketsStreaming/");
+        if (uri_pos != uri_sub.npos) {
+            streaming_uri->clear();
+            streaming_uri->append(uri_sub.data(), uri_sub.length()); 
+            init_uri->clear();
+            init_uri->append(uri_sub.data(), uri_pos);
+            init_uri->append("/buckets/");
+            butil::StringPiece bucket_name = uri_sub;
+            bucket_name.remove_prefix(uri_pos + std::strlen("/bucketsStreaming/"));
+            init_uri->append(bucket_name.data(), bucket_name.length());
+            return true;
+        }
+        uri_pos = uri_sub.find("/buckets/");
+        if (uri_pos != uri_sub.npos) {
+            init_uri->clear();
+            init_uri->append(uri_sub.data(), uri_sub.length()); 
+            streaming_uri->clear();
+            streaming_uri->append(uri_sub.data(), uri_pos);
+            streaming_uri->append("/bucketsStreaming/");
+            butil::StringPiece bucket_name = uri_sub; 
+            bucket_name.remove_prefix(uri_pos + std::strlen("/buckets/"));
+            streaming_uri->append(bucket_name.data(), bucket_name.length());
+            return true;
+        }
+    } while (false);
+    LOG(FATAL) << "Failed to parse listen url \'" <<  listen_url << "\'.";
+    return false;
+}
+
+bool CouchbaseNamingService::ParseNamingServiceUrl(const butil::StringPiece ns_url, 
+                                                   std::string* listen_port) {
+    butil::StringPiece protocol;
+    std::string server_list;
+    const size_t pos = ns_url.find("//");
+    if (pos != ns_url.npos) {
+        protocol = ns_url.substr(0, pos);
+        butil::StringPiece sub = ns_url.substr(pos+2);
+        server_list.append(sub.data(), sub.length());
+    }
+    if (protocol != "couchbase_list:" && server_list.empty()) {
+        LOG(FATAL) << "Invalid couchbase naming service  " << ns_url;
+        return false;
+    }
+    std::vector<std::string> server_array;
+    butil::SplitString(server_list, ',', &server_array);
+    listen_port->clear();
+    for (const std::string& addr_port : server_array) {
+        butil::EndPoint point;
+        if (butil::str2endpoint(addr_port.c_str(), &point) != 0) {
+            LOG(FATAL) << "Failed to get endpoint from \'" << addr_port 
+                       << "\' of the naming server url \'" << ns_url << "\'.";
+            return false;
+        }
+        if (listen_port->empty()) {
+            *listen_port = butil::IntToString(point.port);
+        }
+    }    
+    return true;
+}
+
+int CouchbaseNamingService::GetServers(const char *service_name,
+                                       std::vector<ServerNode>* servers) {
+    servers->clear();
+    // Sort/unique the inserted vector is faster, but may have a different order
+    // of addresses from the file. To make assertions in tests easier, we use
+    // set to de-duplicate and keep the order.
+    std::set<ServerNode> presence;
+    std::string line;
+
+    if (!service_name) {
+        LOG(FATAL) << "Param[service_name] is NULL";
+        return -1;
+    }
+    std::string new_servers(service_name);
+    {
+        BAIDU_SCOPED_LOCK(_mutex);
+        const auto& iter = servers_map.find(new_servers);
+        if (iter != servers_map.end()) {
+            new_servers = iter->second;
+        }
+    }
+    RemoveUniqueSuffix(new_servers);
+    for (butil::StringSplitter sp(new_servers.c_str(), ','); sp != NULL; ++sp) {
+        line.assign(sp.field(), sp.length());
+        butil::StringPiece addr;
+        butil::StringPiece tag;
+        if (!SplitIntoServerAndTag(line, &addr, &tag)) {
+            continue;
+        }
+        const_cast<char*>(addr.data())[addr.size()] = '\0'; // safe
+        butil::EndPoint point;
+        if (str2endpoint(addr.data(), &point) != 0 &&
+            hostname2endpoint(addr.data(), &point) != 0) {
+            LOG(ERROR) << "Invalid address=`" << addr << '\'';
+            continue;
+        }
+        ServerNode node;
+        node.addr = point;
+        tag.CopyToString(&node.tag);
+        if (presence.insert(node).second) {
+            servers->push_back(node);
+        } else {
+            RPC_VLOG << "Duplicated server=" << node;
+        }
+    }
+    RPC_VLOG << "Got " << servers->size()
+             << (servers->size() > 1 ? " servers" : " server");
+    return 0;
+}
+
+void CouchbaseNamingService::Describe(
+    std::ostream& os, const DescribeOptions&) const {
+    os << "Couchbase_list";
+    return;
+}
+
+NamingService* CouchbaseNamingService::New() const {
+    return new CouchbaseNamingService;
+}
+
+void CouchbaseNamingService::Destroy() {
+    delete this;
+}
+
+void CouchbaseNamingService::ResetCouchbaseListenerServers(
+    const std::string& service_name, std::string& new_servers) {
+    BAIDU_SCOPED_LOCK(_mutex);
+    auto iter = servers_map.find(service_name);
+    if (iter != servers_map.end()) {
+        iter->second.swap(new_servers);
+    } else {
+        servers_map.emplace(service_name, new_servers);
+    }
+}
+
+std::string CouchbaseNamingService::AddUniqueSuffix(
+    const char* name_url, const char* unique_id) {
+    std::string couchbase_name_url;
+    couchbase_name_url.append(name_url);
+    couchbase_name_url.append(1, '_');
+    couchbase_name_url.append(unique_id);
+    return std::move(couchbase_name_url);
+}
+
+void CouchbaseNamingService::RemoveUniqueSuffix(std::string& name_service) {
+    const size_t pos = name_service.find('_');
+    if (pos != std::string::npos) {
+        name_service.resize(pos);
+    }
+}
+
+}  // namespace policy
+} // namespace brpc

src/brpc/policy/couchbase_naming_service.h

+// Copyright (c) 2018 Iqiyi, Inc.
+// 
+// Licensed 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.
+
+// Authors: Cai,Daojin (caidaojin@qiyi.com)
+
+#ifndef  BRPC_POLICY_COUCHBASE_NAMING_SERVICE
+#define  BRPC_POLICY_COUCHBASE_NAMING_SERVICE
+
+#include <unordered_map>
+#include "brpc/periodic_naming_service.h"
+
+namespace brpc {
+
+class CouchbaseServerListener;
+
+}
+
+namespace brpc {
+namespace policy {
+
+// It is only used for couchbase channel. It updates servers for listen channel 
+// of CouchbaseServerListener. The naming service format is like
+// "couchbase_list://addr1:port,addr:port_****" where "_****" is a unique id for
+// each couchbase channel since we can not share naming service and "addr*:port"
+// are avalible servers for initializing.
+// After initialization, it get the latest server list periodically from 
+// 'servers_map' by service name as key. 
+class CouchbaseNamingService : public PeriodicNamingService {
+friend brpc::CouchbaseServerListener;
+private:
+    static butil::Mutex _mutex; 
+    // Store the lastest server list for each couchbase channel.
+    // Key is service name of each couchbase channel and value is the latest 
+    // server list. It is like following:
+    // key: addr1:port,addr2:port_****
+    // value: addr1:port,addr2:port,addr3:port 
+    static std::unordered_map<std::string, std::string> servers_map;
+
+    int GetServers(const char *service_name,
+                   std::vector<ServerNode>* servers);
+    
+    static bool ParseNamingServiceUrl(butil::StringPiece ns_url, 
+                                      std::string* listen_port);
+
+    static bool ParseListenUrl(
+        const butil::StringPiece listen_url, std::string* server_address, 
+        std::string* streaming_uri, std::string* init_uri);
+    
+    // Clear naming server data when couchbase channel destroyed.
+    static void ClearNamingServiceData(const std::string& service_name) {
+        BAIDU_SCOPED_LOCK(_mutex);
+        servers_map.erase(service_name);
+    }
+
+    // Called by couchbase listener when vbucekt map changing. 
+    // It set new server list for key 'service_name' in servers_map.
+    static void ResetCouchbaseListenerServers(const std::string& service_name, 
+                                              std::string& new_servers);
+
+    // For couchbase listeners, we should not share this name service object.
+    // So we append couchbase listener address to make name_url unique.
+    // Input: couchbase_list://address1:port1,address2:port2
+    // Output: couchbase_list://address1:port1,address2:port2_****
+    static std::string AddUniqueSuffix(const char* name_url, 
+                                       const char* unique_id);
+
+    // Reserve handling to AddPrefixBeforeAddress.
+    void RemoveUniqueSuffix(std::string& name_service);
+
+    void Describe(std::ostream& os, const DescribeOptions& options) const;
+
+    NamingService* New() const;
+
+    void Destroy();
+};
+
+}  // namespace policy
+} // namespace brpc
+
+#endif  //BRPC_POLICY_COUCHBASE_NAMING_SERVICE

src/brpc/policy/memcache_binary_header.h

@@ -91,7 +91,10 @@ enum MemcacheBinaryCommand {
     MC_BINARY_RINCR     = 0x39,
     MC_BINARY_RINCRQ    = 0x3a,
     MC_BINARY_RDECR     = 0x3b,
-    MC_BINARY_RDECRQ    = 0x3c
+    MC_BINARY_RDECRQ    = 0x3c,
+
+    // Replicas read for couchbase
+    MC_BINARY_REPLICAS_READ = 0x83
     // End Range operations
 };
 

src/brpc/policy/memcache_binary_protocol.cpp

@@ -64,6 +64,7 @@ static void InitSupportedCommandMap() {
     butil::bit_array_set(supported_cmd_map, MC_BINARY_STAT);
     butil::bit_array_set(supported_cmd_map, MC_BINARY_TOUCH);
     butil::bit_array_set(supported_cmd_map, MC_BINARY_SASL_AUTH);
+    butil::bit_array_set(supported_cmd_map, MC_BINARY_REPLICAS_READ);
 }
 
 inline bool IsSupportedCommand(uint8_t command) {

src/butil/third_party/libvbucket/ketama.c

@@ -4,7 +4,7 @@
 
 /* This library uses the reference MD5 implementation from [RFC1321] */
 #define PROTOTYPES 1
-#include "butil/third_party/libvbucket/rfc1321/md5c.c"
+#include "butil/third_party/libvbucket/rfc1321/md5.h"
 #undef  PROTOTYPES
 
 void hash_md5(const char *key, size_t key_length, unsigned char *result)

src/butil/third_party/libvbucket/rfc1321/md5.h

@@ -24,6 +24,9 @@
    */
 
 /* MD5 context. */
+
+#include "butil/third_party/libvbucket/rfc1321/global.h"
+
 typedef struct {
 	UINT4 state[4];                                   /* state (ABCD) */
 	UINT4 count[2];        /* number of bits, modulo 2^64 (lsb first) */

src/butil/third_party/libvbucket/rfc1321/md5c.c

@@ -23,7 +23,6 @@
    documentation and/or software.
    */
 
-#include "butil/third_party/libvbucket/rfc1321/global.h"
 #include "butil/third_party/libvbucket/rfc1321/md5.h"
 
 /* Constants for MD5Transform routine.

src/butil/third_party/libvbucket/vbucket.c

@@ -741,6 +741,10 @@ const char *vbucket_config_get_rest_api_server(VBUCKET_CONFIG_HANDLE vb, int i)
     return vb->servers[i].rest_api_authority;
 }
 
+int vbucket_config_has_forward_vbuckets(VBUCKET_CONFIG_HANDLE vb) {
+    return vb->fvbuckets ? 1 : 0; 
+}
+
 int vbucket_config_is_config_node(VBUCKET_CONFIG_HANDLE vb, int i) {
     return vb->servers[i].config_node;
 }
@@ -782,6 +786,23 @@ int vbucket_get_replica(VBUCKET_CONFIG_HANDLE vb, int vbucket, int i) {
     }
 }
 
+int fvbucket_get_master(VBUCKET_CONFIG_HANDLE vb, int vbucket) {
+    if (vb->fvbuckets) {
+        return vb->fvbuckets[vbucket].servers[0];
+    }
+    return -1;
+}
+
+int fvbucket_get_replica(VBUCKET_CONFIG_HANDLE vb, int vbucket, int i) {
+    if (vb->fvbuckets) {
+        int idx = i + 1;
+        if (idx < vb->num_servers) {
+            return vb->fvbuckets[vbucket].servers[idx];
+        }
+    }
+    return -1;
+}
+
 int vbucket_found_incorrect_master(VBUCKET_CONFIG_HANDLE vb, int vbucket,
                                    int wrongserver) {
     int mappedServer = vb->vbuckets[vbucket].servers[0];

src/butil/third_party/libvbucket/vbucket.h

@@ -352,6 +352,39 @@ namespace butil {
     LIBVBUCKET_PUBLIC_API
     int vbucket_get_replica(VBUCKET_CONFIG_HANDLE h, int id, int n);
 
+    /**
+     * Check whether including forward vbuckets
+     *
+     * @param id the fvbucket identifier
+     *
+     * @return true if forward vbuckets included.
+     */
+    LIBVBUCKET_PUBLIC_API
+    int vbucket_config_has_forward_vbuckets(VBUCKET_CONFIG_HANDLE h);
+
+    /**
+     * Get the master server for the given vbucket.
+     *
+     * @param h the vbucket config
+     * @param id the fvbucket identifier
+     *
+     * @return the server index
+     */
+    LIBVBUCKET_PUBLIC_API
+    int fvbucket_get_master(VBUCKET_CONFIG_HANDLE h, int id);
+
+    /**
+     * Get a given replica for a forward vbucket.
+     *
+     * @param h the vbucket config
+     * @param id the vbucket id
+     * @param n the replica number
+     *
+     * @return the server ID
+     */
+    LIBVBUCKET_PUBLIC_API
+    int fvbucket_get_replica(VBUCKET_CONFIG_HANDLE h, int id, int n);
+
     /**
      * @}
      */