chore: remove unused log

okg-cxf · okg-cxf · commit 4093c19cc311 · 2024-08-01T19:38:33.000+08:00
diff --git a/src/main/java/io/lettuce/core/ContextualChannel.java b/src/main/java/io/lettuce/core/ContextualChannel.java
@@ -1,5 +1,7 @@
 package io.lettuce.core;
 
+import java.net.SocketAddress;
+
 import io.lettuce.core.context.ConnectionContext;
 import io.netty.buffer.ByteBufAllocator;
 import io.netty.channel.Channel;
@@ -15,8 +17,6 @@
 import io.netty.util.AttributeKey;
 import org.jetbrains.annotations.NotNull;
 
-import java.net.SocketAddress;
-
 /**
  * @author chenxiaofan
  */
@@ -30,6 +30,10 @@ public ConnectionContext getContext() {
         return context;
     }
 
+    public ConnectionContext.State getInitialState() {
+        return context.getInitialState();
+    }
+
     public Channel getDelegate() {
         return delegate;
     }
diff --git a/src/main/java/io/lettuce/core/context/BatchFlushEndPointContext.java b/src/main/java/io/lettuce/core/context/BatchFlushEndPointContext.java
@@ -45,6 +45,10 @@ public boolean tryEnterSafeGetVolatile() {
             return safe.get() == 0 && /* rare case if QPS is high */ safe.compareAndSet(0, 1);
         }
 
