Visitar URL original
Fixed - check for invocation of sync methods in async/rx/reactive lis… · redisson/redisson@503f9a8 · GitHub
Skip to content

Commit 503f9a8

Browse files
mrnikomrniko
committed
Fixed - check for invocation of sync methods in async/rx/reactive listeners doesn't work if nettyExecutor or eventLoopGroup setting is defined
1 parent f309455 commit 503f9a8

4 files changed

Lines changed: 118 additions & 3 deletions

File tree

‎redisson/src/main/java/org/redisson/command/CommandAsyncService.java‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -181,7 +181,7 @@ public <V> void transfer(CompletionStage<V> future1, CompletableFuture<V> future
181181

182182
@Override
183183
public <V> V get(RFuture<V> future) {
184-
if (Thread.currentThread().getName().startsWith("redisson-netty")) {
184+
if (connectionManager.getServiceManager().isNettyThread()) {
185185
throw new IllegalStateException("Sync methods can't be invoked from async/rx/reactive listeners");
186186
}
187187

@@ -198,7 +198,7 @@ public <V> V get(RFuture<V> future) {
198198

199199
@Override
200200
public <V> V get(CompletableFuture<V> future) {
201-
if (Thread.currentThread().getName().startsWith("redisson-netty")) {
201+
if (connectionManager.getServiceManager().isNettyThread()) {
202202
throw new IllegalStateException("Sync methods can't be invoked from async/rx/reactive listeners");
203203
}
204204

‎redisson/src/main/java/org/redisson/connection/MasterSlaveConnectionManager.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -520,7 +520,7 @@ protected RedisClientConfig createRedisConfig(NodeType type, RedisURI address, i
520520
.setPingConnectionInterval(config.getPingConnectionInterval())
521521
.setUsername(Objects.toString(serviceCfg.getUsername(), config.getUsername()))
522522
.setPassword(Objects.toString(serviceCfg.getPassword(), config.getPassword()))
523-
.setNettyHook(serviceCfg.getNettyHook())
523+
.setNettyHook(serviceManager.getNettyHook())
524524
.setFailedNodeDetector(failedNodeDetector)
525525
.setProtocol(serviceCfg.getProtocol())
526526
.setCapabilities(serviceCfg.getValkeyCapabilities())

‎redisson/src/main/java/org/redisson/connection/ServiceManager.java‎

Lines changed: 44 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -15,8 +15,11 @@
1515
*/
1616
package org.redisson.connection;
1717

18+
import io.netty.bootstrap.Bootstrap;
1819
import io.netty.buffer.ByteBuf;
1920
import io.netty.buffer.ByteBufUtil;
21+
import io.netty.channel.Channel;
22+
import io.netty.channel.EventLoop;
2023
import io.netty.channel.EventLoopGroup;
2124
import io.netty.channel.MultiThreadIoEventLoopGroup;
2225
import io.netty.channel.epoll.EpollDatagramChannel;
@@ -52,6 +55,7 @@
5255
import org.redisson.RedissonShutdownException;
5356
import org.redisson.api.*;
5457
import org.redisson.cache.LRUCacheMap;
58+
import org.redisson.client.NettyHook;
5559
import org.redisson.client.RedisNodeNotFoundException;
5660
import org.redisson.client.codec.Codec;
5761
import org.redisson.client.protocol.RedisCommand;
@@ -152,6 +156,10 @@ public boolean cancel() {
152156

153157
private volatile boolean bitFieldRODisabled;
154158

159+
private final Map<Thread, EventLoop> nettyThreads = new ConcurrentHashMap<>();
160+
161+
private final NettyHook nettyHook;
162+
155163
private static final Map<InetSocketAddress, Set<String>> SCRIPT_SHA_CACHE = new ConcurrentHashMap<>();
156164

157165
private static final Map<String, String> SHA_CACHE = new LRUCacheMap<>(500, 0, 0);
@@ -260,6 +268,28 @@ public ServiceManager(MasterSlaveServersConfig config, Config cfg) {
260268
this.cfg = cfg;
261269
this.config = config;
262270

271+
if (cfg.getNettyExecutor() == null && cfg.getEventLoopGroup() == null) {
272+
this.nettyHook = cfg.getNettyHook();
273+
} else {
274+
this.nettyHook = new NettyHook() {
275+
@Override
276+
public void afterBoostrapInitialization(Bootstrap bootstrap) {
277+
cfg.getNettyHook().afterBoostrapInitialization(bootstrap);
278+
}
279+
280+
@Override
281+
public void afterChannelInitialization(Channel channel) {
282+
if (nettyThreads.put(Thread.currentThread(), channel.eventLoop()) == null) {
283+
// An event loop can stop its thread and use a new thread later.
284+
// Each new thread removes the threads that no longer run their event loop.
285+
// Thus the map cannot grow without limit.
286+
nettyThreads.entrySet().removeIf(entry -> !entry.getValue().inEventLoop(entry.getKey()));
287+
}
288+
cfg.getNettyHook().afterChannelInitialization(channel);
289+
}
290+
};
291+
}
292+
263293
if (cfg.getConnectionListener() != null) {
264294
this.connectionEventsHub.addListener(cfg.getConnectionListener());
265295
}
@@ -385,6 +415,20 @@ public EventLoopGroup getGroup() {
385415
return group;
386416
}
387417

418+
public NettyHook getNettyHook() {
419+
return nettyHook;
420+
}
421+
422+
public boolean isNettyThread() {
423+
Thread thread = Thread.currentThread();
424+
if (thread.getName().startsWith("redisson-netty")) {
425+
return true;
426+
}
427+
428+
EventLoop eventLoop = nettyThreads.get(thread);
429+
return eventLoop != null && eventLoop.inEventLoop(thread);
430+
}
431+
388432
public CompletableFuture<List<RedisURI>> resolveAll(RedisURI uri) {
389433
if (uri.isIP()) {
390434
RedisURI mappedUri = toURI(uri.getScheme(), uri.getHost(), "" + uri.getPort());

‎redisson/src/test/java/org/redisson/RedissonTest.java‎

Lines changed: 71 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,8 @@
66
import com.github.dockerjava.api.model.Ports;
77
import io.netty.buffer.ByteBuf;
88
import io.netty.buffer.ByteBufAllocator;
9+
import io.netty.channel.EventLoopGroup;
10+
import io.netty.channel.nio.NioEventLoopGroup;
911
import io.netty.channel.socket.DatagramChannel;
1012
import io.netty.channel.socket.SocketChannel;
1113
import io.netty.resolver.AddressResolverGroup;
@@ -58,6 +60,7 @@
5860
import java.util.concurrent.atomic.AtomicReference;
5961

6062
import static org.assertj.core.api.Assertions.assertThat;
63+
import static org.assertj.core.api.Assertions.assertThatThrownBy;
6164
import static org.awaitility.Awaitility.await;
6265

6366
public class RedissonTest extends RedisDockerTest {
@@ -174,6 +177,74 @@ public void testVirtualNettyExecutor() {
174177
r.shutdown();
175178
}
176179

180+
@Test
181+
public void testSyncMethodInListener() throws Exception {
182+
testSyncMethodInListener(createConfig());
183+
}
184+
185+
@Test
186+
public void testSyncMethodInListenerNettyExecutor() throws Exception {
187+
ExecutorService executor = Executors.newCachedThreadPool();
188+
try {
189+
Config c = createConfig();
190+
c.setNettyExecutor(executor);
191+
testSyncMethodInListener(c);
192+
} finally {
193+
executor.shutdown();
194+
}
195+
}
196+
197+
@Test
198+
public void testSyncMethodInListenerVirtualNettyExecutor() throws Exception {
199+
ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor();
200+
try {
201+
Config c = createConfig();
202+
c.setNettyExecutor(executor);
203+
testSyncMethodInListener(c);
204+
} finally {
205+
executor.shutdown();
206+
}
207+
}
208+
209+
@Test
210+
public void testSyncMethodInListenerEventLoopGroup() throws Exception {
211+
EventLoopGroup group = new NioEventLoopGroup();
212+
try {
213+
Config c = createConfig();
214+
c.setEventLoopGroup(group);
215+
testSyncMethodInListener(c);
216+
} finally {
217+
group.shutdownGracefully();
218+
}
219+
}
220+
221+
private void testSyncMethodInListener(Config config) throws Exception {
222+
RedissonClient r = Redisson.create(config);
223+
try {
224+
RBlockingQueue<String> queue = r.getBlockingQueue("queue");
225+
RBucket<String> bucket = r.getBucket("bucket");
226+
227+
CompletableFuture<String> result = new CompletableFuture<>();
228+
// The test adds the listener before the element.
229+
// Thus a Netty thread always invokes the listener, and the test thread never does.
230+
queue.takeAsync().whenComplete((v, e) -> {
231+
try {
232+
result.complete(bucket.get());
233+
} catch (Exception ex) {
234+
result.completeExceptionally(ex);
235+
}
236+
});
237+
queue.add("1");
238+
239+
assertThatThrownBy(() -> result.get(5, TimeUnit.SECONDS))
240+
.isInstanceOf(ExecutionException.class)
241+
.hasCauseInstanceOf(IllegalStateException.class)
242+
.hasRootCauseMessage("Sync methods can't be invoked from async/rx/reactive listeners");
243+
} finally {
244+
r.shutdown();
245+
}
246+
}
247+
177248
@Test
178249
public void testStopThreads() {
179250
Set<Thread> threads = Thread.getAllStackTraces().keySet();

0 commit comments

Comments
 (0)