Merge Decouple ConnectionWatchdog reconnect from timer thread #100

Motivation: Netty's HashedWheelTimer runs single threaded and processes, therefore, one task at a time. lettuce's ConnectionWatchdog uses a blocking call (without a timeout) to the initializing future that may block the timer thread indefinitely. The blocking call stops the reconnection for all connections because the processing thread is blocked.
redis · Jul 8, 2015 · c5411d4 · c5411d4
1 parent fe6f445
commit c5411d4
Show file tree

Hide file tree

Showing 37 changed files with 215 additions and 117 deletions.
diff --git a/src/main/java/com/lambdaworks/redis/PlainChannelInitializer.java b/src/main/java/com/lambdaworks/redis/PlainChannelInitializer.java
@@ -50,11 +50,6 @@ public void channelInactive(ChannelHandlerContext ctx) throws Exception {
                     super.channelInactive(ctx);
                 }
 
-                @Override
-                public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
-                    super.channelRead(ctx, msg);
-                }
-
                 @Override
                 public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
                     if (evt instanceof ConnectionEvents.Close) {

diff --git a/src/main/java/com/lambdaworks/redis/RedisChannelHandler.java b/src/main/java/com/lambdaworks/redis/RedisChannelHandler.java
@@ -197,14 +197,14 @@ public void setOptions(ClientOptions clientOptions) {
         this.clientOptions = clientOptions;
     }
 
-    public TimeUnit getTimeoutUnit() {
-        return unit;
-    }
-
     public long getTimeout() {
         return timeout;
     }
 
+    public TimeUnit getTimeoutUnit() {
+        return unit;
+    }
+
     protected <T> T syncHandler(Object asyncApi, Class<?>... interfaces) {
         FutureSyncInvocationHandler<K, V> h = new FutureSyncInvocationHandler<>((StatefulConnection) this, asyncApi);
         return (T) Proxy.newProxyInstance(AbstractRedisClient.class.getClassLoader(), interfaces, h);

diff --git a/src/main/java/com/lambdaworks/redis/protocol/ChannelLogDescriptor.java b/src/main/java/com/lambdaworks/redis/protocol/ChannelLogDescriptor.java
@@ -0,0 +1,32 @@
+package com.lambdaworks.redis.protocol;
+
+import io.netty.channel.Channel;
+
+/**
+ * @author <a href="mailto:[email protected]">Mark Paluch</a>
+ * @since 08.07.15 09:43
+ */
+class ChannelLogDescriptor {
+
+    static String logDescriptor(Channel channel) {
+
+        if (channel == null) {
+            return "unknown";
+        }
+
+        StringBuffer buffer = new StringBuffer(64);
+
+        if (channel.localAddress() != null) {
+            buffer.append(channel.localAddress()).append(" -> ");
+        }
+        if (channel.remoteAddress() != null) {
+            buffer.append(channel.remoteAddress());
+        }
+
+        if (!channel.isActive()) {
+            buffer.append(" (inactive)");
+        }
+
+        return buffer.toString();
+    }
+}
diff --git a/src/main/java/com/lambdaworks/redis/protocol/CommandHandler.java b/src/main/java/com/lambdaworks/redis/protocol/CommandHandler.java
@@ -2,6 +2,7 @@
 
 package com.lambdaworks.redis.protocol;
 
+import java.nio.channels.ClosedChannelException;
 import java.nio.charset.Charset;
 import java.util.ArrayDeque;
 import java.util.ArrayList;
@@ -14,6 +15,8 @@
 
 import io.netty.buffer.ByteBuf;
 import io.netty.channel.*;
+import io.netty.util.concurrent.Future;
+import io.netty.util.concurrent.GenericFutureListener;
 import io.netty.util.internal.logging.InternalLogger;
 import io.netty.util.internal.logging.InternalLoggerFactory;
 
@@ -28,6 +31,7 @@
 public class CommandHandler<K, V> extends ChannelDuplexHandler implements RedisChannelWriter<K, V> {
 
     private static final InternalLogger logger = InternalLoggerFactory.getInstance(CommandHandler.class);
+    private static final WriteLogListener WRITE_LOG_LISTENER = new WriteLogListener();
 
     protected ClientOptions clientOptions;
     protected Queue<RedisCommand<K, V, ?>> queue;
@@ -176,7 +180,7 @@ public <T, C extends RedisCommand<K, V, T>> C write(C command) {
 
                 if (reliability == Reliability.AT_LEAST_ONCE) {
                     // commands are ok to stay within the queue, reconnect will retrigger them
-                    channel.write(command, channel.voidPromise());
+                    channel.write(command).addListener(WRITE_LOG_LISTENER);
                     channel.flush();
                 }
             } else {
@@ -232,7 +236,6 @@ public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise)
             }
         } catch (Exception e) {
             cmd.completeExceptionally(e);
-            cmd.cancel();
             promise.setFailure(e);
             throw e;
         }
@@ -483,14 +486,8 @@ private String logPrefix() {
         if (logPrefix != null) {
             return logPrefix;
         }
-        StringBuffer buffer = new StringBuffer(16);
-        buffer.append('[');
-        if (channel != null) {
-            buffer.append(channel.remoteAddress());
-        } else {
-            buffer.append("not connected");
-        }
-        buffer.append(']');
+        StringBuffer buffer = new StringBuffer(64);
+        buffer.append('[').append(ChannelLogDescriptor.logDescriptor(channel)).append(']');
         return logPrefix = buffer.toString();
     }
 
@@ -524,4 +521,17 @@ public void operationComplete(ChannelFuture future) throws Exception {
         }
     }
 
