You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by tk...@apache.org on 2025/03/12 12:28:45 UTC
(ignite-3) branch main updated: IGNITE-24562 Avoid "Message handling has been too long" message in custom pools (#5399)
This is an automated email from the ASF dual-hosted git repository.
tkalkirill pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/ignite-3.git
The following commit(s) were added to refs/heads/main by this push:
new f6c5ac911d2 IGNITE-24562 Avoid "Message handling has been too long" message in custom pools (#5399)
f6c5ac911d2 is described below
commit f6c5ac911d24268957fe6250369df055ce9e7902
Author: Kirill Tkalenko <tk...@yandex.ru>
AuthorDate: Wed Mar 12 15:28:40 2025 +0300
IGNITE-24562 Avoid "Message handling has been too long" message in custom pools (#5399)
---
.../ignite/internal/thread/IgniteThread.java | 2 +-
.../network/IgniteMessageServiceThread.java | 38 ++++++++++++
.../network/IgniteMessageServiceThreadFactory.java | 71 ++++++++++++++++++++++
.../network/TrackableNetworkMessageHandler.java | 21 +------
.../CriticalStripedThreadPoolExecutorFactory.java | 3 +-
.../internal/network/DefaultMessagingService.java | 3 +-
6 files changed, 115 insertions(+), 23 deletions(-)
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/thread/IgniteThread.java b/modules/core/src/main/java/org/apache/ignite/internal/thread/IgniteThread.java
index 95f325e3a71..413566512e8 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/thread/IgniteThread.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/thread/IgniteThread.java
@@ -65,7 +65,7 @@ public class IgniteThread extends Thread implements ThreadAttributes {
* @param r Runnable to execute.
* @param allowedOperations Operations which this thread allows to execute.
*/
- IgniteThread(String finalName, Runnable r, ThreadOperation... allowedOperations) {
+ public IgniteThread(String finalName, Runnable r, ThreadOperation... allowedOperations) {
super(r, finalName);
Set<ThreadOperation> operations = EnumSet.noneOf(ThreadOperation.class);
diff --git a/modules/network-api/src/main/java/org/apache/ignite/internal/network/IgniteMessageServiceThread.java b/modules/network-api/src/main/java/org/apache/ignite/internal/network/IgniteMessageServiceThread.java
new file mode 100644
index 00000000000..5ecba01fecd
--- /dev/null
+++ b/modules/network-api/src/main/java/org/apache/ignite/internal/network/IgniteMessageServiceThread.java
@@ -0,0 +1,38 @@
+/*
+ * 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.ignite.internal.network;
+
+import org.apache.ignite.internal.thread.IgniteThread;
+import org.apache.ignite.internal.thread.ThreadOperation;
+
+/**
+ * Extension for use in {@link MessagingService} implementations that will allow, for example, tracking long message processing, which can
+ * be critical for network communications.
+ */
+public class IgniteMessageServiceThread extends IgniteThread {
+ /**
+ * Creates thread with given name.
+ *
+ * @param finalName Name of thread.
+ * @param r Runnable to execute.
+ * @param allowedOperations Operations which this thread allows to execute.
+ */
+ public IgniteMessageServiceThread(String finalName, Runnable r, ThreadOperation... allowedOperations) {
+ super(finalName, r, allowedOperations);
+ }
+}
diff --git a/modules/network-api/src/main/java/org/apache/ignite/internal/network/IgniteMessageServiceThreadFactory.java b/modules/network-api/src/main/java/org/apache/ignite/internal/network/IgniteMessageServiceThreadFactory.java
new file mode 100644
index 00000000000..0c943083700
--- /dev/null
+++ b/modules/network-api/src/main/java/org/apache/ignite/internal/network/IgniteMessageServiceThreadFactory.java
@@ -0,0 +1,71 @@
+/*
+ * 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.ignite.internal.network;
+
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.ignite.internal.logger.IgniteLogger;
+import org.apache.ignite.internal.thread.IgniteThread;
+import org.apache.ignite.internal.thread.LogUncaughtExceptionHandler;
+import org.apache.ignite.internal.thread.ThreadOperation;
+
+/** {@link IgniteMessageServiceThread} factory. */
+public class IgniteMessageServiceThreadFactory implements ThreadFactory {
+ /** Thread prefix (when computing final thread name, no dash will be added to the prefix). */
+ private final String prefix;
+
+ /** Thread counter. */
+ private final AtomicInteger counter = new AtomicInteger(0);
+
+ private final Thread.UncaughtExceptionHandler exceptionHandler;
+
+ private final ThreadOperation[] allowedOperations;
+
+ private IgniteMessageServiceThreadFactory(String prefix, ThreadOperation[] allowedOperations, IgniteLogger log) {
+ this.prefix = prefix;
+ this.allowedOperations = allowedOperations;
+
+ this.exceptionHandler = new LogUncaughtExceptionHandler(log);
+ }
+
+ @Override
+ public Thread newThread(Runnable r) {
+ var t = new IgniteMessageServiceThread(prefix + counter.getAndIncrement(), r, allowedOperations);
+
+ t.setUncaughtExceptionHandler(exceptionHandler);
+
+ return t;
+ }
+
+ /**
+ * Creates a non-daemon thread factory based on a node's name and a name of the pool.
+ *
+ * @param nodeName Node name.
+ * @param poolName Pool name.
+ * @param logger Logger.
+ * @param allowedOperations Operations that are allowed to be executed on threads produced by this factory.
+ */
+ public static IgniteMessageServiceThreadFactory create(
+ String nodeName,
+ String poolName,
+ IgniteLogger logger,
+ ThreadOperation... allowedOperations
+ ) {
+ return new IgniteMessageServiceThreadFactory(IgniteThread.threadPrefix(nodeName, poolName), allowedOperations, logger);
+ }
+}
diff --git a/modules/network-api/src/main/java/org/apache/ignite/internal/network/TrackableNetworkMessageHandler.java b/modules/network-api/src/main/java/org/apache/ignite/internal/network/TrackableNetworkMessageHandler.java
index fcd5423c095..0a4454eddd6 100644
--- a/modules/network-api/src/main/java/org/apache/ignite/internal/network/TrackableNetworkMessageHandler.java
+++ b/modules/network-api/src/main/java/org/apache/ignite/internal/network/TrackableNetworkMessageHandler.java
@@ -19,12 +19,9 @@ package org.apache.ignite.internal.network;
import static org.apache.ignite.internal.tostring.IgniteToStringBuilder.includeSensitive;
-import java.util.Set;
import java.util.concurrent.TimeUnit;
import org.apache.ignite.internal.logger.IgniteLogger;
import org.apache.ignite.internal.logger.Loggers;
-import org.apache.ignite.internal.thread.ThreadAttributes;
-import org.apache.ignite.internal.thread.ThreadOperation;
import org.apache.ignite.network.ClusterNode;
import org.jetbrains.annotations.Nullable;
@@ -51,25 +48,13 @@ public class TrackableNetworkMessageHandler implements NetworkMessageHandler {
targetHandler.onReceived(message, sender, correlationId);
- if (!storageThread()) {
+ if (isNetworkThread()) {
maybeLogLongProcessing(message, startTimeNanos);
}
}
- private static boolean storageThread() {
- Thread currentThread = Thread.currentThread();
-
- if (!(currentThread instanceof ThreadAttributes)) {
- return false;
- }
-
- ThreadAttributes current = (ThreadAttributes) currentThread;
-
- Set<ThreadOperation> allowedOperations = current.allowedOperations();
-
- return allowedOperations.contains(ThreadOperation.STORAGE_READ)
- || allowedOperations.contains(ThreadOperation.STORAGE_WRITE)
- || allowedOperations.contains(ThreadOperation.TX_STATE_STORAGE_ACCESS);
+ private static boolean isNetworkThread() {
+ return Thread.currentThread() instanceof IgniteMessageServiceThread;
}
private static void maybeLogLongProcessing(NetworkMessage message, long startTimeNanos) {
diff --git a/modules/network/src/main/java/org/apache/ignite/internal/network/CriticalStripedThreadPoolExecutorFactory.java b/modules/network/src/main/java/org/apache/ignite/internal/network/CriticalStripedThreadPoolExecutorFactory.java
index 308f64e66dd..d2ee399c759 100644
--- a/modules/network/src/main/java/org/apache/ignite/internal/network/CriticalStripedThreadPoolExecutorFactory.java
+++ b/modules/network/src/main/java/org/apache/ignite/internal/network/CriticalStripedThreadPoolExecutorFactory.java
@@ -21,7 +21,6 @@ import static org.apache.ignite.internal.thread.ThreadOperation.NOTHING_ALLOWED;
import java.util.List;
import org.apache.ignite.internal.logger.IgniteLogger;
-import org.apache.ignite.internal.thread.IgniteThreadFactory;
import org.apache.ignite.internal.worker.CriticalStripedThreadPoolExecutor;
import org.apache.ignite.internal.worker.CriticalWorker;
import org.apache.ignite.internal.worker.CriticalWorkerRegistry;
@@ -62,7 +61,7 @@ class CriticalStripedThreadPoolExecutorFactory {
short channelTypeId = channelType.id();
String poolName = poolNamePrefix + "-" + channelType.name() + "-" + channelTypeId;
- var threadFactory = IgniteThreadFactory.create(nodeName, poolName, log, NOTHING_ALLOWED);
+ var threadFactory = IgniteMessageServiceThreadFactory.create(nodeName, poolName, log, NOTHING_ALLOWED);
var executor = new CriticalStripedThreadPoolExecutor(stripeCountForIndex(channelTypeId), threadFactory, false, 0);
for (CriticalWorker worker : executor.workers()) {
diff --git a/modules/network/src/main/java/org/apache/ignite/internal/network/DefaultMessagingService.java b/modules/network/src/main/java/org/apache/ignite/internal/network/DefaultMessagingService.java
index 892491c8d13..ec6e39de78b 100644
--- a/modules/network/src/main/java/org/apache/ignite/internal/network/DefaultMessagingService.java
+++ b/modules/network/src/main/java/org/apache/ignite/internal/network/DefaultMessagingService.java
@@ -66,7 +66,6 @@ import org.apache.ignite.internal.network.serialization.ClassDescriptorRegistry;
import org.apache.ignite.internal.network.serialization.marshal.UserObjectMarshaller;
import org.apache.ignite.internal.thread.ExecutorChooser;
import org.apache.ignite.internal.thread.IgniteThread;
-import org.apache.ignite.internal.thread.IgniteThreadFactory;
import org.apache.ignite.internal.worker.CriticalSingleThreadExecutor;
import org.apache.ignite.internal.worker.CriticalWorkerRegistry;
import org.apache.ignite.lang.IgniteException;
@@ -156,7 +155,7 @@ public class DefaultMessagingService extends AbstractMessagingService {
this.criticalWorkerRegistry = criticalWorkerRegistry;
outboundExecutor = new CriticalSingleThreadExecutor(
- IgniteThreadFactory.create(nodeName, "MessagingService-outbound", LOG, NOTHING_ALLOWED)
+ IgniteMessageServiceThreadFactory.create(nodeName, "MessagingService-outbound", LOG, NOTHING_ALLOWED)
);
inboundExecutors = new CriticalStripedExecutors(