[incubator-ratis] branch master updated: RATIS-590. Deadlock in ratis client. Contributed by Lokesh Jain

[szetszwo]

This is an automated email from the ASF dual-hosted git repository.

szetszwo pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-ratis.git


The following commit(s) were added to refs/heads/master by this push:
     new 5dc6eee  RATIS-590. Deadlock in ratis client.  Contributed by Lokesh 
Jain
5dc6eee is described below

commit 5dc6eeee0d4029facd5e7e09c2e51482fc4bfb46
Author: Tsz Wo Nicholas Sze <szets...@apache.org>
AuthorDate: Mon Jun 24 14:26:01 2019 +0800

    RATIS-590. Deadlock in ratis client.  Contributed by Lokesh Jain
---
 .../org/apache/ratis/client/impl/OrderedAsync.java | 13 +++++-------
 .../apache/ratis/client/impl/RaftClientImpl.java   | 23 +++++-----------------
 2 files changed, 10 insertions(+), 26 deletions(-)

diff --git 
a/ratis-client/src/main/java/org/apache/ratis/client/impl/OrderedAsync.java 
b/ratis-client/src/main/java/org/apache/ratis/client/impl/OrderedAsync.java
index 3800439..35681b8 100644
--- a/ratis-client/src/main/java/org/apache/ratis/client/impl/OrderedAsync.java
+++ b/ratis-client/src/main/java/org/apache/ratis/client/impl/OrderedAsync.java
@@ -31,7 +31,6 @@ import org.apache.ratis.protocol.RaftException;
 import org.apache.ratis.protocol.RaftPeerId;
 import org.apache.ratis.retry.RetryPolicies;
 import org.apache.ratis.retry.RetryPolicy;
-import org.apache.ratis.util.AutoCloseableLock;
 import org.apache.ratis.util.IOUtils;
 import org.apache.ratis.util.JavaUtils;
 import org.apache.ratis.util.Preconditions;
@@ -201,14 +200,12 @@ class OrderedAsync {
     final RetryPolicy retryPolicy = client.getRetryPolicy();
     final CompletableFuture<RaftClientReply> f;
     final RaftClientRequest request;
-    try (AutoCloseableLock readLock = client.readLock()) {
-      if (getSlidingWindow((RaftPeerId) null).isFirst(pending.getSeqNum())) {
-        pending.setFirstRequest();
-      }
-      request = pending.newRequest();
-      LOG.debug("{}: send* {}", client.getId(), request);
-      f = client.getClientRpc().sendRequestAsync(request);
+    if (getSlidingWindow((RaftPeerId) null).isFirst(pending.getSeqNum())) {
+      pending.setFirstRequest();
     }
+    request = pending.newRequest();
+    LOG.debug("{}: send* {}", client.getId(), request);
+    f = client.getClientRpc().sendRequestAsync(request);
     int attemptCount = pending.getAttemptCount();
     return f.thenApply(reply -> {
       LOG.debug("{}: receive* {}", client.getId(), reply);
diff --git 
a/ratis-client/src/main/java/org/apache/ratis/client/impl/RaftClientImpl.java 
b/ratis-client/src/main/java/org/apache/ratis/client/impl/RaftClientImpl.java
index 34bce9e..feed9d0 100644
--- 
a/ratis-client/src/main/java/org/apache/ratis/client/impl/RaftClientImpl.java
+++ 
b/ratis-client/src/main/java/org/apache/ratis/client/impl/RaftClientImpl.java
@@ -26,7 +26,6 @@ import org.apache.ratis.proto.RaftProtos.ReplicationLevel;
 import org.apache.ratis.proto.RaftProtos.SlidingWindowEntry;
 import org.apache.ratis.protocol.*;
 import org.apache.ratis.retry.RetryPolicy;
-import org.apache.ratis.util.AutoCloseableLock;
 import org.apache.ratis.util.CollectionUtils;
 import org.apache.ratis.util.JavaUtils;
 import org.apache.ratis.util.Preconditions;
@@ -42,7 +41,6 @@ import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ConcurrentLinkedQueue;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicLong;
-import java.util.concurrent.locks.ReentrantReadWriteLock;
 import java.util.function.Consumer;
 import java.util.function.Function;
 import java.util.function.Supplier;
@@ -85,7 +83,6 @@ final class RaftClientImpl implements RaftClient {
   private volatile RaftPeerId leaderId;
 
   private final TimeoutScheduler scheduler;
-  private final ReentrantReadWriteLock lock = new ReentrantReadWriteLock(true);
 
   private final Supplier<OrderedAsync> orderedAsync;
 
@@ -368,25 +365,15 @@ final class RaftClientImpl implements RaftClient {
             && newLeader != null && stillLeader;
     final boolean reconnect = changeLeader || clientRpc.shouldReconnect(ioe);
     if (reconnect) {
-      try(AutoCloseableLock writeLock = writeLock()) {
-        if (changeLeader && oldLeader.equals(leaderId)) {
-          LOG.debug("{} {}: client change Leader from {} to {} ex={}", groupId,
-              clientId, oldLeader, newLeader, ioe.getClass().getName());
-          this.leaderId = newLeader;
-        }
-        clientRpc.handleException(oldLeader, ioe, reconnect);
+      if (changeLeader && oldLeader.equals(leaderId)) {
+        LOG.debug("{} {}: client change Leader from {} to {} ex={}", groupId,
+            clientId, oldLeader, newLeader, ioe.getClass().getName());
+        this.leaderId = newLeader;
       }
+      clientRpc.handleException(oldLeader, ioe, reconnect);
     }
   }
 
-  AutoCloseableLock readLock() {
-    return AutoCloseableLock.acquire(lock.readLock());
-  }
-
-  private AutoCloseableLock writeLock() {
-    return AutoCloseableLock.acquire(lock.writeLock());
-  }
-
   void assertScheduler(int numThreads) {
     Preconditions.assertTrue(scheduler.getNumThreads() == numThreads);
   }