You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@ratis.apache.org by "ivandika3 (via GitHub)" <gi...@apache.org> on 2023/10/28 16:18:48 UTC
[PR] RATIS-1921. Shared worker group in WorkerGroupGetter should be closed [ratis]
ivandika3 opened a new pull request, #955:
URL: https://github.com/apache/ratis/pull/955
## What changes were proposed in this pull request?
### Problem
When benchmarking Ozone streaming pipeline using `ozone freon ockg`, the benchmark would not end although all the keys have been written. `hdds.ratis.raft.netty.dataStream.client.worker-group.share` is set to true to use the shared worker group optimization.
Using Arthas, it's found that non-daemon threads like "NettyClientStreamRpc-workerGroup–thread1" are still running even after the benchmark has finished. The root cause is that the shared worker group will never be closed, causing the JVM shutdown hook to never be triggered. The benchmark was able to shutdown normally if the share configuration is disabled.
### Background
It seems that shared worker group is a lazily instantiated singleton. The shared worker group will be instantiated when the first `WorkerGroupGetter` is instantiated and passed to a new `Connection`during construction. Due to the nature of singleton, this worker group will be shared across all the Raft clients under the same Ozone client (please correct me if I'm wrong).
In the case of Ozone streaming write pipeline, every `BlockDatastreamOutput` (whose scope is a single block) will create a new `RaftClient` which corresponds to a single `NettyClientStreamRpc` instance. These `RaftClient`s will share the shared worker group.
### Solution
The current solution uses a "modified" reference counted `EventLoopGroup` using the `ReferenceCountedObject` interface. Previously, I was trying to use the implementation from `ReferenceCountedObject#wrap`. However, when the reference count is 1 and `release()` is invoked, it will completely release the object and throw exceptions for any further operations. This implementation does not seem to suit the use case.
The "modified" reference count instantiates the shared worker group whenever it's retained and the previous reference count is 0. It will also gracefully shutdown the worker group when it's released and the reference count becomes 0. it's retained whenever a new `WorkerGroupGetter` is instantiated (i.e when connection is created), and released when the connection is closed.
Technically, the worker group is not a singleton anymore, the shared worker group will be shared with all the connections, but will be removed and shutdown when all connections are removed, subsequent connections will use a new shared worker group. This should guarantee that there will be one worker group shared among the connections, but not necessarily the same instance.
The current solution is still a rough solution for reference what the solution might be, any advise is greatly appreciated.
Also enabled the `hdds.ratis.raft.netty.dataStream.client.worker-group.share` according to (https://github.com/szetszwo/ozone-benchmark/blob/master/benchmark-conf.xml).
## What is the link to the Apache JIRA
https://issues.apache.org/jira/browse/RATIS-1921
## How was this patch tested?
Existing unit tests (enable worker group sharing by default). Also tested using `ozone freon ockg` again.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@ratis.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] RATIS-1921. Shared worker group in WorkerGroupGetter should be closed [ratis]
Posted by "szetszwo (via GitHub)" <gi...@apache.org>.
szetszwo commented on code in PR #955:
URL: https://github.com/apache/ratis/pull/955#discussion_r1376632033
##########
ratis-common/src/main/java/org/apache/ratis/util/ReferenceCountedObject.java:
##########
@@ -81,8 +86,11 @@ static <V> ReferenceCountedObject<V> wrap(V value, Runnable retainMethod, Runnab
@Override
public V get() {
- if (count.get() < 0) {
+ final int previous = count.get();
+ if (previous < 0) {
throw new IllegalStateException("Failed to get: object has already been completely released.");
+ } else if (previous == 0) {
Review Comment:
@ivandika3 , Sorry that this change is causing test failures for DataStream tests. Could you include the following fix?
```java
+++ b/ratis-netty/src/main/java/org/apache/ratis/netty/server/DataStreamManagement.java
@@ -325,10 +325,13 @@ public class DataStreamManagement {
long byteWritten = 0;
for (ByteBuffer buffer : buf.nioBuffers()) {
final ReferenceCountedObject<ByteBuffer> wrapped = ReferenceCountedObject.wrap(buffer, buf::retain, buf::release);
+ wrapped.retain();
try {
byteWritten += channel.write(wrapped);
} catch (Throwable t) {
throw new CompletionException(t);
+ } finally {
+ wrapped.release();
}
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@ratis.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] RATIS-1921. Shared worker group in WorkerGroupGetter should be closed [ratis]
Posted by "ivandika3 (via GitHub)" <gi...@apache.org>.
ivandika3 commented on code in PR #955:
URL: https://github.com/apache/ratis/pull/955#discussion_r1376893101
##########
ratis-common/src/main/java/org/apache/ratis/util/ReferenceCountedObject.java:
##########
@@ -81,8 +86,11 @@ static <V> ReferenceCountedObject<V> wrap(V value, Runnable retainMethod, Runnab
@Override
public V get() {
- if (count.get() < 0) {
+ final int previous = count.get();
+ if (previous < 0) {
throw new IllegalStateException("Failed to get: object has already been completely released.");
+ } else if (previous == 0) {
Review Comment:
Thank you, yes I think this should resolve the issue, ran the unit tests in local.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@ratis.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] RATIS-1921. Shared worker group in WorkerGroupGetter should be closed [ratis]
Posted by "szetszwo (via GitHub)" <gi...@apache.org>.
szetszwo commented on PR #955:
URL: https://github.com/apache/ratis/pull/955#issuecomment-1784208441
Another way is to use `CompletableFuture` as below. It may be easier to understand.
```java
private static final AtomicReference<CompletableFuture<ReferenceCountedObject<EventLoopGroup>>> SHARED_WORKER_GROUP
= new AtomicReference<>();
static WorkerGroupGetter newInstance(RaftProperties properties) {
final boolean shared = NettyConfigKeys.DataStream.Client.workerGroupShare(properties);
if (shared) {
final CompletableFuture<ReferenceCountedObject<EventLoopGroup>> created = new CompletableFuture<>();
final CompletableFuture<ReferenceCountedObject<EventLoopGroup>> current
= SHARED_WORKER_GROUP.updateAndGet(g -> g != null ? g : created);
if (current == created) {
created.complete(ReferenceCountedObject.wrap(newWorkerGroup(properties)));
}
return new WorkerGroupGetter(current.join().retain()) {
@Override
void shutdownGracefully() {
final CompletableFuture<ReferenceCountedObject<EventLoopGroup>> returned
= SHARED_WORKER_GROUP.updateAndGet(previous -> {
Preconditions.assertSame(current, previous, "SHARED_WORKER_GROUP");
return previous.join().release() ? null : previous;
});
if (returned == null) {
workerGroup.shutdownGracefully();
}
}
};
} else {
return new WorkerGroupGetter(newWorkerGroup(properties));
}
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@ratis.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] RATIS-1921. Shared worker group in WorkerGroupGetter should be closed [ratis]
Posted by "ivandika3 (via GitHub)" <gi...@apache.org>.
ivandika3 commented on code in PR #955:
URL: https://github.com/apache/ratis/pull/955#discussion_r1376891052
##########
ratis-netty/src/main/java/org/apache/ratis/netty/client/NettyClientStreamRpc.java:
##########
@@ -78,7 +80,33 @@ public class NettyClientStreamRpc implements DataStreamClientRpc {
public static final Logger LOG = LoggerFactory.getLogger(NettyClientStreamRpc.class);
private static class WorkerGroupGetter implements Supplier<EventLoopGroup> {
- private static final AtomicReference<EventLoopGroup> SHARED_WORKER_GROUP = new AtomicReference<>();
+
+ private static final AtomicReference<ReferenceCountedObject<EventLoopGroup>> SHARED_WORKER_GROUP =
+ new AtomicReference<>();
Review Comment:
Thank you for the explanation.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@ratis.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] RATIS-1921. Shared worker group in WorkerGroupGetter should be closed [ratis]
Posted by "szetszwo (via GitHub)" <gi...@apache.org>.
szetszwo commented on PR #955:
URL: https://github.com/apache/ratis/pull/955#issuecomment-1789430482
`TestNettyDataStreamStarTopologyWithGrpcCluster` and `TestNettyDataStreamChainTopologyWithGrpcCluster` have failed in quite a few attempts. It may be related to changing the default of WORKER_GROUP_SHARE_KEY since these tests created quite many streams. We may need to either disable it in the tests or increase the workerGroupSize.
@ivandika3 , could you take a look?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@ratis.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] RATIS-1921. Shared worker group in WorkerGroupGetter should be closed [ratis]
Posted by "ivandika3 (via GitHub)" <gi...@apache.org>.
ivandika3 commented on PR #955:
URL: https://github.com/apache/ratis/pull/955#issuecomment-1785230080
Thank you for the review and guidance @szetszwo.
I have updated it accordingly, but found that Datastream tests threw this error.
```
java.lang.IllegalStateException: Failed to get StreamInfo for DataStreamRequestByteBuf:clientId=client-FF2407A774AC,type=STREAM_DATA,id=0,offset=75998,length=607
at org.apache.ratis.netty.server.DataStreamManagement.lambda$readImpl$11(DataStreamManagement.java:467)
at java.base/java.util.Optional.orElseThrow(Optional.java:408)
at org.apache.ratis.netty.server.DataStreamManagement.readImpl(DataStreamManagement.java:466)
at org.apache.ratis.netty.server.DataStreamManagement.read(DataStreamManagement.java:433)
at org.apache.ratis.netty.server.NettyServerStreamRpc$1.channelRead(NettyServerStreamRpc.java:236)
```
From my understanding, it might be because `STREAM_HEADER` was not sent, before the `STREAM_DATA` was received. I will try to resolve this.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@ratis.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] RATIS-1921. Shared worker group in WorkerGroupGetter should be closed [ratis]
Posted by "szetszwo (via GitHub)" <gi...@apache.org>.
szetszwo commented on code in PR #955:
URL: https://github.com/apache/ratis/pull/955#discussion_r1376632033
##########
ratis-common/src/main/java/org/apache/ratis/util/ReferenceCountedObject.java:
##########
@@ -81,8 +86,11 @@ static <V> ReferenceCountedObject<V> wrap(V value, Runnable retainMethod, Runnab
@Override
public V get() {
- if (count.get() < 0) {
+ final int previous = count.get();
+ if (previous < 0) {
throw new IllegalStateException("Failed to get: object has already been completely released.");
+ } else if (previous == 0) {
Review Comment:
@ivandika3 , Sorry that this change is causing test failures for other code. How about reverting it and changeing this separately.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@ratis.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] RATIS-1921. Shared worker group in WorkerGroupGetter should be closed [ratis]
Posted by "szetszwo (via GitHub)" <gi...@apache.org>.
szetszwo commented on code in PR #955:
URL: https://github.com/apache/ratis/pull/955#discussion_r1376582942
##########
ratis-netty/src/main/java/org/apache/ratis/netty/client/NettyClientStreamRpc.java:
##########
@@ -78,7 +80,33 @@ public class NettyClientStreamRpc implements DataStreamClientRpc {
public static final Logger LOG = LoggerFactory.getLogger(NettyClientStreamRpc.class);
private static class WorkerGroupGetter implements Supplier<EventLoopGroup> {
- private static final AtomicReference<EventLoopGroup> SHARED_WORKER_GROUP = new AtomicReference<>();
+
+ private static final AtomicReference<ReferenceCountedObject<EventLoopGroup>> SHARED_WORKER_GROUP =
+ new AtomicReference<>();
Review Comment:
Yes. When there are two or more threads changing an `AtomicReference` at the same time, the `updateAndGet(..)` method runs in a loop and may call the update function multiple times. In our previous code, it may create multiple `EventLoopGroup` objects and the old objects will not be shutdown.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@ratis.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] RATIS-1921. Shared worker group in WorkerGroupGetter should be closed [ratis]
Posted by "ivandika3 (via GitHub)" <gi...@apache.org>.
ivandika3 commented on code in PR #955:
URL: https://github.com/apache/ratis/pull/955#discussion_r1376893101
##########
ratis-common/src/main/java/org/apache/ratis/util/ReferenceCountedObject.java:
##########
@@ -81,8 +86,11 @@ static <V> ReferenceCountedObject<V> wrap(V value, Runnable retainMethod, Runnab
@Override
public V get() {
- if (count.get() < 0) {
+ final int previous = count.get();
+ if (previous < 0) {
throw new IllegalStateException("Failed to get: object has already been completely released.");
+ } else if (previous == 0) {
Review Comment:
Thank you, yes I think this should resolve the issue, ran the unit tests in local.
Also tested without regression using `ozone freon ockg`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@ratis.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] RATIS-1921. Shared worker group in WorkerGroupGetter should be closed [ratis]
Posted by "ivandika3 (via GitHub)" <gi...@apache.org>.
ivandika3 commented on code in PR #955:
URL: https://github.com/apache/ratis/pull/955#discussion_r1377023019
##########
ratis-netty/src/main/java/org/apache/ratis/netty/server/DataStreamManagement.java:
##########
@@ -326,9 +326,11 @@ static long writeTo(ByteBuf buf, Iterable<WriteOption> options,
for (ByteBuffer buffer : buf.nioBuffers()) {
final ReferenceCountedObject<ByteBuffer> wrapped = ReferenceCountedObject.wrap(buffer, buf::retain, buf::release);
try {
- byteWritten += channel.write(wrapped);
+ byteWritten += channel.write(wrapped.retain());
Review Comment:
My mistake, I have updated it accordingly.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@ratis.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] RATIS-1921. Shared worker group in WorkerGroupGetter should be closed [ratis]
Posted by "szetszwo (via GitHub)" <gi...@apache.org>.
szetszwo commented on code in PR #955:
URL: https://github.com/apache/ratis/pull/955#discussion_r1377017070
##########
ratis-netty/src/main/java/org/apache/ratis/netty/server/DataStreamManagement.java:
##########
@@ -326,9 +326,11 @@ static long writeTo(ByteBuf buf, Iterable<WriteOption> options,
for (ByteBuffer buffer : buf.nioBuffers()) {
final ReferenceCountedObject<ByteBuffer> wrapped = ReferenceCountedObject.wrap(buffer, buf::retain, buf::release);
try {
- byteWritten += channel.write(wrapped);
+ byteWritten += channel.write(wrapped.retain());
Review Comment:
@ivandika3 , we need to pass `ReferenceCountedObject` instead of `ByteBuffer` so that the implementation can also retain the buffer for later use.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@ratis.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] RATIS-1921. Shared worker group in WorkerGroupGetter should be closed [ratis]
Posted by "ivandika3 (via GitHub)" <gi...@apache.org>.
ivandika3 commented on code in PR #955:
URL: https://github.com/apache/ratis/pull/955#discussion_r1376231798
##########
ratis-netty/src/main/java/org/apache/ratis/netty/client/NettyClientStreamRpc.java:
##########
@@ -78,7 +80,33 @@ public class NettyClientStreamRpc implements DataStreamClientRpc {
public static final Logger LOG = LoggerFactory.getLogger(NettyClientStreamRpc.class);
private static class WorkerGroupGetter implements Supplier<EventLoopGroup> {
- private static final AtomicReference<EventLoopGroup> SHARED_WORKER_GROUP = new AtomicReference<>();
+
+ private static final AtomicReference<ReferenceCountedObject<EventLoopGroup>> SHARED_WORKER_GROUP =
+ new AtomicReference<>();
+
+ static WorkerGroupGetter newInstance(RaftProperties properties) {
+ final boolean shared = NettyConfigKeys.DataStream.Client.workerGroupShare(properties);
+ if (shared) {
+ final Supplier<ReferenceCountedObject<EventLoopGroup>> supplier = MemoizedSupplier.valueOf(
+ () -> ReferenceCountedObject.wrap(newWorkerGroup((properties))));
+ final ReferenceCountedObject<EventLoopGroup> sharedWorkerGroup = SHARED_WORKER_GROUP.updateAndGet(
+ g -> g != null ? g : supplier.get());
Review Comment:
Thank you for the review. Updated.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@ratis.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] RATIS-1921. Shared worker group in WorkerGroupGetter should be closed [ratis]
Posted by "szetszwo (via GitHub)" <gi...@apache.org>.
szetszwo commented on PR #955:
URL: https://github.com/apache/ratis/pull/955#issuecomment-1785767736
> ... it might be because STREAM_HEADER was not sent, before the STREAM_DATA was received. ...
@ivandika3 , when reviewing this, I found that we also have used `AtomicReference`/`ConcurrentHashMap` incorrectly in some other case; filed RATIS-1923.
If you could find some other bugs, please also file a JIRA. Thanks!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@ratis.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] RATIS-1921. Shared worker group in WorkerGroupGetter should be closed [ratis]
Posted by "szetszwo (via GitHub)" <gi...@apache.org>.
szetszwo commented on code in PR #955:
URL: https://github.com/apache/ratis/pull/955#discussion_r1375324078
##########
ratis-netty/src/main/java/org/apache/ratis/netty/NettyConfigKeys.java:
##########
@@ -158,7 +158,7 @@ static void setWorkerGroupSize(RaftProperties properties, int clientWorkerGroupS
}
String WORKER_GROUP_SHARE_KEY = PREFIX + ".worker-group.share";
- boolean WORKER_GROUP_SHARE_DEFAULT = false;
+ boolean WORKER_GROUP_SHARE_DEFAULT = true;
Review Comment:
It is good to change the default to true.
##########
ratis-netty/src/main/java/org/apache/ratis/netty/client/NettyClientStreamRpc.java:
##########
@@ -78,7 +80,53 @@ public class NettyClientStreamRpc implements DataStreamClientRpc {
public static final Logger LOG = LoggerFactory.getLogger(NettyClientStreamRpc.class);
private static class WorkerGroupGetter implements Supplier<EventLoopGroup> {
- private static final AtomicReference<EventLoopGroup> SHARED_WORKER_GROUP = new AtomicReference<>();
+
+ private static class RefCountedWorkerGroup implements ReferenceCountedObject<EventLoopGroup> {
+ private final AtomicInteger count = new AtomicInteger();
+ private final AtomicReference<EventLoopGroup> value = new AtomicReference<>();
+
+ @Override
+ public synchronized EventLoopGroup get() {
+ if (count.get() < 0) {
+ throw new IllegalStateException("Failed to get: object has already been completely released.");
+ }
+ return value.get();
+ }
+
+ @Override
+ public synchronized EventLoopGroup retain() {
+ // n < 0: exception
+ // n >= 0: n++
+ final int previous = count.getAndUpdate(n -> n < 0? n : n + 1);
+ if (previous < 0) {
+ throw new IllegalStateException("Failed to retain: object has already been completely released.");
+ } else if (previous == 0) {
+ // TODO: Find a way to pass RaftProperties
+ // New shared worker group will be created when previously there is
+ // no active connections
+ return value.updateAndGet(g -> g != null ? g: newWorkerGroup(new RaftProperties()));
+ }
+ return value.get();
+ }
+
+ @Override
+ public synchronized boolean release() {
+ // n <= 0: exception
+ // n >= 1: n--
+ final int previous = count.getAndUpdate(n -> n <= 0? -1: n - 1);
+ if (previous <= 0) {
+ throw new IllegalStateException("Failed to release: object has already been completely released.");
+ } else if (previous == 1) {
+ // Shutdown the event loop group when there are no active connection,
+ // subsequent retain will create a new shared worker group.
+ EventLoopGroup previousEventLoopGroup = value.getAndSet(null);
+ previousEventLoopGroup.shutdownGracefully();
+ }
+ return previous == 1;
+ }
+ }
+
+ private static final RefCountedWorkerGroup SHARED_WORKER_GROUP = new RefCountedWorkerGroup();
Review Comment:
We needs `AtomicReference` so that a new `SHARED_WORKER_GROUP` can be created after an old one is completely released.
##########
ratis-netty/src/main/java/org/apache/ratis/netty/client/NettyClientStreamRpc.java:
##########
@@ -78,7 +80,53 @@ public class NettyClientStreamRpc implements DataStreamClientRpc {
public static final Logger LOG = LoggerFactory.getLogger(NettyClientStreamRpc.class);
private static class WorkerGroupGetter implements Supplier<EventLoopGroup> {
- private static final AtomicReference<EventLoopGroup> SHARED_WORKER_GROUP = new AtomicReference<>();
+
+ private static class RefCountedWorkerGroup implements ReferenceCountedObject<EventLoopGroup> {
Review Comment:
We should just use `ReferenceCountedObject.warp(..)`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@ratis.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] RATIS-1921. Shared worker group in WorkerGroupGetter should be closed [ratis]
Posted by "ivandika3 (via GitHub)" <gi...@apache.org>.
ivandika3 commented on code in PR #955:
URL: https://github.com/apache/ratis/pull/955#discussion_r1376234658
##########
ratis-netty/src/main/java/org/apache/ratis/netty/client/NettyClientStreamRpc.java:
##########
@@ -78,7 +80,33 @@ public class NettyClientStreamRpc implements DataStreamClientRpc {
public static final Logger LOG = LoggerFactory.getLogger(NettyClientStreamRpc.class);
private static class WorkerGroupGetter implements Supplier<EventLoopGroup> {
- private static final AtomicReference<EventLoopGroup> SHARED_WORKER_GROUP = new AtomicReference<>();
+
+ private static final AtomicReference<ReferenceCountedObject<EventLoopGroup>> SHARED_WORKER_GROUP =
+ new AtomicReference<>();
Review Comment:
> since AtomicReference.updateAndGet(updateFunction) may call the updateFunction without using the result. The current code will then called newWorkerGroup(..) without shutting it down.
Do you mean it's because `AtomicReference.updateAndGet(updateFunction)` might get re-applied, as mentioned in the `AtomicReference` documentation?
> Since it may be re-applied when attempted updates fail due to contention among threads.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@ratis.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] RATIS-1921. Shared worker group in WorkerGroupGetter should be closed [ratis]
Posted by "ivandika3 (via GitHub)" <gi...@apache.org>.
ivandika3 commented on PR #955:
URL: https://github.com/apache/ratis/pull/955#issuecomment-1793353423
Hi @szetszwo , thank you for the suggestions. I have increased the worker group size to 100 (taken from https://github.com/szetszwo/ozone-benchmark/blob/master/benchmark-conf.xml).
TestNettyDataStreamChainTopologyWithGrpcCluster 400x: https://github.com/ivandika3/ratis/actions/runs/6752427405
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@ratis.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] RATIS-1921. Shared worker group in WorkerGroupGetter should be closed [ratis]
Posted by "szetszwo (via GitHub)" <gi...@apache.org>.
szetszwo merged PR #955:
URL: https://github.com/apache/ratis/pull/955
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@ratis.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] RATIS-1921. Shared worker group in WorkerGroupGetter should be closed [ratis]
Posted by "szetszwo (via GitHub)" <gi...@apache.org>.
szetszwo commented on code in PR #955:
URL: https://github.com/apache/ratis/pull/955#discussion_r1375473296
##########
ratis-netty/src/main/java/org/apache/ratis/netty/client/NettyClientStreamRpc.java:
##########
@@ -78,7 +80,33 @@ public class NettyClientStreamRpc implements DataStreamClientRpc {
public static final Logger LOG = LoggerFactory.getLogger(NettyClientStreamRpc.class);
private static class WorkerGroupGetter implements Supplier<EventLoopGroup> {
- private static final AtomicReference<EventLoopGroup> SHARED_WORKER_GROUP = new AtomicReference<>();
+
+ private static final AtomicReference<ReferenceCountedObject<EventLoopGroup>> SHARED_WORKER_GROUP =
+ new AtomicReference<>();
Review Comment:
We need
```java
private static final AtomicReference<ReferenceCountedObject<Supplier<EventLoopGroup>>> SHARED_WORKER_GROUP
= new AtomicReference<>();
```
since `AtomicReference.updateAndGet(updateFunction)` may call the `updateFunction` without using the result. The current code will then called `newWorkerGroup(..)` without shutting it down.
Use a `MemoizedSupplier` inside:
```java
final Supplier<ReferenceCountedObject<Supplier<EventLoopGroup>>> supplier = MemoizedSupplier.valueOf(
() -> ReferenceCountedObject.wrap(MemoizedSupplier.valueOf(() -> newWorkerGroup(properties))));
final ReferenceCountedObject<Supplier<EventLoopGroup>> sharedWorkerGroup = SHARED_WORKER_GROUP.updateAndGet(
g -> g != null ? g : supplier.get());
return new WorkerGroupGetter(sharedWorkerGroup.retain().get()) {
...
};
```
##########
ratis-netty/src/main/java/org/apache/ratis/netty/client/NettyClientStreamRpc.java:
##########
@@ -78,7 +80,33 @@ public class NettyClientStreamRpc implements DataStreamClientRpc {
public static final Logger LOG = LoggerFactory.getLogger(NettyClientStreamRpc.class);
private static class WorkerGroupGetter implements Supplier<EventLoopGroup> {
- private static final AtomicReference<EventLoopGroup> SHARED_WORKER_GROUP = new AtomicReference<>();
+
+ private static final AtomicReference<ReferenceCountedObject<EventLoopGroup>> SHARED_WORKER_GROUP =
+ new AtomicReference<>();
+
+ static WorkerGroupGetter newInstance(RaftProperties properties) {
+ final boolean shared = NettyConfigKeys.DataStream.Client.workerGroupShare(properties);
+ if (shared) {
+ final Supplier<ReferenceCountedObject<EventLoopGroup>> supplier = MemoizedSupplier.valueOf(
+ () -> ReferenceCountedObject.wrap(newWorkerGroup((properties))));
+ final ReferenceCountedObject<EventLoopGroup> sharedWorkerGroup = SHARED_WORKER_GROUP.updateAndGet(
+ g -> g != null ? g : supplier.get());
Review Comment:
We should call `retain()`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@ratis.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] RATIS-1921. Shared worker group in WorkerGroupGetter should be closed [ratis]
Posted by "ivandika3 (via GitHub)" <gi...@apache.org>.
ivandika3 commented on code in PR #955:
URL: https://github.com/apache/ratis/pull/955#discussion_r1376893101
##########
ratis-common/src/main/java/org/apache/ratis/util/ReferenceCountedObject.java:
##########
@@ -81,8 +86,11 @@ static <V> ReferenceCountedObject<V> wrap(V value, Runnable retainMethod, Runnab
@Override
public V get() {
- if (count.get() < 0) {
+ final int previous = count.get();
+ if (previous < 0) {
throw new IllegalStateException("Failed to get: object has already been completely released.");
+ } else if (previous == 0) {
Review Comment:
Thank you, yes I think this should resolve the issue.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@ratis.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org