RATIS-1853. When the stream server channelInactive, all requests in the channel.#889
Merged
szetszwo merged 6 commits intoapache:masterfrom Jul 1, 2023
Merged
Conversation
szetszwo
reviewed
Jun 29, 2023
Contributor
szetszwo
left a comment
szetszwo
left a comment
There was a problem hiding this comment.
@guohao-rosicky , thanks a lot for working on this! Please see the comments inlined. Most of them are cosmetic changes.
| } | ||
| } | ||
|
|
||
| public static class RequestTracker { |
Contributor
There was a problem hiding this comment.
Let's call it ChannelMap.
| } | ||
|
|
||
| public static class RequestTracker { | ||
| private final Map<ChannelId, Set<ClientInvocationId>> ref = new ConcurrentHashMap<>(); |
Contributor
There was a problem hiding this comment.
Since the field is only used locally in this class, let's simply call it map.
|
|
||
| public void add(ChannelId channelId, | ||
| ClientInvocationId clientInvocationId) { | ||
| ref.computeIfAbsent(channelId, (e) -> new CopyOnWriteArraySet<>()).add(clientInvocationId); |
Contributor
There was a problem hiding this comment.
ConcurrentHashMap is more efficient than CopyOnWriteArraySet. Let's declare it as
private final Map<ChannelId, Map<ClientInvocationId, ClientInvocationId>> map = new ConcurrentHashMap<>();and use ConcurrentHashMap.
Comment on lines
+246
to
+248
| public Set<ClientInvocationId> removeAll(ChannelId channelId) { | ||
| return ref.remove(channelId); | ||
| } |
Contributor
There was a problem hiding this comment.
Let's simply call it remove.
Set<ClientInvocationId> remove(ChannelId channelId) {
return Optional.ofNullable(map.remove(channelId))
.map(Map::keySet)
.orElse(Collections.emptySet());
}|
|
||
| void cleanUpOnChannelInactive(ClientInvocationId key) { | ||
| Optional.ofNullable(streams.remove(key)).ifPresent(removed -> removed.getLocal().cleanUp()); | ||
| void cleanUpOnChannelInactive(ChannelHandlerContext ctx, TimeDuration channelInactiveGracePeriod) { |
Contributor
There was a problem hiding this comment.
Let's pass ChannelId instead of the ChannelHandlerContext.
| // Delayed memory garbage cleanup | ||
| Optional.ofNullable(requestTracker.removeAll(ctx.channel().id())).ifPresent(ids -> { | ||
| final String streamIds = Arrays.toString(ids.toArray()); | ||
| LOG.info("server {} is disconnect, cleanup clientInvocationIds={}", ctx.channel(), streamIds); |
Contributor
There was a problem hiding this comment.
Let's use the same wording.
LOG.info("Channel {} is inactive, cleanup clientInvocationIds={}", channelId, ids);
Comment on lines
+418
to
+423
| () -> { | ||
| for (ClientInvocationId clientInvocationId : ids) { | ||
| Optional.ofNullable(streams.remove(clientInvocationId)).ifPresent(removed -> | ||
| removed.getLocal().cleanUp()); | ||
| } | ||
| }, |
Contributor
There was a problem hiding this comment.
It will be easier to read if it is refactored out
void cleanUp(Set<ClientInvocationId> ids) {
for (ClientInvocationId clientInvocationId : ids) {
Optional.ofNullable(streams.remove(clientInvocationId))
.map(StreamInfo::getLocal)
.ifPresent(LocalStream::cleanUp);
}
}
void cleanUpOnChannelInactive(ChannelId channelId, TimeDuration channelInactiveGracePeriod) {
// Delayed memory garbage cleanup
Optional.ofNullable(channels.remove(channelId)).ifPresent(ids -> {
LOG.info("Channel {} is inactive, cleanup clientInvocationIds={}", channelId, ids);
TimeoutExecutor.getInstance().onTimeout(channelInactiveGracePeriod, () -> cleanUp(ids),
LOG, () -> "Timeout check failed, clientInvocationIds=" + ids);
});
}
Contributor
Author
|
Thanks for the review @szetszwo . |
szetszwo
approved these changes
Jun 30, 2023
Contributor
szetszwo
left a comment
szetszwo
left a comment
There was a problem hiding this comment.
+1 the change looks good.
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.
What changes were proposed in this pull request?
When the stream server channelInactive, all requests in the channel.
At the same time, there will be multiple stream sessions, using the same channel, for transmission, and when the connection is broken, we should clean up more stream sessions
What is the link to the Apache JIRA
https://issues.apache.org/jira/browse/RATIS-1853