RATIS-2235. Allow only one thread to perform appendLog #1206
Merged
szetszwo merged 3 commits intoapache:masterfrom Jan 6, 2025
Merged
RATIS-2235. Allow only one thread to perform appendLog #1206szetszwo merged 3 commits intoapache:masterfrom
szetszwo merged 3 commits intoapache:masterfrom
Conversation
szetszwo
reviewed
Jan 2, 2025
Contributor
szetszwo
left a comment
szetszwo
left a comment
There was a problem hiding this comment.
@SzyWilliam , thanks for working this! Please see a comment inlined.
Comment on lines
+1648
to
+1652
| final CompletableFuture<Void> appendLog = appendLogFuture.updateAndGet(f -> f.thenCompose(ignored -> { | ||
| final List<CompletableFuture<Long>> futures = entries.isEmpty() ? Collections.emptyList() | ||
| : state.getLog().append(requestRef.delegate(entries)); | ||
| return JavaUtils.allOf(futures); | ||
| })); |
Contributor
There was a problem hiding this comment.
We should retain the ref first, i.e.
@@ -1641,9 +1645,8 @@ class RaftServerImpl implements RaftServer.Division,
state.updateConfiguration(entries);
}
future.join();
-
- final List<CompletableFuture<Long>> futures = entries.isEmpty() ? Collections.emptyList()
- : state.getLog().append(requestRef.delegate(entries));
+ final CompletableFuture<Void> appendLog = entries.isEmpty()? appendLogFuture.get()
+ : appendLog(requestRef.delegate(entries));
proto.getCommitInfosList().forEach(commitInfoCache::update);
CodeInjectionForTesting.execute(LOG_SYNC, getId(), null);
@@ -1657,7 +1660,7 @@ class RaftServerImpl implements RaftServer.Division,
final long commitIndex = effectiveCommitIndex(proto.getLeaderCommit(), previous, entries.size());
final long matchIndex = isHeartbeat? RaftLog.INVALID_LOG_INDEX: entries.get(entries.size() - 1).getIndex();
- return JavaUtils.allOf(futures).whenCompleteAsync((r, t) -> {
+ return appendLog.whenCompleteAsync((r, t) -> {
followerState.ifPresent(fs -> fs.updateLastRpcTime(FollowerState.UpdateType.APPEND_COMPLETE));
timer.stop();
}, getServerExecutor()).thenApply(v -> {
@@ -1675,6 +1678,13 @@ class RaftServerImpl implements RaftServer.Division,
});
}
+ private CompletableFuture<Void> appendLog(ReferenceCountedObject<List<LogEntryProto>> entriesRef) {
+ entriesRef.retain();
+ return appendLogFuture.updateAndGet(f -> f.thenCompose(
+ ignored -> JavaUtils.allOf(state.getLog().append(entriesRef))))
+ .whenComplete((v, e) -> entriesRef.release());
+ }
+
private long checkInconsistentAppendEntries(TermIndex previous, List<LogEntryProto> entries) {
// Check if a snapshot installation through state machine is in progress.
final long installSnapshot = snapshotInstallationHandler.getInProgressInstallSnapshotIndex();
Member
Author
|
@szetszwo Thanks for reviewing this pr and help with the code. Now I learned how to use zero-copy in async programming, thank you! |
Contributor
|
The test failures seem related. Not yet sure what is the problem. |
szetszwo
reviewed
Jan 4, 2025
| state.updateConfiguration(entries); | ||
| } | ||
| future.join(); | ||
| final CompletableFuture<Void> appendLog = entries.isEmpty()? appendLogFuture.get() |
Contributor
There was a problem hiding this comment.
@SzyWilliam , I see the problem now -- when it is empty, it should not wait for the previous append. Otherwise, it blocks the heartbeats.
final CompletableFuture<Void> appendLog = entries.isEmpty()? CompletableFuture.completedFuture(null)
Member
Author
There was a problem hiding this comment.
That is a tricky one. Thanks for spotting and troubleshooting this!
Member
Author
|
@szetszwo the tests are all passed now, PTAL ~ |
szetszwo
approved these changes
Jan 6, 2025
Contributor
szetszwo
left a comment
szetszwo
left a comment
There was a problem hiding this comment.
+1 the change looks good.
Member
Author
|
@szetszwo Thanks for reviewing and merging this PR! |
SzyWilliam
added a commit
that referenced
this pull request
Jan 9, 2025
SzyWilliam
added a commit
that referenced
this pull request
Jan 9, 2025
SzyWilliam
added a commit
that referenced
this pull request
Jan 9, 2025
SzyWilliam
added a commit
to SzyWilliam/ratis
that referenced
this pull request
Jan 9, 2025
szetszwo
pushed a commit
that referenced
this pull request
Jan 9, 2025
szetszwo
pushed a commit
to szetszwo/ratis
that referenced
this pull request
May 23, 2025
szetszwo
pushed a commit
to szetszwo/ratis
that referenced
this pull request
May 23, 2025
slfan1989
pushed a commit
to slfan1989/ratis
that referenced
this pull request
May 6, 2026
slfan1989
pushed a commit
to slfan1989/ratis
that referenced
this pull request
May 6, 2026
(cherry picked from commit 9b74401)
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
See https://issues.apache.org/jira/browse/RATIS-2235
See the discussion at https://issues.apache.org/jira/browse/RATIS-2208