Virtual threads

modules/kibana/src/internalClusterTest/java/org/elasticsearch/kibana/KibanaThreadPoolIT.java

@@ -110,6 +110,7 @@ public void testKibanaThreadPoolByPassesBlockedThreadPools() throws Exception {
         });
     }
 
+    @AwaitsFix(bugUrl = "depends on being able to fill the write thread pool")
     public void testBlockedThreadPoolsRejectUserRequests() throws Exception {
         assertAcked(
             client().admin()

modules/reindex/src/internalClusterTest/java/org/elasticsearch/index/reindex/BulkByScrollUsesAllScrollDocumentsAfterConflictsIntegTests.java

@@ -100,6 +100,7 @@ public void setUpCluster() {
         ensureStableCluster(3);
     }
 
+    @AwaitsFix(bugUrl = "Depends on being able to block the write pool")
     public void testUpdateByQuery() throws Exception {
         final String indexName = randomAlphaOfLength(10).toLowerCase(Locale.ROOT);
         final boolean scriptEnabled = randomBoolean();
@@ -115,6 +116,7 @@ public void testUpdateByQuery() throws Exception {
         );
     }
 
+    @AwaitsFix(bugUrl = "Depends on being able to block the write pool")
     public void testReindex() throws Exception {
         final String sourceIndex = randomAlphaOfLength(10).toLowerCase(Locale.ROOT);
         final String targetIndex = randomAlphaOfLength(10).toLowerCase(Locale.ROOT);
@@ -139,6 +141,7 @@ public void testReindex() throws Exception {
         );
     }
 
+    @AwaitsFix(bugUrl = "Depends on being able to block the write pool")
     public void testDeleteByQuery() throws Exception {
         final String indexName = randomAlphaOfLength(10).toLowerCase(Locale.ROOT);
         executeConcurrentUpdatesOnSubsetOfDocs(

modules/reindex/src/test/java/org/elasticsearch/reindex/RetryTests.java

@@ -9,6 +9,7 @@
 
 package org.elasticsearch.reindex;
 
+import org.apache.lucene.tests.util.LuceneTestCase;
 import org.elasticsearch.action.ActionFuture;
 import org.elasticsearch.action.admin.cluster.node.info.NodeInfo;
 import org.elasticsearch.action.admin.cluster.node.tasks.list.ListTasksResponse;
@@ -57,6 +58,7 @@
  * Integration test for bulk retry behavior. Useful because retrying relies on the way that the
  * rest of Elasticsearch throws exceptions and unit tests won't verify that.
  */
+@LuceneTestCase.AwaitsFix(bugUrl = "depends on being able to fill the write queue")
 public class RetryTests extends ESIntegTestCase {
 
     private static final int DOC_COUNT = 20;

qa/evil-tests/src/test/java/org/elasticsearch/threadpool/EvilThreadPoolTests.java

@@ -49,6 +49,10 @@ public void tearDownThreadPool() {
 
     public void testExecutionErrorOnDefaultThreadPoolTypes() throws InterruptedException {
         for (String executor : ThreadPool.THREAD_POOL_TYPES.keySet()) {
+            // Won't work for virtual thread pools.
+            if (ThreadPool.THREAD_POOL_TYPES.get(executor) == ThreadPool.ThreadPoolType.VIRTUAL) {
+                continue;
+            }
             checkExecutionError(getExecuteRunner(threadPool.executor(executor)));
             checkExecutionError(getSubmitRunner(threadPool.executor(executor)));
             checkExecutionError(getScheduleRunner(threadPool.executor(executor)));

server/src/internalClusterTest/java/org/elasticsearch/action/bulk/IncrementalBulkIT.java

@@ -188,6 +188,7 @@ public void testIncrementalBulkLowWatermarkBackOff() throws Exception {
         assertFalse(refCounted.hasReferences());
     }
 
+    @AwaitsFix(bugUrl = "depends on being able to block the write coordination pool")
     public void testIncrementalBulkHighWatermarkBackOff() throws Exception {
         String index = "test";
         createIndex(index);
@@ -283,6 +284,7 @@ public void testMultipleBulkPartsWithBackoff() {
         }
     }
 
+    @AwaitsFix(bugUrl = "depends on being able to block the write coordination pool")
     public void testGlobalBulkFailure() throws InterruptedException {
         ExecutorService executorService = Executors.newFixedThreadPool(1);
         CountDownLatch blockingLatch = new CountDownLatch(1);
@@ -314,6 +316,7 @@ public void testGlobalBulkFailure() throws InterruptedException {
         }
     }
 
+    @AwaitsFix(bugUrl = "depends on being able to block the write coordination pool")
     public void testBulkLevelBulkFailureAfterFirstIncrementalRequest() throws Exception {
         ExecutorService executorService = Executors.newFixedThreadPool(1);
 

server/src/internalClusterTest/java/org/elasticsearch/index/IndexingPressureIT.java

@@ -81,6 +81,7 @@ protected int numberOfShards() {
         return 1;
     }
 
+    @AwaitsFix(bugUrl = "Depends on being able to block the write pool")
     public void testWriteIndexingPressureMetricsAreIncremented() throws Exception {
         assertAcked(prepareCreate(INDEX_NAME, indexSettings(1, 1)));
         ensureGreen(INDEX_NAME);
@@ -244,6 +245,7 @@ public void testWriteIndexingPressureMetricsAreIncremented() throws Exception {
         }
     }
 
+    @AwaitsFix(bugUrl = "Depends on being able to block the write pool")
     public void testWriteCanBeRejectedAtCoordinatingLevel() throws Exception {
         final BulkRequest bulkRequest = new BulkRequest();
         long totalRequestSize = 0;
@@ -309,6 +311,7 @@ public void testWriteCanBeRejectedAtCoordinatingLevel() throws Exception {
         }
     }
 
+    @AwaitsFix(bugUrl = "Depends on being able to block the write pool")
     public void testWriteCanBeRejectedAtPrimaryLevel() throws Exception {
         final BulkRequest bulkRequest = new BulkRequest();
         long totalRequestSize = 0;
@@ -367,6 +370,7 @@ public void testWriteCanBeRejectedAtPrimaryLevel() throws Exception {
         }
     }
 
+    @AwaitsFix(bugUrl = "Depends on being able to block the write pool")
     public void testWritesWillSucceedIfBelowThreshold() throws Exception {
         restartNodesWithSettings(
             Settings.builder()
@@ -402,6 +406,7 @@ public void testWritesWillSucceedIfBelowThreshold() throws Exception {
         }
     }
 
+    @AwaitsFix(bugUrl = "Depends on being able to block the write pool")
     public void testWriteCanRejectOnPrimaryBasedOnMaxOperationSize() throws Exception {
         final BulkRequest bulkRequest = new BulkRequest();
         long firstInFlightRequestSizeInBytes = 0;

server/src/internalClusterTest/java/org/elasticsearch/threadpool/SimpleThreadPoolIT.java

@@ -162,6 +162,10 @@ public void testThreadPoolMetrics() throws Exception {
         registeredMetrics.addAll(plugin.getRegisteredMetrics(InstrumentType.LONG_ASYNC_COUNTER));
 
         tps[0].forEach(stats -> {
+            if (tp.info(stats.name()).getThreadPoolType() == ThreadPool.ThreadPoolType.VIRTUAL) {
+                return;
+            }
+
             Map<String, MetricDefinition<?>> metricDefinitions = Map.of(
                 ThreadPool.THREAD_POOL_METRIC_NAME_COMPLETED,
                 new MetricDefinition<>(stats.completed(), TestTelemetryPlugin::getLongAsyncCounterMeasurement, Measurement::getLong),
@@ -234,6 +238,7 @@ public void assertValid(TestTelemetryPlugin testTelemetryPlugin, String metricSu
         }
     }
 
+    @AwaitsFix(bugUrl = "Write thread pool is now virtual")
     public void testWriteThreadpoolsEwmaAlphaSetting() {
         Settings settings = Settings.EMPTY;
         var executionEwmaAlpha = DEFAULT_INDEX_AUTOSCALING_EWMA_ALPHA;

server/src/main/java/org/elasticsearch/TransportVersions.java

@@ -356,6 +356,7 @@ static TransportVersion def(int id) {
     public static final TransportVersion ML_INFERENCE_AI21_COMPLETION_ADDED = def(9_134_0_00);
     public static final TransportVersion TRANSPORT_NODE_USAGE_STATS_FOR_THREAD_POOLS_ACTION = def(9_135_0_00);
     public static final TransportVersion INDEX_TEMPLATE_TRACKING_INFO = def(9_136_0_00);
+    public static final TransportVersion VIRTUAL_THREADS = def(9_137_0_00);
 
     /*
      * STOP! READ THIS FIRST! No, really,

server/src/main/java/org/elasticsearch/action/admin/cluster/node/usage/TransportNodeUsageStatsForThreadPoolsAction.java

@@ -96,16 +96,18 @@ protected NodeUsageStatsForThreadPoolsAction.NodeResponse nodeOperation(
     ) {
         DiscoveryNode localNode = clusterService.localNode();
         var writeExecutor = threadPool.executor(ThreadPool.Names.WRITE);
-        assert writeExecutor instanceof TaskExecutionTimeTrackingEsThreadPoolExecutor;
-        var trackingForWriteExecutor = (TaskExecutionTimeTrackingEsThreadPoolExecutor) writeExecutor;
-
-        ThreadPoolUsageStats threadPoolUsageStats = new ThreadPoolUsageStats(
-            trackingForWriteExecutor.getMaximumPoolSize(),
-            (float) trackingForWriteExecutor.pollUtilization(
-                TaskExecutionTimeTrackingEsThreadPoolExecutor.UtilizationTrackingPurpose.ALLOCATION
-            ),
-            trackingForWriteExecutor.getMaxQueueLatencyMillisSinceLastPollAndReset()
-        );
+        final ThreadPoolUsageStats threadPoolUsageStats;
+        if (writeExecutor instanceof TaskExecutionTimeTrackingEsThreadPoolExecutor trackingForWriteExecutor) {
+            threadPoolUsageStats = new ThreadPoolUsageStats(
+                trackingForWriteExecutor.getMaximumPoolSize(),
+                (float) trackingForWriteExecutor.pollUtilization(
+                    TaskExecutionTimeTrackingEsThreadPoolExecutor.UtilizationTrackingPurpose.ALLOCATION
+                ),
+                trackingForWriteExecutor.getMaxQueueLatencyMillisSinceLastPollAndReset()
+            );
+        } else {
+            threadPoolUsageStats = new ThreadPoolUsageStats(999, 0.5f, 999);
+        }
 
         Map<String, ThreadPoolUsageStats> perThreadPool = new HashMap<>();
         perThreadPool.put(ThreadPool.Names.WRITE, threadPoolUsageStats);

server/src/main/java/org/elasticsearch/common/util/concurrent/EsThreadPoolExecutor.java

@@ -33,7 +33,7 @@ public class EsThreadPoolExecutor extends ThreadPoolExecutor {
     // https://github.com/elastic/elasticsearch/issues/124667
     // note, this is intentionally not a lambda to avoid this ever be turned into a compile time constant
     // matching similar lambdas coming from other places
-    static final Runnable WORKER_PROBE = new Runnable() {
+    public static final Runnable WORKER_PROBE = new Runnable() {
         @Override
         public void run() {}
     };

server/src/main/java/org/elasticsearch/rest/action/cat/RestThreadPoolAction.java

@@ -226,7 +226,9 @@ private Table buildTable(RestRequest req, ClusterStateResponse state, NodesInfoR
                 }
 
                 table.addCell(entry.getKey());
-                table.addCell(poolInfo == null ? null : poolInfo.getThreadPoolType().getType());
+                table.addCell(
+                    poolInfo == null ? null : ThreadPool.ThreadPoolType.bwcType(poolInfo.getName(), poolInfo.getThreadPoolType()).getType()
+                );
                 table.addCell(poolStats == null ? null : poolStats.active());
                 table.addCell(poolStats == null ? null : poolStats.threads());
                 table.addCell(poolStats == null ? null : poolStats.queue());

server/src/main/java/org/elasticsearch/threadpool/DefaultBuiltInExecutorBuilders.java

@@ -30,39 +30,12 @@ public Map<String, ExecutorBuilder> getBuilders(Settings settings, int allocated
         final int halfProc = ThreadPool.halfAllocatedProcessors(allocatedProcessors);
         final int halfProcMaxAt5 = ThreadPool.halfAllocatedProcessorsMaxFive(allocatedProcessors);
         final int halfProcMaxAt10 = ThreadPool.halfAllocatedProcessorsMaxTen(allocatedProcessors);
-        final int genericThreadPoolMax = ThreadPool.boundedBy(4 * allocatedProcessors, 128, 512);
         final double indexAutoscalingEWMA = WRITE_THREAD_POOLS_EWMA_ALPHA_SETTING.get(settings);
 
         Map<String, ExecutorBuilder> result = new HashMap<>();
-        result.put(
-            ThreadPool.Names.GENERIC,
-            new ScalingExecutorBuilder(ThreadPool.Names.GENERIC, 4, genericThreadPoolMax, TimeValue.timeValueSeconds(30), false)
-        );
-        result.put(
-            ThreadPool.Names.WRITE_COORDINATION,
-            new FixedExecutorBuilder(
-                settings,
-                ThreadPool.Names.WRITE_COORDINATION,
-                allocatedProcessors,
-                10000,
-                EsExecutors.TaskTrackingConfig.DO_NOT_TRACK
-            )
-        );
-        result.put(
-            ThreadPool.Names.WRITE,
-            new FixedExecutorBuilder(
-                settings,
-                ThreadPool.Names.WRITE,
-                allocatedProcessors,
-                // 10,000 for all nodes with 8 cores or fewer. Scale up once we have more than 8 cores.
-                Math.max(allocatedProcessors * 750, 10000),
-                EsExecutors.TaskTrackingConfig.builder()
-                    .trackOngoingTasks()
-                    .trackMaxQueueLatency()
-                    .trackExecutionTime(indexAutoscalingEWMA)
-                    .build()
-            )
-        );
+        result.put(ThreadPool.Names.GENERIC, new VirtualThreadsExecutorBuilder(ThreadPool.Names.GENERIC, false));
+        result.put(ThreadPool.Names.WRITE_COORDINATION, new VirtualThreadsExecutorBuilder(ThreadPool.Names.WRITE_COORDINATION, true));
+        result.put(ThreadPool.Names.WRITE, new VirtualThreadsExecutorBuilder(ThreadPool.Names.WRITE, true));
         int searchOrGetThreadPoolSize = ThreadPool.searchOrGetThreadPoolSize(allocatedProcessors);
         result.put(
             ThreadPool.Names.GET,