diff --git a/src/main/java/io/lettuce/core/cluster/ClusterNodeEndpoint.java b/src/main/java/io/lettuce/core/cluster/ClusterNodeEndpoint.java index de50cbafd0..272f9b2ada 100644 --- a/src/main/java/io/lettuce/core/cluster/ClusterNodeEndpoint.java +++ b/src/main/java/io/lettuce/core/cluster/ClusterNodeEndpoint.java @@ -51,11 +51,28 @@ public CompletableFuture closeAsync() { logger.debug("{} closeAsync()", logPrefix()); + Throwable drainError = null; if (clusterChannelWriter != null) { - retriggerCommands(doExclusive(this::drainCommands)); + try { + retriggerCommands(doExclusive(this::drainCommands)); + } catch (Throwable t) { + drainError = t; + } + } + + CompletableFuture superFuture = super.closeAsync(); + + if (drainError != null) { + final Throwable ex = drainError; + return superFuture.handle((v, t) -> { + if (t != null && t != ex) { + ex.addSuppressed(t); + } + throw (ex instanceof RuntimeException) ? (RuntimeException) ex : new RedisException(ex); + }); } - return super.closeAsync(); + return superFuture; } protected void retriggerCommands(Collection> commands) { diff --git a/src/main/java/io/lettuce/core/protocol/DefaultEndpoint.java b/src/main/java/io/lettuce/core/protocol/DefaultEndpoint.java index 002e64e1e5..88955180fd 100644 --- a/src/main/java/io/lettuce/core/protocol/DefaultEndpoint.java +++ b/src/main/java/io/lettuce/core/protocol/DefaultEndpoint.java @@ -605,14 +605,38 @@ public CompletableFuture closeAsync() { connectionWatchdog.prepareClose(); } - cancelBufferedCommands("Close"); + Throwable cancelError = null; + try { + cancelBufferedCommands("Close"); + } catch (Throwable t) { + cancelError = t; + try { + cancelCommands("Close", drainCommands(), RedisCommand::cancel); + } catch (Throwable drainEx) { + cancelError.addSuppressed(drainEx); + } + } Channel channel = getOpenChannel(); if (channel != null) { - Futures.adapt(channel.close(), closeFuture); + if (cancelError != null) { + final Throwable ex = cancelError; + channel.close().addListener(future -> { + if (!future.isSuccess() && future.cause() != null) { + ex.addSuppressed(future.cause()); + } + closeFuture.completeExceptionally(ex); + }); + } else { + Futures.adapt(channel.close(), closeFuture); + } } else { - closeFuture.complete(null); + if (cancelError != null) { + closeFuture.completeExceptionally(cancelError); + } else { + closeFuture.complete(null); + } } } diff --git a/src/main/java/io/lettuce/core/protocol/SharedLock.java b/src/main/java/io/lettuce/core/protocol/SharedLock.java index c5c7e1f02b..c7617b6910 100644 --- a/src/main/java/io/lettuce/core/protocol/SharedLock.java +++ b/src/main/java/io/lettuce/core/protocol/SharedLock.java @@ -1,11 +1,14 @@ package io.lettuce.core.protocol; +import java.time.Duration; import java.util.WeakHashMap; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicLongFieldUpdater; import java.util.concurrent.locks.Lock; import java.util.concurrent.locks.ReentrantLock; import java.util.function.Supplier; +import io.lettuce.core.RedisException; import io.lettuce.core.internal.LettuceAssert; /** @@ -43,6 +46,28 @@ class SharedLock { private volatile Thread exclusiveLockOwner; + private static final Duration DEFAULT_EXCLUSIVE_LOCK_TIMEOUT = Duration.ofSeconds(30); + + private final long exclusiveLockTimeoutNanos; + + /** + * Create a {@link SharedLock} with the {@link #DEFAULT_EXCLUSIVE_LOCK_TIMEOUT default} exclusive-lock timeout. + */ + SharedLock() { + this(DEFAULT_EXCLUSIVE_LOCK_TIMEOUT); + } + + /** + * Create a {@link SharedLock} with a configurable exclusive-lock acquisition timeout. + * + * @param exclusiveLockTimeout the maximum time an exclusive acquisition waits before failing fast with a + * {@link RedisException}; must not be {@code null}. + */ + SharedLock(Duration exclusiveLockTimeout) { + LettuceAssert.notNull(exclusiveLockTimeout, "Exclusive lock timeout must not be null"); + this.exclusiveLockTimeoutNanos = exclusiveLockTimeout.toNanos(); + } + /** * Wait for stateLock and increment writers. Will wait if stateLock is locked and if writer counter is negative. */ @@ -108,12 +133,13 @@ T doExclusive(Supplier supplier) { LettuceAssert.notNull(supplier, "Supplier must not be null"); - lock.lock(); + long deadline = System.nanoTime() + exclusiveLockTimeoutNanos; + acquireExclusiveGuard(deadline); try { try { - lockWritersExclusive(); + lockWritersExclusive(deadline); return supplier.get(); } finally { unlockWritersExclusive(); @@ -124,10 +150,55 @@ T doExclusive(Supplier supplier) { } /** - * Wait for stateLock and no writers. Must be used in an outer {@code synchronized} block to prevent interleaving with other - * methods using writers. Sets writers to a negative value to create a lock for {@link #incrementWriters()}. + * Acquire the guarding {@link #lock} for an exclusive operation, bounded by {@code deadline}. If the lock was leaked by a + * thread that died inside the guarded region (issue #3804) a plain {@code lock.lock()} would park the caller - frequently a + * Netty event-loop thread - forever. Fail fast with a {@link RedisException} instead so the endpoint can be rebuilt. The + * acquisition remains reentrant, so a thread that already holds the lock re-acquires immediately. + *

+ * Acquisition continues in a loop bounded by {@code deadline} while catching {@link InterruptedException} and restoring the + * interrupt flag afterwards, so callers with their interrupt flag already set (such as during + * {@code close()}/{@code closeAsync()}) do not fail an otherwise successful exclusive operation solely due to the interrupt + * under brief contention. + * + * @param deadline the {@link System#nanoTime()} timestamp by which the acquisition must complete. */ - private void lockWritersExclusive() { + private void acquireExclusiveGuard(long deadline) { + + boolean interrupted = false; + boolean acquired = lock.tryLock(); + try { + while (!acquired) { + long timeoutNanos = deadline - System.nanoTime(); + if (timeoutNanos <= 0) { + break; + } + try { + acquired = lock.tryLock(timeoutNanos, TimeUnit.NANOSECONDS); + } catch (InterruptedException e) { + interrupted = true; + } + } + } finally { + if (interrupted) { + Thread.currentThread().interrupt(); + } + } + + if (!acquired) { + throw new RedisException("Timed out after " + TimeUnit.NANOSECONDS.toMillis(exclusiveLockTimeoutNanos) + + "ms acquiring the exclusive SharedLock; the lock holder likely died inside the guarded region. " + + "The endpoint must be rebuilt to recover."); + } + } + + /** + * Wait for stateLock and no writers, bounded by {@code deadline}. Must be used in an outer {@code synchronized} block to + * prevent interleaving with other methods using writers. Sets writers to a negative value to create a lock for + * {@link #incrementWriters()}. + * + * @param deadline the {@link System#nanoTime()} timestamp by which writer draining must complete. + */ + private void lockWritersExclusive(long deadline) { if (exclusiveLockOwner == Thread.currentThread()) { WRITERS.decrementAndGet(this); @@ -138,12 +209,36 @@ private void lockWritersExclusive() { try { for (;;) { - // allow reentrant exclusive lock by comparing writers count and threadWriters count + // allow reentrant exclusive lock by comparing writers count and threadWriters + // count int threadWriterCount = getThreadWriterCount(); if (WRITERS.compareAndSet(this, threadWriterCount, -1)) { exclusiveLockOwner = Thread.currentThread(); return; } + + // A leaked shared writer (a thread that died before decrementWriters()) keeps + // the writer count above this thread's own count, so the CAS above can never + // succeed. + // Or if exclusive mode was abandoned (writers is negative, e.g. a previous + // exclusive holder died), + // the CAS also cannot succeed. Bound the spin and fail fast instead of burning + // a CPU core forever + // (issues #3804, #3880). + if (System.nanoTime() - deadline >= 0) { + long currentWriters = WRITERS.get(this); + if (currentWriters < 0) { + throw new RedisException("Timed out after " + TimeUnit.NANOSECONDS.toMillis(exclusiveLockTimeoutNanos) + + "ms acquiring the exclusive SharedLock; the exclusive lock was likely abandoned by a previous holder. " + + "The endpoint must be rebuilt to recover."); + } + + throw new RedisException("Timed out after " + TimeUnit.NANOSECONDS.toMillis(exclusiveLockTimeoutNanos) + + "ms waiting for " + currentWriters + " shared writer(s) to drain while acquiring the " + + "exclusive SharedLock; a shared writer was likely leaked. The endpoint must be rebuilt to recover."); + } + + Thread.yield(); } } finally { lock.unlock(); diff --git a/src/test/java/io/lettuce/core/cluster/ClusterNodeEndpointUnitTests.java b/src/test/java/io/lettuce/core/cluster/ClusterNodeEndpointUnitTests.java index c11c28c30a..d60f58502f 100644 --- a/src/test/java/io/lettuce/core/cluster/ClusterNodeEndpointUnitTests.java +++ b/src/test/java/io/lettuce/core/cluster/ClusterNodeEndpointUnitTests.java @@ -6,7 +6,9 @@ import static org.mockito.ArgumentMatchers.*; import static org.mockito.Mockito.*; +import java.time.Duration; import java.util.Queue; +import java.util.concurrent.CompletableFuture; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Tag; @@ -145,6 +147,34 @@ void closeWithBufferedCommandsFails() { assertThatThrownBy(() -> TestFutures.awaitOrTimeout(command)).isInstanceOf(RedisException.class); } + @Test + void closeAsyncClosesSuperWhenSharedLockTimesOut() throws Exception { + + Class sharedLockClass = Class.forName("io.lettuce.core.protocol.SharedLock"); + java.lang.reflect.Constructor ctor = sharedLockClass.getDeclaredConstructor(Duration.class); + ctor.setAccessible(true); + Object sharedLock = ctor.newInstance(Duration.ofMillis(50)); + ReflectionTestUtils.setField(sut, "sharedLock", sharedLock); + + java.lang.reflect.Method incrementWriters = sharedLockClass.getDeclaredMethod("incrementWriters"); + incrementWriters.setAccessible(true); + + Thread leaker = new Thread(() -> { + try { + incrementWriters.invoke(sharedLock); + } catch (Exception ignored) { + } + }); + leaker.start(); + leaker.join(1000); + + CompletableFuture closeFuture = sut.closeAsync(); + assertThat(closeFuture).isNotNull(); + assertThat(sut.isClosed()).isTrue(); + assertThat(closeFuture.isCompletedExceptionally()).isTrue(); + assertThatThrownBy(closeFuture::join).hasCauseInstanceOf(RedisException.class); + } + private void prepareNewEndpoint() { sut = new ClusterNodeEndpoint(clientOptions, clientResources, clusterChannelWriter); disconnectedBuffer = (Queue) ReflectionTestUtils.getField(sut, "disconnectedBuffer"); diff --git a/src/test/java/io/lettuce/core/protocol/DefaultEndpointUnitTests.java b/src/test/java/io/lettuce/core/protocol/DefaultEndpointUnitTests.java index 06354e5d61..37ac70c5f6 100644 --- a/src/test/java/io/lettuce/core/protocol/DefaultEndpointUnitTests.java +++ b/src/test/java/io/lettuce/core/protocol/DefaultEndpointUnitTests.java @@ -5,7 +5,9 @@ import static org.mockito.ArgumentMatchers.*; import static org.mockito.Mockito.*; +import java.lang.reflect.Field; import java.nio.channels.ClosedChannelException; +import java.time.Duration; import java.util.Collection; import java.util.Collections; import java.util.Queue; @@ -47,6 +49,7 @@ import io.netty.channel.DefaultChannelPromise; import io.netty.channel.EventLoop; import io.netty.handler.codec.EncoderException; +import io.netty.util.concurrent.GenericFutureListener; import io.netty.util.concurrent.ImmediateEventExecutor; /** @@ -378,6 +381,99 @@ void retryListenerDoesNotRetryCompletedCommands() { verify(channel, never()).writeAndFlush(command); } + @Test + void closeAsyncSucceedsWhenThreadIsInterrupted() { + + when(channel.close()).thenReturn(mock(ChannelFuture.class)); + sut.notifyChannelActive(channel); + + Thread.currentThread().interrupt(); + try { + CompletableFuture closeFuture = sut.closeAsync(); + assertThat(closeFuture).isNotNull(); + assertThat(sut.isClosed()).isTrue(); + assertThat(Thread.currentThread().isInterrupted()).isTrue(); + } finally { + Thread.interrupted(); + } + } + + @Test + @SuppressWarnings("unchecked") + void closeAsyncCompletesExceptionallyWhenSharedLockTimesOut() throws Exception { + + SharedLock sharedLock = new SharedLock(Duration.ofMillis(50)); + Field sharedLockField = DefaultEndpoint.class.getDeclaredField("sharedLock"); + sharedLockField.setAccessible(true); + sharedLockField.set(sut, sharedLock); + + ChannelFuture channelCloseFuture = mock(ChannelFuture.class); + when(channel.close()).thenReturn(channelCloseFuture); + sut.notifyChannelActive(channel); + + Thread leaker = new Thread(sharedLock::incrementWriters); + leaker.start(); + leaker.join(1000); + + CompletableFuture closeFuture = sut.closeAsync(); + assertThat(closeFuture).isNotNull(); + assertThat(sut.isClosed()).isTrue(); + + verify(channel).close(); + + ArgumentCaptor captor = ArgumentCaptor.forClass(GenericFutureListener.class); + verify(channelCloseFuture).addListener(captor.capture()); + when(channelCloseFuture.isSuccess()).thenReturn(true); + captor.getValue().operationComplete(channelCloseFuture); + + assertThat(closeFuture).isCompletedExceptionally(); + assertThatThrownBy(closeFuture::join).hasCauseInstanceOf(RedisException.class).hasMessageContaining("shared writer"); + + assertThat(sut.closeAsync()).isSameAs(closeFuture); + } + + @Test + void closeAsyncCompletesExceptionallyWhenSharedLockTimesOutAndChannelIsNull() throws Exception { + + SharedLock sharedLock = new SharedLock(Duration.ofMillis(50)); + Field sharedLockField = DefaultEndpoint.class.getDeclaredField("sharedLock"); + sharedLockField.setAccessible(true); + sharedLockField.set(sut, sharedLock); + + Thread leaker = new Thread(sharedLock::incrementWriters); + leaker.start(); + leaker.join(1000); + + CompletableFuture closeFuture = sut.closeAsync(); + assertThat(closeFuture).isNotNull(); + assertThat(sut.isClosed()).isTrue(); + assertThat(closeFuture).isCompletedExceptionally(); + assertThatThrownBy(closeFuture::join).hasCauseInstanceOf(RedisException.class).hasMessageContaining("shared writer"); + } + + @Test + void closeAsyncCancelsBufferedCommandsWhenSharedLockTimesOutAndChannelIsNull() throws Exception { + + SharedLock sharedLock = new SharedLock(Duration.ofMillis(50)); + Field sharedLockField = DefaultEndpoint.class.getDeclaredField("sharedLock"); + sharedLockField.setAccessible(true); + sharedLockField.set(sut, sharedLock); + + RedisCommand command = new AsyncCommand<>( + new Command<>(CommandType.PING, new StatusOutput<>(StringCodec.UTF8))); + sut.write(command); + + Thread leaker = new Thread(sharedLock::incrementWriters); + leaker.start(); + leaker.join(1000); + + CompletableFuture closeFuture = sut.closeAsync(); + assertThat(closeFuture).isNotNull(); + assertThat(sut.isClosed()).isTrue(); + assertThat(closeFuture).isCompletedExceptionally(); + assertThat(command.isCancelled()).isTrue(); + } + @Test void shouldWrapActivationCommands() { diff --git a/src/test/java/io/lettuce/core/protocol/SharedLockUnitTests.java b/src/test/java/io/lettuce/core/protocol/SharedLockUnitTests.java index 54387b0c95..be8633c3de 100644 --- a/src/test/java/io/lettuce/core/protocol/SharedLockUnitTests.java +++ b/src/test/java/io/lettuce/core/protocol/SharedLockUnitTests.java @@ -3,11 +3,16 @@ import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.Tag; import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; import java.lang.reflect.Field; +import java.time.Duration; import java.util.WeakHashMap; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; +import java.util.concurrent.locks.Lock; + +import io.lettuce.core.RedisException; import static io.lettuce.TestTags.UNIT_TEST; import java.util.concurrent.atomic.AtomicInteger; @@ -219,4 +224,157 @@ public void singleThreadLocalEntryPerThread() throws Exception { lock3.decrementWriters(); } + /** + * #3804 — a {@code ReentrantLock} leaked by a thread that died inside the guarded region must not park the next exclusive + * caller forever. {@code doExclusive} should fail fast with a {@link RedisException} once the bounded acquire elapses. + */ + @Test + @Timeout(5) + public void doExclusiveFailsFastWhenGuardLockIsLeaked() throws Exception { + final SharedLock sharedLock = new SharedLock(Duration.ofMillis(200)); + + Lock internalLock = extractInternalLock(sharedLock); + CountDownLatch held = new CountDownLatch(1); + CountDownLatch release = new CountDownLatch(1); + + // Hold the internal lock without ever releasing it during the test window — simulates the leaked lock. + Thread holder = new Thread(() -> { + internalLock.lock(); + held.countDown(); + try { + release.await(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } finally { + internalLock.unlock(); + } + }); + holder.setDaemon(true); + holder.start(); + + Assertions.assertTrue(held.await(1, TimeUnit.SECONDS), "holder thread failed to acquire the internal lock"); + + try { + RedisException ex = Assertions.assertThrows(RedisException.class, + () -> sharedLock.doExclusive(() -> "unreachable")); + Assertions.assertTrue(ex.getMessage().contains("Timed out"), "unexpected message: " + ex.getMessage()); + } finally { + release.countDown(); + holder.join(TimeUnit.SECONDS.toMillis(1)); + } + } + + /** + * #3880 — a shared writer leaked by another thread (incremented, never decremented) keeps the writer count above the + * exclusive caller's own count, so the CAS in {@code lockWritersExclusive} can never succeed. Instead of spinning RUNNABLE + * at 100% CPU forever, {@code doExclusive} should bound the spin and fail fast with a {@link RedisException}. + */ + @Test + @Timeout(5) + public void doExclusiveFailsFastWhenSharedWriterIsLeaked() throws Exception { + final SharedLock sharedLock = new SharedLock(Duration.ofMillis(200)); + + // Leak a shared writer from another thread: increment on a thread that finishes without decrementing. + Thread leaker = new Thread(sharedLock::incrementWriters); + leaker.start(); + leaker.join(TimeUnit.SECONDS.toMillis(1)); + + RedisException ex = Assertions.assertThrows(RedisException.class, () -> sharedLock.doExclusive(() -> "unreachable")); + Assertions.assertTrue(ex.getMessage().contains("shared writer"), "unexpected message: " + ex.getMessage()); + } + + /** + * When exclusive mode was abandoned (for example, writers is left at -1 because a previous exclusive holder died), + * {@code lockWritersExclusive} must fail fast and report an abandoned exclusive lock rather than claiming a negative shared + * writer count or attributing the failure to a leaked shared writer (#3880). + */ + @Test + @Timeout(5) + public void doExclusiveFailsFastWhenExclusiveModeIsAbandoned() throws Exception { + final SharedLock sharedLock = new SharedLock(Duration.ofMillis(200)); + + Field writersField = SharedLock.class.getDeclaredField("writers"); + writersField.setAccessible(true); + writersField.set(sharedLock, -1L); + + RedisException ex = Assertions.assertThrows(RedisException.class, () -> sharedLock.doExclusive(() -> "unreachable")); + Assertions.assertTrue(ex.getMessage().contains("abandoned"), "unexpected message: " + ex.getMessage()); + Assertions.assertFalse(ex.getMessage().contains("shared writer"), + "message should not claim a shared writer was leaked: " + ex.getMessage()); + } + + /** + * When the calling thread's interrupt flag is already set, an uncontended {@code doExclusive} must still succeed without + * throwing {@link RedisException}, and must preserve the interrupted status on the thread. + */ + @Test + public void doExclusiveSucceedsWhenThreadIsInterruptedAndLockIsUncontended() { + final SharedLock sharedLock = new SharedLock(); + + Thread.currentThread().interrupt(); + try { + String result = sharedLock.doExclusive(() -> "ok"); + Assertions.assertEquals("ok", result); + Assertions.assertTrue(Thread.currentThread().isInterrupted(), "interrupt flag should be preserved"); + } finally { + Thread.interrupted(); // clear interrupted status + } + } + + /** + * When the calling thread's interrupt flag is already set and the lock experiences brief contention, {@code doExclusive} + * must wait for contention to clear, succeed without throwing {@link RedisException}, and preserve the interrupted status + * on the thread. + */ + @Test + @Timeout(5) + public void doExclusiveSucceedsWhenThreadIsInterruptedAndLockIsContended() throws Exception { + final SharedLock sharedLock = new SharedLock(Duration.ofSeconds(2)); + Lock internalLock = extractInternalLock(sharedLock); + + CountDownLatch held = new CountDownLatch(1); + CountDownLatch release = new CountDownLatch(1); + + Thread holder = new Thread(() -> { + internalLock.lock(); + held.countDown(); + try { + release.await(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } finally { + internalLock.unlock(); + } + }); + holder.start(); + + Assertions.assertTrue(held.await(1, TimeUnit.SECONDS), "holder failed to acquire lock"); + + Thread.currentThread().interrupt(); + try { + // Release after 50ms so contention clears before the 2s timeout + new Thread(() -> { + try { + Thread.sleep(50); + } catch (InterruptedException ignored) { + } + release.countDown(); + }).start(); + + String result = sharedLock.doExclusive(() -> "ok"); + Assertions.assertEquals("ok", result); + Assertions.assertTrue(Thread.currentThread().isInterrupted(), "interrupt flag should be preserved"); + } finally { + release.countDown(); + holder.join(1000); + Thread.interrupted(); + } + } + + private static Lock extractInternalLock(SharedLock sharedLock) throws Exception { + Field field = SharedLock.class.getDeclaredField("lock"); + field.setAccessible(true); + return (Lock) field.get(sharedLock); + } + }