+        public void exitSafe() {
+            safe.set(0);
+        }
+
         /**
          * This method is not thread safe, can only be used from single thread.
          *
@@ -58,10 +62,6 @@ public boolean tryEnterUnsafe() {
             return true;
         }
 
-        public void exitSafe() {
-            safe.set(0);
-        }
-
         public void exitUnsafe() {
             unsafe = false;
         }
diff --git a/src/main/java/io/lettuce/core/protocol/DefaultBatchFlushEndpoint.java b/src/main/java/io/lettuce/core/protocol/DefaultBatchFlushEndpoint.java
@@ -112,8 +112,6 @@ protected static void cancelCommandOnEndpointClose(RedisCommand<?, ?, ?> cmd) {
         cmd.cancel();
     }
 
-    protected volatile @Nonnull ContextualChannel channel = DummyContextualChannelInstances.CHANNEL_CONNECTING;
-
     private final Reliability reliability;
 
     private final ClientOptions clientOptions;
@@ -123,12 +121,13 @@ protected static void cancelCommandOnEndpointClose(RedisCommand<?, ?, ?> cmd) {
     private final boolean boundedQueues;
 
     // access via QUEUE_SIZE
-    @SuppressWarnings("unused")
     private volatile int queueSize = 0;
 
     // access via STATUS
-    @SuppressWarnings("unused")
     private volatile int status = ST_OPEN;
+    // access via CHANNEL
+
+    protected volatile @Nonnull ContextualChannel channel = DummyContextualChannelInstances.CHANNEL_CONNECTING;
 
     private final Consumer<RedisCommand<?, ?, ?>> callbackOnClose;
 
@@ -152,10 +151,6 @@ protected static void cancelCommandOnEndpointClose(RedisCommand<?, ?, ?> cmd) {
 
     private volatile Throwable connectionError;
 
-    // // access via QUEUE_SIZE
-    // @SuppressWarnings("unused")
-    // private volatile int queueSize = 0;
-
     private final String cachedEndpointId;
 
     protected final UnboundedMpscOfferFirstQueue<RedisCommand<?, ?, ?>> taskQueue;
@@ -308,11 +303,8 @@ public void notifyChannelActive(Channel channel) {
         this.connectionError = null;
 
         if (!CHANNEL.compareAndSet(this, DummyContextualChannelInstances.CHANNEL_CONNECTING, contextualChannel)) {
-            logger.error("[unexpected] {} failed to set to CHANNEL_CONNECTING because current state is '{}'", logPrefix(),
-                    CHANNEL.get(this));
             channel.close();
-            onUnexpectedState("notifyChannelActive", ConnectionContext.State.CONNECTING,
-                    this.channel.getContext().getInitialState());
+            onUnexpectedState("notifyChannelActive", ConnectionContext.State.CONNECTING);
             return;
         }
 
@@ -357,21 +349,13 @@ public void notifyChannelActive(Channel channel) {
         }
     }
 
-    private void onUnexpectedState(String caller, ConnectionContext.State exp, ConnectionContext.State actual) {
-        logger.error("[{}][unexpected] {}: unexpected state: exp '{}' got '{}'", caller, logPrefix(), exp, actual);
-        cancelCommands(String.format("%s: state not match: expect '%s', got '%s'", caller, exp, actual));
-    }
-
     @Override
     public void notifyReconnectFailed(Throwable t) {
         this.failedToReconnectReason = t;
 
         if (!CHANNEL.compareAndSet(this, DummyContextualChannelInstances.CHANNEL_CONNECTING,
                 DummyContextualChannelInstances.CHANNEL_RECONNECT_FAILED)) {
-            logger.error("[unexpected] {} failed to set to CHANNEL_CONNECTING because current state is '{}'", logPrefix(),
-                    CHANNEL.get(this));
-            onUnexpectedState("notifyReconnectFailed", ConnectionContext.State.CONNECTING,
-                    this.channel.getContext().getInitialState());
+            onUnexpectedState("notifyReconnectFailed", ConnectionContext.State.CONNECTING);
             return;
         }
 
@@ -396,13 +380,13 @@ public void notifyChannelInactive(Channel channel) {
     @Override
     public void notifyChannelInactiveAfterWatchdogDecision(Channel channel,
             Deque<RedisCommand<?, ?, ?>> retryableQueuedCommands) {
-        final ContextualChannel prevChan = this.channel;
-        if (!prevChan.getContext().getInitialState().isConnected() || prevChan.getDelegate() != channel) {
+        final ContextualChannel inactiveChan = this.channel;
+        if (!inactiveChan.getInitialState().isConnected() || inactiveChan.getDelegate() != channel) {
             logger.error("[unexpected][{}] notifyChannelInactive: channel not match", logPrefix());
             return;
         }
 
-        if (prevChan.getContext().isChannelInactiveEventFired()) {
+        if (inactiveChan.getContext().isChannelInactiveEventFired()) {
             logger.error("[unexpected][{}] notifyChannelInactive: already fired", logPrefix());
             return;
         }
@@ -431,9 +415,9 @@ public void notifyChannelInactiveAfterWatchdogDecision(Channel channel,
         if (!willReconnect) {
             CHANNEL.set(this, DummyContextualChannelInstances.CHANNEL_ENDPOINT_CLOSED);
         }
-        prevChan.getContext()
+        inactiveChan.getContext()
                 .setCloseStatus(new ConnectionContext.CloseStatus(willReconnect, retryableQueuedCommands, exception));
-        trySetEndpointQuiescence(prevChan);
+        trySetEndpointQuiescence(inactiveChan);
     }
 
     @Override
@@ -444,7 +428,7 @@ public void notifyException(Throwable t) {
         }
 
         final ContextualChannel curr = this.channel;
-        if (!curr.getContext().getInitialState().isConnected() || !curr.isActive()) {
+        if (!curr.getInitialState().isConnected() || !curr.isActive()) {
             connectionError = t;
         }
     }
@@ -457,7 +441,7 @@ public void registerConnectionWatchdog(ConnectionWatchdog connectionWatchdog) {
     @Override
     public void flushCommands() {
         final ContextualChannel chan = this.channel;
-        switch (chan.getContext().getInitialState()) {
+        switch (chan.getInitialState()) {
             case ENDPOINT_CLOSED:
                 syncAfterTerminated(() -> {
                     if (isClosed()) {
@@ -486,7 +470,7 @@ public void flushCommands() {
                 scheduleSendJobIfNeeded(chan);
                 return;
             default:
-                throw new IllegalStateException("unexpected state: " + chan.getContext().getInitialState());
+                throw new IllegalStateException("unexpected state: " + chan.getInitialState());
         }
     }
 
@@ -519,8 +503,8 @@ public CompletableFuture<Void> closeAsync() {
                 connectionWatchdog.prepareClose();
             }
 
-            final Channel chan = channel;
-            if (channel.getContext().getInitialState().isConnected()) {
+            final ContextualChannel chan = channel;
+            if (chan.getInitialState().isConnected()) {
                 // 1. STATUS.compareAndSet(this, ST_OPEN, ST_CLOSED) synchronize-before channel == CONNECTED
                 // 2. channel == CONNECTED synchronize-before setting channel to WILL_RECONNECT/ENDPOINT_CLOSED
                 // 3. setting channel to WILL_RECONNECT synchronize-before `isClosed()`, which will cancel all the commands.
@@ -545,7 +529,7 @@ public void disconnect() {
 
         ContextualChannel chan = this.channel;
 
-        if (chan.getContext().getInitialState().isConnected() && chan.isOpen()) {
+        if (chan.getInitialState().isConnected() && chan.isOpen()) {
             chan.disconnect();
         }
     }
@@ -561,9 +545,9 @@ public void reset() {
             logger.debug("{} reset()", logPrefix());
         }
 
-        final ContextualChannel curr = channel;
-        if (curr.getContext().getInitialState().isConnected()) {
-            curr.pipeline().fireUserEventTriggered(new ConnectionEvents.Reset());
+        final ContextualChannel chan = channel;
+        if (chan.getInitialState().isConnected()) {
+            chan.pipeline().fireUserEventTriggered(new ConnectionEvents.Reset());
         }
         // Unsafe to call cancelBufferedCommands() here.
         // cancelBufferedCommands("Reset");
@@ -575,9 +559,9 @@ private void resetInternal() {
             logger.debug("{} reset()", logPrefix());
         }
 
-        ContextualChannel curr = channel;
-        if (curr.getContext().getInitialState().isConnected()) {
-            curr.pipeline().fireUserEventTriggered(new ConnectionEvents.Reset());
+        ContextualChannel chan = channel;
+        if (chan.getInitialState().isConnected()) {
+            chan.pipeline().fireUserEventTriggered(new ConnectionEvents.Reset());
         }
         // Unsafe to call cancelBufferedCommands() here.
         cancelCommands("Reset");
@@ -593,7 +577,7 @@ public void initialState() {
         cancelCommands("initialState");
 
         ContextualChannel currentChannel = this.channel;
-        if (currentChannel.getContext().getInitialState().isConnected()) {
+        if (currentChannel.getInitialState().isConnected()) {
             ChannelFuture close = currentChannel.close();
             if (currentChannel.isOpen()) {
                 close.syncUninterruptibly();
@@ -602,7 +586,7 @@ public void initialState() {
     }
 
     private boolean isClosed() {
-        return STATUS.get(this) == ST_CLOSED;
+        return status == ST_CLOSED;
     }
 
     protected String logPrefix() {
@@ -799,16 +783,14 @@ private void trySetEndpointQuiescence(ContextualChannel chan) {
     }
 
     private void onEndpointQuiescence() {
-        if (channel.getContext().getInitialState() == ConnectionContext.State.ENDPOINT_CLOSED) {
+        if (channel.getInitialState() == ConnectionContext.State.ENDPOINT_CLOSED) {
             return;
         }
 
         // Create happens-before with channelActive()
         if (!CHANNEL.compareAndSet(this, DummyContextualChannelInstances.CHANNEL_WILL_RECONNECT,
                 DummyContextualChannelInstances.CHANNEL_CONNECTING)) {
-
-            onUnexpectedState("onEndpointQuiescence", ConnectionContext.State.WILL_RECONNECT,
-                    this.channel.getContext().getInitialState());
+            onUnexpectedState("onEndpointQuiescence", ConnectionContext.State.WILL_RECONNECT);
             return;
         }
 
@@ -824,17 +806,18 @@ private void onWillReconnect(@Nonnull final ConnectionContext.CloseStatus closeS
             // Save retryable failed tasks
             logger.info(
                     "[onWillReconnect][{}] compensate {} retryableFailedToSendTasks (write failure) for retrying on reconnecting, first write error: {}",
-                    retryableFailedToSendTasks.size(), batchFlushEndPointContext.getFirstDiscontinueReason().getMessage(),
-                    logPrefix());
+                    logPrefix(), retryableFailedToSendTasks.size(),
+                    batchFlushEndPointContext.getFirstDiscontinueReason().getMessage());
             offerFirstAll(retryableFailedToSendTasks);
         }
 
         LettuceAssert.assertState(reliability != Reliability.AT_MOST_ONCE, "unexpected: reliability is AT_MOST_ONCE");
         final Deque<RedisCommand<?, ?, ?>> retryablePendingCommands = closeStatus.getAndClearRetryablePendingCommands();
         if (retryablePendingCommands != null) {
             // Save uncompletedTasks for later retry.
-            logger.info("[onWillReconnect][{}] compensate {} pendingCommands (write success) for retrying on reconnecting",
-                    retryablePendingCommands.size(), logPrefix());
+            logger.info(
+                    "[onWillReconnect][{}] compensate {} retryable pending commands (write success) for retrying on reconnecting",
+                    logPrefix(), retryablePendingCommands.size());
             offerFirstAll(retryablePendingCommands);
         }
 
@@ -961,13 +944,13 @@ private Throwable validateWrite(@SuppressWarnings("unused") int commands) {
             return localConnectionErr;
         }
 
-        if (boundedQueues && QUEUE_SIZE.get(this) + commands > clientOptions.getRequestQueueSize()) {
+        if (boundedQueues && queueSize + commands > clientOptions.getRequestQueueSize()) {
             return new RedisException("Request queue size exceeded: " + clientOptions.getRequestQueueSize()
                     + ". Commands are not accepted until the queue size drops.");
         }
 
         final ContextualChannel chan = this.channel;
-        switch (chan.getContext().getInitialState()) {
+        switch (chan.getInitialState()) {
             case ENDPOINT_CLOSED:
                 return new RedisException("Connection is closed");
             case RECONNECT_FAILED:
@@ -979,10 +962,16 @@ private Throwable validateWrite(@SuppressWarnings("unused") int commands) {
             case CONNECTED:
                 return !chan.isActive() && rejectCommandsWhileDisconnected ? new RedisException("Connection is closed") : null;
             default:
-                throw new IllegalStateException("unexpected state: " + chan.getContext().getInitialState());
+                throw new IllegalStateException("unexpected state: " + chan.getInitialState());
         }
     }
 
+    private void onUnexpectedState(String caller, ConnectionContext.State exp) {
+        final ConnectionContext.State actual = this.channel.getInitialState();
+        logger.error("[{}][unexpected] {}: unexpected state: exp '{}' got '{}'", caller, logPrefix(), exp, actual);
+        cancelCommands(String.format("%s: state not match: expect '%s', got '%s'", caller, exp, actual));
+    }
+
     private void channelFlush(Channel channel) {
         if (debugEnabled) {
             logger.debug("{} write() channelFlush", logPrefix());

Original file line number	Diff line number	Diff line change
`@@ -45,6 +45,10 @@ public boolean tryEnterSafeGetVolatile() {`
`45`	`45`	`return safe.get() == 0 && /* rare case if QPS is high */ safe.compareAndSet(0, 1);`
`46`	`46`	`}`
`47`	`47`
	`48`	`+ public void exitSafe() {`
	`49`	`+ safe.set(0);`
	`50`	`+ }`
	`51`	`+`
`48`	`52`	`/**`
`49`	`53`	`* This method is not thread safe, can only be used from single thread.`
`50`	`54`	`*`
`@@ -58,10 +62,6 @@ public boolean tryEnterUnsafe() {`
`58`	`62`	`return true;`
`59`	`63`	`}`
`60`	`64`
`61`		`- public void exitSafe() {`
`62`		`- safe.set(0);`
`63`		`- }`
`64`		`-`
`65`	`65`	`public void exitUnsafe() {`
`66`	`66`	`unsafe = false;`
`67`	`67`	`}`