Skip to content

Commit

Permalink
fix: revert protectMode upon channelActive event, complete non-retrya…
Browse files Browse the repository at this point in the history
…ble activation command in drainStackUponChannelInactive()
  • Loading branch information
okg-cxf committed Aug 17, 2024
1 parent 7a54c8b commit e328aa0
Show file tree
Hide file tree
Showing 2 changed files with 27 additions and 22 deletions.
13 changes: 9 additions & 4 deletions src/main/java/io/lettuce/core/protocol/CommandHandler.java
Original file line number Diff line number Diff line change
Expand Up @@ -190,13 +190,18 @@ void setBuffer(ByteBuf buffer) {
return drainCommands(stack);
}

private Deque<RedisCommand<?, ?, ?>> drainStack() {
private Deque<RedisCommand<?, ?, ?>> drainStackUponChannelInactive() {
final Deque<RedisCommand<?, ?, ?>> target = new ArrayDeque<>(stack.size());

RedisCommand<?, ?, ?> cmd;
while ((cmd = stack.poll()) != null) {
if (!cmd.isDone() && !ActivationCommand.isActivationCommand(cmd)) {
target.add(cmd);
if (!cmd.isDone()) {
if (!ActivationCommand.isActivationCommand(cmd)) {
target.add(cmd);
} else {
cmd.completeExceptionally(
new RedisConnectionException("activation command won't be retried upon channel inactive"));
}
}
}

Expand Down Expand Up @@ -379,7 +384,7 @@ public void channelInactive(ChannelHandlerContext ctx) throws Exception {
endpoint.notifyChannelInactive(ctx.channel());
Deque<RedisCommand<?, ?, ?>> autoBatchFlushRetryableDrainQueuedCommands = UnmodifiableDeque.emptyDeque();
if (supportsAutoBatchFlush) {
autoBatchFlushRetryableDrainQueuedCommands = drainStack();
autoBatchFlushRetryableDrainQueuedCommands = drainStackUponChannelInactive();
} else {
endpoint.notifyDrainQueuedCommands(this);
}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -133,6 +133,8 @@ protected static void cancelCommandOnEndpointClose(RedisCommand<?, ?, ?> cmd) {

protected volatile @Nonnull ContextualChannel channel = DummyContextualChannelInstances.CHANNEL_CONNECTING;

private volatile Throwable failedToReconnectReason;

private final Consumer<RedisCommand<?, ?, ?>> callbackOnClose;

private final boolean rejectCommandsWhileDisconnected;
Expand All @@ -153,19 +155,17 @@ protected static void cancelCommandOnEndpointClose(RedisCommand<?, ?, ?> cmd) {

private ConnectionFacade connectionFacade;

private volatile Throwable connectionError;

private final String cachedEndpointId;

protected final UnboundedOfferFirstQueue<Object> taskQueue;

private final boolean canFire;

private volatile boolean inProtectMode;
private volatile EventLoop lastEventLoop = null;

private volatile Throwable failedToReconnectReason;
private volatile Throwable connectionError;

private volatile EventLoop lastEventLoop = null;
private volatile boolean inProtectMode;

private final int writeSpinCount;

Expand Down Expand Up @@ -316,17 +316,16 @@ private <V, K> void writeAndFlushActivationCommands(ContextualChannel chan,
@Override
public void notifyChannelActive(Channel channel) {
final ContextualChannel contextualChannel = new ContextualChannel(channel, ConnectionContext.State.CONNECTED);

this.logPrefix = null;
this.connectionError = null;

if (!CHANNEL.compareAndSet(this, DummyContextualChannelInstances.CHANNEL_CONNECTING, contextualChannel)) {
channel.close();
onUnexpectedState("notifyChannelActive", ConnectionContext.State.CONNECTING);
return;
}

lastEventLoop = channel.eventLoop();
this.lastEventLoop = channel.eventLoop();
this.connectionError = null;
this.inProtectMode = false;
this.logPrefix = null;

// Created a synchronize-before with set channel to CHANNEL_CONNECTING,
if (isClosed()) {
Expand Down Expand Up @@ -398,7 +397,7 @@ public void notifyChannelInactive(Channel channel) {

@Override
public void notifyChannelInactiveAfterWatchdogDecision(Channel channel,
Deque<RedisCommand<?, ?, ?>> retryableQueuedCommands) {
Deque<RedisCommand<?, ?, ?>> retryablePendingCommands) {
final ContextualChannel inactiveChan = this.channel;
if (!inactiveChan.context.initialState.isConnected()) {
logger.error("[unexpected][{}] notifyChannelInactive: channel initial state not connected", logPrefix());
Expand Down Expand Up @@ -446,7 +445,7 @@ public void notifyChannelInactiveAfterWatchdogDecision(Channel channel,
CHANNEL.set(this, DummyContextualChannelInstances.CHANNEL_ENDPOINT_CLOSED);
}
inactiveChan.context
.setCloseStatus(new ConnectionContext.CloseStatus(willReconnect, retryableQueuedCommands, exception));
.setCloseStatus(new ConnectionContext.CloseStatus(willReconnect, retryablePendingCommands, exception));
trySetEndpointQuiescence(inactiveChan);
}

Expand Down Expand Up @@ -945,11 +944,11 @@ private <K, V, T> RedisCommand<K, V, T> processActivationCommand(RedisCommand<K,

private Throwable validateWrite(ContextualChannel chan, int commands, boolean isActivationCommand) {
if (isClosed()) {
return new RedisException("Connection is closed");
return new RedisException("Endpoint is closed");
}

final Throwable localConnectionErr = connectionError;
if (localConnectionErr != null /* different logic of DefaultEndpoint */) {
if (localConnectionErr != null /* attention: different logic of DefaultEndpoint */) {
return localConnectionErr;
}

Expand All @@ -961,18 +960,19 @@ private Throwable validateWrite(ContextualChannel chan, int commands, boolean is

final ConnectionContext.State initialState = chan.context.initialState;
final boolean rejectCommandsWhileDisconnectedLocal = this.rejectCommandsWhileDisconnected || isActivationCommand;
final String rejectDesc = isActivationCommand ? "isActivationCommand" : "rejectCommandsWhileDisconnected";
switch (initialState) {
case ENDPOINT_CLOSED:
return new RedisException("Connection is closed");
case RECONNECT_FAILED:
return failedToReconnectReason;
return getFailedToReconnectReason();
case WILL_RECONNECT:
case CONNECTING:
return rejectCommandsWhileDisconnectedLocal
? new RedisException("Currently not connected. Commands are rejected.")
return rejectCommandsWhileDisconnectedLocal ? new RedisException("Currently not connected and " + rejectDesc)
: null;
case CONNECTED:
return !chan.isActive() && rejectCommandsWhileDisconnectedLocal ? new RedisException("Channel is closed")
return !chan.isActive() && rejectCommandsWhileDisconnectedLocal
? new RedisException("Channel is inactive and " + rejectDesc)
: null;
default:
throw new IllegalStateException("unexpected state: " + initialState);
Expand Down

0 comments on commit e328aa0

Please sign in to comment.