diff --git a/CHANGES.txt b/CHANGES.txt index 84a20e6c71..a28736122f 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -12,6 +12,7 @@ * Ban the usage of "var" instead of full types in the production code (CASSANDRA-20038) * Suppress CVE-2024-45772 from lucene-core-9.7.0.jar (CASSANDRA-20024) Merged from 4.1: + * Fix WaitQueue.Signal.awaitUninterruptibly may block forever if invoking thread is interrupted (CASSANDRA-20084) * Run audit_logging_options through santiation and validation on startup (CASSANDRA-20208) * Enforce CQL message size limit on multiframe messages (CASSANDRA-20052) * Add nodetool checktokenmetadata command that checks TokenMetadata is insync with Gossip endpointState (CASSANDRA-18758) diff --git a/src/java/org/apache/cassandra/utils/concurrent/WaitQueue.java b/src/java/org/apache/cassandra/utils/concurrent/WaitQueue.java index e9dcdf86e9..e97d4a0c74 100644 --- a/src/java/org/apache/cassandra/utils/concurrent/WaitQueue.java +++ b/src/java/org/apache/cassandra/utils/concurrent/WaitQueue.java @@ -74,7 +74,7 @@ import static org.apache.cassandra.utils.Shared.Scope.SIMULATION; * to be met that we no longer need. *
5. This scheme is not fair
*6. Only the thread that calls register() may call await()
- * + *7. A signal can be cancelled while it is within await() if the invoking thread is interrupted
* TODO: this class should not be backed by CLQ (should use an intrusive linked-list with lower overhead) */ @Shared(scope = SIMULATION, inner = INTERFACES) @@ -109,9 +109,9 @@ public interface WaitQueue public boolean isSet(); /** - * atomically: cancels the Signal if !isSet(), or returns true if isSignalled() + * atomically: cancels the Signal if !isSet(), or returns true if isSet() * - * @return true if isSignalled() + * @return true if isSet() */ public boolean checkAndClear(); @@ -120,6 +120,23 @@ public interface WaitQueue * and if signalled propagates the signal to another waiting thread */ public abstract void cancel(); + + /** + * Await indefinitely, throwing any interrupt. + * No spurious wakeups. + * Important: the signal can be cancelled if the thread executing await() is interrupted + * @throws InterruptedException if interrupted + */ + Awaitable await() throws InterruptedException; + + /** + * Await until the deadline (in nanoTime), throwing any interrupt. + * No spurious wakeups. + * @return true if we were signalled, false if the deadline elapsed + * Important: the signal can be cancelled if the thread executing await() is interrupted + * @throws InterruptedException if interrupted + */ + boolean awaitUntil(long nanoTimeDeadline) throws InterruptedException; } /** @@ -281,9 +298,24 @@ public interface WaitQueue */ public static abstract class AbstractSignal extends AbstractAwaitable implements Signal { + public Signal awaitUninterruptibly() + { + boolean interrupted = false; + while (!isSet()) + { + if (Thread.interrupted()) + interrupted = true; + LockSupport.park(); + } + if (interrupted) + Thread.currentThread().interrupt(); + checkAndClear(); + return this; + } + public Signal await() throws InterruptedException { - while (!isSignalled()) + while (!isSet()) { checkInterrupted(); LockSupport.park(); @@ -292,10 +324,27 @@ public interface WaitQueue return this; } + public boolean awaitUntilUninterruptibly(long nanoTimeDeadline) + { + boolean interrupted = false; + long now; + while (nanoTimeDeadline > (now = nanoTime()) && !isSet()) + { + if (Thread.interrupted()) + interrupted = true; + long delta = nanoTimeDeadline - now; + LockSupport.parkNanos(delta); + } + if (interrupted) + Thread.currentThread().interrupt(); + + return checkAndClear(); + } + public boolean awaitUntil(long nanoTimeDeadline) throws InterruptedException { long now; - while (nanoTimeDeadline > (now = nanoTime()) && !isSignalled()) + while (nanoTimeDeadline > (now = nanoTime()) && !isSet()) { checkInterrupted(); long delta = nanoTimeDeadline - now; diff --git a/test/unit/org/apache/cassandra/concurrent/WaitQueueTest.java b/test/unit/org/apache/cassandra/concurrent/WaitQueueTest.java index a9049a9d81..721c9094a9 100644 --- a/test/unit/org/apache/cassandra/concurrent/WaitQueueTest.java +++ b/test/unit/org/apache/cassandra/concurrent/WaitQueueTest.java @@ -26,6 +26,7 @@ import org.apache.cassandra.utils.concurrent.WaitQueue; import org.junit.*; import java.util.concurrent.ThreadLocalRandom; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; @@ -119,4 +120,73 @@ public class WaitQueueTest assertFalse(fail.get()); } + @Test + public void testInterruptOfSignalAwaitingThread() throws InterruptedException + { + final WaitQueue waitQueue = newWaitQueue(); + Thread writerAwaitThread = createThread(() -> { + Thread.currentThread().interrupt(); + WaitQueue.Signal signal = waitQueue.register(); + signal.awaitUninterruptibly(); + + }, "writer.await"); + + writerAwaitThread.start(); + + Thread.sleep(1_000); // wait to enter signal.awaitUninterruptibly() + waitQueue.signalAll(); + + writerAwaitThread.join(4_000); + if (writerAwaitThread.isAlive()) + { + printThreadStackTrace(writerAwaitThread); + fail("signal.awaitUninterruptibly() is stuck"); + } + } + + @Test + public void testInterruptOfSignalAwaitingWithTimeoutThread() throws InterruptedException + { + final WaitQueue waitQueue = newWaitQueue(); + Thread writerAwaitThread = createThread(() -> { + Thread.currentThread().interrupt(); + WaitQueue.Signal signal = waitQueue.register(); + signal.awaitUninterruptibly(100_000, TimeUnit.MILLISECONDS); + }, "writer.await"); + + writerAwaitThread.start(); + + Thread.sleep(1_000); // wait to enter signal.awaitUninterruptibly() + waitQueue.signalAll(); + + writerAwaitThread.join(4_000); + if (writerAwaitThread.isAlive()) + { + printThreadStackTrace(writerAwaitThread); + fail("signal.awaitUninterruptibly() is stuck"); + } + } + + private static Thread createThread(Runnable job, String name) + { + Thread thread = new Thread(job, name); + thread.setDaemon(true); + return thread; + } + + private static void printThreadStackTrace(Thread thread) + { + System.out.println("Stack trace for thread: " + thread.getName()); + StackTraceElement[] stackTrace = thread.getStackTrace(); + if (stackTrace.length == 0) + { + System.out.println("The thread is not currently running or has no stack trace."); + } else + { + for (StackTraceElement element : stackTrace) + { + System.out.println("\tat " + element); + } + } + } }