Pass Allocator as RedisStateMachine constructor argument #1053

Use provided allocator instead of using a static reference to PooledByteBufAllocator.DEFAULT.
redis · Nov 8, 2019 · 9c583b9 · 9c583b9
1 parent caa2748
commit 9c583b9
Show file tree

Hide file tree

Showing 4 changed files with 30 additions and 21 deletions.
diff --git a/src/main/java/io/lettuce/core/protocol/CommandHandler.java b/src/main/java/io/lettuce/core/protocol/CommandHandler.java
@@ -75,7 +75,6 @@ public class CommandHandler extends ChannelDuplexHandler implements HasQueuedCom
     private final ArrayDeque<RedisCommand<?, ?, ?>> stack = new ArrayDeque<>();
     private final long commandHandlerId = COMMAND_HANDLER_COUNTER.incrementAndGet();
 
-    private final RedisStateMachine rsm = new RedisStateMachine();
     private final boolean traceEnabled = logger.isTraceEnabled();
     private final boolean debugEnabled = logger.isDebugEnabled();
     private final boolean latencyMetricsEnabled;
@@ -85,7 +84,8 @@ public class CommandHandler extends ChannelDuplexHandler implements HasQueuedCom
     private final boolean boundedQueues;
     private final BackpressureSource backpressureSource = new BackpressureSource();
 
-    Channel channel;
+    private RedisStateMachine rsm;
+    private Channel channel;
     private ByteBuf buffer;
     private LifecycleState lifecycleState = LifecycleState.NOT_CONNECTED;
     private String logPrefix;
@@ -165,6 +165,7 @@ public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
         setState(LifecycleState.REGISTERED);
 
         buffer = ctx.alloc().directBuffer(8192 * 8);
+        rsm = new RedisStateMachine(ctx.alloc());
         ctx.fireChannelRegistered();
     }
 
@@ -186,11 +187,12 @@ public void channelUnregistered(ChannelHandlerContext ctx) throws Exception {
 
         channel = null;
         buffer.release();
+        rsm.close();
+        rsm = null;
 
         reset();
 
         setState(LifecycleState.CLOSED);
-        rsm.close();
 
         ctx.fireChannelUnregistered();
     }
@@ -232,7 +234,8 @@ public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws E
             try {
                 command.completeExceptionally(cause);
             } catch (Exception ex) {
-                logger.warn("{} Unexpected exception during command completion exceptionally: {}", logPrefix, ex.toString(), ex);
+                logger.warn("{} Unexpected exception during command completion exceptionally: {}", logPrefix, ex.toString(),
+                        ex);
             }
         }
 
@@ -328,8 +331,6 @@ public void channelInactive(ChannelHandlerContext ctx) throws Exception {
             onProtectedMode(command.getOutput().getError());
         }
 
-        rsm.reset();
-
         if (debugEnabled) {
             logger.debug("{} channelInactive() done", logPrefix());
         }
@@ -429,8 +430,8 @@ private void writeBatch(ChannelHandlerContext ctx, Collection<RedisCommand<?, ?,
 
             if (isWriteable(command) && !deduplicated.add(command)) {
                 deduplicated.remove(command);
-                command.completeExceptionally(new RedisException(
-                        "Attempting to write duplicate command that is already enqueued: " + command));
+                command.completeExceptionally(
+                        new RedisException("Attempting to write duplicate command that is already enqueued: " + command));
             }
         }
 
