[FLASH-332] Add backoff to getting gc safe point from pd

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

@@ -9,7 +9,7 @@
 #include <common/CltException.h>
 
 namespace pingcap {
-namespace kv {
+namespace common {
 
 enum Jitter {
     NoJitter = 1,

contrib/client-c/include/pd/Client.h

@@ -7,6 +7,7 @@
 #include <condition_variable>
 #include <kvproto/pdpb.grpc.pb.h>
 #include <common/Log.h>
+#include <common/Backoff.h>
 #include "IClient.h"
 
 namespace pingcap{
@@ -27,6 +28,8 @@ class Client : public IClient {
 
     const std::chrono::seconds update_leader_interval;
 
+    const size_t get_gc_safe_point_timeout; // ms
+
 public:
 
     Client(const std::vector<std::string> & addrs);

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

@@ -6,7 +6,7 @@
 #include <pd/Client.h>
 #include <pd/MockPDClient.h>
 #include <kvproto/metapb.pb.h>
-#include <tikv/Backoff.h>
+#include <common/Backoff.h>
 #include <common/Log.h>
 #include <kvproto/errorpb.pb.h>
 
@@ -123,6 +123,7 @@ struct RPCContext {
 };
 
 using RPCContextPtr = std::shared_ptr<RPCContext>;
+using Backoffer = common::Backoffer;
 
 class RegionCache {
 public:

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

@@ -2,7 +2,7 @@
 
 #include<tikv/Rpc.h>
 #include<tikv/Region.h>
-#include<tikv/Backoff.h>
+#include<common/Backoff.h>
 
 namespace pingcap {
 namespace kv {
@@ -19,7 +19,7 @@ struct RegionClient {
 
     int64_t getReadIndex() {
         auto request = new kvrpcpb::ReadIndexRequest();
-        Backoffer bo(readIndexMaxBackoff);
+        Backoffer bo(common::readIndexMaxBackoff);
         auto rpc_call = std::make_shared<RpcCall<kvrpcpb::ReadIndexRequest>>(request);
         sendReqToRegion(bo, rpc_call, true);
         return rpc_call -> getResp() -> read_index();
@@ -57,10 +57,10 @@ struct RegionClient {
             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"));
+                bo.backoff(common::boUpdateLeader, Exception("not leader"));
             } else {
                 cache -> dropRegion(rpc_ctx->region);
-                bo.backoff(boRegionMiss, Exception("not leader"));
+                bo.backoff(common::boRegionMiss, Exception("not leader"));
             }
             return;
         }
@@ -76,7 +76,7 @@ struct RegionClient {
         }
 
         if (err.has_server_is_busy()) {
-            bo.backoff(boServerBusy, Exception("server busy"));
+            bo.backoff(common::boServerBusy, Exception("server busy"));
             return;
         }
 
@@ -93,12 +93,12 @@ struct RegionClient {
 
     void onGetLearnerFail(Backoffer & bo, const Exception & e) {
         log -> error("error found, retrying. The error msg is: "+ e.message());
-        bo.backoff(boTiKVRPC, e);
+        bo.backoff(common::boTiKVRPC, e);
     }
 
     void onSendFail(Backoffer & bo, const Exception & e, RPCContextPtr rpc_ctx) {
         cache->dropStoreOnSendReqFail(rpc_ctx, e);
-        bo.backoff(boTiKVRPC, e);
+        bo.backoff(common::boTiKVRPC, e);
     }
 };
 

contrib/client-c/src/CMakeLists.txt

@@ -3,7 +3,7 @@ set(kvClient_sources)
 list(APPEND kvClient_sources pd/Client.cc)
 list(APPEND kvClient_sources tikv/Region.cc)
 list(APPEND kvClient_sources tikv/RegionClient.cc)
-list(APPEND kvClient_sources tikv/Backoff.cc)
+list(APPEND kvClient_sources common/Backoff.cc)
 list(APPEND kvClient_sources tikv/Rpc.cc)
 
 set(kvClient_INCLUDE_DIR ${kvClient_SOURCE_DIR}/include)

contrib/client-c/src/tikv/Backoff.cc → contrib/client-c/src/common/Backoff.cc

@@ -1,8 +1,8 @@
-#include <tikv/Backoff.h>
+#include <common/Backoff.h>
 #include <common/CltException.h>
 
 namespace pingcap {
-namespace kv {
+namespace common {
 
 BackoffPtr newBackoff(BackoffType tp) {
     switch(tp) {

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

@@ -25,6 +25,7 @@ Client::Client(const std::vector<std::string> & addrs)
      pd_timeout(3),
      loop_interval(100),
      update_leader_interval(60),
+     get_gc_safe_point_timeout(20000),
      urls(addrsToUrls(addrs)),
      log(&Logger::get("pingcap.pd"))
 {
@@ -187,11 +188,12 @@ uint64_t Client::getGCSafePoint() {
     pdpb::GetGCSafePointRequest request{};
     pdpb::GetGCSafePointResponse response{};
     request.set_allocated_header(requestHeader());
-;
+
+    common::Backoffer bo(get_gc_safe_point_timeout);
     ::grpc::Status status;
     std::string err_msg;
 
-    for (int i = 0; i < max_init_cluster_retries; i++) {
+    while (true) {
         grpc::ClientContext context;
 
         context.set_deadline(std::chrono::system_clock::now() + pd_timeout);
@@ -202,10 +204,9 @@ uint64_t Client::getGCSafePoint() {
         err_msg = "get safe point failed: " + std::to_string(status.error_code()) + ": " + status.error_message();
         log->error(err_msg);
         check_leader.store(true);
-        usleep(100000);
-        // TODO retry outside.
+        bo.backoff(common::BackoffType::boPDRPC, Exception(err_msg, status.error_code()));
     }
-    throw Exception(err_msg, status.error_code());
+
 }
 
 std::tuple<metapb::Region, metapb::Peer, std::vector<metapb::Peer>> Client::getRegion(std::string key) {

contrib/client-c/src/tikv/Region.cc

@@ -70,7 +70,7 @@ RegionPtr RegionCache::loadRegionByID(Backoffer & bo, uint64_t region_id) {
             }
             return region;
         } catch (const Exception & e) {
-            bo.backoff(boPDRPC, e);
+            bo.backoff(common::boPDRPC, e);
         }
     }
 }
@@ -103,7 +103,7 @@ RegionPtr RegionCache::loadRegion(Backoffer & bo, std::string key) {
             }
             return region;
         } catch (const Exception & e) {
-            bo.backoff(boPDRPC, e);
+            bo.backoff(common::boPDRPC, e);
         }
     }
 }
@@ -114,7 +114,7 @@ metapb::Store RegionCache::loadStore(Backoffer & bo, uint64_t id) {
             const auto & store = pdClient->getStore(id);
             return store;
         } catch (Exception & e) {
-            bo.backoff(boPDRPC, e);
+            bo.backoff(common::boPDRPC, e);
         }
     }
 }