Merge branch raft into master (#30)

.gitignore

@@ -236,3 +236,9 @@ node_modules
 public
 website/docs
 website/presentations
+
+# TiCS built and test files
+build_docker
+docker/builder/tics
+tests/docker/data
+tests/docker/log

cmake/find_kvproto.cmake

@@ -1,7 +1,11 @@
 # Currently kvproto should always use bundled library.
 
 if (NOT EXISTS "${ClickHouse_SOURCE_DIR}/contrib/kvproto/cpp/kvproto/errorpb.pb.h")
-   message (FATAL_ERROR "kvproto submodule in contrib/kvproto is missing.")
+	if (EXISTS "${ClickHouse_SOURCE_DIR}/contrib/kvproto/proto/errorpb.proto")
+		message (FATAL_ERROR "kvproto cpp files in contrib/kvproto is missing. Try go to contrib/kvproto, and run ./generate_cpp.sh")
+	else()
+		message (FATAL_ERROR "kvproto submodule in contrib/kvproto is missing. Try run 'git submodule update --init --recursive', and go to contrib/kvproto, and run ./generate_cpp.sh")
+	endif()
 endif ()
 
 message(STATUS "Using kvproto: ${ClickHouse_SOURCE_DIR}/contrib/kvproto/cpp")

contrib/client-c/.gitignore

@@ -38,6 +38,7 @@ build
 .idea
 *.iml
 *.swp
+*.swo
 tags
 .clang-format
 

contrib/client-c/CMakeLists.txt

@@ -1,5 +1,7 @@
 cmake_minimum_required(VERSION 2.8)
 project(kvClient)
+set (CMAKE_CXX_STANDARD 11)
+set (CMAKE_CXX_FLAGS  "${CMAKE_CXX_FLAGS} -Wno-narrowing")
 
 if (NOT gRPC_FOUND)
     include (cmake/find_grpc.cmake)

contrib/client-c/cmake/find_kvproto.cmake

@@ -1,7 +1,7 @@
 # Currently kvproto should always use bundled library.
 
 if (NOT EXISTS "${kvClient_SOURCE_DIR}/third_party/kvproto/cpp/kvproto/errorpb.pb.h")
-   message (FATAL_ERROR "kvproto submodule in thrid_party/kvproto is missing.")
+   message (FATAL_ERROR "kvproto submodule in third_party/kvproto is missing.")
 endif ()
 
 message(STATUS "Using kvproto: ${kvClient_SOURCE_DIR}/third_party/kvproto/cpp")

contrib/client-c/include/tikv/Exception.h → contrib/client-c/include/common/CltException.h

@@ -5,9 +5,13 @@
 #include <Poco/Exception.h>
 
 namespace pingcap {
-namespace kv {
 
 const int MismatchClusterIDCode = 1;
+const int GRPCErrorCode = 2;
+const int InitClusterIDFailed = 3;
+const int UpdatePDLeaderFailed = 4;
+const int TimeoutError = 5;
+const int RegionUnavailable = 6;
 
 class Exception : public Poco::Exception
 {
@@ -23,6 +27,4 @@ class Exception : public Poco::Exception
 
 };
 
-
-}
 }

contrib/client-c/include/tikv/Backoff.h

@@ -6,7 +6,7 @@
 #include <map>
 #include <memory>
 
-#include <tikv/Exception.h>
+#include <common/CltException.h>
 
 namespace pingcap {
 namespace kv {
@@ -71,6 +71,8 @@ struct Backoff {
     }
 };
 
+constexpr int readIndexMaxBackoff = 20000;
+
 using BackoffPtr = std::shared_ptr<Backoff>;
 
 struct Backoffer {

contrib/client-c/include/tikv/Region.h

@@ -7,6 +7,8 @@
 #include <pd/MockPDClient.h>
 #include <kvproto/metapb.pb.h>
 #include <tikv/Backoff.h>
+#include <common/Log.h>
+#include <kvproto/errorpb.pb.h>
 
 namespace pingcap {
 namespace kv {
@@ -31,6 +33,8 @@ struct RegionVerID {
     uint64_t confVer;
     uint64_t ver;
 
+    RegionVerID(uint64_t id_, uint64_t conf_ver, uint64_t ver_): id(id_), confVer(conf_ver), ver(ver_){}
+
     bool operator == (const RegionVerID & rhs) const {
         return id == rhs.id && confVer == rhs.confVer && ver == rhs.ver;
     }
@@ -56,6 +60,8 @@ struct Region {
     metapb::Peer   peer;
     metapb::Peer   learner;
 
+    Region(const metapb::Region & meta_, const metapb::Peer & peer_) : meta(meta_), peer(peer_), learner(metapb::Peer::default_instance()) {}
+
     Region(const metapb::Region & meta_, const metapb::Peer & peer_, const metapb::Peer & learner_)
         : meta(meta_), peer(peer_), learner(learner_) {}
 
@@ -118,12 +124,22 @@ using RPCContextPtr = std::shared_ptr<RPCContext>;
 
 class RegionCache {
 public:
-    RegionCache(pd::ClientPtr pdClient_) : pdClient(pdClient_) {
+    RegionCache(pd::ClientPtr pdClient_) : pdClient(pdClient_), log(&Logger::get("pingcap.tikv")) {
     }
 
     RPCContextPtr getRPCContext(Backoffer & bo, const RegionVerID & id, bool is_learner);
 
+    void updateLeader(Backoffer & bo, const RegionVerID & region_id, uint64_t leader_store_id);
+
     //KeyLocation locateKey(Backoffer & bo, std::string key);
+    //
+    void dropRegion(const RegionVerID &);
+
+    void dropStore(uint64_t failed_store_id);
+
+    void dropStoreOnSendReqFail(RPCContextPtr & ctx, const Exception & exc);
+
+    void onRegionStale(RPCContextPtr ctx, const errorpb::StaleEpoch & stale_epoch);
 
 private:
     RegionPtr getCachedRegion(Backoffer & bo, const RegionVerID & id);
@@ -153,6 +169,8 @@ class RegionCache {
     std::mutex region_mutex;
 
     std::mutex store_mutex;
+
+    Logger * log;
 };
 
 using RegionCachePtr = std::shared_ptr<RegionCache>;

contrib/client-c/include/tikv/RegionClient.h

@@ -17,28 +17,75 @@ struct RegionClient {
 
     int64_t getReadIndex() {
         auto request = new kvrpcpb::ReadIndexRequest();
-        Backoffer bo(10000);
+        Backoffer bo(readIndexMaxBackoff);
         auto rpc_call = std::make_shared<RpcCall<kvrpcpb::ReadIndexRequest>>(request);
-        auto ctx = cache -> getRPCContext(bo, region_id, true);
-        store_addr = ctx->addr;
-        sendReqToRegion(bo, rpc_call, ctx);
+        sendReqToRegion(bo, rpc_call, true);
         return rpc_call -> getResp() -> read_index();
     }
 
     template<typename T>
-    void sendReqToRegion(Backoffer & bo, RpcCallPtr<T> rpc, RPCContextPtr rpc_ctx) {
-        try {
-            rpc -> setCtx(rpc_ctx);
-            client -> sendRequest(store_addr, rpc);
+    void sendReqToRegion(Backoffer & bo, RpcCallPtr<T> rpc, bool learner) {
+        for (;;) {
+            auto ctx = cache -> getRPCContext(bo, region_id, learner);
+            store_addr = ctx->addr;
+            rpc -> setCtx(ctx);
+            try {
+                client -> sendRequest(store_addr, rpc);
+            } catch(const Exception & e) {
+                onSendFail(bo, e, ctx);
+                continue;
+            }
+            auto resp = rpc -> getResp();
+            if (resp -> has_region_error()) {
+                onRegionError(bo, ctx, resp->region_error());
+            } else {
+                return;
+            }
         }
-        catch(const Exception & e) {
-            onSendFail(bo, e);
+    }
+
+    void onRegionError(Backoffer & bo, RPCContextPtr rpc_ctx, const errorpb::Error & err) {
+        if (err.has_not_leader()) {
+            auto not_leader = err.not_leader();
+            if (not_leader.has_leader()) {
+                cache -> updateLeader(bo, rpc_ctx->region, not_leader.leader().store_id());
+                bo.backoff(boUpdateLeader, Exception("not leader"));
+            } else {
+                cache -> dropRegion(rpc_ctx->region);
+                bo.backoff(boRegionMiss, Exception("not leader"));
+            }
+            return;
+        }
+
+        if (err.has_store_not_match()) {
+            cache -> dropStore(rpc_ctx->peer.store_id());
+            return;
+        }
+
+        if (err.has_stale_epoch()) {
+            cache -> onRegionStale(rpc_ctx, err.stale_epoch());
+            return;
+        }
+
+        if (err.has_server_is_busy()) {
+            bo.backoff(boServerBusy, Exception("server busy"));
+            return;
+        }
+
+        if (err.has_stale_command()) {
+            return;
+        }
+
+        if (err.has_raft_entry_too_large()) {
+            throw Exception("entry too large");
         }
 
+        cache -> dropRegion(rpc_ctx -> region);
     }
 
-    void onSendFail(Backoffer & ,const Exception & e) {
-        e.rethrow();
+    void onSendFail(Backoffer & bo, const Exception & e, RPCContextPtr rpc_ctx) {
+        cache->dropStoreOnSendReqFail(rpc_ctx, e);
+        bo.backoff(boTiKVRPC, e);
     }
 };
 

contrib/client-c/include/tikv/Rpc.h

@@ -78,9 +78,12 @@ class RpcCall {
     void call(std::unique_ptr<tikvpb::Tikv::Stub> stub) {
         if constexpr(std::is_same<T, kvrpcpb::ReadIndexRequest>::value) {
             grpc::ClientContext context;
+            context.set_deadline(std::chrono::system_clock::now() + std::chrono::seconds(3));
             auto status = stub->ReadIndex(&context, *req, resp);
             if (!status.ok()) {
-                log -> error("read index failed: " + std::to_string(status.error_code()) + ": " + status.error_message());
+                std::string err_msg = ("read index failed: " + std::to_string(status.error_code()) + ": " + status.error_message());
+                log->error(err_msg);
+                throw Exception(err_msg, GRPCErrorCode);
             }
         }
     }

contrib/client-c/src/CMakeLists.txt

@@ -1,5 +1,3 @@
-add_subdirectory (pd)
-
 set(kvClient_sources)
 
 list(APPEND kvClient_sources pd/Client.cc)
@@ -13,3 +11,5 @@ set(kvClient_INCLUDE_DIR ${kvClient_SOURCE_DIR}/include)
 add_library(kv_client ${kvClient_sources})
 target_include_directories(kv_client PUBLIC ${KVPROTO_INCLUDE_DIR} ${kvClient_INCLUDE_DIR})
 target_link_libraries(kv_client kvproto ${Poco_Foundation_LIBRARY})
+
+add_subdirectory (test)

contrib/client-c/src/pd/Client.cc

@@ -1,4 +1,5 @@
 #include <pd/Client.h>
+#include <common/CltException.h>
 #include <grpcpp/security/credentials.h>
 #include <grpcpp/create_channel.h>
 #include <Poco/URI.h>
@@ -76,7 +77,9 @@ pdpb::GetMembersResponse Client::getMembers(std::string url)
 
     auto status = pdpb::PD::NewStub(cc)->GetMembers(&context, pdpb::GetMembersRequest{}, &resp);
     if (!status.ok()) {
-        log->error("get member failed: " + std::to_string(status.error_code()) + ": " + status.error_message());
+        std::string err_msg = "get member failed: " + std::to_string(status.error_code()) + ": " + status.error_message();
+        log->error(err_msg);
+        throw Exception(err_msg, GRPCErrorCode);
     }
     return resp;
 }
@@ -100,7 +103,7 @@ void Client::initClusterID() {
         };
         std::this_thread::sleep_for(std::chrono::seconds(1));
     }
-    throw "failed to init cluster id";
+    throw Exception("failed to init cluster id", InitClusterIDFailed);
 }
 
 void Client::updateLeader() {
@@ -115,7 +118,7 @@ void Client::updateLeader() {
         switchLeader(resp.leader().client_urls());
         return;
     }
-    throw "failed to update leader";
+    throw Exception("failed to update leader", UpdatePDLeaderFailed);
 }
 
 void Client::switchLeader(const ::google::protobuf::RepeatedPtrField<std::string>& leader_urls) {
@@ -162,8 +165,8 @@ void Client::leaderLoop() {
             try {
                 check_leader = false;
                 updateLeader();
-            } catch (...) {
-                log->error("update leader failed.");
+            } catch (Exception & e) {
+                log->error(e.displayText());
             }
         }
     }
@@ -186,7 +189,9 @@ uint64_t Client::getGCSafePoint() {
 
     auto status = leaderStub()->GetGCSafePoint(&context, request, &response);
     if (!status.ok()) {
-        log->error("get safe point failed: " + std::to_string(status.error_code()) + ": " + status.error_message());
+        std::string err_msg = "get safe point failed: " + std::to_string(status.error_code()) + ": " + status.error_message();
+        log->error(err_msg);
+        throw Exception(err_msg, GRPCErrorCode);
     }
     return response.safe_point();
 }
@@ -204,7 +209,9 @@ std::tuple<metapb::Region, metapb::Peer, metapb::Peer> Client::getRegion(std::st
 
     auto status = leaderStub()->GetRegion(&context, request, &response);
     if (!status.ok()) {
-        log->error("get region failed: " + std::to_string(status.error_code()) + " : " + status.error_message());
+        std::string err_msg = ("get region failed: " + std::to_string(status.error_code()) + " : " + status.error_message());
+        log->error(err_msg);
+        throw Exception(err_msg, GRPCErrorCode);
     }
     if (response.slaves_size() == 0) {
         return std::make_tuple(response.region(), response.leader(), metapb::Peer::default_instance());
@@ -225,7 +232,9 @@ std::tuple<metapb::Region, metapb::Peer, metapb::Peer> Client::getRegionByID(uin
 
     auto status = leaderStub()->GetRegionByID(&context, request, &response);
     if (!status.ok()) {
-        log-> error("get region by id failed: " + std::to_string (status.error_code())  + ": " + status.error_message());
+        std::string err_msg = ("get region by id failed: " + std::to_string (status.error_code())  + ": " + status.error_message());
+        log->error(err_msg);
+        throw Exception(err_msg, GRPCErrorCode);
     }
     if (response.slaves_size() == 0) {
         return std::make_tuple(response.region(), response.leader(), metapb::Peer::default_instance());
@@ -246,7 +255,9 @@ metapb::Store Client::getStore(uint64_t store_id) {
 
     auto status = leaderStub()->GetStore(&context, request, &response);
     if (!status.ok()) {
-        log-> error("get store failed: " + std::to_string (status.error_code())  + ": " + status.error_message());
+        std::string err_msg = ("get store failed: " + std::to_string (status.error_code())  + ": " + status.error_message());
+        log->error(err_msg);
+        throw Exception(err_msg, GRPCErrorCode);
     }
     return response.store();
 }