synchronized blocks were replaced with ReentrantLock in WriterImpl

Author: Igor Melnichenko

topic/src/main/java/tech/ydb/topic/write/impl/WriterImpl.java

@@ -12,6 +12,7 @@
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.atomic.AtomicReference;
+import java.util.concurrent.locks.ReentrantLock;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -44,6 +45,7 @@ public abstract class WriterImpl extends GrpcStreamRetrier {
     private final AtomicReference<CompletableFuture<InitResult>> initResultFutureRef = new AtomicReference<>(null);
     // Messages that are waiting for being put into sending queue due to queue overflow
     private final Queue<IncomingMessage> incomingQueue = new LinkedList<>();
+    private final ReentrantLock incomingQueueLock = new ReentrantLock();
     // Messages that are currently encoding
     private final Queue<EnqueuedMessage> encodingMessages = new LinkedList<>();
     // Messages that are taken into send buffer, are already compressed and are waiting for being sent
@@ -99,7 +101,9 @@ private IncomingMessage(EnqueuedMessage message) {
     }
 
     public CompletableFuture<Void> tryToEnqueue(EnqueuedMessage message, boolean instant) {
-        synchronized (incomingQueue) {
+        incomingQueueLock.lock();
+        
+        try {
             if (currentInFlightCount >= settings.getMaxSendBufferMessagesCount()) {
                 if (instant) {
                     logger.info("[{}] Rejecting a message due to reaching message queue in-flight limit of {}", id,
@@ -137,10 +141,12 @@ public CompletableFuture<Void> tryToEnqueue(EnqueuedMessage message, boolean ins
             IncomingMessage incomingMessage = new IncomingMessage(message);
             incomingQueue.add(incomingMessage);
             return incomingMessage.future;
+        } finally {
+            incomingQueueLock.unlock();
         }
     }
 
-    // should be done under synchronized incomingQueue
+    // should be done under incomingQueueLock
     private void acceptMessageIntoSendingQueue(EnqueuedMessage message) {
         this.lastAcceptedMessageFuture = message.getFuture();
         this.currentInFlightCount++;
@@ -187,7 +193,9 @@ private void moveEncodedMessagesToSendingQueue() {
         boolean haveNewMessagesToSend = false;
         // Working with encodingMessages under synchronized incomingQueue to prevent deadlocks
         // while working with free method
-        synchronized (incomingQueue) {
+        incomingQueueLock.lock();
+        
+        try {
             // Taking all encoded messages to sending queue
             while (true) {
                 EnqueuedMessage encodedMessage = encodingMessages.peek();
@@ -217,6 +225,8 @@ private void moveEncodedMessagesToSendingQueue() {
                     break;
                 }
             }
+        } finally {
+            incomingQueueLock.unlock();
         }
         if (haveNewMessagesToSend) {
             session.sendDataRequestIfNeeded();
@@ -264,15 +274,21 @@ protected CompletableFuture<Void> flushImpl() {
         if (this.lastAcceptedMessageFuture == null) {
             return CompletableFuture.completedFuture(null);
         }
-        synchronized (incomingQueue) {
+        incomingQueueLock.lock();
+        
+        try {
             return this.lastAcceptedMessageFuture.isDone()
                     ? CompletableFuture.completedFuture(null)
                     : this.lastAcceptedMessageFuture.thenApply(v -> null);
+        } finally {
+            incomingQueueLock.unlock();
         }
     }
 
     private void free(int messageCount, long sizeBytes) {
-        synchronized (incomingQueue) {
+        incomingQueueLock.lock();
+        
+        try {
             currentInFlightCount -= messageCount;
             availableSizeBytes += sizeBytes;
             if (logger.isTraceEnabled()) {
@@ -301,6 +317,8 @@ private void free(int messageCount, long sizeBytes) {
                 }
                 logger.trace("[{}] All messages from incomingQueue are accepted into send buffer", id);
             }
+        } finally {
+            incomingQueueLock.unlock();
         }
     }