synchronized blocks were replaced with ReentrantLock in ReadStreamCall

Author: Igor Melnichenko

core/src/main/java/tech/ydb/core/impl/call/ReadStreamCall.java

@@ -3,6 +3,7 @@
 import java.util.concurrent.CancellationException;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.atomic.AtomicReference;
+import java.util.concurrent.locks.ReentrantLock;
 
 import javax.annotation.Nullable;
 
@@ -29,6 +30,7 @@ public class ReadStreamCall<ReqT, RespT> extends ClientCall.Listener<RespT> impl
 
     private final String traceId;
     private final ClientCall<ReqT, RespT> call;
+    private final ReentrantLock callLock = new ReentrantLock();
     private final GrpcStatusHandler statusConsumer;
     private final ReqT request;
     private final Metadata headers;
@@ -56,34 +58,40 @@ public CompletableFuture<Status> start(Observer<RespT> observer) {
             throw new IllegalStateException("Read stream call is already started");
         }
 
-        synchronized (call) {
+        callLock.lock();
+
+        try {
+            call.start(this, headers);
+            call.request(1);
+            if (logger.isTraceEnabled()) {
+                logger.trace("ReadStreamCall[{}] --> {}", traceId, TextFormat.shortDebugString((Message) request));
+            }
+            call.sendMessage(request);
+            // close stream by client side
+            call.halfClose();
+        } catch (Throwable t) {
             try {
-                call.start(this, headers);
-                call.request(1);
-                if (logger.isTraceEnabled()) {
-                    logger.trace("ReadStreamCall[{}] --> {}", traceId, TextFormat.shortDebugString((Message) request));
-                }
-                call.sendMessage(request);
-                // close stream by client side
-                call.halfClose();
-            } catch (Throwable t) {
-                try {
-                    call.cancel(null, t);
-                } catch (Throwable ex) {
-                    logger.error("ReadStreamCall[{}] got exception while canceling", traceId, ex);
-                }
-
-                statusFuture.completeExceptionally(t);
+                call.cancel(null, t);
+            } catch (Throwable ex) {
+                logger.error("ReadStreamCall[{}] got exception while canceling", traceId, ex);
             }
+
+            statusFuture.completeExceptionally(t);
+        } finally {
+            callLock.unlock();
         }
 
         return statusFuture;
     }
 
     @Override
     public void cancel() {
-        synchronized (call) {
+        callLock.lock();
+
+        try {
             call.cancel("Cancelled on user request", new CancellationException());
+        } finally {
+            callLock.unlock();
         }
     }
 
@@ -95,18 +103,23 @@ public void onMessage(RespT message) {
             }
             observerReference.get().onNext(message);
             // request delivery of the next inbound message.
-            synchronized (call) {
+            callLock.lock();
+
+            try {
                 call.request(1);
+            } finally {
+                callLock.unlock();
             }
         } catch (Exception ex) {
             statusFuture.completeExceptionally(ex);
+            callLock.lock();
 
             try {
-                synchronized (call) {
-                    call.cancel("Canceled by exception from observer", ex);
-                }
+                call.cancel("Canceled by exception from observer", ex);
             } catch (Throwable th) {
                 logger.error("ReadStreamCall[{}] got exception while canceling", traceId, th);
+            } finally {
+                callLock.unlock();
             }
         }
     }