You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ratis.apache.org by lj...@apache.org on 2019/10/24 07:18:32 UTC
[incubator-ratis] branch master updated: RATIS-704. Invoke
sendAsync as soon as OrderedAsync is created. Contributed by Tsz Wo
Nicholas Sze.
This is an automated email from the ASF dual-hosted git repository.
ljain 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 9c1638d RATIS-704. Invoke sendAsync as soon as OrderedAsync is created. Contributed by Tsz Wo Nicholas Sze.
9c1638d is described below
commit 9c1638dbe6ef31246d780331b0cc8d27a8dc0c60
Author: Lokesh Jain <lj...@apache.org>
AuthorDate: Thu Oct 24 12:48:08 2019 +0530
RATIS-704. Invoke sendAsync as soon as OrderedAsync is created. Contributed by Tsz Wo Nicholas Sze.
---
.../apache/ratis/client/RaftClientConfigKeys.java | 15 ++++++++++++-
.../org/apache/ratis/client/impl/OrderedAsync.java | 26 ++++++++++++++++------
.../apache/ratis/client/impl/RaftClientImpl.java | 2 +-
.../apache/ratis/protocol/RaftClientRequest.java | 5 +++++
.../java/org/apache/ratis/util/PeerProxyMap.java | 1 +
.../apache/ratis/grpc/client/GrpcClientRpc.java | 2 +-
.../org/apache/ratis/RaftAsyncExceptionTests.java | 20 ++++++++++-------
.../test/java/org/apache/ratis/RaftAsyncTests.java | 14 +++++++++---
.../test/java/org/apache/ratis/RaftBasicTests.java | 5 +++--
9 files changed, 67 insertions(+), 23 deletions(-)
diff --git a/ratis-client/src/main/java/org/apache/ratis/client/RaftClientConfigKeys.java b/ratis-client/src/main/java/org/apache/ratis/client/RaftClientConfigKeys.java
index bb01249..d2c3679 100644
--- a/ratis-client/src/main/java/org/apache/ratis/client/RaftClientConfigKeys.java
+++ b/ratis-client/src/main/java/org/apache/ratis/client/RaftClientConfigKeys.java
@@ -1,4 +1,4 @@
-/**
+/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
@@ -62,6 +62,19 @@ public interface RaftClientConfigKeys {
static void setMaxOutstandingRequests(RaftProperties properties, int outstandingRequests) {
setInt(properties::setInt, MAX_OUTSTANDING_REQUESTS_KEY, outstandingRequests);
}
+
+ interface Experimental {
+ String PREFIX = Async.PREFIX + "." + Experimental.class.getSimpleName().toLowerCase();
+
+ String SEND_DUMMY_REQUEST_KEY = PREFIX + ".send-dummy-request";
+ boolean SEND_DUMMY_REQUEST_DEFAULT = true;
+ static boolean sendDummyRequest(RaftProperties properties) {
+ return getBoolean(properties::getBoolean, SEND_DUMMY_REQUEST_KEY, SEND_DUMMY_REQUEST_DEFAULT, getDefaultLog());
+ }
+ static void setSendDummyRequest(RaftProperties properties, boolean sendDummyRequest) {
+ setBoolean(properties::setBoolean, SEND_DUMMY_REQUEST_KEY, sendDummyRequest);
+ }
+ }
}
static void main(String[] args) {
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 7694450..79ee050 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
@@ -53,8 +53,8 @@ import java.util.function.Function;
import java.util.function.LongFunction;
/** Send ordered asynchronous requests to a raft service. */
-class OrderedAsync {
- private static final Logger LOG = LoggerFactory.getLogger(OrderedAsync.class);
+public final class OrderedAsync {
+ public static final Logger LOG = LoggerFactory.getLogger(OrderedAsync.class);
static class PendingOrderedRequest extends PendingClientRequest
implements SlidingWindow.ClientSideRequest<RaftClientReply> {
@@ -110,13 +110,23 @@ class OrderedAsync {
}
}
+ static OrderedAsync newInstance(RaftClientImpl client, RaftProperties properties) {
+ final OrderedAsync ordered = new OrderedAsync(client, properties);
+ // send a dummy watch request to establish the connection
+ // TODO: this is a work around, it is better to fix the underlying RPC implementation
+ if (RaftClientConfigKeys.Async.Experimental.sendDummyRequest(properties)) {
+ ordered.send(RaftClientRequest.watchRequestType(), null, null);
+ }
+ return ordered;
+ }
+
private final RaftClientImpl client;
/** Map: id -> {@link SlidingWindow}, in order to support async calls to the Raft service or individual servers. */
private final ConcurrentMap<String, SlidingWindow.Client<PendingOrderedRequest, RaftClientReply>> slidingWindows
= new ConcurrentHashMap<>();
private final Semaphore requestSemaphore;
- OrderedAsync(RaftClientImpl client, RaftProperties properties) {
+ private OrderedAsync(RaftClientImpl client, RaftProperties properties) {
this.client = Objects.requireNonNull(client, "client == null");
this.requestSemaphore = new Semaphore(RaftClientConfigKeys.Async.maxOutstandingRequests(properties));
}
@@ -170,6 +180,7 @@ class OrderedAsync {
final RaftClientRequest request = pending.newRequestImpl();
if (request == null) { // already done
+ LOG.debug("{} newRequestImpl returns null", pending);
return;
}
@@ -198,14 +209,15 @@ class OrderedAsync {
private void scheduleWithTimeout(PendingOrderedRequest pending, RaftClientRequest request, RetryPolicy retryPolicy) {
final int attempt = pending.getAttemptCount();
- LOG.debug("schedule* attempt #{} with policy {} for {}", attempt, retryPolicy, request);
final TimeDuration sleepTime = retryPolicy.getSleepTime(attempt, request);
- scheduleWithTimeout(pending, request.getServerId(), sleepTime);
+ LOG.debug("schedule* attempt #{} with sleep {} and policy {} for {}", attempt, sleepTime, retryPolicy, request);
+ scheduleWithTimeout(pending, sleepTime, getSlidingWindow(request));
}
- private void scheduleWithTimeout(PendingOrderedRequest pending, RaftPeerId serverId, TimeDuration sleepTime) {
+ private void scheduleWithTimeout(PendingOrderedRequest pending, TimeDuration sleepTime,
+ SlidingWindow.Client<PendingOrderedRequest, RaftClientReply> slidingWindow) {
client.getScheduler().onTimeout(sleepTime,
- () -> getSlidingWindow(serverId).retry(pending, this::sendRequestWithRetry),
+ () -> slidingWindow.retry(pending, this::sendRequestWithRetry),
LOG, () -> "Failed* to retry " + pending);
}
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 582b6b6..e1df86b 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
@@ -99,7 +99,7 @@ final class RaftClientImpl implements RaftClient {
scheduler = TimeoutScheduler.getInstance();
clientRpc.addServers(peers);
- this.orderedAsync = JavaUtils.memoize(() -> new OrderedAsync(this, properties));
+ this.orderedAsync = JavaUtils.memoize(() -> OrderedAsync.newInstance(this, properties));
}
@Override
diff --git a/ratis-common/src/main/java/org/apache/ratis/protocol/RaftClientRequest.java b/ratis-common/src/main/java/org/apache/ratis/protocol/RaftClientRequest.java
index 4c10c0c..a9e5064 100644
--- a/ratis-common/src/main/java/org/apache/ratis/protocol/RaftClientRequest.java
+++ b/ratis-common/src/main/java/org/apache/ratis/protocol/RaftClientRequest.java
@@ -30,6 +30,8 @@ import static org.apache.ratis.proto.RaftProtos.RaftClientRequestProto.TypeCase.
*/
public class RaftClientRequest extends RaftClientMessage {
private static final Type WRITE_DEFAULT = new Type(WriteRequestTypeProto.getDefaultInstance());
+ private static final Type WATCH_DEFAULT = new Type(
+ WatchRequestTypeProto.newBuilder().setIndex(0L).setReplication(ReplicationLevel.MAJORITY).build());
private static final Type DEFAULT_READ = new Type(ReadRequestTypeProto.getDefaultInstance());
private static final Type DEFAULT_STALE_READ = new Type(StaleReadRequestTypeProto.getDefaultInstance());
@@ -47,6 +49,9 @@ public class RaftClientRequest extends RaftClientMessage {
: new Type(StaleReadRequestTypeProto.newBuilder().setMinIndex(minIndex).build());
}
+ public static Type watchRequestType() {
+ return WATCH_DEFAULT;
+ }
public static Type watchRequestType(long index, ReplicationLevel replication) {
return new Type(WatchRequestTypeProto.newBuilder().setIndex(index).setReplication(replication).build());
}
diff --git a/ratis-common/src/main/java/org/apache/ratis/util/PeerProxyMap.java b/ratis-common/src/main/java/org/apache/ratis/util/PeerProxyMap.java
index ea0c2f1..6327adb 100644
--- a/ratis-common/src/main/java/org/apache/ratis/util/PeerProxyMap.java
+++ b/ratis-common/src/main/java/org/apache/ratis/util/PeerProxyMap.java
@@ -98,6 +98,7 @@ public class PeerProxyMap<PROXY extends Closeable> implements Closeable {
}
public PROXY getProxy(RaftPeerId id) throws IOException {
+ Objects.requireNonNull(id, "id == null");
PeerAndProxy p = peers.get(id);
if (p == null) {
synchronized (resetLock) {
diff --git a/ratis-grpc/src/main/java/org/apache/ratis/grpc/client/GrpcClientRpc.java b/ratis-grpc/src/main/java/org/apache/ratis/grpc/client/GrpcClientRpc.java
index 5883598..1fceb1b 100644
--- a/ratis-grpc/src/main/java/org/apache/ratis/grpc/client/GrpcClientRpc.java
+++ b/ratis-grpc/src/main/java/org/apache/ratis/grpc/client/GrpcClientRpc.java
@@ -66,7 +66,7 @@ public class GrpcClientRpc extends RaftClientRpcWithProxy<GrpcClientProtocolClie
final GrpcClientProtocolClient proxy = getProxies().getProxy(serverId);
// Reuse the same grpc stream for all async calls.
return proxy.getOrderedStreamObservers().onNext(request);
- } catch (IOException e) {
+ } catch (Throwable e) {
return JavaUtils.completeExceptionally(e);
}
}
diff --git a/ratis-server/src/test/java/org/apache/ratis/RaftAsyncExceptionTests.java b/ratis-server/src/test/java/org/apache/ratis/RaftAsyncExceptionTests.java
index cdb0e6b..4e8f68e 100644
--- a/ratis-server/src/test/java/org/apache/ratis/RaftAsyncExceptionTests.java
+++ b/ratis-server/src/test/java/org/apache/ratis/RaftAsyncExceptionTests.java
@@ -17,9 +17,11 @@
*/
package org.apache.ratis;
-import org.apache.ratis.client.RaftClient;
+import org.apache.log4j.Level;
import org.apache.ratis.RaftTestUtil.SimpleMessage;
+import org.apache.ratis.client.RaftClient;
import org.apache.ratis.client.RaftClientConfigKeys;
+import org.apache.ratis.client.impl.OrderedAsync;
import org.apache.ratis.protocol.AlreadyClosedException;
import org.apache.ratis.protocol.GroupMismatchException;
import org.apache.ratis.protocol.RaftClientReply;
@@ -27,7 +29,7 @@ import org.apache.ratis.protocol.RaftGroup;
import org.apache.ratis.protocol.RaftGroupId;
import org.apache.ratis.statemachine.SimpleStateMachine4Testing;
import org.apache.ratis.statemachine.StateMachine;
-import org.apache.ratis.util.TimeDuration;
+import org.apache.ratis.util.LogUtils;
import org.junit.Assert;
import org.junit.Test;
@@ -36,20 +38,22 @@ import java.util.Iterator;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
-import java.util.concurrent.TimeUnit;
public abstract class RaftAsyncExceptionTests<CLUSTER extends MiniRaftCluster>
extends BaseTest
implements MiniRaftCluster.Factory.Get<CLUSTER> {
{
+ LogUtils.setLogLevel(OrderedAsync.LOG, Level.DEBUG);
getProperties().setClass(MiniRaftCluster.STATEMACHINE_CLASS_KEY,
SimpleStateMachine4Testing.class, StateMachine.class);
}
@Test
public void testGroupMismatchException() throws Exception {
+ RaftClientConfigKeys.Async.Experimental.setSendDummyRequest(getProperties(), false);
runWithNewCluster(1, this::runTestGroupMismatchException);
+ RaftClientConfigKeys.Async.Experimental.setSendDummyRequest(getProperties(), true);
}
private void runTestGroupMismatchException(CLUSTER cluster) throws Exception {
@@ -95,24 +99,24 @@ public abstract class RaftAsyncExceptionTests<CLUSTER extends MiniRaftCluster>
private void runTestTimeoutException(CLUSTER cluster) throws Exception {
// send a message to make sure the cluster is working
try(RaftClient client = cluster.createClient()) {
- client.send(new SimpleMessage("m0"));
+ final RaftClientReply reply = client.send(new SimpleMessage("m0"));
+ Assert.assertTrue(reply.isSuccess());
- RaftClientConfigKeys.Rpc.setRequestTimeout(properties.get(),
- TimeDuration.valueOf(3, TimeUnit.SECONDS));
+ RaftClientConfigKeys.Rpc.setRequestTimeout(properties.get(), ONE_SECOND);
// Block StartTransaction
cluster.getServers().stream()
.map(cluster::getRaftServerImpl)
.map(SimpleStateMachine4Testing::get)
.forEach(SimpleStateMachine4Testing::blockStartTransaction);
final CompletableFuture<RaftClientReply> replyFuture = client.sendAsync(new SimpleMessage("m1"));
- Thread.sleep(10000);
+ FIVE_SECONDS.sleep();
// Unblock StartTransaction
cluster.getServers().stream()
.map(cluster::getRaftServerImpl)
.map(SimpleStateMachine4Testing::get)
.forEach(SimpleStateMachine4Testing::unblockStartTransaction);
// The request should succeed after start transaction is unblocked
- Assert.assertTrue(replyFuture.get().isSuccess());
+ Assert.assertTrue(replyFuture.get(FIVE_SECONDS.getDuration(), FIVE_SECONDS.getUnit()).isSuccess());
}
}
}
diff --git a/ratis-server/src/test/java/org/apache/ratis/RaftAsyncTests.java b/ratis-server/src/test/java/org/apache/ratis/RaftAsyncTests.java
index 4ab5282..8760147 100644
--- a/ratis-server/src/test/java/org/apache/ratis/RaftAsyncTests.java
+++ b/ratis-server/src/test/java/org/apache/ratis/RaftAsyncTests.java
@@ -105,12 +105,18 @@ public abstract class RaftAsyncTests<CLUSTER extends MiniRaftCluster> extends Ba
@Test
public void testRequestAsyncWithRetryFailure() throws Exception {
- runWithNewCluster(1, false, cluster -> runTestRequestAsyncWithRetryFailure(false, cluster));
+ runTestRequestAsyncWithRetryFailure(false);
}
@Test
public void testRequestAsyncWithRetryFailureAfterInitialMessages() throws Exception {
- runWithNewCluster(1, true, cluster -> runTestRequestAsyncWithRetryFailure(true, cluster));
+ runTestRequestAsyncWithRetryFailure(true);
+ }
+
+ void runTestRequestAsyncWithRetryFailure(boolean initialMessages) throws Exception {
+ RaftClientConfigKeys.Async.Experimental.setSendDummyRequest(getProperties(), false);
+ runWithNewCluster(1, initialMessages, cluster -> runTestRequestAsyncWithRetryFailure(initialMessages, cluster));
+ RaftClientConfigKeys.Async.Experimental.setSendDummyRequest(getProperties(), true);
}
void runTestRequestAsyncWithRetryFailure(boolean initialMessages, CLUSTER cluster) throws Exception {
@@ -345,11 +351,13 @@ public abstract class RaftAsyncTests<CLUSTER extends MiniRaftCluster> extends Ba
@Test
public void testRequestTimeout() throws Exception {
final TimeDuration oldExpiryTime = RaftServerConfigKeys.RetryCache.expiryTime(getProperties());
- RaftServerConfigKeys.RetryCache.setExpiryTime(getProperties(), TimeDuration.valueOf(5, TimeUnit.SECONDS));
+ RaftServerConfigKeys.RetryCache.setExpiryTime(getProperties(), FIVE_SECONDS);
+ RaftClientConfigKeys.Async.Experimental.setSendDummyRequest(getProperties(), false);
runWithNewCluster(NUM_SERVERS, cluster -> RaftBasicTests.testRequestTimeout(true, cluster, LOG));
//reset for the other tests
RaftServerConfigKeys.RetryCache.setExpiryTime(getProperties(), oldExpiryTime);
+ RaftClientConfigKeys.Async.Experimental.setSendDummyRequest(getProperties(), true);
}
@Test
diff --git a/ratis-server/src/test/java/org/apache/ratis/RaftBasicTests.java b/ratis-server/src/test/java/org/apache/ratis/RaftBasicTests.java
index a3cab64..f07eb7c 100644
--- a/ratis-server/src/test/java/org/apache/ratis/RaftBasicTests.java
+++ b/ratis-server/src/test/java/org/apache/ratis/RaftBasicTests.java
@@ -37,6 +37,7 @@ import org.apache.ratis.util.ExitUtils;
import org.apache.ratis.util.JavaUtils;
import org.apache.ratis.util.LogUtils;
import org.apache.ratis.util.TimeDuration;
+import org.apache.ratis.util.Timestamp;
import org.junit.Assert;
import org.junit.Test;
import org.slf4j.Logger;
@@ -407,7 +408,7 @@ public abstract class RaftBasicTests<CLUSTER extends MiniRaftCluster>
public static void testRequestTimeout(boolean async, MiniRaftCluster cluster, Logger LOG) throws Exception {
waitForLeader(cluster);
- long time = System.currentTimeMillis();
+ final Timestamp startTime = Timestamp.currentTime();
try (final RaftClient client = cluster.createClient()) {
// Get the next callId to be used by the client
long callId = RaftClientTestUtil.getCallId(client);
@@ -428,7 +429,7 @@ public abstract class RaftBasicTests<CLUSTER extends MiniRaftCluster>
// Eventually the request would be accepted by the server
// when the retry cache entry is invalidated.
// The duration for which the client waits should be more than the retryCacheExpiryDuration.
- TimeDuration duration = TimeDuration.valueOf(System.currentTimeMillis() - time, TimeUnit.MILLISECONDS);
+ final TimeDuration duration = startTime.elapsedTime();
TimeDuration retryCacheExpiryDuration = RaftServerConfigKeys.RetryCache.expiryTime(cluster.getProperties());
Assert.assertTrue(duration.compareTo(retryCacheExpiryDuration) >= 0);
}