Refactor Region Cache to reduce lock

src/main/java/org/tikv/common/ConfigUtils.java

@@ -55,10 +55,10 @@ public class ConfigUtils {
   public static final String TIKV_ENABLE_ATOMIC_FOR_CAS = "tikv.enable_atomic_for_cas";
 
   public static final String DEF_PD_ADDRESSES = "127.0.0.1:2379";
-  public static final String DEF_TIMEOUT = "150ms";
+  public static final String DEF_TIMEOUT = "300ms";
   public static final String DEF_FORWARD_TIMEOUT = "600ms";
   public static final String DEF_SCAN_TIMEOUT = "20s";
-  public static final int DEF_CHECK_HEALTH_TIMEOUT = 40;
+  public static final int DEF_CHECK_HEALTH_TIMEOUT = 100;
   public static final int DEF_SCAN_BATCH_SIZE = 10240;
   public static final int DEF_MAX_FRAME_SIZE = 268435456 * 2; // 256 * 2 MB
   public static final int DEF_INDEX_SCAN_BATCH_SIZE = 20000;

src/main/java/org/tikv/common/PDClient.java

@@ -93,6 +93,7 @@
 public class PDClient extends AbstractGRPCClient<PDBlockingStub, PDStub>
     implements ReadOnlyPDClient {
   private static final String TIFLASH_TABLE_SYNC_PROGRESS_PATH = "/tiflash/table/sync";
+  private static final long MIN_TRY_UPDATE_DURATION = 50;
   private final Logger logger = LoggerFactory.getLogger(PDClient.class);
   private RequestHeader header;
   private TsoRequest tsoReq;
@@ -103,6 +104,7 @@ public class PDClient extends AbstractGRPCClient<PDBlockingStub, PDStub>
   private Client etcdClient;
   private ConcurrentMap<Long, Double> tiflashReplicaMap;
   private HostMapping hostMapping;
+  private long lastUpdateLeaderTime;
 
   public static final Histogram PD_GET_REGION_BY_KEY_REQUEST_LATENCY =
       Histogram.build()
@@ -392,6 +394,9 @@ synchronized boolean createFollowerClientWrapper(String followerUrlStr, String l
   }
 
   public synchronized void updateLeaderOrforwardFollower() {
+    if (System.currentTimeMillis() - lastUpdateLeaderTime < MIN_TRY_UPDATE_DURATION) {
+      return;
+    }
     for (URI url : this.pdAddrs) {
       // since resp is null, we need update leader's address by walking through all pd server.
       GetMembersResponse resp = getMembers(url);
@@ -407,6 +412,7 @@ public synchronized void updateLeaderOrforwardFollower() {
 
       // if leader is switched, just return.
       if (checkHealth(leaderUrlStr, hostMapping) && trySwitchLeader(leaderUrlStr)) {
+        lastUpdateLeaderTime = System.currentTimeMillis();
         return;
       }
 
@@ -441,6 +447,7 @@ public synchronized void updateLeaderOrforwardFollower() {
         }
       }
     }
+    lastUpdateLeaderTime = System.currentTimeMillis();
     if (pdClientWrapper == null) {
       throw new TiClientInternalException(
           "already tried all address on file, but not leader found yet.");
@@ -470,6 +477,7 @@ public void tryUpdateLeader() {
         return;
       }
     }
+    lastUpdateLeaderTime = System.currentTimeMillis();
     if (pdClientWrapper == null) {
       throw new TiClientInternalException(
           "already tried all address on file, but not leader found yet.");

src/main/java/org/tikv/common/operation/RegionErrorHandler.java

@@ -36,7 +36,6 @@ public RegionErrorHandler(
   public boolean handleResponseError(BackOffer backOffer, RespT resp) {
     if (resp == null) {
       String msg = String.format("Request Failed with unknown reason for [%s]", recv.getRegion());
-      logger.warn(msg);
       return handleRequestError(backOffer, new GrpcException(msg));
     }
     // Region error handling logic
@@ -171,6 +170,7 @@ public boolean handleRequestError(BackOffer backOffer, Exception e) {
       return true;
     }
 
+    logger.warn("request failed because of: " + e.getMessage());
     backOffer.doBackOff(
         BackOffFunction.BackOffFuncType.BoTiKVRPC,
         new GrpcException(

src/main/java/org/tikv/common/policy/RetryPolicy.java

@@ -79,6 +79,8 @@ public RespT callWithRetry(Callable<RespT> proc, String methodName) {
         if (retry) {
           GRPC_REQUEST_RETRY_NUM.labels(methodName).inc();
           continue;
+        } else {
+          return result;
         }
       }
 

src/main/java/org/tikv/common/region/AbstractRegionStoreClient.java

@@ -105,6 +105,7 @@ public boolean onNotLeader(TiRegion newRegion) {
     }
     region = newRegion;
     targetStore = regionManager.getStoreById(region.getLeader().getStoreId());
+    originStore = null;
     String addressStr = targetStore.getStore().getAddress();
     ManagedChannel channel =
         channelFactory.getChannel(addressStr, regionManager.getPDClient().getHostMapping());
@@ -128,7 +129,7 @@ public boolean onStoreUnreachable() {
     } else if (retryTimes > region.getFollowerList().size()) {
       logger.warn(
           String.format(
-              "retry time exceed for region[%d], invalid this region and store[%d]",
+              "retry time exceed for region[%d], invalid this region[%d]",
               region.getId(), targetStore.getId()));
       regionManager.onRequestFail(region);
       return false;
@@ -139,6 +140,7 @@ public boolean onStoreUnreachable() {
           String.format(
               "no forward store can be selected for store [%s] and region[%d]",
               targetStore.getStore().getAddress(), region.getId()));
+      regionManager.onRequestFail(region);
       return false;
     }
     if (originStore == null) {
@@ -168,6 +170,10 @@ public boolean onStoreUnreachable() {
   @Override
   protected void tryUpdateProxy() {
     if (originStore != null) {
+      logger.warn(
+          String.format(
+              "update store [%s] by proxy-store [%s]",
+              targetStore.getStore().getAddress(), targetStore.getProxyStore().getAddress()));
       regionManager.updateStore(originStore, targetStore);
     }
   }