Radix-sorted, lookup-free, typesafe, nonblocking, priority-boosting pending tasks

Today the master's pending task queue is just the
`PriorityBlockingQueue<Runnable>` belonging to the underlying
`ThreadPoolExecutor`. The reasons for this date back a long way but it
doesn't really reflect the structure of the queue as it exists today. In
particular, we must keep track of batches independently of the queue
itself, and must do various bits of unchecked casting to process
multiple items of the same type at once.

This commit introduces an new queueing mechanism, independent of the
executor's queue, which better represents the conceptual structure of
the master's pending tasks:

* Today we use a priority queue to allow important tasks to preempt
less-important ones. However there are only a small number of priority
levels, so it is simpler to maintain a queue for each priority,
effectively replacing the sorting within the priority queue with a
radix sort.

* Today when a task is submitted we perform a map lookup to see if it
can be added to an existing batch or not. With this change we allow
client code to create its own dedicated queue of tasks. The entries in
the per-priority-level queues are themselves queues, one for each
executor, representing the batches to be run.

* Today each task in the queue holds a reference to its executor, but
the executor used to run a task may belong to a different task in the
same batch. In practice we know they're the same executor (that's how
batches are defined) but we cannot express this knowledge in the type
system so we have to do a bunch of unchecked casting to work around it.
With this change we associate each per-executor queue directly with its
executor, avoiding the need to do all this unchecked casting.

* Today the master service must block its thread while waiting for each
task to complete, because otherwise the executor would start to process
the next task in the queue. This makes testing using a
`DeterministicTaskQueue` harder (see `FakeThreadPoolMasterService`).
With this change we avoid submitting a task to the `ThreadPoolExecutor`
until the previous task is complete, which means we can make the
implementation avoid blocking while a task is running and therefore run
the whole production implementation in deterministic tests[^1].

* Today it's possible for a steady drip of high-priority tasks to starve
low-priority tasks of access to the master for an extended period of
time. With this change we separate the queue of tasks from the queue
which determines the execution order. This allows us to implement more
intelligent execution policies. For instance, if we detect that the
queue has not processed any low-priority tasks for too long then we can
make the decision to boost their priorities[^2].

[^1]: Not done yet but this is a step in the right direction.

[^2]: Not done yet but this is a step in the right direction.

Author: DaveCTurner

server/src/main/java/org/elasticsearch/action/admin/cluster/desirednodes/TransportDeleteDesiredNodesAction.java

@@ -19,14 +19,15 @@
 import org.elasticsearch.cluster.metadata.DesiredNodesMetadata;
 import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
 import org.elasticsearch.cluster.service.ClusterService;
