diff --git a/CHANGES.txt b/CHANGES.txt index 0ecb0bb11e..04ab37acf0 100644 --- a/CHANGES.txt +++ b/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) diff --git a/src/java/org/apache/cassandra/concurrent/CassandraThread.java b/src/java/org/apache/cassandra/concurrent/CassandraThread.java index 5afec8815b..62cbcdb6a4 100644 --- a/src/java/org/apache/cassandra/concurrent/CassandraThread.java +++ b/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() diff --git a/src/java/org/apache/cassandra/concurrent/ImmediateTaskHolder.java b/src/java/org/apache/cassandra/concurrent/ImmediateTaskHolder.java new file mode 100644 index 0000000000..99c1afa397 --- /dev/null +++ b/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(); +} diff --git a/src/java/org/apache/cassandra/concurrent/SEPExecutor.java b/src/java/org/apache/cassandra/concurrent/SEPExecutor.java index ccce797be7..c467b1da22 100644 --- a/src/java/org/apache/cassandra/concurrent/SEPExecutor.java +++ b/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) { diff --git a/src/java/org/apache/cassandra/concurrent/SEPWorker.java b/src/java/org/apache/cassandra/concurrent/SEPWorker.java index 8eb34303e2..f3cdbc345b 100644 --- a/src/java/org/apache/cassandra/concurrent/SEPWorker.java +++ b/src/java/org/apache/cassandra/concurrent/SEPWorker.java @@ -51,6 +51,35 @@ final class SEPWorker extends AtomicReference implements Runnabl private final AtomicReference currentTask = new AtomicReference<>(); + private class ImmediateDebuggableTaskRunner implements DebuggableTask.DebuggableTaskRunner, ImmediateTaskHolder + { + private final AtomicReference 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 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; diff --git a/src/java/org/apache/cassandra/concurrent/SharedExecutorPool.java b/src/java/org/apache/cassandra/concurrent/SharedExecutorPool.java index 3ad848d325..8dbe563deb 100644 --- a/src/java/org/apache/cassandra/concurrent/SharedExecutorPool.java +++ b/src/java/org/apache/cassandra/concurrent/SharedExecutorPool.java @@ -123,7 +123,7 @@ public class SharedExecutorPool public Stream workers() { - return allWorkers.stream(); + return allWorkers.stream().flatMap(worker -> Stream.of(worker, worker.immediateRunner())); } void maybeStartSpinningWorker() diff --git a/test/distributed/org/apache/cassandra/distributed/test/QueriesTableTest.java b/test/distributed/org/apache/cassandra/distributed/test/QueriesTableTest.java index 934987076a..627fe5e409 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/QueriesTableTest.java +++ b/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);