@@ -779,13 +780,14 @@ protected void afterDecode(ChannelHandlerContext ctx, RedisCommand<?, ?, ?> comm
 
     private void recordLatency(WithLatency withLatency, ProtocolKeyword commandType) {
 
-        if (withLatency != null && clientResources.commandLatencyCollector().isEnabled() && channel != null && remote() != null) {
+        if (withLatency != null && clientResources.commandLatencyCollector().isEnabled() && channel != null
+                && remote() != null) {
 
             long firstResponseLatency = withLatency.getFirstResponse() - withLatency.getSent();
             long completionLatency = nanoTime() - withLatency.getSent();
 
-            clientResources.commandLatencyCollector().recordCommandLatency(local(), remote(), commandType,
-                    firstResponseLatency, completionLatency);
+            clientResources.commandLatencyCollector().recordCommandLatency(local(), remote(), commandType, firstResponseLatency,
+                    completionLatency);
         }
     }
 
@@ -813,7 +815,9 @@ private void reset() {
 
     private void resetInternals() {
 
-        rsm.reset();
+        if (rsm != null) {
+            rsm.reset();
+        }
 
         if (buffer.refCnt() > 0) {
             buffer.clear();

diff --git a/src/main/java/io/lettuce/core/protocol/RedisStateMachine.java b/src/main/java/io/lettuce/core/protocol/RedisStateMachine.java
@@ -25,7 +25,7 @@
 import io.lettuce.core.RedisException;
 import io.lettuce.core.output.CommandOutput;
 import io.netty.buffer.ByteBuf;
-import io.netty.buffer.PooledByteBufAllocator;
+import io.netty.buffer.ByteBufAllocator;
 import io.netty.util.ByteProcessor;
 import io.netty.util.internal.logging.InternalLogger;
 import io.netty.util.internal.logging.InternalLoggerFactory;
@@ -55,15 +55,16 @@ enum Type {
     private final State[] stack = new State[32];
     private final boolean debugEnabled = logger.isDebugEnabled();
     private final LongProcessor longProcessor = new LongProcessor();
-    private final ByteBuf responseElementBuffer = PooledByteBufAllocator.DEFAULT.directBuffer(1024);
+    private final ByteBuf responseElementBuffer;
     private final AtomicBoolean closed = new AtomicBoolean();
 
     private int stackElements;
 
     /**
      * Initialize a new instance.
      */
-    public RedisStateMachine() {
+    public RedisStateMachine(ByteBufAllocator alloc) {
+        this.responseElementBuffer = alloc.directBuffer(1024);
     }
 
     /**

diff --git a/src/test/java/io/lettuce/core/protocol/StateMachineUnitTests.java b/src/test/java/io/lettuce/core/protocol/StateMachineUnitTests.java
@@ -28,16 +28,14 @@
 import org.apache.logging.log4j.core.LoggerContext;
 import org.apache.logging.log4j.core.config.Configuration;
 import org.apache.logging.log4j.core.config.LoggerConfig;
-import org.junit.jupiter.api.AfterAll;
-import org.junit.jupiter.api.BeforeAll;
-import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.*;
 
 import io.lettuce.core.RedisException;
 import io.lettuce.core.codec.RedisCodec;
 import io.lettuce.core.codec.StringCodec;
 import io.lettuce.core.output.*;
 import io.netty.buffer.ByteBuf;
+import io.netty.buffer.ByteBufAllocator;
 import io.netty.buffer.Unpooled;
 
 /**
@@ -69,7 +67,12 @@ static void afterClass() {
     @BeforeEach
     final void createStateMachine() {
         output = new StatusOutput<>(codec);
-        rsm = new RedisStateMachine();
+        rsm = new RedisStateMachine(ByteBufAllocator.DEFAULT);
+    }
+
+    @AfterEach
+    void tearDown() {
+        rsm.close();
     }
 
     @Test

diff --git a/src/test/jmh/io/lettuce/core/protocol/RedisStateMachineBenchmark.java b/src/test/jmh/io/lettuce/core/protocol/RedisStateMachineBenchmark.java
@@ -23,6 +23,7 @@
 import io.lettuce.core.codec.ByteArrayCodec;
 import io.lettuce.core.output.ArrayOutput;
 import io.netty.buffer.ByteBuf;
+import io.netty.buffer.ByteBufAllocator;
 import io.netty.buffer.PooledByteBufAllocator;
 
 /**
@@ -59,7 +60,7 @@ public void set(long integer) {
 
     private ByteBuf masterBuffer;
 
-    private final RedisStateMachine stateMachine = new RedisStateMachine();
+    private final RedisStateMachine stateMachine = new RedisStateMachine(ByteBufAllocator.DEFAULT);
     private final byte[] payload = ("*3\r\n" + //
             "$4\r\n" + //
             "LLEN\r\n" + //