Merge branch 'cassandra-4.1' into cassandra-5.0

# Conflicts:
#	CHANGES.txt

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)

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.
  * <p>5. This scheme is not fair</p>
  * <p>6. Only the thread that calls register() may call await()</p>
- *
+ * <p>7. A signal can be cancelled while it is within await() if the invoking thread is interrupted</p>
  * 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;

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);
+            }
+        }
+    }
 }