* thread: replace all synchronized with ReentrantLock/Condition, incl…

…uding RateControl/ShutdownHandling/Test

Signed-off-by: neo <1100909+neowu@users.noreply.github.com>

CHANGELOG.md

@@ -4,7 +4,7 @@
 
 * mysql: updated and patched to 8.3.0, fixed CJException should be wrapped as SQLException
   > make sure use "core.framework.mysql:mysql-connector-j:8.3.0-r2"
-* rate: RateControl, replace synchronized with ReentrantLock
+* thread: replace all synchronized with ReentrantLock/Condition, including RateControl/ShutdownHandling/Test
 
 ### 9.0.8 (1/29/2024 - 3/7/2024)
 

core-ng-mongo-test/src/main/java/core/framework/mongo/module/TestMongoConfig.java

@@ -7,11 +7,14 @@
 import de.bwaldvogel.mongo.backend.memory.MemoryBackend;
 
 import java.net.InetSocketAddress;
+import java.util.concurrent.locks.ReentrantLock;
 
 /**
  * @author neo
  */
 public class TestMongoConfig extends MongoConfig {
+    private static final ReentrantLock LOCK = new ReentrantLock();
+
     // only start one mongo server for testing to reduce resource overhead,
     // only breaking case is that multiple mongo() using same collection name, then if one unit test operates both MongoCollection may result in conflict or merged results
     // this can be avoided by designing test differently
@@ -27,13 +30,16 @@ protected void initialize(ModuleContext context, String name) {
     }
 
     private void startLocalMongoServer(ModuleContext context) {
-        synchronized (TestMongoConfig.class) {
+        LOCK.lock();
+        try {
             // in test env, config is initialized in order and within same thread, so no threading issue
             if (localMongoAddress == null) {
                 var server = new MongoServer(new MemoryBackend());
                 localMongoAddress = server.bind();
                 context.shutdownHook.add(ShutdownHook.STAGE_6, timeout -> server.shutdown());
             }
+        } finally {
+            LOCK.unlock();
         }
     }
 

core-ng-search-test/src/main/java/core/framework/search/module/TestSearchConfig.java

@@ -9,10 +9,14 @@
 import org.elasticsearch.common.logging.internal.LoggerFactoryImpl;
 import org.elasticsearch.logging.internal.spi.LoggerFactory;
 
+import java.util.concurrent.locks.ReentrantLock;
+
 /**
  * @author neo
  */
 public class TestSearchConfig extends SearchConfig {
+    private static final ReentrantLock LOCK = new ReentrantLock();
+
     // only start one local node for testing to reduce resource overhead,
     // only breaking case is that multiple search() using same index name, then if one unit test operates both ElasticSearchType will result in conflict or merged results
     // this can be avoided by designing test differently
@@ -26,13 +30,16 @@ protected void initialize(ModuleContext context, String name) {
     }
 
     private void startLocalElasticSearch(ModuleContext context) {
-        synchronized (TestSearchConfig.class) {
+        LOCK.lock();
+        try {
             // in test env, config is initialized in order and within same thread, so no threading issue
             if (localESHost == null) {
                 var server = new LocalElasticSearch();
                 localESHost = server.start();
                 context.shutdownHook.add(ShutdownHook.STAGE_6, timeout -> server.close());
             }
+        } finally {
+            LOCK.unlock();
         }
     }
 

core-ng/src/main/java/core/framework/internal/kafka/MessageListenerThread.java

@@ -29,6 +29,9 @@
 import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.Semaphore;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.Condition;
+import java.util.concurrent.locks.ReentrantLock;
 
 import static core.framework.log.Markers.errorCode;
 
@@ -46,9 +49,11 @@ class MessageListenerThread extends Thread {
     private final Semaphore semaphore;
     private final int concurrency;
 
-    private final Object lock = new Object();
+    private final ReentrantLock lock = new ReentrantLock();
+    private final Condition processingCondition = lock.newCondition();
+    private boolean processing;
+
     private volatile boolean shutdown;
-    private volatile boolean processing;
 
     MessageListenerThread(String name, Consumer<String, byte[]> consumer, MessageListener listener) {
         super(name);
@@ -67,9 +72,16 @@ public void run() {
             process();
         } finally {
             processing = false;
-            synchronized (lock) {
-                lock.notifyAll();
-            }
+            notifyProcessingCondition();
+        }
+    }
+
+    private void notifyProcessingCondition() {
+        lock.lock();
+        try {
+            processingCondition.notifyAll();
+        } finally {
+            lock.unlock();
         }
     }
 
@@ -120,15 +132,18 @@ void shutdown() {
 
     boolean awaitTermination(long timeoutInMs) throws InterruptedException {
         long end = System.currentTimeMillis() + timeoutInMs;
-        synchronized (lock) {
+        lock.lock();
+        try {
             while (processing) {
                 long left = end - System.currentTimeMillis();
                 if (left <= 0) {
                     return false;
                 }
-                lock.wait(left);
+                processingCondition.await(left, TimeUnit.MILLISECONDS);
             }
             return true;
+        } finally {
+            lock.unlock();
         }
     }
 

core-ng/src/main/java/core/framework/internal/web/ShutdownHandler.java

@@ -7,14 +7,21 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.Condition;
+import java.util.concurrent.locks.ReentrantLock;
+
 /**
  * @author neo
  */
 public class ShutdownHandler implements ExchangeCompletionListener {
     final Counter activeRequests = new Counter();
 
     private final Logger logger = LoggerFactory.getLogger(ShutdownHandler.class);
-    private final Object lock = new Object();
+
+    private final ReentrantLock lock = new ReentrantLock();
+    private final Condition activeRequestCondition = lock.newCondition();
+
     private volatile boolean shutdown;
 
     boolean handle(HttpServerExchange exchange) {
@@ -41,15 +48,18 @@ void shutdown() {
 
     boolean awaitTermination(long timeoutInMs) throws InterruptedException {
         long end = System.currentTimeMillis() + timeoutInMs;
-        synchronized (lock) {
+        lock.lock();
+        try {
             while (activeRequests.get() > 0) {
                 long left = end - System.currentTimeMillis();
                 if (left <= 0) {
                     return false;
                 }
-                lock.wait(left);
+                activeRequestCondition.await(left, TimeUnit.MILLISECONDS);
             }
             return true;
+        } finally {
+            lock.unlock();
         }
     }
 
@@ -58,12 +68,19 @@ public void exchangeEvent(HttpServerExchange exchange, NextListener next) {
         try {
             int count = activeRequests.decrease();
             if (count <= 0 && shutdown) {
-                synchronized (lock) {
-                    lock.notifyAll();
-                }
+                notifyActiveRequestCondition();
             }
         } finally {
             next.proceed();
         }
     }
+
+    private void notifyActiveRequestCondition() {
+        lock.lock();
+        try {
+            activeRequestCondition.signalAll();
+        } finally {
+            lock.unlock();
+        }
+    }
 }