+import org.elasticsearch.cluster.service.MasterServiceTaskQueue;
 import org.elasticsearch.common.Priority;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.tasks.Task;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.transport.TransportService;
 
 public class TransportDeleteDesiredNodesAction extends TransportMasterNodeAction<DeleteDesiredNodesAction.Request, ActionResponse.Empty> {
-    private final DesiredNodesClusterStateTaskExecutor taskExecutor;
+    private final MasterServiceTaskQueue<ClusterStateUpdateTask> taskQueue;
 
     @Inject
     public TransportDeleteDesiredNodesAction(
@@ -47,7 +48,7 @@ public TransportDeleteDesiredNodesAction(
             in -> ActionResponse.Empty.INSTANCE,
             ThreadPool.Names.SAME
         );
-        this.taskExecutor = new DesiredNodesClusterStateTaskExecutor();
+        this.taskQueue = clusterService.getTaskQueue("delete-desired-nodes", Priority.HIGH, new DesiredNodesClusterStateTaskExecutor());
     }
 
     @Override
@@ -57,7 +58,7 @@ protected void masterOperation(
         ClusterState state,
         ActionListener<ActionResponse.Empty> listener
     ) throws Exception {
-        final var clusterStateUpdateTask = new ClusterStateUpdateTask(Priority.HIGH) {
+        taskQueue.submitTask("delete-desired-nodes", new ClusterStateUpdateTask(Priority.HIGH) {
             @Override
             public ClusterState execute(ClusterState currentState) {
                 return currentState.copyAndUpdateMetadata(metadata -> metadata.removeCustom(DesiredNodesMetadata.TYPE));
@@ -72,8 +73,7 @@ public void onFailure(Exception e) {
             public void clusterStateProcessed(ClusterState oldState, ClusterState newState) {
                 listener.onResponse(ActionResponse.Empty.INSTANCE);
             }
-        };
-        clusterService.submitStateUpdateTask("delete-desired-nodes", clusterStateUpdateTask, clusterStateUpdateTask, taskExecutor);
+        }, null);
     }
 
     @Override

server/src/main/java/org/elasticsearch/action/admin/cluster/desirednodes/TransportUpdateDesiredNodesAction.java

@@ -12,7 +12,6 @@
 import org.elasticsearch.action.support.ActionFilters;
 import org.elasticsearch.action.support.master.TransportMasterNodeAction;
 import org.elasticsearch.cluster.ClusterState;
-import org.elasticsearch.cluster.ClusterStateTaskExecutor;
 import org.elasticsearch.cluster.ClusterStateUpdateTask;
 import org.elasticsearch.cluster.block.ClusterBlockException;
 import org.elasticsearch.cluster.block.ClusterBlockLevel;
@@ -22,6 +21,7 @@
 import org.elasticsearch.cluster.metadata.DesiredNodesMetadata;
 import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
 import org.elasticsearch.cluster.service.ClusterService;
+import org.elasticsearch.cluster.service.MasterServiceTaskQueue;
 import org.elasticsearch.common.Priority;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.tasks.Task;
@@ -34,7 +34,7 @@
 
 public class TransportUpdateDesiredNodesAction extends TransportMasterNodeAction<UpdateDesiredNodesRequest, UpdateDesiredNodesResponse> {
     private final DesiredNodesSettingsValidator settingsValidator;
-    private final ClusterStateTaskExecutor<ClusterStateUpdateTask> taskExecutor;
+    private final MasterServiceTaskQueue<ClusterStateUpdateTask> taskQueue;
 
     @Inject
     public TransportUpdateDesiredNodesAction(
@@ -58,7 +58,7 @@ public TransportUpdateDesiredNodesAction(
             ThreadPool.Names.SAME
         );
         this.settingsValidator = settingsValidator;
-        this.taskExecutor = new DesiredNodesClusterStateTaskExecutor();
+        this.taskQueue = clusterService.getTaskQueue("delete-desired-nodes", Priority.URGENT, new DesiredNodesClusterStateTaskExecutor());
     }
 
     @Override
@@ -77,7 +77,7 @@ protected void masterOperation(
             DesiredNodes proposedDesiredNodes = new DesiredNodes(request.getHistoryID(), request.getVersion(), request.getNodes());
             settingsValidator.validate(proposedDesiredNodes);
 
-            final var clusterStateUpdateTask = new ClusterStateUpdateTask(Priority.URGENT, request.masterNodeTimeout()) {
+            taskQueue.submitTask("update-desired-nodes", new ClusterStateUpdateTask(Priority.URGENT, request.masterNodeTimeout()) {
                 volatile boolean replacedExistingHistoryId = false;
 
                 @Override
@@ -99,8 +99,7 @@ public void onFailure(Exception e) {
                 public void clusterStateProcessed(ClusterState oldState, ClusterState newState) {
                     listener.onResponse(new UpdateDesiredNodesResponse(replacedExistingHistoryId));
                 }
-            };
-            clusterService.submitStateUpdateTask("update-desired-nodes", clusterStateUpdateTask, clusterStateUpdateTask, taskExecutor);
+            }, request.masterNodeTimeout());
         } catch (Exception e) {
             listener.onFailure(e);
         }

server/src/main/java/org/elasticsearch/action/admin/indices/create/AutoCreateAction.java

@@ -20,7 +20,6 @@
 import org.elasticsearch.action.support.master.TransportMasterNodeAction;
 import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.cluster.ClusterStateAckListener;
-import org.elasticsearch.cluster.ClusterStateTaskConfig;
 import org.elasticsearch.cluster.ClusterStateTaskExecutor;
 import org.elasticsearch.cluster.ClusterStateTaskListener;
 import org.elasticsearch.cluster.block.ClusterBlockException;
@@ -35,6 +34,7 @@
 import org.elasticsearch.cluster.node.DiscoveryNode;
 import org.elasticsearch.cluster.routing.allocation.AllocationService;
 import org.elasticsearch.cluster.service.ClusterService;
+import org.elasticsearch.cluster.service.MasterServiceTaskQueue;
 import org.elasticsearch.common.Priority;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.settings.Settings;
@@ -76,7 +76,7 @@ public static final class TransportAction extends TransportMasterNodeAction<Crea
         private final AutoCreateIndex autoCreateIndex;
         private final SystemIndices systemIndices;
 
-        private final ClusterStateTaskExecutor<CreateIndexTask> executor;
+        private final MasterServiceTaskQueue<CreateIndexTask> taskQueue;
 
         @Inject
         public TransportAction(
@@ -107,7 +107,7 @@ public TransportAction(
             this.createIndexService = createIndexService;
             this.metadataCreateDataStreamService = metadataCreateDataStreamService;
             this.autoCreateIndex = autoCreateIndex;
-            executor = (currentState, taskContexts) -> {
+            this.taskQueue = clusterService.getTaskQueue("auto-create", Priority.URGENT, (currentState, taskContexts) -> {
                 ClusterState state = currentState;
                 final Map<CreateIndexRequest, String> successfulRequests = Maps.newMapWithExpectedSize(taskContexts.size());
                 for (final var taskContext : taskContexts) {
@@ -123,7 +123,7 @@ public TransportAction(
                     state = allocationService.reroute(state, "auto-create");
                 }
                 return state;
-            };
+            });
         }
 
         @Override
@@ -133,11 +133,10 @@ protected void masterOperation(
             ClusterState state,
             ActionListener<CreateIndexResponse> listener
         ) {
-            clusterService.submitStateUpdateTask(
+            taskQueue.submitTask(
                 "auto create [" + request.index() + "]",
                 new CreateIndexTask(request, listener),
-                ClusterStateTaskConfig.build(Priority.URGENT, request.masterNodeTimeout()),
-                executor
+                request.masterNodeTimeout()
             );
         }
 

server/src/main/java/org/elasticsearch/action/admin/indices/rollover/TransportRolloverAction.java

@@ -22,7 +22,6 @@
 import org.elasticsearch.action.support.master.TransportMasterNodeAction;
 import org.elasticsearch.client.internal.Client;
 import org.elasticsearch.cluster.ClusterState;
-import org.elasticsearch.cluster.ClusterStateTaskConfig;
 import org.elasticsearch.cluster.ClusterStateTaskExecutor;
 import org.elasticsearch.cluster.ClusterStateTaskListener;
 import org.elasticsearch.cluster.block.ClusterBlockException;
@@ -32,6 +31,7 @@
 import org.elasticsearch.cluster.metadata.Metadata;
 import org.elasticsearch.cluster.routing.allocation.AllocationService;
 import org.elasticsearch.cluster.service.ClusterService;
+import org.elasticsearch.cluster.service.MasterServiceTaskQueue;
 import org.elasticsearch.common.Priority;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.inject.Inject;
@@ -61,7 +61,7 @@ public class TransportRolloverAction extends TransportMasterNodeAction<RolloverR
     private static final Logger logger = LogManager.getLogger(TransportRolloverAction.class);
 
     private final Client client;
-    private final RolloverExecutor rolloverTaskExecutor;
+    private final MasterServiceTaskQueue<RolloverTask> rolloverTaskQueue;
 
     @Inject
     public TransportRolloverAction(
@@ -86,10 +86,10 @@ public TransportRolloverAction(
             ThreadPool.Names.SAME
         );
         this.client = client;
-        this.rolloverTaskExecutor = new RolloverExecutor(
-            allocationService,
-            rolloverService,
-            new ActiveShardsObserver(clusterService, threadPool)
+        this.rolloverTaskQueue = clusterService.getTaskQueue(
+            "rollover",
+            Priority.NORMAL,
+            new RolloverExecutor(allocationService, rolloverService, new ActiveShardsObserver(clusterService, threadPool))
         );
     }
 
@@ -181,8 +181,7 @@ protected void masterOperation(
                 if (trialConditionResults.size() == 0 || trialMetConditions.size() > 0) {
                     String source = "rollover_index source [" + trialRolloverIndexName + "] to target [" + trialRolloverIndexName + "]";
                     RolloverTask rolloverTask = new RolloverTask(rolloverRequest, statsResponse, trialRolloverResponse, listener);
-                    ClusterStateTaskConfig config = ClusterStateTaskConfig.build(Priority.NORMAL, rolloverRequest.masterNodeTimeout());
-                    clusterService.submitStateUpdateTask(source, rolloverTask, config, rolloverTaskExecutor);
+                    rolloverTaskQueue.submitTask(source, rolloverTask, rolloverRequest.masterNodeTimeout());
                 } else {
                     // conditions not met
                     listener.onResponse(trialRolloverResponse);

server/src/main/java/org/elasticsearch/cluster/ClusterState.java

@@ -67,8 +67,8 @@
  * across master elections (and therefore is preserved in a rolling restart).
  * <p>
  * Updates are triggered by submitting tasks to the {@link MasterService} on the elected master, typically using a {@link
- * TransportMasterNodeAction} to route a request to the master on which the task is submitted with {@link
- * ClusterService#submitStateUpdateTask}. Submitted tasks have an associated {@link ClusterStateTaskConfig} which defines a priority and a
+ * TransportMasterNodeAction} to route a request to the master on which the task is submitted via a queue obtained with {@link
+ * ClusterService#getTaskQueue}, which has an associated priority. Submitted tasks have an associated
  * timeout. Tasks are processed in priority order, so a flood of higher-priority tasks can starve lower-priority ones from running.
  * Therefore, avoid priorities other than {@link Priority#NORMAL} where possible. Tasks associated with client actions should typically have
  * a timeout, or otherwise be sensitive to client cancellations, to avoid surprises caused by the execution of stale tasks long after they

server/src/main/java/org/elasticsearch/cluster/ClusterStateTaskConfig.java

@@ -15,7 +15,7 @@
 /**
  * A task that can update the cluster state.
  */
-public abstract class ClusterStateUpdateTask implements ClusterStateTaskConfig, ClusterStateTaskListener {
+public abstract class ClusterStateUpdateTask implements ClusterStateTaskListener {
 
     private final Priority priority;
 
@@ -64,7 +64,6 @@ public final TimeValue timeout() {
         return timeout;
     }
 
-    @Override
     public final Priority priority() {
         return priority;
     }

server/src/main/java/org/elasticsearch/cluster/ClusterStateUpdateTask.java

@@ -34,45 +34,40 @@ public final void clusterStateProcessed(ClusterState oldState, ClusterState newS
     protected void onPublicationComplete() {}
 
     public void submit(MasterService masterService, String source) {
-        masterService.submitStateUpdateTask(
-            source,
-            this,
-            ClusterStateTaskConfig.build(priority),
-            // Uses a new executor each time so that these tasks are not batched, but they never change the cluster state anyway so they
-            // don't trigger the publication process and hence batching isn't really needed.
-            new ClusterStateTaskExecutor<>() {
+        // Uses a new queue each time so that these tasks are not batched, but they never change the cluster state anyway so they
+        // don't trigger the publication process and hence batching isn't really needed.
+        masterService.getTaskQueue("local-master-service-task", priority, new ClusterStateTaskExecutor<LocalMasterServiceTask>() {
 
-                @Override
-                public boolean runOnlyOnMaster() {
-                    return false;
-                }
+            @Override
+            public boolean runOnlyOnMaster() {
+                return false;
+            }
 
-                @Override
-                public String describeTasks(List<LocalMasterServiceTask> tasks) {
-                    return ""; // only one task in the batch so the source is enough
-                }
+            @Override
+            public String describeTasks(List<LocalMasterServiceTask> tasks) {
+                return ""; // only one task in the batch so the source is enough
+            }
 
-                @Override
-                public ClusterState execute(ClusterState currentState, List<TaskContext<LocalMasterServiceTask>> taskContexts)
-                    throws Exception {
-                    final LocalMasterServiceTask thisTask = LocalMasterServiceTask.this;
-                    assert taskContexts.size() == 1 && taskContexts.get(0).getTask() == thisTask
-                        : "expected one-element task list containing current object but was " + taskContexts;
-                    thisTask.execute(currentState);
-                    taskContexts.get(0).success(new ActionListener<>() {
-                        @Override
-                        public void onResponse(ClusterState clusterState) {
-                            onPublicationComplete();
-                        }
+            @Override
+            public ClusterState execute(ClusterState currentState, List<TaskContext<LocalMasterServiceTask>> taskContexts)
+                throws Exception {
+                final LocalMasterServiceTask thisTask = LocalMasterServiceTask.this;
+                assert taskContexts.size() == 1 && taskContexts.get(0).getTask() == thisTask
+                    : "expected one-element task list containing current object but was " + taskContexts;
+                thisTask.execute(currentState);
+                taskContexts.get(0).success(new ActionListener<>() {
+                    @Override
+                    public void onResponse(ClusterState clusterState) {
+                        onPublicationComplete();
+                    }
 
-                        @Override
-                        public void onFailure(Exception e) {
-                            LocalMasterServiceTask.this.onFailure(e);
-                        }
-                    });
-                    return currentState;
-                }
+                    @Override
+                    public void onFailure(Exception e) {
+                        LocalMasterServiceTask.this.onFailure(e);
+                    }
+                });
+                return currentState;
             }
-        );
+        }).submitTask(source, this, null);
     }
 }