fix: use ReentrantLock instead of synchronized in Connection.

Author: shedfreez

polaris-plugins/polaris-plugins-connector/connector-polaris-grpc/src/main/java/com/tencent/polaris/plugins/connector/grpc/Connection.java

@@ -28,6 +28,8 @@
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.locks.ReentrantLock;
+
 import org.slf4j.Logger;
 
 /**
@@ -74,7 +76,7 @@ public class Connection {
     /**
      * 申请锁
      */
-    private final Object lock = new Object();
+    private final ReentrantLock lock = new ReentrantLock();
 
     /**
      * 连接是否已经关闭
@@ -108,23 +110,40 @@ public static boolean isAvailableConnection(Connection connection) {
      */
     public boolean acquire(String opKey) {
         if (lazyDestroy.get()) {
+            LOG.warn("connection {}: acquired for op {} need destroy, curRef is {}", connID, opKey, ref.get());
             return false;
         }
-        synchronized (lock) {
-            if (lazyDestroy.get()) {
+        boolean locked = false;
+        try {
+            // retry outside.
+            locked = lock.tryLock(1, TimeUnit.SECONDS);
+            if (locked) {
+                if (lazyDestroy.get()) {
+                    return false;
+                }
+                int curRef = ref.incrementAndGet();
+                LOG.debug("connection {}: acquired for op {}, curRef is {}", connID, opKey, curRef);
+                return true;
+            } else {
+                LOG.warn("connection {}: acquired for op {} timeout, curRef is {}", connID, opKey, ref.get());
                 return false;
             }
-            int curRef = ref.incrementAndGet();
-            LOG.debug("connection {}: acquired for op {}, curRef is {}", connID, opKey, curRef);
-            return true;
+        } catch (Exception e) {
+            LOG.warn("connection {}: acquired for op {} occur exception, curRef is {}, msg:{}", connID, opKey, ref.get(), e.getMessage());
+            return false;
+        } finally {
+            if (locked) {
+                lock.unlock();
+            }
         }
     }
 
     /**
      * 关闭连接
      */
     public void closeConnection() {
-        synchronized (lock) {
+        lock.lock();
+        try {
             if (ref.get() <= 0 && !closed) {
                 LOG.info("connection {}: closed", connID);
                 closed = true;
@@ -155,6 +174,8 @@ public void closeConnection() {
                     LOG.debug("Success to gracefully shutdown connection: {}. ", connID);
                 }
             }
+        } finally {
+            lock.unlock();
         }
     }
 

polaris-plugins/polaris-plugins-connector/connector-polaris-grpc/src/main/java/com/tencent/polaris/plugins/connector/grpc/GrpcConnector.java

@@ -504,20 +504,34 @@ public void heartbeat(CommonProviderRequest req) throws PolarisException {
         ServiceKey serviceKey = new ServiceKey(req.getNamespace(), req.getService());
         long startTimestamp = 0L;
         try {
+            long t1 = System.currentTimeMillis();
             waitDiscoverReady();
+            long t2 = System.currentTimeMillis();
             connection = connectionManager
                     .getConnection(GrpcUtil.OP_KEY_INSTANCE_HEARTBEAT, ClusterType.HEALTH_CHECK_CLUSTER);
+            long t3 = System.currentTimeMillis();
             req.setTargetServer(connectionToTargetNode(connection));
+            long t4 = System.currentTimeMillis();
             PolarisGRPCGrpc.PolarisGRPCBlockingStub stub = PolarisGRPCGrpc.newBlockingStub(connection.getChannel());
+            long t5 = System.currentTimeMillis();
             stub = GrpcUtil.attachRequestHeader(stub, GrpcUtil.nextHeartbeatReqId());
+            long t6 = System.currentTimeMillis();
             stub = GrpcUtil.attachAccessToken(connectorConfig.getToken(), stub);
             startTimestamp = System.currentTimeMillis();
+            if (startTimestamp - t1 > 1000) {
+                LOG.warn("waitDiscoverReady cost {} ms, getConnection cost {} ms, connectionToTargetNode cost {} ms, newBlockingStub cost {} ms, attachRequestHeader cost {} ms, attachAccessToken cost {} ms",
+                        t2 - t1, t3 - t2, t4 - t3, t5 - t4, t6 - t5, startTimestamp - t6);
+            } else if (LOG.isDebugEnabled()) {
+                LOG.debug("waitDiscoverReady cost {} ms, getConnection cost {} ms, connectionToTargetNode cost {} ms, newBlockingStub cost {} ms, attachRequestHeader cost {} ms, attachAccessToken cost {} ms",
+                        t2 - t1, t3 - t2, t4 - t3, t5 - t4, t6 - t5, startTimestamp - t6);
+            }
             LOG.debug("start heartbeat at {} ms.", startTimestamp);
             ResponseProto.Response heartbeatResponse = stub.withDeadlineAfter(req.getTimeoutMs(),
                     TimeUnit.MILLISECONDS).heartbeat(buildHeartbeatRequest(req));
             GrpcUtil.checkResponse(heartbeatResponse);
             LOG.debug("received heartbeat response {}", heartbeatResponse);
         } catch (Throwable t) {
+            LOG.warn("heartbeat fail, req: {}, msg: {}", req, t.getMessage());
             if (t instanceof PolarisException) {
                 //服务端异常不进行重试
                 throw t;
@@ -531,7 +545,12 @@ public void heartbeat(CommonProviderRequest req) throws PolarisException {
                             req.getInstanceID(), req.getHost(), req.getPort(), serviceKey), t);
         } finally {
             long endTimestamp = System.currentTimeMillis();
-            LOG.debug("end heartbeat at {} ms. Diff {} ms", endTimestamp, endTimestamp - startTimestamp);
+            long cost = endTimestamp - startTimestamp;
+            if (cost > 1000) {
+                LOG.warn("end heartbeat at {} ms. Diff {} ms", endTimestamp, cost);
+            } else if (LOG.isDebugEnabled()) {
+                LOG.debug("end heartbeat at {} ms. Diff {} ms", endTimestamp, cost);
+            }
             if (null != connection) {
                 connection.release(GrpcUtil.OP_KEY_INSTANCE_HEARTBEAT);
             }