From 901dc62c8d18309a7bf9db31a4cf62236ac0e66c Mon Sep 17 00:00:00 2001 From: David Smiley Date: Mon, 27 Jul 2026 11:28:33 -0400 Subject: [PATCH] Executors/Threads: Enforce reasonable rules. One ExecutorUtil.newMDCAwareCachedThreadPool method is misleading as-named, and should be newMDCAwareFixedThreadPool. Renamed. --- .../org/apache/solr/core/CoreContainer.java | 10 +- .../java/org/apache/solr/core/SolrCores.java | 2 +- .../solr/handler/IncrementalShardBackup.java | 7 +- .../org/apache/solr/handler/RestoreCore.java | 7 +- .../solr/handler/admin/CoreAdminHandler.java | 6 +- .../solr/update/UpdateShardHandler.java | 13 +- .../solr/handler/TestHttpRequestId.java | 12 +- .../ParallelHttpShardHandlerTest.java | 13 +- .../request/TestUnInvertedFieldException.java | 5 +- .../update/TestInPlaceUpdatesDistrib.java | 37 ++--- .../apache/solr/util/OrderedExecutorTest.java | 5 +- .../solrj/jetty/HttpJettySolrClient.java | 7 +- .../client/solrj/impl/HttpJdkSolrClient.java | 15 +- .../apache/solr/common/util/ExecutorUtil.java | 148 ++++++++++++++++-- .../solr/common/util/ExecutorUtilTest.java | 35 ++++- 15 files changed, 210 insertions(+), 112 deletions(-) diff --git a/solr/core/src/java/org/apache/solr/core/CoreContainer.java b/solr/core/src/java/org/apache/solr/core/CoreContainer.java index d79c3b7b794a..cdd2d4bd3f70 100644 --- a/solr/core/src/java/org/apache/solr/core/CoreContainer.java +++ b/solr/core/src/java/org/apache/solr/core/CoreContainer.java @@ -425,10 +425,10 @@ public CoreContainer(NodeConfig config, CoresLocator locator, boolean asyncSolrC this.replayUpdatesExecutor = new OrderedExecutor<>( cfg.getReplayUpdatesThreads(), - ExecutorUtil.newMDCAwareCachedThreadPool( + ExecutorUtil.newMDCAwareFixedThreadPool( cfg.getReplayUpdatesThreads(), // thread count cfg.getReplayUpdatesThreads(), // queue size - new SolrNamedThreadFactory("replayUpdatesExecutor"))); + "replayUpdatesExecutor")); this.appHandlersByConfigSetId = new JerseyAppHandlerCache(); SolrPaths.AllowPathBuilder allowPathBuilder = new SolrPaths.AllowPathBuilder(); @@ -451,10 +451,8 @@ public CoreContainer(NodeConfig config, CoresLocator locator, boolean asyncSolrC this.indexSearcherExecutor = SolrIndexSearcher.initCollectorExecutor(cfg); this.indexFingerprintExecutor = - ExecutorUtil.newMDCAwareCachedThreadPool( - EXECUTOR_MAX_CPU_THREADS, - Integer.MAX_VALUE, - new SolrNamedThreadFactory("IndexFingerprintPool")); + ExecutorUtil.newMDCAwareFixedThreadPool( + EXECUTOR_MAX_CPU_THREADS, Integer.MAX_VALUE, "IndexFingerprintPool"); } @SuppressWarnings({"unchecked"}) diff --git a/solr/core/src/java/org/apache/solr/core/SolrCores.java b/solr/core/src/java/org/apache/solr/core/SolrCores.java index 3a8d7a058bab..2d455fdf0849 100644 --- a/solr/core/src/java/org/apache/solr/core/SolrCores.java +++ b/solr/core/src/java/org/apache/solr/core/SolrCores.java @@ -109,7 +109,7 @@ public void close() { ExecutorService coreCloseExecutor = ExecutorUtil.newMDCAwareFixedThreadPool( - Integer.MAX_VALUE, new SolrNamedThreadFactory("coreCloseExecutor")); + 64, new SolrNamedThreadFactory("coreCloseExecutor")); try { for (SolrCore core : coreList) { coreCloseExecutor.execute( diff --git a/solr/core/src/java/org/apache/solr/handler/IncrementalShardBackup.java b/solr/core/src/java/org/apache/solr/handler/IncrementalShardBackup.java index 6c94d6955a5b..4caaaa88f819 100644 --- a/solr/core/src/java/org/apache/solr/handler/IncrementalShardBackup.java +++ b/solr/core/src/java/org/apache/solr/handler/IncrementalShardBackup.java @@ -41,7 +41,6 @@ import org.apache.solr.common.SolrException; import org.apache.solr.common.util.EnvUtils; import org.apache.solr.common.util.ExecutorUtil; -import org.apache.solr.common.util.SolrNamedThreadFactory; import org.apache.solr.core.DirectoryFactory; import org.apache.solr.core.IndexDeletionPolicyWrapper; import org.apache.solr.core.SolrCore; @@ -218,10 +217,8 @@ private BackupStats incrementalCopy(Collection indexFiles, Directory dir "BackupUploadExecutor", ExecutorService.class, s -> - ExecutorUtil.newMDCAwareCachedThreadPool( - MAX_PARALLEL_UPLOADS, - Integer.MAX_VALUE, - new SolrNamedThreadFactory("BackupUploadExecutor"))); + ExecutorUtil.newMDCAwareFixedThreadPool( + MAX_PARALLEL_UPLOADS, Integer.MAX_VALUE, "BackupUploadExecutor")); List> uploadFutures = new ArrayList<>(); for (String fileName : indexFiles) { diff --git a/solr/core/src/java/org/apache/solr/handler/RestoreCore.java b/solr/core/src/java/org/apache/solr/handler/RestoreCore.java index 2d5e1f2fa92d..ba61598ab03b 100644 --- a/solr/core/src/java/org/apache/solr/handler/RestoreCore.java +++ b/solr/core/src/java/org/apache/solr/handler/RestoreCore.java @@ -40,7 +40,6 @@ import org.apache.solr.common.SolrException; import org.apache.solr.common.util.EnvUtils; import org.apache.solr.common.util.ExecutorUtil; -import org.apache.solr.common.util.SolrNamedThreadFactory; import org.apache.solr.core.DirectoryFactory; import org.apache.solr.core.SolrCore; import org.apache.solr.core.backup.BackupFilePaths; @@ -136,10 +135,8 @@ public boolean doRestore() throws Exception { "RestoreDownloadExecutor", ExecutorService.class, s -> - ExecutorUtil.newMDCAwareCachedThreadPool( - MAX_PARALLEL_DOWNLOADS, - Integer.MAX_VALUE, - new SolrNamedThreadFactory("RestoreDownloadExecutor"))); + ExecutorUtil.newMDCAwareFixedThreadPool( + MAX_PARALLEL_DOWNLOADS, Integer.MAX_VALUE, "RestoreDownloadExecutor")); // Move all files from backupDir to restoreIndexDir for (String filename : repository.listAllFiles()) { diff --git a/solr/core/src/java/org/apache/solr/handler/admin/CoreAdminHandler.java b/solr/core/src/java/org/apache/solr/handler/admin/CoreAdminHandler.java index 494e3f898414..fab48de35a69 100644 --- a/solr/core/src/java/org/apache/solr/handler/admin/CoreAdminHandler.java +++ b/solr/core/src/java/org/apache/solr/handler/admin/CoreAdminHandler.java @@ -383,10 +383,8 @@ public static class CoreAdminAsyncTracker { // Executor for expensive tasks // We keep the number of max threads very low to have throttling for expensive tasks private ExecutorService expensiveExecutor = - ExecutorUtil.newMDCAwareCachedThreadPool( - 5, - Integer.MAX_VALUE, - new SolrNamedThreadFactory("parallelCoreAdminAPIExpensiveExecutor")); + ExecutorUtil.newMDCAwareFixedThreadPool( + 5, Integer.MAX_VALUE, "parallelCoreAdminAPIExpensiveExecutor"); public CoreAdminAsyncTracker() { this( diff --git a/solr/core/src/java/org/apache/solr/update/UpdateShardHandler.java b/solr/core/src/java/org/apache/solr/update/UpdateShardHandler.java index ed0f4f4e850a..d4b516b0fed6 100644 --- a/solr/core/src/java/org/apache/solr/update/UpdateShardHandler.java +++ b/solr/core/src/java/org/apache/solr/update/UpdateShardHandler.java @@ -22,7 +22,6 @@ import java.util.Set; import java.util.concurrent.ExecutorService; import java.util.concurrent.SynchronousQueue; -import java.util.concurrent.ThreadFactory; import java.util.concurrent.TimeUnit; import org.apache.solr.client.solrj.impl.SolrHttpConstants; import org.apache.solr.client.solrj.jetty.HttpJettySolrClient; @@ -120,17 +119,15 @@ public UpdateShardHandler(UpdateShardHandlerConfig cfg) { recoveryOnlyClient = recoveryOnlyClientBuilder.build(); - ThreadFactory recoveryThreadFactory = new SolrNamedThreadFactory("recoveryExecutor"); - if (cfg != null && cfg.getMaxRecoveryThreads() > 0) { - if (log.isDebugEnabled()) { - log.debug("Creating recoveryExecutor with pool size {}", cfg.getMaxRecoveryThreads()); - } + int maxRecoveryThreads = cfg == null ? -1 : cfg.getMaxRecoveryThreads(); + if (maxRecoveryThreads > 0) { + log.debug("Creating recoveryExecutor with pool size {}", maxRecoveryThreads); recoveryExecutor = ExecutorUtil.newMDCAwareFixedThreadPool( - cfg.getMaxRecoveryThreads(), recoveryThreadFactory); + maxRecoveryThreads, new SolrNamedThreadFactory("recoveryExecutor")); } else { log.debug("Creating recoveryExecutor with unbounded pool"); - recoveryExecutor = ExecutorUtil.newMDCAwareCachedThreadPool(recoveryThreadFactory); + recoveryExecutor = ExecutorUtil.newMDCAwareCachedThreadPool("recoveryExecutor"); } } diff --git a/solr/core/src/test/org/apache/solr/handler/TestHttpRequestId.java b/solr/core/src/test/org/apache/solr/handler/TestHttpRequestId.java index ef0f486fa30e..83c88ab00fec 100644 --- a/solr/core/src/test/org/apache/solr/handler/TestHttpRequestId.java +++ b/solr/core/src/test/org/apache/solr/handler/TestHttpRequestId.java @@ -55,33 +55,33 @@ public static void beforeTest() throws Exception { public void mdcContextTest() { String collection = "/collection1"; BlockingQueue workQueue = new SynchronousQueue<>(false); - setupClientAndRun(collection, workQueue, 0); + setupClientAndRun(collection, workQueue, 0, Integer.MAX_VALUE); } @Test public void mdcContextFailureTest() { String collection = "/doesnotexist"; BlockingQueue workQueue = new SynchronousQueue<>(false); - setupClientAndRun(collection, workQueue, 0); + setupClientAndRun(collection, workQueue, 0, Integer.MAX_VALUE); } @Test public void mdcContextTest2() { String collection = "/collection1"; BlockingQueue workQueue = new ArrayBlockingQueue<>(10, false); - setupClientAndRun(collection, workQueue, 3); + setupClientAndRun(collection, workQueue, 3, 3); } @Test public void mdcContextFailureTest2() { String collection = "/doesnotexist"; BlockingQueue workQueue = new ArrayBlockingQueue<>(10, false); - setupClientAndRun(collection, workQueue, 3); + setupClientAndRun(collection, workQueue, 3, 3); } @SuppressForbidden(reason = "We need to use log4J2 classes directly to test MDC impacts") private void setupClientAndRun( - String collection, BlockingQueue workQueue, int corePoolSize) { + String collection, BlockingQueue workQueue, int corePoolSize, int maximumPoolSize) { final String key = "mdcContextTestKey" + System.nanoTime(); final String value = "TestHttpRequestId" + System.nanoTime(); @@ -91,7 +91,7 @@ private void setupClientAndRun( ThreadPoolExecutor commExecutor = new ExecutorUtil.MDCAwareThreadPoolExecutor( corePoolSize, - Integer.MAX_VALUE, + maximumPoolSize, 1, TimeUnit.SECONDS, workQueue, diff --git a/solr/core/src/test/org/apache/solr/handler/component/ParallelHttpShardHandlerTest.java b/solr/core/src/test/org/apache/solr/handler/component/ParallelHttpShardHandlerTest.java index 454f1a8f9fa9..06ac4856e52e 100644 --- a/solr/core/src/test/org/apache/solr/handler/component/ParallelHttpShardHandlerTest.java +++ b/solr/core/src/test/org/apache/solr/handler/component/ParallelHttpShardHandlerTest.java @@ -281,22 +281,17 @@ public void testTakeDoesNotHangUnderAsyncInnerFutureCompletion() throws Exceptio ExecutorService mockIoThreads = ExecutorUtil.newMDCAwareFixedThreadPool(2, new SolrNamedThreadFactory("testMockIo")); - ExecutorService takeExecutor = - ExecutorUtil.newMDCAwareCachedThreadPool(new SolrNamedThreadFactory("testTakeRunner")); - + ExecutorService takeExecutor = ExecutorUtil.newMDCAwareCachedThreadPool("testTakeRunner"); try { for (int i = 0; i < iterations; i++) { runAsyncRaceCycle(commExecutor, mockIoThreads, takeExecutor, i, perIterationTimeoutMs); } } finally { - takeExecutor.shutdownNow(); - takeExecutor.awaitTermination(5, TimeUnit.SECONDS); - mockIoThreads.shutdownNow(); - mockIoThreads.awaitTermination(5, TimeUnit.SECONDS); + ExecutorUtil.shutdownNowAndAwaitTermination(takeExecutor); + ExecutorUtil.shutdownNowAndAwaitTermination(mockIoThreads); commExecutor.shutdown(); if (!commExecutor.awaitTermination(15, TimeUnit.SECONDS)) { - commExecutor.shutdownNow(); - commExecutor.awaitTermination(5, TimeUnit.SECONDS); + ExecutorUtil.shutdownNowAndAwaitTermination(commExecutor); } } } diff --git a/solr/core/src/test/org/apache/solr/request/TestUnInvertedFieldException.java b/solr/core/src/test/org/apache/solr/request/TestUnInvertedFieldException.java index 5c1d11e6d614..02fab5e41ddf 100644 --- a/solr/core/src/test/org/apache/solr/request/TestUnInvertedFieldException.java +++ b/solr/core/src/test/org/apache/solr/request/TestUnInvertedFieldException.java @@ -84,10 +84,11 @@ public void testConcurrentInit() throws Exception { initCallables.add(() -> UnInvertedField.getUnInvertedField(proto.field(), searcher)); } + final int numThreads = TestUtil.nextInt(random(), 3, 6); final ThreadPoolExecutor pool = new MDCAwareThreadPoolExecutor( - 3, - TestUtil.nextInt(random(), 3, 6), + numThreads, + numThreads, 10, TimeUnit.MILLISECONDS, new LinkedBlockingQueue(), diff --git a/solr/core/src/test/org/apache/solr/update/TestInPlaceUpdatesDistrib.java b/solr/core/src/test/org/apache/solr/update/TestInPlaceUpdatesDistrib.java index 8a1dd5fd46c4..0b041703903e 100644 --- a/solr/core/src/test/org/apache/solr/update/TestInPlaceUpdatesDistrib.java +++ b/solr/core/src/test/org/apache/solr/update/TestInPlaceUpdatesDistrib.java @@ -59,7 +59,6 @@ import org.apache.solr.common.params.SolrParams; import org.apache.solr.common.util.ExecutorUtil; import org.apache.solr.common.util.NamedList; -import org.apache.solr.common.util.SolrNamedThreadFactory; import org.apache.solr.common.util.TimeSource; import org.apache.solr.embedded.JettySolrRunner; import org.apache.solr.index.NoMergePolicyFactory; @@ -343,9 +342,7 @@ private void reorderedDBQsSimpleTest() throws Exception { } // Reordering needs to happen using parallel threads - ExecutorService threadpool = - ExecutorUtil.newMDCAwareFixedThreadPool( - updates.size() + 1, new SolrNamedThreadFactory(getTestName())); + ExecutorService threadpool = newThreadPerUpdatePool(updates); // re-order the updates for NONLEADER 0 List reorderedUpdates = new ArrayList<>(updates); @@ -418,9 +415,7 @@ private void reorderedDBQIndividualReplicaTest() throws Exception { "inplace_updatable_float:" + (newinplace_updatable_float + 1), version0 + 3)); // Reordering needs to happen using parallel threads - ExecutorService threadpool = - ExecutorUtil.newMDCAwareFixedThreadPool( - updates.size() + 1, new SolrNamedThreadFactory(getTestName())); + ExecutorService threadpool = newThreadPerUpdatePool(updates); // re-order the updates by swapping the last two List reorderedUpdates = new ArrayList<>(updates); @@ -932,9 +927,7 @@ private void outOfOrderUpdatesIndividualReplicaTest() throws Exception { // Reordering needs to happen using parallel threads, since some of these updates will // be blocking calls, waiting for some previous update operations to arrive on which it depends. - ExecutorService threadpool = - ExecutorUtil.newMDCAwareFixedThreadPool( - updates.size() + 1, new SolrNamedThreadFactory(getTestName())); + ExecutorService threadpool = newThreadPerUpdatePool(updates); // re-order the updates for NONLEADER 0 List reorderedUpdates = new ArrayList<>(updates); @@ -1036,9 +1029,7 @@ private void reorderedDeletesTest() throws Exception { } // Reordering needs to happen using parallel threads - ExecutorService threadpool = - ExecutorUtil.newMDCAwareFixedThreadPool( - updates.size() + 1, new SolrNamedThreadFactory(getTestName())); + ExecutorService threadpool = newThreadPerUpdatePool(updates); // re-order the updates for NONLEADER 0 List reorderedUpdates = new ArrayList<>(updates); @@ -1143,9 +1134,7 @@ private void reorderedDBQsResurrectionTest() throws Exception { } // Reordering needs to happen using parallel threads - ExecutorService threadpool = - ExecutorUtil.newMDCAwareFixedThreadPool( - updates.size() + 1, new SolrNamedThreadFactory(getTestName())); + ExecutorService threadpool = newThreadPerUpdatePool(updates); // re-order the last two updates for NONLEADER 0 List reorderedUpdates = new ArrayList<>(updates); Collections.swap(reorderedUpdates, 2, 3); @@ -1248,9 +1237,7 @@ private void delayedReorderingFetchesMissingUpdateFromLeaderTest() throws Except .getFilter(ServletFixtures.DelayServlet.class) .addDelay("Waiting for dependant update to timeout", 1, 6000); - ExecutorService threadpool = - ExecutorUtil.newMDCAwareFixedThreadPool( - updates.size() + 1, new SolrNamedThreadFactory(getTestName())); + ExecutorService threadpool = newThreadPerUpdatePool(updates); for (UpdateRequest update : updates) { AsyncUpdateWithRandomCommit task = new AsyncUpdateWithRandomCommit(update, cloudClient, random().nextLong()); @@ -1353,9 +1340,7 @@ private void delayedReorderingFetchesMissingUpdateFromLeaderTest() throws Except .getFilter(ServletFixtures.DelayServlet.class) .addDelay("Waiting for dependant update to timeout", 4, 5998); // the delete update - threadpool = - ExecutorUtil.newMDCAwareFixedThreadPool( - updates.size() + 1, new SolrNamedThreadFactory(getTestName())); + threadpool = newThreadPerUpdatePool(updates); for (UpdateRequest update : updates) { AsyncUpdateWithRandomCommit task = new AsyncUpdateWithRandomCommit(update, cloudClient, random().nextLong()); @@ -1419,6 +1404,10 @@ public void checkExpectedSchemaField(Map expected) throws Except assertEquals("Field: " + fieldName, expected, rsp.getField()); } + private ExecutorService newThreadPerUpdatePool(List updates) { + return ExecutorUtil.newMDCAwareFixedThreadPool(updates.size() + 1, 1, getTestName()); + } + private class AsyncUpdateWithRandomCommit implements Callable { UpdateRequest update; SolrClient solrClient; @@ -1650,9 +1639,7 @@ private void reorderedDBQsUsingUpdatedValueFromADroppedUpdate() throws Exception .getFilter(ServletFixtures.DelayServlet.class) .addDelay("Waiting for dependant update to timeout", 2, 8000); - ExecutorService threadpool = - ExecutorUtil.newMDCAwareFixedThreadPool( - updates.size() + 1, new SolrNamedThreadFactory(getTestName())); + ExecutorService threadpool = newThreadPerUpdatePool(updates); for (UpdateRequest update : updates) { AsyncUpdateWithRandomCommit task = new AsyncUpdateWithRandomCommit(update, cloudClient, random().nextLong()); diff --git a/solr/core/src/test/org/apache/solr/util/OrderedExecutorTest.java b/solr/core/src/test/org/apache/solr/util/OrderedExecutorTest.java index 199062ba1a40..553437bb24b0 100644 --- a/solr/core/src/test/org/apache/solr/util/OrderedExecutorTest.java +++ b/solr/core/src/test/org/apache/solr/util/OrderedExecutorTest.java @@ -30,7 +30,6 @@ import java.util.concurrent.TimeoutException; import org.apache.solr.SolrTestCase; import org.apache.solr.common.util.ExecutorUtil; -import org.apache.solr.common.util.SolrNamedThreadFactory; import org.junit.Test; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -42,10 +41,10 @@ private static OrderedExecutor newOrderedExecutor(int numThreads) { // initialize exactly as done in CoreContainer so we test realistically return new OrderedExecutor<>( numThreads, - ExecutorUtil.newMDCAwareCachedThreadPool( + ExecutorUtil.newMDCAwareFixedThreadPool( numThreads, // thread count numThreads, // queue size - new SolrNamedThreadFactory("testOrderedExecutor"))); + "testOrderedExecutor")); } @Test diff --git a/solr/solrj-jetty/src/java/org/apache/solr/client/solrj/jetty/HttpJettySolrClient.java b/solr/solrj-jetty/src/java/org/apache/solr/client/solrj/jetty/HttpJettySolrClient.java index 0e5c885febae..2e34760a2d5c 100644 --- a/solr/solrj-jetty/src/java/org/apache/solr/client/solrj/jetty/HttpJettySolrClient.java +++ b/solr/solrj-jetty/src/java/org/apache/solr/client/solrj/jetty/HttpJettySolrClient.java @@ -55,7 +55,6 @@ import org.apache.solr.common.util.ExecutorUtil; import org.apache.solr.common.util.NamedList; import org.apache.solr.common.util.ObjectReleaseTracker; -import org.apache.solr.common.util.SolrNamedThreadFactory; import org.eclipse.jetty.client.AuthenticationStore; import org.eclipse.jetty.client.HttpClient; import org.eclipse.jetty.client.HttpClientTransport; @@ -83,7 +82,6 @@ import org.eclipse.jetty.http2.client.HTTP2Client; import org.eclipse.jetty.http2.client.transport.HttpClientTransportOverHTTP2; import org.eclipse.jetty.io.ClientConnector; -import org.eclipse.jetty.util.BlockingArrayQueue; import org.eclipse.jetty.util.ssl.KeyStoreScanner; import org.eclipse.jetty.util.ssl.SslContextFactory; import org.slf4j.Logger; @@ -220,10 +218,7 @@ public HttpClient getHttpClient() { private HttpClient createHttpClient(Builder builder) { executor = builder.getExecutor(); if (executor == null) { - BlockingArrayQueue queue = BlockingArrayQueue.newInstance(256, Integer.MAX_VALUE); - this.executor = - new ExecutorUtil.MDCAwareThreadPoolExecutor( - 32, 256, 60, TimeUnit.SECONDS, queue, new SolrNamedThreadFactory("h2sc")); + this.executor = ExecutorUtil.newMDCAwareCachedThreadPool("h2sc"); // nocommmit new PR shutdownExecutor = true; } else { shutdownExecutor = false; diff --git a/solr/solrj/src/java/org/apache/solr/client/solrj/impl/HttpJdkSolrClient.java b/solr/solrj/src/java/org/apache/solr/client/solrj/impl/HttpJdkSolrClient.java index 111bfd1bc92e..5fffd9199b82 100644 --- a/solr/solrj/src/java/org/apache/solr/client/solrj/impl/HttpJdkSolrClient.java +++ b/solr/solrj/src/java/org/apache/solr/client/solrj/impl/HttpJdkSolrClient.java @@ -37,12 +37,9 @@ import java.util.HashMap; import java.util.Locale; import java.util.Map; -import java.util.concurrent.BlockingQueue; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ExecutorService; import java.util.concurrent.Future; -import java.util.concurrent.LinkedBlockingQueue; -import java.util.concurrent.TimeUnit; import java.util.regex.Matcher; import java.util.regex.Pattern; import javax.net.ssl.SSLContext; @@ -60,7 +57,6 @@ import org.apache.solr.common.util.ExecutorUtil; import org.apache.solr.common.util.NamedList; import org.apache.solr.common.util.ObjectReleaseTracker; -import org.apache.solr.common.util.SolrNamedThreadFactory; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -106,15 +102,8 @@ protected HttpJdkSolrClient(String serverBaseUrl, HttpJdkSolrClient.Builder buil this.executor = builder.getExecutor(); this.shutdownExecutor = false; } else { - BlockingQueue queue = new LinkedBlockingQueue<>(1024); - this.executor = - new ExecutorUtil.MDCAwareThreadPoolExecutor( - 4, - 256, - 60, - TimeUnit.SECONDS, - queue, - new SolrNamedThreadFactory(this.getClass().getSimpleName())); + // nocommit new PR + this.executor = ExecutorUtil.newMDCAwareCachedThreadPool(this.getClass().getSimpleName()); this.shutdownExecutor = true; } b.executor(this.executor); diff --git a/solr/solrj/src/java/org/apache/solr/common/util/ExecutorUtil.java b/solr/solrj/src/java/org/apache/solr/common/util/ExecutorUtil.java index 48ac61f55d87..ffdb6f8fc5f8 100644 --- a/solr/solrj/src/java/org/apache/solr/common/util/ExecutorUtil.java +++ b/solr/solrj/src/java/org/apache/solr/common/util/ExecutorUtil.java @@ -35,11 +35,17 @@ import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicReference; -import java.util.stream.Collectors; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.slf4j.MDC; +/** + * Factories for {@link ExecutorService}s that propagate the SLF4J {@link MDC} to their tasks, plus + * helpers to shut them down. + * + *

To configure a pool that no factory here covers, see {@link MDCAwareThreadPoolExecutor} for + * how {@code corePoolSize}, {@code maximumPoolSize} and the queue interact. + */ public class ExecutorUtil { private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass()); @@ -210,7 +216,15 @@ public static void awaitTerminationForever(ExecutorService pool) { } } - /** See {@link java.util.concurrent.Executors#newFixedThreadPool(int, ThreadFactory)} */ + /** + * See {@link java.util.concurrent.Executors#newFixedThreadPool(int, ThreadFactory)}. The queue is + * unbounded, so work beyond {@code nThreads} is never rejected; it accumulates in memory instead. + * Use {@link #newMDCAwareFixedThreadPool(int, int, String)} to bound it — note that + * overload also reclaims idle threads, whereas these live as long as the pool. + * + *

This is good for fixed-size workloads, like for heavy/intensive work. + */ + @Deprecated(since = "10.1") // prefer the overloaded one, thus explicit about queue capacity public static ExecutorService newMDCAwareFixedThreadPool( int nThreads, ThreadFactory threadFactory) { return new MDCAwareThreadPoolExecutor( @@ -245,6 +259,10 @@ public static ExecutorService newMDCAwareCachedThreadPool(String name) { * Create a new pool of threads, with no limit for the number of threads. The pool has no task * queue. Each submitted task is executed immediately, either by reusing an existing thread if one * is available, or by starting a new thread. Unused threads will be closed after 60 seconds. + * + *

Thread count tracks how many tasks run concurrently, but nothing bounds it: a burst of + * simultaneous tasks starts a thread apiece. Only use this where something upstream already + * limits how many tasks can be in flight. */ public static ExecutorService newMDCAwareCachedThreadPool(ThreadFactory threadFactory) { return new MDCAwareThreadPoolExecutor( @@ -252,29 +270,93 @@ public static ExecutorService newMDCAwareCachedThreadPool(ThreadFactory threadFa } /** - * Create a new pool of threads. Threads are created for new work if there is room to do so up to - * {@code maxThreads}. Beyond that, the queue is used up to {@code queueCapacity}. Beyond that, - * work is rejected with an exception. Unused threads will be closed after 60 seconds. + * Create a new pool of at most {@code nThreads} threads. Each submitted task starts a thread + * until {@code nThreads} exist — even when an idle thread could have taken it — so a + * pool that sees many tasks reaches {@code nThreads} regardless of how many run at once. Beyond + * that, the queue is used up to {@code queueCapacity}. Beyond that, work is rejected with an + * exception. Unused threads will be closed after 60 seconds. + * + *

This is good for limiting heavy/intensive workloads, and that which need to reject tasks + * when the queue is full. */ - public static ExecutorService newMDCAwareCachedThreadPool( - int maxThreads, int queueCapacity, ThreadFactory threadFactory) { - // Create an executor with same value of core size and max total size. With an unbounded queue, - // the ThreadPoolExecutor ignores the configured max value and only considers core pool size. - // Since we allow core threads to die when idle for too long, this ends in having a pool with - // lazily-initialized and cached threads. + public static ExecutorService newMDCAwareFixedThreadPool( + int nThreads, int queueCapacity, String poolName) { + if (nThreads > 1024) { // likely wrong choice + assert false : "Creating a thread pool " + poolName + " with " + nThreads + " threads"; + log.warn("Creating a thread pool {} with {} threads", poolName, nThreads); + } + // Core and max must be equal: any queue with spare capacity is filled before threads are added + // beyond the core size, which would cap the pool below maxThreads. Core threads are then + // allowed to die so that an idle pool still releases its threads. MDCAwareThreadPoolExecutor executor = new MDCAwareThreadPoolExecutor( - maxThreads, - maxThreads, + nThreads, + nThreads, 60L, TimeUnit.SECONDS, new LinkedBlockingQueue<>(queueCapacity), - threadFactory); + new SolrNamedThreadFactory(poolName)); // Allow core threads to die executor.allowCoreThreadTimeOut(true); return executor; } + @Deprecated(since = "10.1") + public static ExecutorService newMDCAwareCachedThreadPool( + int maxThreads, int queueCapacity, ThreadFactory threadFactory) { + return newMDCAwareFixedThreadPool( + maxThreads, queueCapacity, ((SolrNamedThreadFactory) threadFactory).getPoolName()); + } + + /** + * A {@link ThreadPoolExecutor} that propagates the SLF4J {@link MDC} to its tasks. Prefer a + * factory in {@link ExecutorUtil} where one fits; construct this directly only to configure a + * pool they don't cover. + * + *

Choosing a pool shape

+ * + * {@link ThreadPoolExecutor} grows by two rules that are easy to get wrong: + * + *
    + *
  1. Below {@code corePoolSize}, every submitted task starts a new thread, even if other + * threads are idle. So the pool grows to {@code corePoolSize} in proportion to tasks + * submitted, not to how many run at once. + *
  2. At or above {@code corePoolSize}, a thread is created only once the queue is + * full. A queue with spare capacity therefore caps the pool at {@code + * corePoolSize} and makes {@code maximumPoolSize} unreachable; this class rejects that + * combination outright. + *
+ * + * Which leaves three shapes, chosen by what should happen when every thread is busy: + * + * + * + * + * + * + * + * + * + * + * + * + * + * + * + * + *
Pool shapes
When all threads are busyConfiguration
Start another thread, without limit{@code core=0}, {@code max=MAX_VALUE}, {@link SynchronousQueue} — see {@link + * ExecutorUtil#newMDCAwareCachedThreadPool(ThreadFactory)}
Start another thread up to N, then reject{@code core=0}, {@code max=N}, {@link SynchronousQueue}
Queue the task; concurrency stays at N{@code core=max=N} plus a queue — see {@link + * ExecutorUtil#newMDCAwareFixedThreadPool(int, int, String)}
+ * + * A {@link SynchronousQueue} holds nothing, so it hands a task to an idle thread when there is + * one and otherwise forces a new thread; that is what makes the first two shapes track actual + * concurrency. Set {@code corePoolSize} above 0 in those shapes only to keep that many threads + * warm for the life of the pool. + * + *

Threads beyond {@code corePoolSize} are reclaimed after {@code keepAliveTime} idle. Core + * threads live as long as the pool unless {@link #allowCoreThreadTimeOut(boolean)} is set, which + * subjects them to the same timeout. + */ @SuppressForbidden(reason = "class customizes ThreadPoolExecutor so it can be used instead") public static class MDCAwareThreadPoolExecutor extends ThreadPoolExecutor { @@ -291,6 +373,7 @@ public MDCAwareThreadPoolExecutor( ThreadFactory threadFactory, RejectedExecutionHandler handler) { super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, handler); + checkPoolConfig(corePoolSize, maximumPoolSize, workQueue); this.enableSubmitterStackTrace = true; } @@ -301,6 +384,7 @@ public MDCAwareThreadPoolExecutor( TimeUnit unit, BlockingQueue workQueue) { super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue); + checkPoolConfig(corePoolSize, maximumPoolSize, workQueue); this.enableSubmitterStackTrace = true; } @@ -323,6 +407,7 @@ public MDCAwareThreadPoolExecutor( ThreadFactory threadFactory, boolean enableSubmitterStackTrace) { super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory); + checkPoolConfig(corePoolSize, maximumPoolSize, workQueue); this.enableSubmitterStackTrace = enableSubmitterStackTrace; } @@ -334,9 +419,41 @@ public MDCAwareThreadPoolExecutor( BlockingQueue workQueue, RejectedExecutionHandler handler) { super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, handler); + checkPoolConfig(corePoolSize, maximumPoolSize, workQueue); this.enableSubmitterStackTrace = true; } + /** + * Rejects a pool that can never reach {@code maximumPoolSize}. {@link ThreadPoolExecutor} only + * creates threads beyond {@code corePoolSize} once the queue is full, so a queue with spare + * capacity silently caps the pool at {@code corePoolSize} (or at one thread, which is always + * created to rescue a queued task when {@code corePoolSize} is 0). + */ + private static void checkPoolConfig( + int corePoolSize, int maximumPoolSize, BlockingQueue workQueue) { + int queueCapacity = workQueue.remainingCapacity(); // exact; the queue is empty + int effectiveMaximum = Math.max(corePoolSize, 1); + if (effectiveMaximum < maximumPoolSize && queueCapacity > 0) { + throw new IllegalArgumentException( + "maximumPoolSize " + + maximumPoolSize + + " is unreachable: a queue of capacity " + + queueCapacity + + " is filled before threads are created beyond corePoolSize " + + corePoolSize + + ", capping the pool at " + + effectiveMaximum + + ". Use corePoolSize == maximumPoolSize (with allowCoreThreadTimeOut(true) to" + + " reclaim idle threads), or a SynchronousQueue."); + } + if (maximumPoolSize == Integer.MAX_VALUE + && !(workQueue instanceof SynchronousQueue)) { + // harmless to allow, but it's illogical; maybe the caller is confused + throw new IllegalArgumentException( + "maximumPoolSize is Integer.MAX_VALUE but the workQueue is not a SynchronousQueue."); + } + } + /** When the thread factory is a {@link SolrNamedThreadFactory}, prefixes the pool name. */ @Override public String toString() { @@ -473,8 +590,7 @@ public static Collection submitAllAndAwaitAggregatingExceptions( // Could alternatively use service.invokeAll, but this way we can start looping over futures // before all are done - List> futures = - tasks.stream().map(service::submit).collect(Collectors.toUnmodifiableList()); + List> futures = tasks.stream().map(service::submit).toList(); for (Future f : futures) { try { results.add(f.get()); diff --git a/solr/solrj/src/test/org/apache/solr/common/util/ExecutorUtilTest.java b/solr/solrj/src/test/org/apache/solr/common/util/ExecutorUtilTest.java index ed739632e84b..a42ebec7169c 100644 --- a/solr/solrj/src/test/org/apache/solr/common/util/ExecutorUtilTest.java +++ b/solr/solrj/src/test/org/apache/solr/common/util/ExecutorUtilTest.java @@ -25,10 +25,11 @@ import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutorService; import java.util.concurrent.Future; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.SynchronousQueue; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicLong; -import org.apache.lucene.util.NamedThreadFactory; import org.apache.solr.SolrTestCase; import org.apache.solr.logging.MDCLoggingContext; import org.apache.solr.util.TimeOut; @@ -115,8 +116,7 @@ public void testExecutorUtilAwaitsTerminationWhenTaskRespectsInterupt() throws E public void testCMDCAwareCachedThreadPool() throws Exception { // 5 threads max, unbounded queue ExecutorService executor = - ExecutorUtil.newMDCAwareCachedThreadPool( - 5, Integer.MAX_VALUE, new NamedThreadFactory("test")); + ExecutorUtil.newMDCAwareFixedThreadPool(5, Integer.MAX_VALUE, "test"); AtomicInteger concurrentTasks = new AtomicInteger(); AtomicInteger maxConcurrentTasks = new AtomicInteger(); @@ -276,6 +276,35 @@ public void submitAllWithExceptionsTest() { } } + @Test + public void unreachableMaximumPoolSizeIsRejected() { + // a queue with spare capacity is filled before threads are created beyond corePoolSize, + // so maximumPoolSize is unreachable and the pool is silently capped at corePoolSize + expectThrows( + IllegalArgumentException.class, + () -> + new ExecutorUtil.MDCAwareThreadPoolExecutor( + 4, 256, 60, TimeUnit.SECONDS, new LinkedBlockingQueue<>(1024))); + expectThrows( + IllegalArgumentException.class, + () -> + new ExecutorUtil.MDCAwareThreadPoolExecutor( + 4, 256, 60, TimeUnit.SECONDS, new LinkedBlockingQueue<>())); + + // a SynchronousQueue never holds a task, so maximumPoolSize is reachable + new ExecutorUtil.MDCAwareThreadPoolExecutor( + 4, 256, 60, TimeUnit.SECONDS, new SynchronousQueue<>()) + .shutdown(); + // core == max needs no growth beyond core + new ExecutorUtil.MDCAwareThreadPoolExecutor( + 256, 256, 60, TimeUnit.SECONDS, new LinkedBlockingQueue<>(1024)) + .shutdown(); + // a queued task is always rescued by one thread, so a maximum of 1 is reachable + ExecutorUtil.shutdownNowAndAwaitTermination( + ExecutorUtil.newMDCAwareSingleLazyThreadExecutor( + new SolrNamedThreadFactory("test"), 1, TimeUnit.SECONDS)); + } + @Test public void mdcAwarePoolToStringIncludesPoolName() { ExecutorService service = ExecutorUtil.newMDCAwareCachedThreadPool("test-async-task");