+    /**
+     * A generic future listener which logs unsuccessful writes.
+     *
+     */
+    static class WriteLogListener implements GenericFutureListener<Future<Void>> {
+
+        @Override
+        public void operationComplete(Future<Void> future) throws Exception {
+            if (!future.isSuccess() && !(future.cause() instanceof ClosedChannelException))
+                logger.warn(future.cause().getMessage(), future.cause());
+        }
+
+    }
 }
diff --git a/src/main/java/com/lambdaworks/redis/protocol/ConnectionWatchdog.java b/src/main/java/com/lambdaworks/redis/protocol/ConnectionWatchdog.java
@@ -3,11 +3,13 @@
 package com.lambdaworks.redis.protocol;
 
 import java.net.SocketAddress;
+import java.util.concurrent.Callable;
 import java.util.concurrent.TimeUnit;
 
 import com.google.common.base.Supplier;
 import com.lambdaworks.redis.ClientOptions;
 import com.lambdaworks.redis.ConnectionEvents;
+import com.lambdaworks.redis.RedisChannelHandler;
 import com.lambdaworks.redis.RedisChannelInitializer;
 
 import io.netty.bootstrap.Bootstrap;
@@ -50,6 +52,7 @@ public class ConnectionWatchdog extends ChannelInboundHandlerAdapter implements
     private SocketAddress remoteAddress;
     private int attempts;
     private long lastReconnectionLogging = -1;
+    private String logPrefix;
 
     /**
      * Create a new watchdog that adds to new connections to the supplied {@link ChannelGroup} and establishes a new
@@ -82,7 +85,7 @@ public ConnectionWatchdog(ClientOptions clientOptions, Bootstrap bootstrap, Time
 
     @Override
     public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
-        logger.debug("userEventTriggered(" + ctx + ", " + evt + ")");
+        logger.debug("{} userEventTriggered({}, {})", logPrefix(), ctx, evt);
         if (evt instanceof ConnectionEvents.PrepareClose) {
             ConnectionEvents.PrepareClose prepareClose = (ConnectionEvents.PrepareClose) evt;
             setListenOnChannelInactive(false);
@@ -94,7 +97,7 @@ public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exc
     @Override
     public void channelActive(ChannelHandlerContext ctx) throws Exception {
 
-        logger.debug("channelActive(" + ctx + ")");
+        logger.debug("{} channelActive({})", logPrefix(), ctx);
         channel = ctx.channel();
         attempts = 0;
         remoteAddress = channel.remoteAddress();
@@ -104,10 +107,13 @@ public void channelActive(ChannelHandlerContext ctx) throws Exception {
     @Override
     public void channelInactive(ChannelHandlerContext ctx) throws Exception {
 
-        logger.debug("channelInactive(" + ctx + ")");
+        logger.debug("{} channelInactive({})", logPrefix(), ctx);
         channel = null;
         if (listenOnChannelInactive && !reconnectSuspended) {
             scheduleReconnect();
+        } else {
+            logger.debug("{} Reconnect scheduling disabled", logPrefix(), ctx);
+            logger.debug("");
         }
         super.channelInactive(ctx);
     }
@@ -116,9 +122,10 @@ public void channelInactive(ChannelHandlerContext ctx) throws Exception {
      * Schedule reconnect if channel is not available/not active.
      */
     public void scheduleReconnect() {
-        logger.debug("scheduleReconnect()");
+        logger.debug("{} scheduleReconnect()", logPrefix());
 
         if (!isEventLoopGroupActive()) {
+            logger.debug("isEventLoopGroupActive() == false");
             return;
         }
 
@@ -127,9 +134,26 @@ public void scheduleReconnect() {
                 attempts++;
             }
             int timeout = 2 << attempts;
-            timer.newTimeout(this, timeout, TimeUnit.MILLISECONDS);
+            timer.newTimeout(new TimerTask() {
+                @Override
+                public void run(Timeout timeout) throws Exception {
+
+                    if (!isEventLoopGroupActive()) {
+                        logger.debug("isEventLoopGroupActive() == false");
+                        return;
+                    }
+
+                    bootstrap.group().submit(new Callable<Object>() {
+                        @Override
+                        public Object call() throws Exception {
+                            ConnectionWatchdog.this.run(null);
+                            return null;
+                        }
+                    });
+                }
+            }, timeout, TimeUnit.MILLISECONDS);
         } else {
-            logger.debug("Skipping scheduleReconnect() because I have an active channel");
+            logger.debug("{} Skipping scheduleReconnect() because I have an active channel", logPrefix());
         }
     }
 
