Merge branch 'cassandra-6.0' into trunk

* cassandra-6.0:
  Expose immediately-executed tasks in the queries virtual table

CHANGES.txt

@@ -4,6 +4,7 @@
  * Allow nodetool garbagecollect to take a user defined list of SSTables (CASSANDRA-16767)
  * Add a guardrail for misprepared statements (CASSANDRA-21139)
 Merged from 6.0:
+ * Expose immediately-executed tasks in the queries virtual table (CASSANDRA-21471)
  * Avoid potential deadlock between GlobalLogFollower and GossipStage (CASSANDRA-21384)
  * Add CMS membership as a field in ClusterMetadata (CASSANDRA-20736)
  * Fix maven remote publishing of Accord artifacts (CASSANDRA-21261)

src/java/org/apache/cassandra/concurrent/CassandraThread.java

@@ -27,19 +27,34 @@ public class CassandraThread extends FastThreadLocalThread
     private ThreadLocalMetrics threadLocalMetrics;
     private ExecutorLocals executorLocals;
 
-    public CassandraThread(ThreadGroup group, Runnable target, String name)
+    private final ImmediateTaskHolder immediateTaskHolder;
+
+    public CassandraThread(ThreadGroup group, Runnable target, String name, ImmediateTaskHolder immediateTaskHolder)
     {
         super(group, target, name);
+        assert immediateTaskHolder != null;
+        this.immediateTaskHolder = immediateTaskHolder;
+    }
+    public CassandraThread(ThreadGroup group, Runnable target, String name)
+    {
+        this(group, target, name, ImmediateTaskHolder.NO_OP);
     }
 
     public CassandraThread()
     {
         super();
+        this.immediateTaskHolder = ImmediateTaskHolder.NO_OP;
     }
 
     public CassandraThread(Runnable target)
     {
         super(target);
+        this.immediateTaskHolder = ImmediateTaskHolder.NO_OP;
+    }
+
+    public ImmediateTaskHolder getImmediateTaskHolder()
+    {
+        return immediateTaskHolder;
     }
 
     public ThreadLocalMetrics getThreadLocalMetrics()

src/java/org/apache/cassandra/concurrent/ImmediateTaskHolder.java

@@ -0,0 +1,41 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.concurrent;
+
+public interface ImmediateTaskHolder
+{
+    /**
+     * @param currentTask to set as a current immediate task
+     * @return the previously set immediate task, can be null
+     */
+    Runnable setImmediateTask(Runnable currentTask);
+
+    class NoOp implements ImmediateTaskHolder
+    {
+
+        @Override
+        public Runnable setImmediateTask(Runnable currentTask)
+        {
+            // nothing to do
+            return null;
+        }
+    }
+
+    NoOp NO_OP = new NoOp();
+}

src/java/org/apache/cassandra/concurrent/SEPExecutor.java

@@ -211,12 +211,15 @@ public class SEPExecutor implements LocalAwareExecutorPlus, SEPExecutorMBean
         }
         else
         {
+            ImmediateTaskHolder taskHolder = getNestedCurrentTaskHolder();
+            Runnable previousTask = taskHolder.setImmediateTask(task);
             try
             {
                 task.run();
             }
             finally
             {
+                taskHolder.setImmediateTask(previousTask);
                 returnWorkPermit();
                 // we have to maintain our invariant of always scheduling after any work is performed
                 // in this case in particular we are not processing the rest of the queue anyway, and so
@@ -226,6 +229,16 @@ public class SEPExecutor implements LocalAwareExecutorPlus, SEPExecutorMBean
         }
     }
 
