From 048e955a4412dbf48dbab300a9bbe60381725468 Mon Sep 17 00:00:00 2001 From: Carlos Delgado <6339205+carlosdelest@users.noreply.github.com> Date: Fri, 25 Oct 2024 07:45:40 +0200 Subject: [PATCH] Identify system threads using a Thread subclass (#113562) --- .../common/util/concurrent/EsExecutors.java | 35 ++++++++++--- .../DefaultBuiltInExecutorBuilders.java | 12 +++-- .../threadpool/ExecutorBuilder.java | 7 ++- .../threadpool/FixedExecutorBuilder.java | 49 +++++++++++++++++-- .../threadpool/ScalingExecutorBuilder.java | 4 +- .../util/concurrent/EsExecutorsTests.java | 8 ++- 6 files changed, 98 insertions(+), 17 deletions(-) diff --git a/server/src/main/java/org/elasticsearch/common/util/concurrent/EsExecutors.java b/server/src/main/java/org/elasticsearch/common/util/concurrent/EsExecutors.java index b10db7d4d1dd3..9120576815bac 100644 --- a/server/src/main/java/org/elasticsearch/common/util/concurrent/EsExecutors.java +++ b/server/src/main/java/org/elasticsearch/common/util/concurrent/EsExecutors.java @@ -326,16 +326,25 @@ public static String executorName(Thread thread) { } public static ThreadFactory daemonThreadFactory(Settings settings, String namePrefix) { - return daemonThreadFactory(threadName(settings, namePrefix)); + return createDaemonThreadFactory(threadName(settings, namePrefix), false); } public static ThreadFactory daemonThreadFactory(String nodeName, String namePrefix) { + return daemonThreadFactory(nodeName, namePrefix, false); + } + + public static ThreadFactory daemonThreadFactory(String nodeName, String namePrefix, boolean isSystemThread) { assert nodeName != null && false == nodeName.isEmpty(); - return daemonThreadFactory(threadName(nodeName, namePrefix)); + return createDaemonThreadFactory(threadName(nodeName, namePrefix), isSystemThread); } - public static ThreadFactory daemonThreadFactory(String namePrefix) { - return new EsThreadFactory(namePrefix); + public static ThreadFactory daemonThreadFactory(String name) { + assert name != null && name.isEmpty() == false; + return createDaemonThreadFactory(name, false); + } + + private static ThreadFactory createDaemonThreadFactory(String namePrefix, boolean isSystemThread) { + return new EsThreadFactory(namePrefix, isSystemThread); } static class EsThreadFactory implements ThreadFactory { @@ -343,22 +352,36 @@ static class EsThreadFactory implements ThreadFactory { final ThreadGroup group; final AtomicInteger threadNumber = new AtomicInteger(1); final String namePrefix; + final boolean isSystem; - EsThreadFactory(String namePrefix) { + EsThreadFactory(String namePrefix, boolean isSystem) { this.namePrefix = namePrefix; SecurityManager s = System.getSecurityManager(); group = (s != null) ? s.getThreadGroup() : Thread.currentThread().getThreadGroup(); + this.isSystem = isSystem; } @Override public Thread newThread(Runnable r) { return AccessController.doPrivileged((PrivilegedAction) () -> { - Thread t = new Thread(group, r, namePrefix + "[T#" + threadNumber.getAndIncrement() + "]", 0); + Thread t = new EsThread(group, r, namePrefix + "[T#" + threadNumber.getAndIncrement() + "]", 0, isSystem); t.setDaemon(true); return t; }); } + } + public static class EsThread extends Thread { + private final boolean isSystem; + + EsThread(ThreadGroup group, Runnable target, String name, long stackSize, boolean isSystem) { + super(group, target, name, stackSize); + this.isSystem = isSystem; + } + + public boolean isSystem() { + return isSystem; + } } /** diff --git a/server/src/main/java/org/elasticsearch/threadpool/DefaultBuiltInExecutorBuilders.java b/server/src/main/java/org/elasticsearch/threadpool/DefaultBuiltInExecutorBuilders.java index c3a24d012c013..a97d22a976631 100644 --- a/server/src/main/java/org/elasticsearch/threadpool/DefaultBuiltInExecutorBuilders.java +++ b/server/src/main/java/org/elasticsearch/threadpool/DefaultBuiltInExecutorBuilders.java @@ -170,7 +170,8 @@ public Map getBuilders(Settings settings, int allocated ThreadPool.Names.SYSTEM_READ, halfProcMaxAt5, 2000, - EsExecutors.TaskTrackingConfig.DO_NOT_TRACK + EsExecutors.TaskTrackingConfig.DO_NOT_TRACK, + true ) ); result.put( @@ -180,7 +181,8 @@ public Map getBuilders(Settings settings, int allocated ThreadPool.Names.SYSTEM_WRITE, halfProcMaxAt5, 1000, - new EsExecutors.TaskTrackingConfig(true, indexAutoscalingEWMA) + new EsExecutors.TaskTrackingConfig(true, indexAutoscalingEWMA), + true ) ); result.put( @@ -190,7 +192,8 @@ public Map getBuilders(Settings settings, int allocated ThreadPool.Names.SYSTEM_CRITICAL_READ, halfProcMaxAt5, 2000, - EsExecutors.TaskTrackingConfig.DO_NOT_TRACK + EsExecutors.TaskTrackingConfig.DO_NOT_TRACK, + true ) ); result.put( @@ -200,7 +203,8 @@ public Map getBuilders(Settings settings, int allocated ThreadPool.Names.SYSTEM_CRITICAL_WRITE, halfProcMaxAt5, 1500, - new EsExecutors.TaskTrackingConfig(true, indexAutoscalingEWMA) + new EsExecutors.TaskTrackingConfig(true, indexAutoscalingEWMA), + true ) ); return unmodifiableMap(result); diff --git a/server/src/main/java/org/elasticsearch/threadpool/ExecutorBuilder.java b/server/src/main/java/org/elasticsearch/threadpool/ExecutorBuilder.java index 2337d51d07571..c259feb1c978e 100644 --- a/server/src/main/java/org/elasticsearch/threadpool/ExecutorBuilder.java +++ b/server/src/main/java/org/elasticsearch/threadpool/ExecutorBuilder.java @@ -24,9 +24,11 @@ public abstract class ExecutorBuilder { private final String name; + private final boolean isSystemThread; - public ExecutorBuilder(String name) { + public ExecutorBuilder(String name, boolean isSystemThread) { this.name = name; + this.isSystemThread = isSystemThread; } protected String name() { @@ -90,4 +92,7 @@ abstract static class ExecutorSettings { } + public boolean isSystemThread() { + return isSystemThread; + } } diff --git a/server/src/main/java/org/elasticsearch/threadpool/FixedExecutorBuilder.java b/server/src/main/java/org/elasticsearch/threadpool/FixedExecutorBuilder.java index 07db563da39a1..9c723f241f1d0 100644 --- a/server/src/main/java/org/elasticsearch/threadpool/FixedExecutorBuilder.java +++ b/server/src/main/java/org/elasticsearch/threadpool/FixedExecutorBuilder.java @@ -51,7 +51,28 @@ public final class FixedExecutorBuilder extends ExecutorBuilder( sizeKey, @@ -102,7 +145,7 @@ FixedExecutorSettings getSettings(Settings settings) { ThreadPool.ExecutorHolder build(final FixedExecutorSettings settings, final ThreadContext threadContext) { int size = settings.size; int queueSize = settings.queueSize; - final ThreadFactory threadFactory = EsExecutors.daemonThreadFactory(EsExecutors.threadName(settings.nodeName, name())); + final ThreadFactory threadFactory = EsExecutors.daemonThreadFactory(settings.nodeName, name(), isSystemThread()); final ExecutorService executor = EsExecutors.newFixed( settings.nodeName + "/" + name(), size, diff --git a/server/src/main/java/org/elasticsearch/threadpool/ScalingExecutorBuilder.java b/server/src/main/java/org/elasticsearch/threadpool/ScalingExecutorBuilder.java index a31f940cdb2dc..1017d41a77444 100644 --- a/server/src/main/java/org/elasticsearch/threadpool/ScalingExecutorBuilder.java +++ b/server/src/main/java/org/elasticsearch/threadpool/ScalingExecutorBuilder.java @@ -104,7 +104,7 @@ public ScalingExecutorBuilder( final String prefix, final EsExecutors.TaskTrackingConfig trackingConfig ) { - super(name); + super(name, false); this.coreSetting = Setting.intSetting(settingsKey(prefix, "core"), core, Setting.Property.NodeScope); this.maxSetting = Setting.intSetting(settingsKey(prefix, "max"), max, Setting.Property.NodeScope); this.keepAliveSetting = Setting.timeSetting(settingsKey(prefix, "keep_alive"), keepAlive, Setting.Property.NodeScope); @@ -131,7 +131,7 @@ ThreadPool.ExecutorHolder build(final ScalingExecutorSettings settings, final Th int core = settings.core; int max = settings.max; final ThreadPool.Info info = new ThreadPool.Info(name(), ThreadPool.ThreadPoolType.SCALING, core, max, keepAlive, null); - final ThreadFactory threadFactory = EsExecutors.daemonThreadFactory(EsExecutors.threadName(settings.nodeName, name())); + final ThreadFactory threadFactory = EsExecutors.daemonThreadFactory(settings.nodeName, name()); ExecutorService executor; executor = EsExecutors.newScaling( settings.nodeName + "/" + name(), diff --git a/server/src/test/java/org/elasticsearch/common/util/concurrent/EsExecutorsTests.java b/server/src/test/java/org/elasticsearch/common/util/concurrent/EsExecutorsTests.java index bdfec9dfaa630..2867c9e007937 100644 --- a/server/src/test/java/org/elasticsearch/common/util/concurrent/EsExecutorsTests.java +++ b/server/src/test/java/org/elasticsearch/common/util/concurrent/EsExecutorsTests.java @@ -635,15 +635,19 @@ public void testParseExecutorName() throws InterruptedException { final var executorName = randomAlphaOfLength(10); final String nodeName = rarely() ? null : randomIdentifier(); final ThreadFactory threadFactory; + final boolean isSystem; if (nodeName == null) { + isSystem = false; threadFactory = EsExecutors.daemonThreadFactory(Settings.EMPTY, executorName); } else if (randomBoolean()) { + isSystem = false; threadFactory = EsExecutors.daemonThreadFactory( Settings.builder().put(Node.NODE_NAME_SETTING.getKey(), nodeName).build(), executorName ); } else { - threadFactory = EsExecutors.daemonThreadFactory(nodeName, executorName); + isSystem = randomBoolean(); + threadFactory = EsExecutors.daemonThreadFactory(nodeName, executorName, isSystem); } final var thread = threadFactory.newThread(() -> {}); @@ -652,6 +656,8 @@ public void testParseExecutorName() throws InterruptedException { assertThat(EsExecutors.executorName(thread), equalTo(executorName)); assertThat(EsExecutors.executorName("TEST-" + thread.getName()), is(nullValue())); assertThat(EsExecutors.executorName("LuceneTestCase" + thread.getName()), is(nullValue())); + assertThat(EsExecutors.executorName("LuceneTestCase" + thread.getName()), is(nullValue())); + assertThat(((EsExecutors.EsThread) thread).isSystem(), equalTo(isSystem)); } finally { thread.join(); }