From d03e0de847f0d5cac23cf3b916164cf74d77b51c Mon Sep 17 00:00:00 2001 From: Nikita Date: Fri, 8 Apr 2016 12:02:19 +0300 Subject: [PATCH] Fixed probably thread blocking issues #455 --- .../org/redisson/RedissonCountDownLatch.java | 4 +- src/main/java/org/redisson/RedissonLock.java | 4 +- .../java/org/redisson/RedissonObject.java | 6 +++ .../java/org/redisson/RedissonSemaphore.java | 5 ++- .../org/redisson/client/RedisConnection.java | 40 +++++++++++++------ .../client/handler/CommandDecoder.java | 7 +--- .../client/handler/CommandsQueue.java | 4 +- .../command/CommandAsyncExecutor.java | 3 ++ .../redisson/command/CommandAsyncService.java | 29 +++++++++++++- .../connection/pool/ConnectionPool.java | 2 +- 10 files changed, 76 insertions(+), 28 deletions(-) diff --git a/src/main/java/org/redisson/RedissonCountDownLatch.java b/src/main/java/org/redisson/RedissonCountDownLatch.java index 8f3eeef1f..5a4d660a6 100644 --- a/src/main/java/org/redisson/RedissonCountDownLatch.java +++ b/src/main/java/org/redisson/RedissonCountDownLatch.java @@ -53,7 +53,7 @@ public class RedissonCountDownLatch extends RedissonObject implements RCountDown public void await() throws InterruptedException { Future promise = subscribe(); try { - promise.await(); + get(promise); while (getCount() > 0) { // waiting for open state @@ -71,7 +71,7 @@ public class RedissonCountDownLatch extends RedissonObject implements RCountDown public boolean await(long time, TimeUnit unit) throws InterruptedException { Future promise = subscribe(); try { - if (!promise.await(time, unit)) { + if (!await(promise, time, unit)) { return false; } diff --git a/src/main/java/org/redisson/RedissonLock.java b/src/main/java/org/redisson/RedissonLock.java index 3bc14bf9b..eaf060496 100644 --- a/src/main/java/org/redisson/RedissonLock.java +++ b/src/main/java/org/redisson/RedissonLock.java @@ -110,7 +110,7 @@ public class RedissonLock extends RedissonExpirable implements RLock { } Future future = subscribe(); - future.sync(); + get(future); try { while (true) { @@ -229,7 +229,7 @@ public class RedissonLock extends RedissonExpirable implements RLock { } Future future = subscribe(); - if (!future.await(time, TimeUnit.MILLISECONDS)) { + if (!await(future, time, TimeUnit.MILLISECONDS)) { future.addListener(new FutureListener() { @Override public void operationComplete(Future future) throws Exception { diff --git a/src/main/java/org/redisson/RedissonObject.java b/src/main/java/org/redisson/RedissonObject.java index 7e12d8ac8..6cc3d6000 100644 --- a/src/main/java/org/redisson/RedissonObject.java +++ b/src/main/java/org/redisson/RedissonObject.java @@ -15,6 +15,8 @@ */ package org.redisson; +import java.util.concurrent.TimeUnit; + import org.redisson.client.codec.Codec; import org.redisson.client.protocol.RedisCommands; import org.redisson.command.CommandAsyncExecutor; @@ -45,6 +47,10 @@ abstract class RedissonObject implements RObject { this(commandExecutor.getConnectionManager().getCodec(), commandExecutor, name); } + protected boolean await(Future future, long timeout, TimeUnit timeoutUnit) throws InterruptedException { + return commandExecutor.await(future, timeout, timeoutUnit); + } + protected V get(Future future) { return commandExecutor.get(future); } diff --git a/src/main/java/org/redisson/RedissonSemaphore.java b/src/main/java/org/redisson/RedissonSemaphore.java index e29a4e0a0..a23cd966b 100644 --- a/src/main/java/org/redisson/RedissonSemaphore.java +++ b/src/main/java/org/redisson/RedissonSemaphore.java @@ -70,7 +70,8 @@ public class RedissonSemaphore extends RedissonExpirable implements RSemaphore { return; } - Future future = subscribe().sync(); + Future future = subscribe(); + get(future); try { while (true) { if (tryAcquire(permits)) { @@ -113,7 +114,7 @@ public class RedissonSemaphore extends RedissonExpirable implements RSemaphore { long time = unit.toMillis(waitTime); Future future = subscribe(); - if (!future.await(time, TimeUnit.MILLISECONDS)) { + if (!await(future, time, TimeUnit.MILLISECONDS)) { return false; } diff --git a/src/main/java/org/redisson/client/RedisConnection.java b/src/main/java/org/redisson/client/RedisConnection.java index 4f6c49dc4..fc98647fd 100644 --- a/src/main/java/org/redisson/client/RedisConnection.java +++ b/src/main/java/org/redisson/client/RedisConnection.java @@ -15,6 +15,7 @@ */ package org.redisson.client; +import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import org.redisson.client.codec.Codec; @@ -111,21 +112,34 @@ public class RedisConnection implements RedisCommands { return redisClient; } - public R await(Future cmd) { - // TODO change connectTimeout to timeout - if (!cmd.awaitUninterruptibly(redisClient.getTimeout(), TimeUnit.MILLISECONDS)) { - Promise promise = (Promise)cmd; - RedisTimeoutException ex = new RedisTimeoutException("Command execution timeout for " + redisClient.getAddr()); - promise.setFailure(ex); - throw ex; - } - if (!cmd.isSuccess()) { - if (cmd.cause() instanceof RedisException) { - throw (RedisException) cmd.cause(); + public R await(Future future) { + final CountDownLatch l = new CountDownLatch(1); + future.addListener(new FutureListener() { + @Override + public void operationComplete(Future future) throws Exception { + l.countDown(); + } + }); + + try { + // TODO change connectTimeout to timeout + if (!l.await(redisClient.getTimeout(), TimeUnit.MILLISECONDS)) { + Promise promise = (Promise)future; + RedisTimeoutException ex = new RedisTimeoutException("Command execution timeout for " + redisClient.getAddr()); + promise.setFailure(ex); + throw ex; + } + if (!future.isSuccess()) { + if (future.cause() instanceof RedisException) { + throw (RedisException) future.cause(); + } + throw new RedisException("Unexpected exception while processing command", future.cause()); } - throw new RedisException("Unexpected exception while processing command", cmd.cause()); + return future.getNow(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + return null; } - return cmd.getNow(); } public T sync(RedisStrictCommand command, Object ... params) { diff --git a/src/main/java/org/redisson/client/handler/CommandDecoder.java b/src/main/java/org/redisson/client/handler/CommandDecoder.java index f0b5ce652..6e4849a00 100644 --- a/src/main/java/org/redisson/client/handler/CommandDecoder.java +++ b/src/main/java/org/redisson/client/handler/CommandDecoder.java @@ -1,5 +1,4 @@ /** - * Copyright 2014 Nikita Koksharov, Nickolay Borbit * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -173,7 +172,7 @@ public class CommandDecoder extends ReplayingDecoder { if (code == '+') { String result = in.readBytes(in.bytesBefore((byte) '\r')).toString(CharsetUtil.UTF_8); in.skipBytes(2); - + handleResult(data, parts, result, false, channel); } else if (code == '-') { String error = in.readBytes(in.bytesBefore((byte) '\r')).toString(CharsetUtil.UTF_8); @@ -206,9 +205,7 @@ public class CommandDecoder extends ReplayingDecoder { } } } else if (code == ':') { - String status = in.readBytes(in.bytesBefore((byte) '\r')).toString(CharsetUtil.UTF_8); - in.skipBytes(2); - Object result = Long.valueOf(status); + Long result = readLong(in); handleResult(data, parts, result, false, channel); } else if (code == '$') { ByteBuf buf = readBytes(in); diff --git a/src/main/java/org/redisson/client/handler/CommandsQueue.java b/src/main/java/org/redisson/client/handler/CommandsQueue.java index ff65b9d4c..96446c90e 100644 --- a/src/main/java/org/redisson/client/handler/CommandsQueue.java +++ b/src/main/java/org/redisson/client/handler/CommandsQueue.java @@ -23,10 +23,10 @@ import org.redisson.client.protocol.QueueCommand; import org.redisson.client.protocol.QueueCommandHolder; import io.netty.channel.Channel; -import io.netty.channel.ChannelDuplexHandler; import io.netty.channel.ChannelFuture; import io.netty.channel.ChannelFutureListener; import io.netty.channel.ChannelHandlerContext; +import io.netty.channel.ChannelOutboundHandlerAdapter; import io.netty.channel.ChannelPromise; import io.netty.util.AttributeKey; import io.netty.util.internal.PlatformDependent; @@ -37,7 +37,7 @@ import io.netty.util.internal.PlatformDependent; * @author Nikita Koksharov * */ -public class CommandsQueue extends ChannelDuplexHandler { +public class CommandsQueue extends ChannelOutboundHandlerAdapter { public static final AttributeKey CURRENT_COMMAND = AttributeKey.valueOf("promise"); diff --git a/src/main/java/org/redisson/command/CommandAsyncExecutor.java b/src/main/java/org/redisson/command/CommandAsyncExecutor.java index cddeb2a8b..a218b8f19 100644 --- a/src/main/java/org/redisson/command/CommandAsyncExecutor.java +++ b/src/main/java/org/redisson/command/CommandAsyncExecutor.java @@ -18,6 +18,7 @@ package org.redisson.command; import java.net.InetSocketAddress; import java.util.Collection; import java.util.List; +import java.util.concurrent.TimeUnit; import org.redisson.SlotCallback; import org.redisson.client.RedisException; @@ -38,6 +39,8 @@ public interface CommandAsyncExecutor { RedisException convertException(Future future); + boolean await(Future future, long timeout, TimeUnit timeoutUnit) throws InterruptedException; + V get(Future future); Future writeAsync(Integer slot, Codec codec, RedisCommand command, Object ... params); diff --git a/src/main/java/org/redisson/command/CommandAsyncService.java b/src/main/java/org/redisson/command/CommandAsyncService.java index ab0b9e18b..cf1a506a9 100644 --- a/src/main/java/org/redisson/command/CommandAsyncService.java +++ b/src/main/java/org/redisson/command/CommandAsyncService.java @@ -22,9 +22,11 @@ import java.util.Collection; import java.util.Collections; import java.util.List; import java.util.Set; +import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; import org.redisson.RedisClientResult; import org.redisson.RedissonShutdownException; @@ -82,13 +84,38 @@ public class CommandAsyncService implements CommandAsyncExecutor { @Override public V get(Future future) { - future.awaitUninterruptibly(); + final CountDownLatch l = new CountDownLatch(1); + future.addListener(new FutureListener() { + @Override + public void operationComplete(Future future) throws Exception { + l.countDown(); + } + }); + try { + l.await(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + // commented out due to blocking issues up to 200 ms per minute for each thread + // future.awaitUninterruptibly(); if (future.isSuccess()) { return future.getNow(); } throw convertException(future); } + @Override + public boolean await(Future future, long timeout, TimeUnit timeoutUnit) throws InterruptedException { + final CountDownLatch l = new CountDownLatch(1); + future.addListener(new FutureListener() { + @Override + public void operationComplete(Future future) throws Exception { + l.countDown(); + } + }); + return l.await(timeout, timeoutUnit); + } + @Override public Future readAsync(InetSocketAddress client, String key, Codec codec, RedisCommand command, Object ... params) { Promise mainPromise = connectionManager.newPromise(); diff --git a/src/main/java/org/redisson/connection/pool/ConnectionPool.java b/src/main/java/org/redisson/connection/pool/ConnectionPool.java index 00c1c6886..87dc8a594 100644 --- a/src/main/java/org/redisson/connection/pool/ConnectionPool.java +++ b/src/main/java/org/redisson/connection/pool/ConnectionPool.java @@ -153,7 +153,7 @@ abstract class ConnectionPool { } } - StringBuilder errorMsg = new StringBuilder("Publish/Subscribe connection pool exhausted! All connections are busy. Try to increase Publish/Subscribe connection pool size."); + StringBuilder errorMsg = new StringBuilder("Connection pool exhausted! All connections are busy. Increase connection pool size."); // if (!freezed.isEmpty()) { // errorMsg.append(" Disconnected hosts: " + freezed); // }