+    private static ImmediateTaskHolder getNestedCurrentTaskHolder()
+    {
+        Thread currentThread = Thread.currentThread();
+        if (currentThread instanceof CassandraThread)
+        {
+            return ((CassandraThread) currentThread).getImmediateTaskHolder();
+        }
+        return ImmediateTaskHolder.NO_OP;
+    }
+
     @Override
     public void execute(Runnable run)
     {

src/java/org/apache/cassandra/concurrent/SEPWorker.java

@@ -51,6 +51,35 @@ final class SEPWorker extends AtomicReference<SEPWorker.Work> implements Runnabl
 
     private final AtomicReference<Runnable> currentTask = new AtomicReference<>();
 
+    private class ImmediateDebuggableTaskRunner implements DebuggableTask.DebuggableTaskRunner, ImmediateTaskHolder
+    {
+        private final AtomicReference<Runnable> immediateCurrentTask = new AtomicReference<>();
+
+        @Override
+        public DebuggableTask running()
+        {
+            return getDebuggableTask(immediateCurrentTask.get());
+        }
+
+        @Override
+        public String id()
+        {
+            // derive from the current thread name so the nested row tracks renames and correlates with the worker's main row
+            return thread.getName() + "(immediate)";
+        }
+
+        @Override
+        public Runnable setImmediateTask(Runnable currentTask)
+        {
+            // plain is used to reduce overheads, the method is expected to be invoked only by a single thread
+            Runnable previousTask = immediateCurrentTask.getPlain();
+            immediateCurrentTask.lazySet(currentTask);
+            return previousTask;
+        }
+    }
+
+    private final ImmediateDebuggableTaskRunner immediateDebuggableTaskRunner;
+
     private String lastUsedExecutorName;
 
     SEPWorker(ThreadGroup threadGroup, Long workerId, Work initialState, SharedExecutorPool pool)
@@ -58,18 +87,27 @@ final class SEPWorker extends AtomicReference<SEPWorker.Work> implements Runnabl
         this.pool = pool;
         this.workerId = workerId;
         this.workerIdThreadSuffix = '-' + workerId.toString();
-        thread = new CassandraThread(threadGroup, this, threadGroup.getName() + "-Worker-" + workerId);
+        String threadName = threadGroup.getName() + "-Worker-" + workerId;
+        this.immediateDebuggableTaskRunner = new ImmediateDebuggableTaskRunner();
+        thread = new CassandraThread(threadGroup, this, threadName, immediateDebuggableTaskRunner);
         thread.setDaemon(true);
         set(initialState);
         thread.start();
     }
 
+    public DebuggableTask.DebuggableTaskRunner immediateRunner()
+    {
+        return immediateDebuggableTaskRunner;
+    }
+
     @Override
     public DebuggableTask running()
     {
-        // can change after null check so go off local reference
-        Runnable task = currentTask.get();
+        return getDebuggableTask(currentTask.get());
+    }
 
+    private static DebuggableTask getDebuggableTask(Runnable task)
+    {
         // Local read and mutation Runnables are themselves debuggable
         if (task instanceof DebuggableTask)
             return (DebuggableTask) task;

src/java/org/apache/cassandra/concurrent/SharedExecutorPool.java

@@ -123,7 +123,7 @@ public class SharedExecutorPool
 
     public Stream<? extends DebuggableTaskRunner> workers()
     {
-        return allWorkers.stream();
+        return allWorkers.stream().flatMap(worker -> Stream.of(worker, worker.immediateRunner()));
     }
 
     void maybeStartSpinningWorker()

test/distributed/org/apache/cassandra/distributed/test/QueriesTableTest.java

@@ -121,13 +121,14 @@ public class QueriesTableTest extends TestBaseImpl
             String task = row.get("task").toString();
 
             boolean localReaderThread = threadId.contains("Read") || threadId.contains("SharedPool-Worker");
-            readVisible |= localReaderThread && task.contains("SELECT");
+            // QUERY in a task name is used as a way to differentiate a local replica read vs a coordinator read
+            readVisible |= localReaderThread && task.contains("SELECT") && !task.contains("QUERY");
             boolean coordReaderThread = threadId.contains("Native-Transport-Requests") || threadId.contains("SharedPool-Worker");
-            coordinatorReadVisible |= coordReaderThread && task.contains("SELECT");
+            coordinatorReadVisible |= coordReaderThread && task.contains("SELECT") && task.contains("QUERY");
             boolean localWriterThread = threadId.contains("Mutation") || threadId.contains("SharedPool-Worker");
-            writeVisible |= localWriterThread && task.contains("Mutation");
+            writeVisible |= localWriterThread && task.contains("Mutation") && !task.contains("QUERY");
             boolean coordWriterThread = threadId.contains("Native-Transport-Requests") || threadId.contains("SharedPool-Worker");
-            coordinatorWriteVisible |= coordWriterThread && task.contains("INSERT");
+            coordinatorWriteVisible |= coordWriterThread && task.contains("INSERT") && task.contains("QUERY");
         }
 
         assertTrue(readVisible);
@@ -169,9 +170,10 @@ public class QueriesTableTest extends TestBaseImpl
             String task = row.get("task").toString();
 
             boolean localReaderThread = threadId.contains("Read") || threadId.contains("SharedPool-Worker");
-            readVisible |= localReaderThread && task.contains("SELECT");
+            // QUERY in a task name is used as a way to differentiate a local replica read vs a coordinator read
+            readVisible |= localReaderThread && task.contains("SELECT") && !task.contains("QUERY");
             boolean coordUpdateThread = threadId.contains("Native-Transport-Requests") || threadId.contains("SharedPool-Worker");
-            coordinatorUpdateVisible |= coordUpdateThread && task.contains("UPDATE");
+            coordinatorUpdateVisible |= coordUpdateThread && task.contains("UPDATE") && task.contains("QUERY");
         }
 
         assertTrue(readVisible);