Remove Synchronized Statements(Loom-friendly) (#173)

Motivation:
Synchronized statements tend to pin virtual threads to carrier threads.

Modification:
Replaced synchronized statements with ReentrantLock

Result:
Loom-friendly

Author: jchrys

src/main/java/io/asyncer/r2dbc/mysql/MySqlConnectionFactory.java

@@ -35,6 +35,7 @@
 import java.net.InetSocketAddress;
 import java.net.SocketAddress;
 import java.util.Objects;
+import java.util.concurrent.locks.ReentrantLock;
 import java.util.function.Predicate;
 
 import static io.asyncer.r2dbc.mysql.internal.util.AssertUtils.requireNonNull;
@@ -157,6 +158,8 @@ private static final class LazyQueryCache {
 
         private final int capacity;
 
+        private final ReentrantLock lock = new ReentrantLock();
+
         @Nullable
         private volatile QueryCache cache;
 
@@ -167,11 +170,14 @@ private LazyQueryCache(int capacity) {
         public QueryCache get() {
             QueryCache cache = this.cache;
             if (cache == null) {
-                synchronized (this) {
+                lock.lock();
+                try {
                     if ((cache = this.cache) == null) {
                         this.cache = cache = Caches.createQueryCache(capacity);
                     }
                     return cache;
+                } finally {
+                    lock.unlock();
                 }
             }
             return cache;

src/main/java/io/asyncer/r2dbc/mysql/cache/PrepareBoundedCache.java

@@ -17,10 +17,11 @@
 package io.asyncer.r2dbc.mysql.cache;
 
 import java.util.HashMap;
+import java.util.concurrent.locks.ReentrantLock;
 import java.util.function.IntConsumer;
 
 /**
- * A bounded implementation of {@link PrepareCache} that uses synchronized methods to ensure correctness, even
+ * A bounded implementation of {@link PrepareCache} that uses {@link ReentrantLock} to ensure correctness, even
  * it should not be used thread concurrently.
  */
 final class PrepareBoundedCache extends HashMap<String, Lru.Node<Integer>> implements PrepareCache {
@@ -33,6 +34,8 @@ final class PrepareBoundedCache extends HashMap<String, Lru.Node<Integer>> imple
 
     private final Lru<Integer> protection;
 
+    private final ReentrantLock lock = new ReentrantLock();
+
     PrepareBoundedCache(int capacity) {
         int windowSize = Math.max(1, capacity / 100);
         int protectionSize = Math.max(1, (int) ((capacity - windowSize) * 0.8));
@@ -45,29 +48,39 @@ final class PrepareBoundedCache extends HashMap<String, Lru.Node<Integer>> imple
     }
 
     @Override
-    public synchronized Integer getIfPresent(String key) {
-        Lru.Node<Integer> node = super.get(key);
-
-        if (node == null) {
-            return null;
+    public Integer getIfPresent(String key) {
+        lock.lock();
+        try {
+            Lru.Node<Integer> node = super.get(key);
+
+            if (node == null) {
+                return null;
+            }
+
+            drainRead(node);
+            return node.getValue();
+        } finally {
+            lock.unlock();
         }
-
-        drainRead(node);
-        return node.getValue();
     }
 
     @Override
-    public synchronized boolean putIfAbsent(String key, int value, IntConsumer evict) {
-        Lru.Node<Integer> wantAdd = new Lru.Node<>(key, value);
-        Lru.Node<Integer> present = super.putIfAbsent(key, wantAdd);
-
-        if (present == null) {
-            drainAdded(wantAdd, evict);
-            return true;
+    public boolean putIfAbsent(String key, int value, IntConsumer evict) {
+        lock.lock();
+        try {
+            Lru.Node<Integer> wantAdd = new Lru.Node<>(key, value);
+            Lru.Node<Integer> present = super.putIfAbsent(key, wantAdd);
+
+            if (present == null) {
+                drainAdded(wantAdd, evict);
+                return true;
+            }
+
+            drainRead(present);
+            return false;
+        } finally {
+            lock.unlock();
         }
-
-        drainRead(present);
-        return false;
     }
 
     @Override

src/main/java/io/asyncer/r2dbc/mysql/client/RequestQueue.java

@@ -21,6 +21,7 @@
 
 import java.util.Queue;
 import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
+import java.util.concurrent.locks.ReentrantLock;
 
 abstract class LeftPadding {
 
@@ -57,6 +58,8 @@ final class RequestQueue extends ActiveStatus implements Runnable {
 
     private final Queue<RequestTask<?>> queue = Queues.<RequestTask<?>>small().get();
 
+    private final ReentrantLock lock = new ReentrantLock();
+
     @Nullable
     private volatile RuntimeException disposed;
 
@@ -145,14 +148,17 @@ private RuntimeException requireDisposed() {
         RuntimeException disposed = this.disposed;
 
         if (disposed == null) {
-            synchronized (this) {
+            lock.lock();
+            try {
                 disposed = this.disposed;
 
                 if (disposed == null) {
                     this.disposed = disposed = new IllegalStateException("Request queue was disposed");
                 }
 
                 return disposed;
+            } finally {
+                lock.unlock();
             }
         }
 

src/main/java/io/asyncer/r2dbc/mysql/codec/DefaultCodecs.java

@@ -34,6 +34,7 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.locks.ReentrantLock;
 
 import static io.asyncer.r2dbc.mysql.internal.util.AssertUtils.requireNonNull;
 
@@ -319,16 +320,19 @@ static final class Builder implements CodecsBuilder {
 
         private final ByteBufAllocator allocator;
 
-        @GuardedBy("this")
+        @GuardedBy("lock")
         private final ArrayList<Codec<?>> codecs = new ArrayList<>();
 
+        private final ReentrantLock lock = new ReentrantLock();
+
         Builder(ByteBufAllocator allocator) {
             this.allocator = allocator;
         }
 
         @Override
         public CodecsBuilder addFirst(Codec<?> codec) {
-            synchronized (this) {
+            lock.lock();
+            try {
                 if (codecs.isEmpty()) {
                     Codec<?>[] defaultCodecs = defaultCodecs(allocator);
 
@@ -339,24 +343,30 @@ public CodecsBuilder addFirst(Codec<?> codec) {
                 } else {
                     codecs.add(0, codec);
                 }
+            } finally {
+                lock.unlock();
             }
             return this;
         }
 
         @Override
         public CodecsBuilder addLast(Codec<?> codec) {
-            synchronized (this) {
+            lock.lock();
+            try {
                 if (codecs.isEmpty()) {
                     codecs.addAll(InternalArrays.asImmutableList(defaultCodecs(allocator)));
                 }
                 codecs.add(codec);
+            } finally {
+                lock.unlock();
             }
             return this;
         }
 
         @Override
         public Codecs build() {
-            synchronized (this) {
+            lock.lock();
+            try {
                 try {
                     if (codecs.isEmpty()) {
                         return new DefaultCodecs(defaultCodecs(allocator));
@@ -366,6 +376,8 @@ public Codecs build() {
                     codecs.clear();
                     codecs.trimToSize();
                 }
+            } finally {
+                lock.unlock();
             }
         }
     }

src/main/java/io/asyncer/r2dbc/mysql/collation/LazyInitCharCollation.java

@@ -16,7 +16,10 @@
 
 package io.asyncer.r2dbc.mysql.collation;
 
+import org.jetbrains.annotations.Nullable;
+
 import java.nio.charset.Charset;
+import java.util.concurrent.locks.ReentrantLock;
 
 /**
  * Character collation those NOT use cached {@link CharsetTarget} of MySQL, it will be initialized and cached
@@ -26,6 +29,9 @@
  */
 final class LazyInitCharCollation extends AbstractCharCollation {
 
+    private final ReentrantLock lock = new ReentrantLock();
+
+    @Nullable
     private volatile Charset cached;
 
     LazyInitCharCollation(int id, String name, CharsetTarget target) {
@@ -37,7 +43,8 @@ public Charset getCharset() {
         Charset cached = this.cached;
 
         if (cached == null) {
-            synchronized (this) {
+            lock.lock();
+            try {
                 cached = this.cached;
 
                 if (cached == null) {
@@ -46,6 +53,8 @@ public Charset getCharset() {
                 }
 
                 return cached;
+            } finally {
+                lock.unlock();
             }
         }