Fix: make sure FIFO order between write and notify channel active (#2597)

okg-cxf · tishun · web-flow · commit b76f985e7532 · 2024-12-06T18:02:53.000+02:00
* fix: make sure FIFO order for write() when notifyChannelActive(), also make sure channel access thread-safe and avoid potential NPE

* Formatting issues

---------

Co-authored-by: Tihomir Mateev &lt;tihomir.mateev@gmail.com&gt;
diff --git a/src/main/java/io/lettuce/core/protocol/DefaultEndpoint.java b/src/main/java/io/lettuce/core/protocol/DefaultEndpoint.java
@@ -195,9 +195,9 @@ public <K, V, T> RedisCommand<K, V, T> write(RedisCommand<K, V, T> command) {
             }
 
             if (autoFlushCommands) {
-
-                if (isConnected()) {
-                    writeToChannelAndFlush(command);
+                Channel channel = this.channel;
+                if (isConnected(channel)) {
+                    writeToChannelAndFlush(channel, command);
                 } else {
                     writeToDisconnectedBuffer(command);
                 }
@@ -236,9 +236,9 @@ public <K, V, T> RedisCommand<K, V, T> write(RedisCommand<K, V, T> command) {
             }
 
             if (autoFlushCommands) {
-
-                if (isConnected()) {
-                    writeToChannelAndFlush(commands);
+                Channel channel = this.channel;
+                if (isConnected(channel)) {
+                    writeToChannelAndFlush(channel, commands);
                 } else {
                     writeToDisconnectedBuffer(commands);
                 }
@@ -288,10 +288,9 @@ private RedisException validateWrite(int commands) {
             return new RedisException("Connection is closed");
         }
 
+        final boolean connected = isConnected(this.channel);
         if (usesBoundedQueues()) {
 
-            boolean connected = isConnected();
-
             if (QUEUE_SIZE.get(this) + commands > clientOptions.getRequestQueueSize()) {
                 return new RedisException("Request queue size exceeded: " + clientOptions.getRequestQueueSize()
                         + ". Commands are not accepted until the queue size drops.");
@@ -308,7 +307,7 @@ private RedisException validateWrite(int commands) {
             }
         }
 
-        if (!isConnected() && rejectCommandsWhileDisconnected) {
+        if (!connected && rejectCommandsWhileDisconnected) {
             return new RedisException("Currently not connected. Commands are rejected.");
         }
 
@@ -370,11 +369,11 @@ private void writeToDisconnectedBuffer(RedisCommand<?, ?, ?> command) {
         commandBuffer.add(command);
     }
 
-    private void writeToChannelAndFlush(RedisCommand<?, ?, ?> command) {
+    private void writeToChannelAndFlush(Channel channel, RedisCommand<?, ?, ?> command) {
 
         QUEUE_SIZE.incrementAndGet(this);
 
-        ChannelFuture channelFuture = channelWriteAndFlush(command);
+        ChannelFuture channelFuture = channelWriteAndFlush(channel, command);
 
         if (reliability == Reliability.AT_MOST_ONCE) {
             // cancel on exceptions and remove from queue, because there is no housekeeping
@@ -387,30 +386,30 @@ private void writeToChannelAndFlush(RedisCommand<?, ?, ?> command) {
         }
     }
 
-    private void writeToChannelAndFlush(Collection<? extends RedisCommand<?, ?, ?>> commands) {
+    private void writeToChannelAndFlush(Channel channel, Collection<? extends RedisCommand<?, ?, ?>> commands) {
 
         QUEUE_SIZE.addAndGet(this, commands.size());
 
         if (reliability == Reliability.AT_MOST_ONCE) {
 
             // cancel on exceptions and remove from queue, because there is no housekeeping
             for (RedisCommand<?, ?, ?> command : commands) {
-                channelWrite(command).addListener(AtMostOnceWriteListener.newInstance(this, command));
+                channelWrite(channel, command).addListener(AtMostOnceWriteListener.newInstance(this, command));
             }
         }
 
         if (reliability == Reliability.AT_LEAST_ONCE) {
 
             // commands are ok to stay within the queue, reconnect will retrigger them
             for (RedisCommand<?, ?, ?> command : commands) {
-                channelWrite(command).addListener(RetryListener.newInstance(this, command));
+                channelWrite(channel, command).addListener(RetryListener.newInstance(this, command));
             }
         }
 
-        channelFlush();
+        channelFlush(channel);
     }
 
-    private void channelFlush() {
+    private void channelFlush(Channel channel) {
 
         if (debugEnabled) {
             logger.debug("{} write() channelFlush", logPrefix());
@@ -419,7 +418,7 @@ private void channelFlush() {
         channel.flush();
     }
 
-    private ChannelFuture channelWrite(RedisCommand<?, ?, ?> command) {
+    private ChannelFuture channelWrite(Channel channel, RedisCommand<?, ?, ?> command) {
 
         if (debugEnabled) {
             logger.debug("{} write() channelWrite command {}", logPrefix(), command);
@@ -428,7 +427,7 @@ private ChannelFuture channelWrite(RedisCommand<?, ?, ?> command) {
         return channel.write(command);
     }
 
-    private ChannelFuture channelWriteAndFlush(RedisCommand<?, ?, ?> command) {
+    private ChannelFuture channelWriteAndFlush(Channel channel, RedisCommand<?, ?, ?> command) {
 
         if (debugEnabled) {
             logger.debug("{} write() writeAndFlush command {}", logPrefix(), command);
@@ -441,7 +440,6 @@ private ChannelFuture channelWriteAndFlush(RedisCommand<?, ?, ?> command) {
     public void notifyChannelActive(Channel channel) {
 
         this.logPrefix = null;
-        this.channel = channel;
         this.connectionError = null;
 
         if (isClosed()) {
@@ -456,6 +454,7 @@ public void notifyChannelActive(Channel channel) {
         }
 
         sharedLock.doExclusive(() -> {
+            this.channel = channel;
 
             try {
                 // Move queued commands to buffer before issuing any commands because of connection activation.
@@ -478,7 +477,7 @@ public void notifyChannelActive(Channel channel) {
                     inActivation = false;
                 }
 
-                flushCommands(disconnectedBuffer);
+                flushCommands(channel, disconnectedBuffer);
             } catch (Exception e) {
 
                 if (debugEnabled) {
@@ -525,7 +524,7 @@ public void notifyException(Throwable t) {
             doExclusive(this::drainCommands).forEach(cmd -> cmd.completeExceptionally(t));
         }
 
-        if (!isConnected()) {
+        if (!isConnected(this.channel)) {
             connectionError = t;
         }
     }
@@ -538,16 +537,16 @@ public void registerConnectionWatchdog(ConnectionWatchdog connectionWatchdog) {
     @Override
     @SuppressWarnings({ "rawtypes", "unchecked" })
     public void flushCommands() {
-        flushCommands(commandBuffer);
+        flushCommands(this.channel, commandBuffer);
     }
 
-    private void flushCommands(Queue<RedisCommand<?, ?, ?>> queue) {
+    private void flushCommands(Channel channel, Queue<RedisCommand<?, ?, ?>> queue) {
 
         if (debugEnabled) {
             logger.debug("{} flushCommands()", logPrefix());
         }
 
-        if (isConnected()) {
+        if (isConnected(channel)) {
 
             List<RedisCommand<?, ?, ?>> commands = sharedLock.doExclusive(() -> {
 
@@ -563,7 +562,7 @@ private void flushCommands(Queue<RedisCommand<?, ?, ?>> queue) {
             }
 
             if (!commands.isEmpty()) {
-                writeToChannelAndFlush(commands);
+                writeToChannelAndFlush(channel, commands);
             }
         }
     }
@@ -626,10 +625,10 @@ public void disconnect() {
 
     private Channel getOpenChannel() {
 
-        Channel currentChannel = this.channel;
+        Channel channel = this.channel;
 
-        if (currentChannel != null) {
-            return currentChannel;
+        if (channel != null /* && channel.isOpen() is this deliberately omitted? */) {
+            return channel;
         }
 
         return null;
@@ -646,6 +645,7 @@ public void reset() {
             logger.debug("{} reset()", logPrefix());
         }
 
+        Channel channel = this.channel;
         if (channel != null) {
             channel.pipeline().fireUserEventTriggered(new ConnectionEvents.Reset());
         }
@@ -718,9 +718,7 @@ public void notifyDrainQueuedCommands(HasQueuedCommands queuedCommands) {
                 }
             }
 
-            if (isConnected()) {
-                flushCommands(disconnectedBuffer);
-            }
+            flushCommands(this.channel, disconnectedBuffer);
         });
     }
 
@@ -802,9 +800,7 @@ private void cancelCommands(String message, Iterable<? extends RedisCommand<?, ?
         }
     }
 
-    private boolean isConnected() {
-
-        Channel channel = this.channel;
+    private boolean isConnected(Channel channel) {
         return channel != null && channel.isActive();
     }
 
diff --git a/src/test/java/io/lettuce/core/protocol/DefaultEndpointUnitTests.java b/src/test/java/io/lettuce/core/protocol/DefaultEndpointUnitTests.java
@@ -124,6 +124,34 @@ void before() {
         sut.setConnectionFacade(connectionFacade);
     }
 
+    @Test
+    void writeShouldGuaranteeFIFOOrder() {
+        sut.write(Collections.singletonList(new Command<>(CommandType.SELECT, new StatusOutput<>(StringCodec.UTF8))));
+
+        sut.registerConnectionWatchdog(connectionWatchdog);
+        doAnswer(i -> sut.write(new Command<>(CommandType.AUTH, new StatusOutput<>(StringCodec.UTF8)))).when(connectionWatchdog)
+                .arm();
+        when(channel.isActive()).thenReturn(true);
+
+        sut.notifyChannelActive(channel);
+
+        DefaultChannelPromise promise = new DefaultChannelPromise(channel, ImmediateEventExecutor.INSTANCE);
+
+        when(channel.writeAndFlush(any())).thenAnswer(invocation -> {
+            if (invocation.getArguments()[0] instanceof RedisCommand) {
+                queue.add((RedisCommand) invocation.getArguments()[0]);
+            }
+
+            if (invocation.getArguments()[0] instanceof Collection) {
+                queue.addAll((Collection) invocation.getArguments()[0]);
+            }
+            return promise;
+        });
+
+        assertThat(queue).hasSize(2).first().hasFieldOrPropertyWithValue("type", CommandType.SELECT);
+        assertThat(queue).hasSize(2).last().hasFieldOrPropertyWithValue("type", CommandType.AUTH);
+    }
+
     @Test
     void writeConnectedShouldWriteCommandToChannel() {
 

Original file line number	Diff line number	Diff line change
`@@ -195,9 +195,9 @@ public <K, V, T> RedisCommand<K, V, T> write(RedisCommand<K, V, T> command) {`
`195`	`195`	`}`
`196`	`196`
`197`	`197`	`if (autoFlushCommands) {`
`198`		`-`
`199`		`- if (isConnected()) {`
`200`		`- writeToChannelAndFlush(command);`
	`198`	`+ Channel channel = this.channel;`
	`199`	`+ if (isConnected(channel)) {`
	`200`	`+ writeToChannelAndFlush(channel, command);`
`201`	`201`	`} else {`
`202`	`202`	`writeToDisconnectedBuffer(command);`
`203`	`203`	`}`
`@@ -236,9 +236,9 @@ public <K, V, T> RedisCommand<K, V, T> write(RedisCommand<K, V, T> command) {`
`236`	`236`	`}`
`237`	`237`
`238`	`238`	`if (autoFlushCommands) {`
`239`		`-`
`240`		`- if (isConnected()) {`
`241`		`- writeToChannelAndFlush(commands);`
	`239`	`+ Channel channel = this.channel;`
	`240`	`+ if (isConnected(channel)) {`
	`241`	`+ writeToChannelAndFlush(channel, commands);`
`242`	`242`	`} else {`
`243`	`243`	`writeToDisconnectedBuffer(commands);`
`244`	`244`	`}`
`@@ -288,10 +288,9 @@ private RedisException validateWrite(int commands) {`
`288`	`288`	`return new RedisException("Connection is closed");`
`289`	`289`	`}`
`290`	`290`
	`291`	`+ final boolean connected = isConnected(this.channel);`
`291`	`292`	`if (usesBoundedQueues()) {`
`292`	`293`
`293`		`- boolean connected = isConnected();`
`294`		`-`
`295`	`294`	`if (QUEUE_SIZE.get(this) + commands > clientOptions.getRequestQueueSize()) {`
`296`	`295`	`return new RedisException("Request queue size exceeded: " + clientOptions.getRequestQueueSize()`
`297`	`296`	`+ ". Commands are not accepted until the queue size drops.");`
`@@ -308,7 +307,7 @@ private RedisException validateWrite(int commands) {`
`308`	`307`	`}`
`309`	`308`	`}`
`310`	`309`
`311`		`- if (!isConnected() && rejectCommandsWhileDisconnected) {`
	`310`	`+ if (!connected && rejectCommandsWhileDisconnected) {`
`312`	`311`	`return new RedisException("Currently not connected. Commands are rejected.");`
`313`	`312`	`}`
`314`	`313`
`@@ -370,11 +369,11 @@ private void writeToDisconnectedBuffer(RedisCommand<?, ?, ?> command) {`
`370`	`369`	`commandBuffer.add(command);`
`371`	`370`	`}`
`372`	`371`
`373`		`- private void writeToChannelAndFlush(RedisCommand<?, ?, ?> command) {`
	`372`	`+ private void writeToChannelAndFlush(Channel channel, RedisCommand<?, ?, ?> command) {`
`374`	`373`
`375`	`374`	`QUEUE_SIZE.incrementAndGet(this);`
`376`	`375`
`377`		`- ChannelFuture channelFuture = channelWriteAndFlush(command);`
	`376`	`+ ChannelFuture channelFuture = channelWriteAndFlush(channel, command);`
`378`	`377`
`379`	`378`	`if (reliability == Reliability.AT_MOST_ONCE) {`
`380`	`379`	`// cancel on exceptions and remove from queue, because there is no housekeeping`
`@@ -387,30 +386,30 @@ private void writeToChannelAndFlush(RedisCommand<?, ?, ?> command) {`
`387`	`386`	`}`
`388`	`387`	`}`
`389`	`388`
`390`		`- private void writeToChannelAndFlush(Collection<? extends RedisCommand<?, ?, ?>> commands) {`
	`389`	`+ private void writeToChannelAndFlush(Channel channel, Collection<? extends RedisCommand<?, ?, ?>> commands) {`
`391`	`390`
`392`	`391`	`QUEUE_SIZE.addAndGet(this, commands.size());`
`393`	`392`
`394`	`393`	`if (reliability == Reliability.AT_MOST_ONCE) {`
`395`	`394`
`396`	`395`	`// cancel on exceptions and remove from queue, because there is no housekeeping`
`397`	`396`	`for (RedisCommand<?, ?, ?> command : commands) {`
`398`		`- channelWrite(command).addListener(AtMostOnceWriteListener.newInstance(this, command));`
	`397`	`+ channelWrite(channel, command).addListener(AtMostOnceWriteListener.newInstance(this, command));`
`399`	`398`	`}`
`400`	`399`	`}`
`401`	`400`
`402`	`401`	`if (reliability == Reliability.AT_LEAST_ONCE) {`
`403`	`402`
`404`	`403`	`// commands are ok to stay within the queue, reconnect will retrigger them`
`405`	`404`	`for (RedisCommand<?, ?, ?> command : commands) {`
`406`		`- channelWrite(command).addListener(RetryListener.newInstance(this, command));`
	`405`	`+ channelWrite(channel, command).addListener(RetryListener.newInstance(this, command));`
`407`	`406`	`}`
`408`	`407`	`}`
`409`	`408`
`410`		`- channelFlush();`
	`409`	`+ channelFlush(channel);`
`411`	`410`	`}`
`412`	`411`
`413`		`- private void channelFlush() {`
	`412`	`+ private void channelFlush(Channel channel) {`
`414`	`413`
`415`	`414`	`if (debugEnabled) {`
`416`	`415`	`logger.debug("{} write() channelFlush", logPrefix());`
`@@ -419,7 +418,7 @@ private void channelFlush() {`
`419`	`418`	`channel.flush();`
`420`	`419`	`}`
`421`	`420`
`422`		`- private ChannelFuture channelWrite(RedisCommand<?, ?, ?> command) {`
	`421`	`+ private ChannelFuture channelWrite(Channel channel, RedisCommand<?, ?, ?> command) {`
`423`	`422`
`424`	`423`	`if (debugEnabled) {`
`425`	`424`	`logger.debug("{} write() channelWrite command {}", logPrefix(), command);`
`@@ -428,7 +427,7 @@ private ChannelFuture channelWrite(RedisCommand<?, ?, ?> command) {`
`428`	`427`	`return channel.write(command);`
`429`	`428`	`}`
`430`	`429`
`431`		`- private ChannelFuture channelWriteAndFlush(RedisCommand<?, ?, ?> command) {`
	`430`	`+ private ChannelFuture channelWriteAndFlush(Channel channel, RedisCommand<?, ?, ?> command) {`
`432`	`431`
`433`	`432`	`if (debugEnabled) {`
`434`	`433`	`logger.debug("{} write() writeAndFlush command {}", logPrefix(), command);`
`@@ -441,7 +440,6 @@ private ChannelFuture channelWriteAndFlush(RedisCommand<?, ?, ?> command) {`
`441`	`440`	`public void notifyChannelActive(Channel channel) {`
`442`	`441`
`443`	`442`	`this.logPrefix = null;`
`444`		`- this.channel = channel;`
`445`	`443`	`this.connectionError = null;`
`446`	`444`
`447`	`445`	`if (isClosed()) {`
`@@ -456,6 +454,7 @@ public void notifyChannelActive(Channel channel) {`
`456`	`454`	`}`
`457`	`455`
`458`	`456`	`sharedLock.doExclusive(() -> {`
	`457`	`+ this.channel = channel;`
`459`	`458`
`460`	`459`	`try {`
`461`	`460`	`// Move queued commands to buffer before issuing any commands because of connection activation.`
`@@ -478,7 +477,7 @@ public void notifyChannelActive(Channel channel) {`
`478`	`477`	`inActivation = false;`
`479`	`478`	`}`
`480`	`479`
`481`		`- flushCommands(disconnectedBuffer);`
	`480`	`+ flushCommands(channel, disconnectedBuffer);`
`482`	`481`	`} catch (Exception e) {`
`483`	`482`
`484`	`483`	`if (debugEnabled) {`
`@@ -525,7 +524,7 @@ public void notifyException(Throwable t) {`
`525`	`524`	`doExclusive(this::drainCommands).forEach(cmd -> cmd.completeExceptionally(t));`
`526`	`525`	`}`
`527`	`526`
`528`		`- if (!isConnected()) {`
	`527`	`+ if (!isConnected(this.channel)) {`
`529`	`528`	`connectionError = t;`
`530`	`529`	`}`
`531`	`530`	`}`
`@@ -538,16 +537,16 @@ public void registerConnectionWatchdog(ConnectionWatchdog connectionWatchdog) {`
`538`	`537`	`@Override`
`539`	`538`	`@SuppressWarnings({ "rawtypes", "unchecked" })`
`540`	`539`	`public void flushCommands() {`
`541`		`- flushCommands(commandBuffer);`
	`540`	`+ flushCommands(this.channel, commandBuffer);`
`542`	`541`	`}`
`543`	`542`
`544`		`- private void flushCommands(Queue<RedisCommand<?, ?, ?>> queue) {`
	`543`	`+ private void flushCommands(Channel channel, Queue<RedisCommand<?, ?, ?>> queue) {`
`545`	`544`
`546`	`545`	`if (debugEnabled) {`
`547`	`546`	`logger.debug("{} flushCommands()", logPrefix());`
`548`	`547`	`}`
`549`	`548`
`550`		`- if (isConnected()) {`
	`549`	`+ if (isConnected(channel)) {`
`551`	`550`
`552`	`551`	`List<RedisCommand<?, ?, ?>> commands = sharedLock.doExclusive(() -> {`
`553`	`552`
`@@ -563,7 +562,7 @@ private void flushCommands(Queue<RedisCommand<?, ?, ?>> queue) {`
`563`	`562`	`}`
`564`	`563`
`565`	`564`	`if (!commands.isEmpty()) {`
`566`		`- writeToChannelAndFlush(commands);`
	`565`	`+ writeToChannelAndFlush(channel, commands);`
`567`	`566`	`}`
`568`	`567`	`}`
`569`	`568`	`}`
`@@ -626,10 +625,10 @@ public void disconnect() {`
`626`	`625`
`627`	`626`	`private Channel getOpenChannel() {`
`628`	`627`
`629`		`- Channel currentChannel = this.channel;`
	`628`	`+ Channel channel = this.channel;`
`630`	`629`
`631`		`- if (currentChannel != null) {`
`632`		`- return currentChannel;`
	`630`	`+ if (channel != null /* && channel.isOpen() is this deliberately omitted? */) {`
	`631`	`+ return channel;`
`633`	`632`	`}`
`634`	`633`
`635`	`634`	`return null;`
`@@ -646,6 +645,7 @@ public void reset() {`
`646`	`645`	`logger.debug("{} reset()", logPrefix());`
`647`	`646`	`}`
`648`	`647`
	`648`	`+ Channel channel = this.channel;`
`649`	`649`	`if (channel != null) {`
`650`	`650`	`channel.pipeline().fireUserEventTriggered(new ConnectionEvents.Reset());`
`651`	`651`	`}`
`@@ -718,9 +718,7 @@ public void notifyDrainQueuedCommands(HasQueuedCommands queuedCommands) {`
`718`	`718`	`}`
`719`	`719`	`}`
`720`	`720`
`721`		`- if (isConnected()) {`
`722`		`- flushCommands(disconnectedBuffer);`
`723`		`- }`
	`721`	`+ flushCommands(this.channel, disconnectedBuffer);`
`724`	`722`	`});`
`725`	`723`	`}`
`726`	`724`
`@@ -802,9 +800,7 @@ private void cancelCommands(String message, Iterable<? extends RedisCommand<?, ?`
`802`	`800`	`}`
`803`	`801`	`}`
`804`	`802`
`805`		`- private boolean isConnected() {`
`806`		`-`
`807`		`- Channel channel = this.channel;`
	`803`	`+ private boolean isConnected(Channel channel) {`
`808`	`804`	`return channel != null && channel.isActive();`
`809`	`805`	`}`
`810`	`806`