RATIS-2018. Zero-copy buffers are not released - 2nd chunk #1032
Conversation
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
szetszwo
reviewed
Jan 29, 2024
There was a problem hiding this comment.
@duongkame , thanks for working on this! Please see the comments inlined.
| @@ -117,7 +117,13 @@ public V retain() { | |||
|
|
|||
| @Override | |||
| public boolean release() { | |||
| return fromRefs.stream().map(ReferenceCountedObject::release).allMatch(r -> r); | |||
| boolean allReleased = true; | |||
| for (ReferenceCountedObject ref : fromRefs) { | |||
There was a problem hiding this comment.
Add type parameter:
for (ReferenceCountedObject<T> ref : fromRefs) {
Comment on lines
464
to
466
| final boolean isRetry = requests.putIfAbsent(request); | ||
| LOG.debug("Received seq={}, isRetry? {}, {}", seqNum, isRetry, this); | ||
| if (isRetry || request.getSeqNum() < nextToProcess) { |
There was a problem hiding this comment.
If request.getSeqNum() < nextToProcess, the request should not be added then.
public synchronized void receivedRequest(REQUEST request, Consumer<REQUEST> processingMethod) {
final long seqNum = request.getSeqNum();
final boolean accepted;
if (nextToProcess == -1 && (request.isFirstRequest() || seqNum == 0)) {
nextToProcess = seqNum;
requests.putNewRequest(request);
LOG.debug("Received seq={} (first request), {}", seqNum, this);
accepted = true;
} else if (request.getSeqNum() < nextToProcess) {
LOG.debug("Received seq={} < nextToProcess {}, {}", seqNum, nextToProcess, this);
accepted = false;
} else {
final boolean isRetry = requests.putIfAbsent(request);
LOG.debug("Received seq={}, isRetry? {}, {}", seqNum, isRetry, this);
accepted = !isRetry;
}
if (accepted) {
processRequestsFromHead(processingMethod);
} else {
request.release();
}
}| @@ -320,11 +320,13 @@ void processClientRequest(ReferenceCountedObject<RaftClientRequest> requestRef) | |||
|
|
|||
| final CompletableFuture<Void> f = processClientRequest(requestRef, reply -> { | |||
| if (!reply.isSuccess()) { | |||
| LOG.info("Failed {}, reply={}", request, reply); | |||
| LOG.info("Failed {}, reply={}", request.toStringShort(), reply); | |||
There was a problem hiding this comment.
We cannot use request here
final SlidingWindowEntry slidingWindowEntry = request.getSlidingWindowEntry();
final CompletableFuture<Void> f = processClientRequest(requestRef, reply -> {
if (!reply.isSuccess()) {
LOG.info("Failed request cid={}, {}, reply={}", callId, slidingWindowEntry, reply);
}
final RaftClientReplyProto proto = ClientProtoUtils.toRaftClientReplyProto(reply);
responseNext(proto);
});
Comment on lines
54
to
56
| private final REQUEST request; | ||
| private final ReferenceCountedObject<REQUEST> requestRef; | ||
| private AtomicBoolean released = new AtomicBoolean(false); |
There was a problem hiding this comment.
Let's combine these three fields:
static class PendingServerRequest<REQUEST> {
private final AtomicReference<ReferenceCountedObject<REQUEST>> requestRef;
private final CompletableFuture<Void> future = new CompletableFuture<>();
PendingServerRequest(ReferenceCountedObject<REQUEST> requestRef) {
requestRef.retain();
this.requestRef = new AtomicReference<>(requestRef);
}
REQUEST getRequest() {
return Optional.ofNullable(requestRef.get())
.map(ReferenceCountedObject::get)
.orElse(null);
}
CompletableFuture<Void> getFuture() {
return future;
}
void release() {
Optional.ofNullable(requestRef.getAndSet(null))
.ifPresent(ReferenceCountedObject::release);
}
}
|
One more comments: The line above should not use request since it may be released. We should pass |
|
Thanks for the detailed review, @szetszwo. We should completely avoid using requests in the asynchronous contexts (reply, error handler) as you suggest. |
szetszwo
approved these changes
Jan 30, 2024
symious
pushed a commit
to symious/ratis
that referenced
this pull request
Apr 5, 2024
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
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.
What changes were proposed in this pull request?
Zero-copy buffers are not released - 2nd chunk, including fixes in:
will NEVER be processed, requests coming with seqNum < nextToProcess.whenComplete.What is the link to the Apache JIRA
https://issues.apache.org/jira/browse/RATIS-2018
How was this patch tested?
CI