@@ -145,6 +169,7 @@ public void scheduleReconnect() {
     public void run(Timeout timeout) throws Exception {
 
         if (!isEventLoopGroupActive()) {
+            logger.debug("isEventLoopGroupActive() == false");
             return;
         }
 
@@ -163,7 +188,7 @@ public void run(Timeout timeout) throws Exception {
         try {
             reconnect(infoLevel, warnLevel);
         } catch (Exception e) {
-            logger.log(warnLevel, "Cannot connect: " + e.toString());
+            logger.log(warnLevel, "Cannot connect: {}", e.toString());
             scheduleReconnect();
         }
     }
@@ -184,10 +209,11 @@ private void reconnect(InternalLogLevel infoLevel, InternalLogLevel warnLevel) t
         connect.sync().await();
 
         RedisChannelInitializer channelInitializer = connect.channel().pipeline().get(RedisChannelInitializer.class);
-        CommandHandler<?, ?> commandHandler = connect.channel().pipeline().get(CommandHandler.class);
-        try {
+        CommandHandler commandHandler = connect.channel().pipeline().get(CommandHandler.class);
+        RedisChannelHandler channelHandler = connect.channel().pipeline().get(RedisChannelHandler.class);
 
-            channelInitializer.channelInitialized().get();
+        try {
+            channelInitializer.channelInitialized().get(channelHandler.getTimeout(), channelHandler.getTimeoutUnit());
             logger.log(infoLevel, "Reconnected to " + remoteAddress);
         } catch (Exception e) {
 
@@ -248,4 +274,14 @@ public boolean isReconnectSuspended() {
     public void setReconnectSuspended(boolean reconnectSuspended) {
         this.reconnectSuspended = reconnectSuspended;
     }
+
+    private String logPrefix() {
+        if (logPrefix != null) {
+            return logPrefix;
+        }
+        StringBuffer buffer = new StringBuffer(64);
+        buffer.append('[').append(ChannelLogDescriptor.logDescriptor(channel)).append(']');
+        return logPrefix = buffer.toString();
+    }
+
 }
diff --git a/src/test/java/biz/paluch/redis/extensibility/MyExtendedRedisClientTest.java b/src/test/java/biz/paluch/redis/extensibility/MyExtendedRedisClientTest.java
@@ -1,18 +1,17 @@
 package biz.paluch.redis.extensibility;
 
-import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.*;
 
-import java.util.concurrent.TimeUnit;
-
-import com.lambdaworks.redis.pubsub.RedisPubSubAsyncCommandsImpl;
-import com.lambdaworks.redis.pubsub.api.async.RedisPubSubAsyncCommands;
 import org.apache.log4j.Logger;
 import org.junit.AfterClass;
 import org.junit.BeforeClass;
 import org.junit.Test;
 
+import com.lambdaworks.redis.FastShutdown;
 import com.lambdaworks.redis.RedisConnection;
 import com.lambdaworks.redis.TestSettings;
+import com.lambdaworks.redis.pubsub.RedisPubSubAsyncCommandsImpl;
+import com.lambdaworks.redis.pubsub.api.async.RedisPubSubAsyncCommands;
 
 /**
  * Test for override/extensability of RedisClient
@@ -38,7 +37,7 @@ protected static MyExtendedRedisClient getRedisClient() {
 
     @AfterClass
     public static void shutdownClient() {
-        client.shutdown(0, 0, TimeUnit.MILLISECONDS);
+        FastShutdown.shutdown(client);
     }
 
     @Test

diff --git a/src/test/java/com/lambdaworks/SslTest.java b/src/test/java/com/lambdaworks/SslTest.java
@@ -35,7 +35,7 @@ public void before() throws Exception {
 
     @AfterClass
     public static void afterClass() {
-        redisClient.shutdown();
+        FastShutdown.shutdown(redisClient);
     }
 
     @Test

diff --git a/src/test/java/com/lambdaworks/redis/AbstractRedisClientTest.java b/src/test/java/com/lambdaworks/redis/AbstractRedisClientTest.java
@@ -56,7 +56,7 @@ public WithPasswordRequired() throws Exception {
                 try {
                     run(client);
                 } finally {
-                    client.shutdown(0, 0, TimeUnit.MILLISECONDS);
+                    FastShutdown.shutdown(client);
                 }
             } finally {
 

diff --git a/src/test/java/com/lambdaworks/redis/AllTheAPIsTest.java b/src/test/java/com/lambdaworks/redis/AllTheAPIsTest.java
@@ -25,7 +25,7 @@ public static void beforeClass() throws Exception {
     @BeforeClass
     public static void afterClass() throws Exception {
         if (clusterClient != null) {
-            clusterClient.shutdown();
+            FastShutdown.shutdown(clusterClient);
         }
     }
 

diff --git a/src/test/java/com/lambdaworks/redis/ClientTest.java b/src/test/java/com/lambdaworks/redis/ClientTest.java
@@ -232,7 +232,7 @@ public void cancelCommandsOnReconnectFailure() throws Exception {
 
             assertThat(connection.isOpen()).isFalse();
             connectionWatchdog.setReconnectSuspended(false);
-            connectionWatchdog.scheduleReconnect();
+            connectionWatchdog.run(null);
             Thread.sleep(500);
             assertThat(connection.isOpen()).isFalse();
 
@@ -309,7 +309,7 @@ public boolean isSatisfied() {
         assertThat(listener.onConnected).isEqualTo(statefulRedisConnection);
         assertThat(listener.onDisconnected).isEqualTo(statefulRedisConnection);
 
-        client.shutdown();
+        FastShutdown.shutdown(client);
     }
 
     @Test
@@ -338,7 +338,7 @@ public boolean isSatisfied() {
         assertThat(removedListener.onDisconnected).isNull();
         assertThat(removedListener.onException).isNull();
 
-        client.shutdown();
+        FastShutdown.shutdown(client);
 
     }
 
@@ -436,7 +436,7 @@ public void emptyClient() throws Exception {
         } catch (IllegalArgumentException e) {
             assertThat(e).hasMessageContaining("RedisURI");
         }
-        client.shutdown();
+        FastShutdown.shutdown(client);
     }
 
     @Test

diff --git a/src/test/java/com/lambdaworks/redis/ConnectionCommandTest.java b/src/test/java/com/lambdaworks/redis/ConnectionCommandTest.java
@@ -42,7 +42,7 @@ public void run(RedisClient client) {
                 RedisConnection<String, String> authConnection = redisClient.connect().sync();
                 authConnection.ping();
                 authConnection.close();
-                redisClient.shutdown();
+                FastShutdown.shutdown(redisClient);
             }
         };
     }

diff --git a/src/test/java/com/lambdaworks/redis/DefaultRedisClient.java b/src/test/java/com/lambdaworks/redis/DefaultRedisClient.java
@@ -17,7 +17,7 @@ public DefaultRedisClient() {
         Runtime.getRuntime().addShutdownHook(new Thread() {
             @Override
             public void run() {
-                redisClient.shutdown(0, 0, TimeUnit.MILLISECONDS);
+                FastShutdown.shutdown(redisClient);
             }
         });
     }