Repository: incubator-ratis Updated Branches: refs/heads/master 09bb6cd16 -> fef6574c2
RATIS-236. Use TimeoutScheduler in RaftClientImpl. Project: http://git-wip-us.apache.org/repos/asf/incubator-ratis/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-ratis/commit/fef6574c Tree: http://git-wip-us.apache.org/repos/asf/incubator-ratis/tree/fef6574c Diff: http://git-wip-us.apache.org/repos/asf/incubator-ratis/diff/fef6574c Branch: refs/heads/master Commit: fef6574c2887d99d1f04618ade6dec1c67eeb6bd Parents: 09bb6cd Author: Tsz Sze <[email protected]> Authored: Fri May 4 10:55:45 2018 -0700 Committer: Tsz Sze <[email protected]> Committed: Fri May 4 10:55:45 2018 -0700 ---------------------------------------------------------------------- .../ratis/client/impl/RaftClientImpl.java | 13 +++---- .../org/apache/ratis/util/TimeoutScheduler.java | 38 +++++++++++++------- .../grpc/client/RaftClientProtocolClient.java | 6 ++-- .../ratis/grpc/server/GRpcLogAppender.java | 5 +-- .../java/org/apache/ratis/RaftAsyncTests.java | 22 ++++++++---- .../java/org/apache/ratis/RaftBasicTests.java | 20 ++++++++--- .../apache/ratis/util/TestTimeoutScheduler.java | 8 ++--- 7 files changed, 73 insertions(+), 39 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/incubator-ratis/blob/fef6574c/ratis-client/src/main/java/org/apache/ratis/client/impl/RaftClientImpl.java ---------------------------------------------------------------------- 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 44df5c3..eb78463 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 @@ -97,7 +97,7 @@ final class RaftClientImpl implements RaftClient { /** Map: id -> {@link SlidingWindow}, in order to support async calls to the RAFT service or individual servers. */ private final ConcurrentMap<String, SlidingWindow.Client<PendingAsyncRequest, RaftClientReply>> slidingWindows = new ConcurrentHashMap<>(); - private final ScheduledExecutorService scheduler; + private final TimeoutScheduler scheduler; private final Semaphore asyncRequestSemaphore; RaftClientImpl(ClientId clientId, RaftGroup group, RaftPeerId leaderId, @@ -111,7 +111,7 @@ final class RaftClientImpl implements RaftClient { this.retryInterval = RaftClientConfigKeys.Rpc.retryInterval(properties); asyncRequestSemaphore = new Semaphore(RaftClientConfigKeys.Async.maxOutstandingRequests(properties)); - scheduler = Executors.newScheduledThreadPool(RaftClientConfigKeys.Async.schedulerThreads(properties)); + scheduler = TimeoutScheduler.newInstance(RaftClientConfigKeys.Async.schedulerThreads(properties)); clientRpc.addServers(peers); } @@ -237,9 +237,10 @@ final class RaftClientImpl implements RaftClient { final CompletableFuture<RaftClientReply> f = pending.getReplyFuture(); return sendRequestAsync(request).thenCompose(reply -> { if (reply == null) { - final TimeUnit unit = retryInterval.getUnit(); - scheduler.schedule(() -> getSlidingWindow(request).retry(pending, this::sendRequestWithRetryAsync), - retryInterval.toLong(unit), unit); + LOG.debug("schedule a retry in {} for {}", retryInterval, request); + scheduler.onTimeout(retryInterval, + () -> getSlidingWindow(request).retry(pending, this::sendRequestWithRetryAsync), + LOG, () -> "Failed to retry " + request); } else { f.complete(reply); } @@ -383,7 +384,7 @@ final class RaftClientImpl implements RaftClient { } void assertScheduler(int numThreads) { - Preconditions.assertTrue(((ScheduledThreadPoolExecutor) scheduler).getCorePoolSize() == numThreads); + Preconditions.assertTrue(scheduler.getNumThreads() == numThreads); } long getCallId() { http://git-wip-us.apache.org/repos/asf/incubator-ratis/blob/fef6574c/ratis-common/src/main/java/org/apache/ratis/util/TimeoutScheduler.java ---------------------------------------------------------------------- diff --git a/ratis-common/src/main/java/org/apache/ratis/util/TimeoutScheduler.java b/ratis-common/src/main/java/org/apache/ratis/util/TimeoutScheduler.java index 7a7d16c..7007a53 100644 --- a/ratis-common/src/main/java/org/apache/ratis/util/TimeoutScheduler.java +++ b/ratis-common/src/main/java/org/apache/ratis/util/TimeoutScheduler.java @@ -22,6 +22,7 @@ import org.slf4j.LoggerFactory; import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ScheduledThreadPoolExecutor; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicReference; import java.util.function.Consumer; @@ -32,14 +33,10 @@ public final class TimeoutScheduler { private static final TimeDuration DEFAULT_GRACE_PERIOD = TimeDuration.valueOf(1, TimeUnit.MINUTES); - private static final Supplier<TimeoutScheduler> INSTANCE = JavaUtils.memoize(TimeoutScheduler::new); - - public static TimeoutScheduler getInstance() { - return INSTANCE.get(); + public static TimeoutScheduler newInstance(int numThreads) { + return new TimeoutScheduler(numThreads); } - private TimeoutScheduler() {} - /** When there is no tasks, the time period to wait before shutting down the scheduler. */ private final AtomicReference<TimeDuration> gracePeriod = new AtomicReference<>(DEFAULT_GRACE_PERIOD); @@ -47,7 +44,18 @@ public final class TimeoutScheduler { private int numTasks = 0; /** The scheduleID for each task */ private int scheduleID = 0; - private ScheduledExecutorService scheduler = null; + + private final int numThreads; + private volatile ScheduledExecutorService scheduler = null; + + private TimeoutScheduler(int numThreads) { + this.numThreads = numThreads; + } + + public int getNumThreads() { + final ScheduledExecutorService s = scheduler; + return s instanceof ScheduledThreadPoolExecutor? ((ScheduledThreadPoolExecutor)s).getCorePoolSize(): numThreads; + } TimeDuration getGracePeriod() { return gracePeriod.get(); @@ -86,13 +94,13 @@ public final class TimeoutScheduler { if (scheduler == null) { Preconditions.assertTrue(numTasks == 0); LOG.debug("Initialize scheduler"); - scheduler = Executors.newScheduledThreadPool(1); + scheduler = Executors.newScheduledThreadPool(numThreads); } numTasks++; final int sid = scheduleID++; LOG.debug("schedule a task: timeout {}, sid {}", timeout, sid); - scheduler.schedule(() -> toSchedule.accept(sid), timeout.getDuration(), timeout.getUnit()); + schedule(scheduler, () -> toSchedule.accept(sid), () -> "task #" + sid, timeout); } private synchronized void onTaskCompleted() { @@ -100,10 +108,15 @@ public final class TimeoutScheduler { final int sid = scheduleID; final TimeDuration grace = getGracePeriod(); LOG.debug("Schedule a shutdown task: grace {}, sid {}", grace, sid); - scheduler.schedule(() -> tryShutdownScheduler(sid), grace.getDuration(), grace.getUnit()); + schedule(scheduler, () -> tryShutdownScheduler(sid), () -> "shutdown task #" + sid, grace); } } + static void schedule(ScheduledExecutorService service, Runnable task, Supplier<String> name, TimeDuration timeDuration) { + service.schedule(LogUtils.newRunnable(LOG, task, name), + timeDuration.getDuration(), timeDuration.getUnit()); + } + private synchronized void tryShutdownScheduler(int sid) { if (sid == scheduleID) { // No new tasks submitted, shutdown the scheduler. @@ -116,8 +129,7 @@ public final class TimeoutScheduler { } /** When timeout, run the task. Log the error, if there is any. */ - public static <THROWABLE extends Throwable> void onTimeout( - TimeDuration timeout, CheckedRunnable<THROWABLE> task, Logger log, Supplier<String> errorMessage) { - getInstance().onTimeout(timeout, task, t -> log.error(errorMessage.get(), t)); + public void onTimeout(TimeDuration timeout, CheckedRunnable<?> task, Logger log, Supplier<String> errorMessage) { + onTimeout(timeout, task, t -> log.error(errorMessage.get(), t)); } } http://git-wip-us.apache.org/repos/asf/incubator-ratis/blob/fef6574c/ratis-grpc/src/main/java/org/apache/ratis/grpc/client/RaftClientProtocolClient.java ---------------------------------------------------------------------- diff --git a/ratis-grpc/src/main/java/org/apache/ratis/grpc/client/RaftClientProtocolClient.java b/ratis-grpc/src/main/java/org/apache/ratis/grpc/client/RaftClientProtocolClient.java index d01bbe8..2d095ab 100644 --- a/ratis-grpc/src/main/java/org/apache/ratis/grpc/client/RaftClientProtocolClient.java +++ b/ratis-grpc/src/main/java/org/apache/ratis/grpc/client/RaftClientProtocolClient.java @@ -48,7 +48,6 @@ import java.util.Map; import java.util.Optional; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicReference; import java.util.function.Consumer; import java.util.function.Supplier; @@ -59,7 +58,10 @@ public class RaftClientProtocolClient implements Closeable { private final Supplier<String> name; private final RaftPeer target; private final ManagedChannel channel; + private final TimeDuration requestTimeoutDuration; + private final TimeoutScheduler scheduler = TimeoutScheduler.newInstance(1); + private final RaftClientProtocolServiceBlockingStub blockingStub; private final RaftClientProtocolServiceStub asyncStub; private final AdminProtocolServiceBlockingStub adminBlockingStub; @@ -188,7 +190,7 @@ public class RaftClientProtocolClient implements Closeable { () -> getName() + ":" + getClass().getSimpleName()); try { requestStreamObserver.onNext(ClientProtoUtils.toRaftClientRequestProto(request)); - TimeoutScheduler.onTimeout(requestTimeoutDuration, () -> timeoutCheck(request), LOG, + scheduler.onTimeout(requestTimeoutDuration, () -> timeoutCheck(request), LOG, () -> "Timeout check failed for client request: " + request); } catch(Throwable t) { handleReplyFuture(request.getCallId(), future -> future.completeExceptionally(t)); http://git-wip-us.apache.org/repos/asf/incubator-ratis/blob/fef6574c/ratis-grpc/src/main/java/org/apache/ratis/grpc/server/GRpcLogAppender.java ---------------------------------------------------------------------- diff --git a/ratis-grpc/src/main/java/org/apache/ratis/grpc/server/GRpcLogAppender.java b/ratis-grpc/src/main/java/org/apache/ratis/grpc/server/GRpcLogAppender.java index cb18caa..9a2163c 100644 --- a/ratis-grpc/src/main/java/org/apache/ratis/grpc/server/GRpcLogAppender.java +++ b/ratis-grpc/src/main/java/org/apache/ratis/grpc/server/GRpcLogAppender.java @@ -48,7 +48,8 @@ public class GRpcLogAppender extends LogAppender { private long callId = 0; private volatile boolean firstResponseReceived = false; - private static TimeDuration requestTimeoutDuration; + private final TimeDuration requestTimeoutDuration; + private final TimeoutScheduler scheduler = TimeoutScheduler.newInstance(1); private volatile StreamObserver<AppendEntriesRequestProto> appendLogRequestObserver; @@ -172,7 +173,7 @@ public class GRpcLogAppender extends LogAppender { server.getId(), null, request); s.onNext(request); - TimeoutScheduler.onTimeout(requestTimeoutDuration, () -> timeoutAppendRequest(request), LOG, + scheduler.onTimeout(requestTimeoutDuration, () -> timeoutAppendRequest(request), LOG, () -> "Timeout check failed for append entry request: " + request); follower.updateLastRpcSendTime(); } http://git-wip-us.apache.org/repos/asf/incubator-ratis/blob/fef6574c/ratis-server/src/test/java/org/apache/ratis/RaftAsyncTests.java ---------------------------------------------------------------------- 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 d8909de..a343f76 100644 --- a/ratis-server/src/test/java/org/apache/ratis/RaftAsyncTests.java +++ b/ratis-server/src/test/java/org/apache/ratis/RaftAsyncTests.java @@ -63,8 +63,6 @@ public abstract class RaftAsyncTests<CLUSTER extends MiniRaftCluster> extends Ba public void setup() { getProperties().setClass(MiniRaftCluster.STATEMACHINE_CLASS_KEY, SimpleStateMachine4Testing.class, StateMachine.class); - TimeDuration retryCacheExpiryDuration = TimeDuration.valueOf(5, TimeUnit.SECONDS); - RaftServerConfigKeys.RetryCache.setExpiryTime(getProperties(), retryCacheExpiryDuration); } @Test @@ -144,12 +142,12 @@ public abstract class RaftAsyncTests<CLUSTER extends MiniRaftCluster> extends Ba cluster.shutdown(); } - void runTestBasicAppendEntriesAsync(ReplicationLevel replication) throws Exception { - final CLUSTER cluster = newCluster(NUM_SERVERS); + void runTestBasicAppendEntriesAsync(ReplicationLevel replication, boolean killLeader) throws Exception { + final CLUSTER cluster = newCluster(killLeader? 5: 3); try { cluster.start(); waitForLeader(cluster); - RaftBasicTests.runTestBasicAppendEntries(true, replication, 1000, cluster, LOG); + RaftBasicTests.runTestBasicAppendEntries(true, replication, killLeader, 1000, cluster, LOG); } finally { cluster.shutdown(); } @@ -157,12 +155,17 @@ public abstract class RaftAsyncTests<CLUSTER extends MiniRaftCluster> extends Ba @Test public void testBasicAppendEntriesAsync() throws Exception { - runTestBasicAppendEntriesAsync(ReplicationLevel.MAJORITY); + runTestBasicAppendEntriesAsync(ReplicationLevel.MAJORITY, false); + } + + @Test + public void testBasicAppendEntriesAsyncKillLeader() throws Exception { + runTestBasicAppendEntriesAsync(ReplicationLevel.MAJORITY, true); } @Test public void testBasicAppendEntriesAsyncWithAllReplication() throws Exception { - runTestBasicAppendEntriesAsync(ReplicationLevel.ALL); + runTestBasicAppendEntriesAsync(ReplicationLevel.ALL, false); } @Test @@ -253,10 +256,15 @@ 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)); final CLUSTER cluster = newCluster(NUM_SERVERS); cluster.start(); RaftBasicTests.testRequestTimeout(true, cluster, LOG); cluster.shutdown(); + + //reset for the other tests + RaftServerConfigKeys.RetryCache.setExpiryTime(getProperties(), oldExpiryTime); } @Test http://git-wip-us.apache.org/repos/asf/incubator-ratis/blob/fef6574c/ratis-server/src/test/java/org/apache/ratis/RaftBasicTests.java ---------------------------------------------------------------------- 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 64b3c7e..f35878c 100644 --- a/ratis-server/src/test/java/org/apache/ratis/RaftBasicTests.java +++ b/ratis-server/src/test/java/org/apache/ratis/RaftBasicTests.java @@ -98,12 +98,17 @@ public abstract class RaftBasicTests extends BaseTest { @Test public void testBasicAppendEntries() throws Exception { - runTestBasicAppendEntries(false, ReplicationLevel.MAJORITY, 10, getCluster(), LOG); + runTestBasicAppendEntries(false, ReplicationLevel.MAJORITY, false, 10, getCluster(), LOG); + } + + @Test + public void testBasicAppendEntriesKillLeader() throws Exception { + runTestBasicAppendEntries(false, ReplicationLevel.MAJORITY, true, 10, getCluster(), LOG); } @Test public void testBasicAppendEntriesWithAllReplication() throws Exception { - runTestBasicAppendEntries(false, ReplicationLevel.ALL, 10, getCluster(), LOG); + runTestBasicAppendEntries(false, ReplicationLevel.ALL, false, 10, getCluster(), LOG); } static void killAndRestartServer(RaftPeerId id, long killSleepMs, long restartSleepMs, MiniRaftCluster cluster, Logger LOG) { @@ -119,9 +124,10 @@ public abstract class RaftBasicTests extends BaseTest { } static void runTestBasicAppendEntries( - boolean async, ReplicationLevel replication, int numMessages, MiniRaftCluster cluster, Logger LOG) throws Exception { - LOG.info("runTestBasicAppendEntries: async? {}, replication={}, numMessages={}", - async, replication, numMessages); + boolean async, ReplicationLevel replication, boolean killLeader, int numMessages, MiniRaftCluster cluster, Logger LOG) + throws Exception { + LOG.info("runTestBasicAppendEntries: async? {}, replication={}, killLeader={}, numMessages={}", + async, replication, killLeader, numMessages); for (RaftServer s : cluster.getServers()) { cluster.restartServer(s.getId(), false); } @@ -129,6 +135,10 @@ public abstract class RaftBasicTests extends BaseTest { final long term = leader.getState().getCurrentTerm(); new Thread(() -> killAndRestartServer(cluster.getFollowers().get(0).getId(), 0, 1000, cluster, LOG)).start(); + if (killLeader) { + LOG.info("killAndRestart leader " + leader.getId()); + new Thread(() -> killAndRestartServer(leader.getId(), 2000, 4000, cluster, LOG)).start(); + } LOG.info(cluster.printServers()); http://git-wip-us.apache.org/repos/asf/incubator-ratis/blob/fef6574c/ratis-server/src/test/java/org/apache/ratis/util/TestTimeoutScheduler.java ---------------------------------------------------------------------- diff --git a/ratis-server/src/test/java/org/apache/ratis/util/TestTimeoutScheduler.java b/ratis-server/src/test/java/org/apache/ratis/util/TestTimeoutScheduler.java index 7c4ef4f..6a63569 100644 --- a/ratis-server/src/test/java/org/apache/ratis/util/TestTimeoutScheduler.java +++ b/ratis-server/src/test/java/org/apache/ratis/util/TestTimeoutScheduler.java @@ -46,7 +46,7 @@ public class TestTimeoutScheduler { @Test(timeout = 1000) public void testSingleTask() throws Exception { - final TimeoutScheduler scheduler = TimeoutScheduler.getInstance(); + final TimeoutScheduler scheduler = TimeoutScheduler.newInstance(1); final TimeDuration grace = TimeDuration.valueOf(100, TimeUnit.MILLISECONDS); scheduler.setGracePeriod(grace); Assert.assertFalse(scheduler.hasScheduler()); @@ -81,7 +81,7 @@ public class TestTimeoutScheduler { @Test(timeout = 1000) public void testMultipleTasks() throws Exception { - final TimeoutScheduler scheduler = TimeoutScheduler.getInstance(); + final TimeoutScheduler scheduler = TimeoutScheduler.newInstance(1); final TimeDuration grace = TimeDuration.valueOf(100, TimeUnit.MILLISECONDS); scheduler.setGracePeriod(grace); Assert.assertFalse(scheduler.hasScheduler()); @@ -127,7 +127,7 @@ public class TestTimeoutScheduler { @Test(timeout = 1000) public void testExtendingGracePeriod() throws Exception { - final TimeoutScheduler scheduler = TimeoutScheduler.getInstance(); + final TimeoutScheduler scheduler = TimeoutScheduler.newInstance(1); final TimeDuration grace = TimeDuration.valueOf(100, TimeUnit.MILLISECONDS); scheduler.setGracePeriod(grace); Assert.assertFalse(scheduler.hasScheduler()); @@ -177,7 +177,7 @@ public class TestTimeoutScheduler { @Test(timeout = 1000) public void testRestartingScheduler() throws Exception { - final TimeoutScheduler scheduler = TimeoutScheduler.getInstance(); + final TimeoutScheduler scheduler = TimeoutScheduler.newInstance(1); final TimeDuration grace = TimeDuration.valueOf(100, TimeUnit.MILLISECONDS); scheduler.setGracePeriod(grace); Assert.assertFalse(scheduler.hasScheduler());
