inner-1898: release switch & inner-1897:prevent sync release hang

src/main/java/com/actiontech/dble/backend/mysql/nio/MySQLConnection.java

@@ -13,6 +13,7 @@ import com.actiontech.dble.backend.mysql.xa.TxState;
 import com.actiontech.dble.btrace.provider.XaDelayProvider;
 import com.actiontech.dble.config.Capabilities;
 import com.actiontech.dble.config.Isolations;
+import com.actiontech.dble.config.model.SystemConfig;
 import com.actiontech.dble.net.AbstractConnection;
 import com.actiontech.dble.net.NIOProcessor;
 import com.actiontech.dble.net.handler.BackEndCleaner;
@@ -560,8 +561,10 @@ public class MySQLConnection extends AbstractConnection implements
             return;
         }
 
+        this.setRowDataFlowing(false);
         if (recycler != null) {
             recycler.signal();
+            recycler = null;
         }
     }
 
@@ -785,7 +788,12 @@ public class MySQLConnection extends AbstractConnection implements
             if (logResponse.compareAndSet(false, true)) {
                 session.setBackendResponseEndTime(this);
             }
-            DbleServer.getInstance().getComplexQueryExecutor().execute(new BackEndRecycleRunnable(this));
+            SystemConfig systemConfig = DbleServer.getInstance().getConfig().getSystem();
+            if (systemConfig.getEnableAsyncRelease() == 1) {
+                DbleServer.getInstance().getComplexQueryExecutor().execute(new BackEndRecycleRunnable(this));
+            } else {
+                new BackEndRecycleRunnable(this).run();
+            }
             return;
         }
         complexQuery = false;

src/main/java/com/actiontech/dble/config/model/SystemConfig.java

@@ -169,6 +169,26 @@ public final class SystemConfig {
     private String routePenetrationRules = "";
     private boolean closeHeartBeatRecord = false;
 
+    private int enableAsyncRelease = 1;
+    //unit: ms
+    private long releaseTimeout = 10L;
+
+    public int getEnableAsyncRelease() {
+        return enableAsyncRelease;
+    }
+
+    public void setEnableAsyncRelease(int enableAsyncRelease) {
+        this.enableAsyncRelease = enableAsyncRelease;
+    }
+
+    public long getReleaseTimeout() {
+        return releaseTimeout;
+    }
+
+    public void setReleaseTimeout(long releaseTimeout) {
+        this.releaseTimeout = releaseTimeout;
+    }
+
 
     public SystemConfig(ProblemReporter problemReporter) {
         this.problemReporter = problemReporter;
@@ -1315,6 +1335,8 @@ public final class SystemConfig {
                 ", enableRoutePenetration=" + enableRoutePenetration +
                 ", routePenetrationRules='" + routePenetrationRules + '\'' +
                 ", closeHeartBeatRecord=" + closeHeartBeatRecord +
+                ", releaseTimeout=" + releaseTimeout +
+                ", enableAsyncRelease=" + enableAsyncRelease +
                 "]";
     }
 

src/main/java/com/actiontech/dble/net/handler/BackEndRecycleRunnable.java

@@ -5,7 +5,9 @@
 
 package com.actiontech.dble.net.handler;
 
+import com.actiontech.dble.DbleServer;
 import com.actiontech.dble.backend.mysql.nio.MySQLConnection;
+import com.actiontech.dble.config.model.SystemConfig;
 
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.locks.Condition;
@@ -31,27 +33,30 @@ public class BackEndRecycleRunnable implements Runnable, BackEndCleaner {
         if (backendConnection.isClosed()) {
             return;
         }
+        boolean awaitTimeout = false;
 
         try {
             lock.lock();
             try {
                 if (backendConnection.isRowDataFlowing()) {
-
-                    if (!condRelease.await(10, TimeUnit.MILLISECONDS)) {
-                        if (!backendConnection.isClosed()) {
-                            backendConnection.close("recycle time out");
-                        }
-                    } else {
-                        backendConnection.release();
+                    SystemConfig systemConfig = DbleServer.getInstance().getConfig().getSystem();
+                    if (!condRelease.await(systemConfig.getReleaseTimeout(), TimeUnit.MILLISECONDS)) {
+                        awaitTimeout = true;
                     }
-                } else {
-                    backendConnection.release();
                 }
             } catch (Exception e) {
                 backendConnection.close("recycle exception");
             } finally {
                 lock.unlock();
             }
+            if (backendConnection.isClosed()) {
+                return;
+            }
+            if (awaitTimeout) {
+                backendConnection.close("recycle time out");
+            } else {
+                backendConnection.release();
+            }
         } catch (Throwable e) {
             backendConnection.close("recycle exception");
         }
@@ -59,13 +64,14 @@ public class BackEndRecycleRunnable implements Runnable, BackEndCleaner {
 
 
     public void signal() {
-        if (lock.tryLock()) {
-            try {
-                condRelease.signal();
-            } finally {
-                lock.unlock();
-            }
+        lock.lock();
+        try {
+            backendConnection.setRowDataFlowing(false);
+            condRelease.signal();
+        } finally {
+            lock.unlock();
         }
+
     }
 
 }