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/ratis.git
The following commit(s) were added to refs/heads/master by this push:
new 8c9c801e5 RATIS-2080. Reuse LeaderElection executor. (#1082)
8c9c801e5 is described below
commit 8c9c801e5b8edc68971911e5819e89e74e114683
Author: Tsz-Wo Nicholas Sze <[email protected]>
AuthorDate: Fri May 10 08:09:20 2024 -0700
RATIS-2080. Reuse LeaderElection executor. (#1082)
---
.../apache/ratis/server/impl/LeaderElection.java | 53 ++++++++--------------
.../apache/ratis/server/impl/RaftServerImpl.java | 24 ++++++++--
2 files changed, 40 insertions(+), 37 deletions(-)
diff --git
a/ratis-server/src/main/java/org/apache/ratis/server/impl/LeaderElection.java
b/ratis-server/src/main/java/org/apache/ratis/server/impl/LeaderElection.java
index d738c8757..2f672c17e 100644
---
a/ratis-server/src/main/java/org/apache/ratis/server/impl/LeaderElection.java
+++
b/ratis-server/src/main/java/org/apache/ratis/server/impl/LeaderElection.java
@@ -42,15 +42,16 @@ import java.io.IOException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
+import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
-import java.util.concurrent.ExecutorCompletionService;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
+import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.Set;
import java.util.stream.Collectors;
@@ -130,28 +131,22 @@ class LeaderElection implements Runnable {
}
static class Executor {
- private final ExecutorCompletionService<RequestVoteReplyProto> service;
private final ExecutorService executor;
private final AtomicInteger count = new AtomicInteger();
Executor(Object name, int size) {
Preconditions.assertTrue(size > 0);
- executor = Executors.newFixedThreadPool(size, r ->
+ executor = Executors.newCachedThreadPool(r ->
Daemon.newBuilder().setName(name + "-" +
count.incrementAndGet()).setRunnable(r).build());
- service = new ExecutorCompletionService<>(executor);
}
- void shutdown() {
- executor.shutdown();
+ ExecutorService getExecutor() {
+ return executor;
}
- void submit(Callable<RequestVoteReplyProto> task) {
- service.submit(task);
- }
-
- Future<RequestVoteReplyProto> poll(TimeDuration waitTime) throws
InterruptedException {
- return service.poll(waitTime.getDuration(), waitTime.getUnit());
+ Future<RequestVoteReplyProto> submit(Callable<RequestVoteReplyProto> task)
{
+ return executor.submit(task);
}
}
@@ -290,13 +285,9 @@ class LeaderElection implements Runnable {
r = new ResultAndTerm(Result.PASSED, electionTerm);
} else {
final TermIndex lastEntry = server.getState().getLastEntry();
- final Executor voteExecutor = new Executor(this, others.size());
- try {
- final int submitted = submitRequests(phase, electionTerm, lastEntry,
others, voteExecutor);
- r = waitForResults(phase, electionTerm, submitted, conf, voteExecutor);
- } finally {
- voteExecutor.shutdown();
- }
+ final List<Future<RequestVoteReplyProto>> submitted = submitRequests(
+ phase, electionTerm, lastEntry, others,
server.getLeaderElectionExecutor());
+ r = waitForResults(phase, electionTerm, submitted, conf);
}
return r;
@@ -348,14 +339,13 @@ class LeaderElection implements Runnable {
}
}
- private int submitRequests(Phase phase, long electionTerm, TermIndex
lastEntry,
+ private List<Future<RequestVoteReplyProto>> submitRequests(Phase phase, long
electionTerm, TermIndex lastEntry,
Collection<RaftPeer> others, Executor voteExecutor) {
- int submitted = 0;
+ final List<Future<RequestVoteReplyProto>> submitted = new
ArrayList<>(others.size());
for (final RaftPeer peer : others) {
final RequestVoteRequestProto r =
ServerProtoUtils.toRequestVoteRequestProto(
server.getMemberId(), peer.getId(), electionTerm, lastEntry, phase
== Phase.PRE_VOTE);
- voteExecutor.submit(() -> server.getServerRpc().requestVote(r));
- submitted++;
+ submitted.add(voteExecutor.submit(() ->
server.getServerRpc().requestVote(r)));
}
return submitted;
}
@@ -369,18 +359,17 @@ class LeaderElection implements Runnable {
.collect(Collectors.toSet());
}
- private ResultAndTerm waitForResults(Phase phase, long electionTerm, int
submitted,
- RaftConfigurationImpl conf, Executor voteExecutor) throws
InterruptedException {
+ private ResultAndTerm waitForResults(Phase phase, long electionTerm,
List<Future<RequestVoteReplyProto>> submitted,
+ RaftConfigurationImpl conf) throws InterruptedException {
final Timestamp timeout =
Timestamp.currentTime().addTime(server.getRandomElectionTimeout());
final Map<RaftPeerId, RequestVoteReplyProto> responses = new HashMap<>();
final List<Exception> exceptions = new ArrayList<>();
- int waitForNum = submitted;
Collection<RaftPeerId> votedPeers = new ArrayList<>();
Collection<RaftPeerId> rejectedPeers = new ArrayList<>();
Set<RaftPeerId> higherPriorityPeers = getHigherPriorityPeers(conf);
final boolean singleMode = conf.isSingleMode(server.getId());
- while (waitForNum > 0 && shouldRun(electionTerm)) {
+ for(Iterator<Future<RequestVoteReplyProto>> i = submitted.iterator();
i.hasNext() && shouldRun(electionTerm); ) {
final TimeDuration waitTime = timeout.elapsedTime().apply(n -> -n);
if (waitTime.isNonPositive()) {
if (conf.hasMajority(votedPeers, server.getId())) {
@@ -396,12 +385,9 @@ class LeaderElection implements Runnable {
}
try {
- final Future<RequestVoteReplyProto> future =
voteExecutor.poll(waitTime);
- if (future == null) {
- continue; // poll timeout, continue to return Result.TIMEOUT
- }
+ final Future<RequestVoteReplyProto> future = i.next();
+ final RequestVoteReplyProto r = future.get(waitTime.getDuration(),
waitTime.getUnit());
- final RequestVoteReplyProto r = future.get();
final RaftPeerId replierId =
RaftPeerId.valueOf(r.getServerReply().getReplyId());
final RequestVoteReplyProto previous =
responses.putIfAbsent(replierId, r);
if (previous != null) {
@@ -444,8 +430,9 @@ class LeaderElection implements Runnable {
} catch(ExecutionException e) {
LogUtils.infoOrTrace(LOG, () -> this + " got exception when requesting
votes", e);
exceptions.add(e);
+ } catch (TimeoutException e) {
+ // get timeout, continue to return Result.TIMEOUT
}
- waitForNum--;
}
// received all the responses
if (conf.hasMajority(votedPeers, server.getId())) {
diff --git
a/ratis-server/src/main/java/org/apache/ratis/server/impl/RaftServerImpl.java
b/ratis-server/src/main/java/org/apache/ratis/server/impl/RaftServerImpl.java
index 4512a2c22..c6a27ba8c 100644
---
a/ratis-server/src/main/java/org/apache/ratis/server/impl/RaftServerImpl.java
+++
b/ratis-server/src/main/java/org/apache/ratis/server/impl/RaftServerImpl.java
@@ -243,6 +243,7 @@ class RaftServerImpl implements RaftServer.Division,
private final ExecutorService serverExecutor;
private final ExecutorService clientExecutor;
+ private final MemoizedSupplier<LeaderElection.Executor>
leaderElectionExecutor;
private final AtomicBoolean firstElectionSinceStartup = new
AtomicBoolean(true);
private final ThreadGroup threadGroup;
@@ -281,14 +282,17 @@ class RaftServerImpl implements RaftServer.Division,
this.snapshotRequestHandler = new SnapshotManagementRequestHandler(this);
this.snapshotInstallationHandler = new SnapshotInstallationHandler(this,
properties);
+ final RaftGroupMemberId memberId = getMemberId();
this.serverExecutor = ConcurrentUtils.newThreadPoolWithMax(
RaftServerConfigKeys.ThreadPool.serverCached(properties),
RaftServerConfigKeys.ThreadPool.serverSize(properties),
- id + "-server");
+ memberId + "-server");
this.clientExecutor = ConcurrentUtils.newThreadPoolWithMax(
RaftServerConfigKeys.ThreadPool.clientCached(properties),
RaftServerConfigKeys.ThreadPool.clientSize(properties),
- id + "-client");
+ memberId + "-client");
+ this.leaderElectionExecutor = MemoizedSupplier.valueOf(
+ () -> new LeaderElection.Executor(memberId + "-election" ,
group.getPeers().size()));
}
private long getCommitIndex(RaftPeerId id) {
@@ -534,12 +538,20 @@ class RaftServerImpl implements RaftServer.Division,
try {
ConcurrentUtils.shutdownAndWait(clientExecutor);
} catch (Exception e) {
- LOG.warn(getMemberId() + ": Failed to shutdown clientExecutor", e);
+ LOG.warn("{}: Failed to shutdown clientExecutor", getMemberId(), e);
}
try {
ConcurrentUtils.shutdownAndWait(serverExecutor);
} catch (Exception e) {
- LOG.warn(getMemberId() + ": Failed to shutdown serverExecutor", e);
+ LOG.warn("{}: Failed to shutdown serverExecutor", getMemberId(), e);
+ }
+
+ if (leaderElectionExecutor.isInitialized()) {
+ try {
+
ConcurrentUtils.shutdownAndWait(leaderElectionExecutor.get().getExecutor());
+ } catch (Exception e) {
+ LOG.warn("{}: Failed to shutdown leaderElectionExecutor",
getMemberId(), e);
+ }
}
});
}
@@ -1525,6 +1537,10 @@ class RaftServerImpl implements RaftServer.Division,
return serverExecutor;
}
+ LeaderElection.Executor getLeaderElectionExecutor() {
+ return leaderElectionExecutor.get();
+ }
+
private CompletableFuture<AppendEntriesReplyProto>
appendEntriesAsync(RaftPeerId leaderId, long callId,
TermIndex previous, ReferenceCountedObject<AppendEntriesRequestProto>
requestRef) throws IOException {
final AppendEntriesRequestProto proto = requestRef.get();
