This is an automated email from the ASF dual-hosted git repository.
jiangtian pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new 68da097d2fb Use global sink executor pool for IoTV2 (#16163)
68da097d2fb is described below
commit 68da097d2fb619bbb7a2c9bc003faf71135f0de6
Author: Peng Junzhi <[email protected]>
AuthorDate: Wed Sep 3 11:44:08 2025 -0500
Use global sink executor pool for IoTV2 (#16163)
* use global sink executor pool for IoTV2
* resolve conflict
* resolve conflict
* fix review
* fix
* fix comment
* fix review
* fix ci
* fix
* fix it npe
* fix
---------
Co-authored-by: 彭俊植 <[email protected]>
---
.../apache/iotdb/consensus/ConsensusFactory.java | 2 +-
.../apache/iotdb/consensus/pipe/PipeConsensus.java | 2 +-
.../task/builder/PipeDataNodeTaskBuilder.java | 2 +-
.../task/execution/PipeSubtaskExecutorManager.java | 15 +++++----
.../task/subtask/sink/PipeSinkSubtaskManager.java | 1 -
.../consensus/PipeConsensusSubtaskExecutor.java | 39 +++++++++++++++++-----
.../pipeconsensus/PipeConsensusAsyncSink.java | 2 +-
.../pipeconsensus/PipeConsensusSyncSink.java | 2 +-
.../db/storageengine/dataregion/DataRegion.java | 2 +-
.../container/IoTV2GlobalComponentContainer.java | 23 ++++++++++---
10 files changed, 64 insertions(+), 26 deletions(-)
diff --git
a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ConsensusFactory.java
b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ConsensusFactory.java
index a9dd2c1e5ea..9d7a30279ad 100644
---
a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ConsensusFactory.java
+++
b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ConsensusFactory.java
@@ -19,7 +19,7 @@
package org.apache.iotdb.consensus;
-import org.apache.iotdb.commons.client.container.IoTV2GlobalComponentContainer;
+import
org.apache.iotdb.commons.consensus.iotv2.container.IoTV2GlobalComponentContainer;
import org.apache.iotdb.consensus.config.ConsensusConfig;
import org.apache.iotdb.consensus.pipe.metric.PipeConsensusSyncLagManager;
diff --git
a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/pipe/PipeConsensus.java
b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/pipe/PipeConsensus.java
index 56792236c23..762f338ad96 100644
---
a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/pipe/PipeConsensus.java
+++
b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/pipe/PipeConsensus.java
@@ -23,9 +23,9 @@ import org.apache.iotdb.common.rpc.thrift.TEndPoint;
import org.apache.iotdb.common.rpc.thrift.TSStatus;
import org.apache.iotdb.commons.client.IClientManager;
import org.apache.iotdb.commons.client.async.AsyncPipeConsensusServiceClient;
-import org.apache.iotdb.commons.client.container.IoTV2GlobalComponentContainer;
import org.apache.iotdb.commons.client.sync.SyncPipeConsensusServiceClient;
import org.apache.iotdb.commons.consensus.ConsensusGroupId;
+import
org.apache.iotdb.commons.consensus.iotv2.container.IoTV2GlobalComponentContainer;
import org.apache.iotdb.commons.exception.StartupException;
import org.apache.iotdb.commons.pipe.agent.task.meta.PipeStatus;
import org.apache.iotdb.commons.service.RegisterManager;
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/builder/PipeDataNodeTaskBuilder.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/builder/PipeDataNodeTaskBuilder.java
index 7ff2f3052dc..5ca08e840fb 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/builder/PipeDataNodeTaskBuilder.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/builder/PipeDataNodeTaskBuilder.java
@@ -123,7 +123,7 @@ public class PipeDataNodeTaskBuilder {
regionId,
pipeType.equals(PipeType.USER)
?
PipeSubtaskExecutorManager.getInstance().getConnectorExecutorSupplier()
- :
PipeSubtaskExecutorManager.getInstance().getConsensusExecutorSupplier());
+ :
PipeSubtaskExecutorManager.getInstance()::getConsensusExecutor);
}
// The processor connects the extractor and connector.
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/execution/PipeSubtaskExecutorManager.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/execution/PipeSubtaskExecutorManager.java
index 61edbe5d60f..bed009d37c9 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/execution/PipeSubtaskExecutorManager.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/execution/PipeSubtaskExecutorManager.java
@@ -19,6 +19,7 @@
package org.apache.iotdb.db.pipe.agent.task.execution;
+import
org.apache.iotdb.commons.consensus.iotv2.container.IoTV2GlobalComponentContainer;
import org.apache.iotdb.commons.subscription.config.SubscriptionConfig;
import org.apache.iotdb.db.pipe.consensus.PipeConsensusSubtaskExecutor;
import
org.apache.iotdb.db.subscription.task.execution.SubscriptionSubtaskExecutor;
@@ -33,7 +34,6 @@ public class PipeSubtaskExecutorManager {
private final PipeProcessorSubtaskExecutor processorExecutor;
private final Supplier<PipeSinkSubtaskExecutor> connectorExecutorSupplier;
private final SubscriptionSubtaskExecutor subscriptionExecutor;
- private final Supplier<PipeConsensusSubtaskExecutor>
consensusExecutorSupplier;
public PipeProcessorSubtaskExecutor getProcessorExecutor() {
return processorExecutor;
@@ -43,12 +43,13 @@ public class PipeSubtaskExecutorManager {
return connectorExecutorSupplier;
}
- public SubscriptionSubtaskExecutor getSubscriptionExecutor() {
- return subscriptionExecutor;
+ public PipeConsensusSubtaskExecutor getConsensusExecutor() {
+ return (PipeConsensusSubtaskExecutor)
+ IoTV2GlobalComponentContainer.getInstance().getConsensusExecutor();
}
- public Supplier<PipeConsensusSubtaskExecutor> getConsensusExecutorSupplier()
{
- return consensusExecutorSupplier;
+ public SubscriptionSubtaskExecutor getSubscriptionExecutor() {
+ return subscriptionExecutor;
}
///////////////////////// Singleton Instance Holder
/////////////////////////
@@ -60,7 +61,9 @@ public class PipeSubtaskExecutorManager {
SubscriptionConfig.getInstance().getSubscriptionEnabled()
? new SubscriptionSubtaskExecutor()
: null;
- consensusExecutorSupplier = PipeConsensusSubtaskExecutor::new;
+ // IoTV2 uses global singleton executor pool.
+ IoTV2GlobalComponentContainer.getInstance()
+ .setConsensusExecutor(new PipeConsensusSubtaskExecutor());
}
private static class PipeTaskExecutorHolder {
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/sink/PipeSinkSubtaskManager.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/sink/PipeSinkSubtaskManager.java
index c2a5575e4ce..a5e217d8919 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/sink/PipeSinkSubtaskManager.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/sink/PipeSinkSubtaskManager.java
@@ -114,7 +114,6 @@ public class PipeSinkSubtaskManager {
if
(!attributeSortedString2SubtaskLifeCycleMap.containsKey(attributeSortedString))
{
final PipeSinkSubtaskExecutor executor = executorSupplier.get();
-
final List<PipeSinkSubtaskLifeCycle> pipeSinkSubtaskLifeCycleList =
new ArrayList<>(connectorNum);
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/consensus/PipeConsensusSubtaskExecutor.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/consensus/PipeConsensusSubtaskExecutor.java
index 8b91c8d4089..32cb9be23db 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/consensus/PipeConsensusSubtaskExecutor.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/consensus/PipeConsensusSubtaskExecutor.java
@@ -19,18 +19,41 @@
package org.apache.iotdb.db.pipe.consensus;
import org.apache.iotdb.commons.concurrent.ThreadName;
-import org.apache.iotdb.commons.pipe.config.PipeConfig;
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.pipe.agent.task.execution.PipeSinkSubtaskExecutor;
-
-import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.iotdb.db.storageengine.dataregion.DataRegion;
public class PipeConsensusSubtaskExecutor extends PipeSinkSubtaskExecutor {
-
- private static final AtomicInteger id = new AtomicInteger(0);
-
public PipeConsensusSubtaskExecutor() {
super(
- PipeConfig.getInstance().getPipeSubtaskExecutorMaxThreadNum(),
- ThreadName.PIPE_CONSENSUS_EXECUTOR_POOL + "-" + id.getAndIncrement());
+ DataRegion.getAcquireDirectBufferMemCost() == 0
+ ? Runtime.getRuntime().availableProcessors()
+ :
+ // The number of data regions for a datanode is limited by
offHeapMemory. At the same
+ // time,
+ // in order to ensure multicore performance, the number of data
regions usually does not
+ // exceed the number of cores. To prevent the thread from
exploding, we take the min of
+ // both.
+ (int)
+ Math.min(
+ // NOTE1: The number of data regions is limited by the
number of cores.
+ Runtime.getRuntime().availableProcessors(),
+ // NOTE2: The number of data regions is also limited by
the offHeapMemory. In
+ // fact,
+ // schema regions also take up off-heap memory. So in
fact, the soft cap of
+ // data region will be smaller than what is calculated
here, but that's okay, we
+ // can
+ // set the core size quota of the Executor pool a little
bit higher slightly
+ Math.max(
+ IoTDBDescriptor.getInstance()
+ .getMemoryConfig()
+ .getDirectBufferMemoryManager()
+ .getTotalMemorySizeInBytes(),
+ IoTDBDescriptor.getInstance()
+ .getMemoryConfig()
+ .getOffHeapMemoryManager()
+ .getTotalMemorySizeInBytes())
+ / DataRegion.getAcquireDirectBufferMemCost()),
+ ThreadName.PIPE_CONSENSUS_EXECUTOR_POOL.getName());
}
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/pipeconsensus/PipeConsensusAsyncSink.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/pipeconsensus/PipeConsensusAsyncSink.java
index fe110ff6645..a481f7cde0d 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/pipeconsensus/PipeConsensusAsyncSink.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/pipeconsensus/PipeConsensusAsyncSink.java
@@ -24,9 +24,9 @@ import org.apache.iotdb.common.rpc.thrift.TConsensusGroupType;
import org.apache.iotdb.common.rpc.thrift.TEndPoint;
import org.apache.iotdb.commons.client.IClientManager;
import org.apache.iotdb.commons.client.async.AsyncPipeConsensusServiceClient;
-import org.apache.iotdb.commons.client.container.IoTV2GlobalComponentContainer;
import org.apache.iotdb.commons.consensus.ConsensusGroupId;
import org.apache.iotdb.commons.consensus.index.ProgressIndex;
+import
org.apache.iotdb.commons.consensus.iotv2.container.IoTV2GlobalComponentContainer;
import
org.apache.iotdb.commons.exception.pipe.PipeRuntimeSinkRetryTimesConfigurableException;
import org.apache.iotdb.commons.pipe.event.EnrichedEvent;
import org.apache.iotdb.commons.pipe.sink.protocol.IoTDBSink;
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/pipeconsensus/PipeConsensusSyncSink.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/pipeconsensus/PipeConsensusSyncSink.java
index 4ee9e1f2352..bb0dcf2dc09 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/pipeconsensus/PipeConsensusSyncSink.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/pipeconsensus/PipeConsensusSyncSink.java
@@ -24,9 +24,9 @@ import org.apache.iotdb.common.rpc.thrift.TConsensusGroupType;
import org.apache.iotdb.common.rpc.thrift.TEndPoint;
import org.apache.iotdb.common.rpc.thrift.TSStatus;
import org.apache.iotdb.commons.client.IClientManager;
-import org.apache.iotdb.commons.client.container.IoTV2GlobalComponentContainer;
import org.apache.iotdb.commons.client.sync.SyncPipeConsensusServiceClient;
import org.apache.iotdb.commons.consensus.index.ProgressIndex;
+import
org.apache.iotdb.commons.consensus.iotv2.container.IoTV2GlobalComponentContainer;
import
org.apache.iotdb.commons.exception.pipe.PipeRuntimeSinkRetryTimesConfigurableException;
import org.apache.iotdb.commons.pipe.config.PipeConfig;
import
org.apache.iotdb.commons.pipe.sink.payload.pipeconsensus.response.PipeConsensusTransferFilePieceResp;
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java
index 10b5d969ed3..2d9024112b3 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java
@@ -4294,7 +4294,7 @@ public class DataRegion implements IDataRegionForQuery {
this.directBufferMemoryCost = acquireDirectBufferMemCost;
}
- private static long getAcquireDirectBufferMemCost() {
+ public static long getAcquireDirectBufferMemCost() {
long acquireDirectBufferMemCost = 0;
if
(config.getDataRegionConsensusProtocolClass().equals(ConsensusFactory.IOT_CONSENSUS)
||
config.getDataRegionConsensusProtocolClass().equals(ConsensusFactory.IOT_CONSENSUS_V2))
{
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/client/container/IoTV2GlobalComponentContainer.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/consensus/iotv2/container/IoTV2GlobalComponentContainer.java
similarity index 86%
rename from
iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/client/container/IoTV2GlobalComponentContainer.java
rename to
iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/consensus/iotv2/container/IoTV2GlobalComponentContainer.java
index f6f66b9c9ef..94eb313a010 100644
---
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/client/container/IoTV2GlobalComponentContainer.java
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/consensus/iotv2/container/IoTV2GlobalComponentContainer.java
@@ -17,7 +17,7 @@
* under the License.
*/
-package org.apache.iotdb.commons.client.container;
+package org.apache.iotdb.commons.consensus.iotv2.container;
import org.apache.iotdb.common.rpc.thrift.TEndPoint;
import
org.apache.iotdb.commons.client.ClientPoolFactory.AsyncPipeConsensusServiceClientPoolFactory;
@@ -30,6 +30,7 @@ import
org.apache.iotdb.commons.concurrent.IoTDBThreadPoolFactory;
import org.apache.iotdb.commons.concurrent.ThreadName;
import org.apache.iotdb.commons.conf.CommonConfig;
import org.apache.iotdb.commons.conf.CommonDescriptor;
+import org.apache.iotdb.commons.pipe.agent.task.execution.PipeSubtaskExecutor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -52,6 +53,7 @@ public class IoTV2GlobalComponentContainer {
private final IClientManager<TEndPoint, AsyncPipeConsensusServiceClient>
asyncClientManager;
private final IClientManager<TEndPoint, SyncPipeConsensusServiceClient>
syncClientManager;
private final ScheduledExecutorService backgroundTaskService;
+ private PipeSubtaskExecutor consensusExecutor;
private IoTV2GlobalComponentContainer() {
// load rpc client config
@@ -96,10 +98,18 @@ public class IoTV2GlobalComponentContainer {
}
}
- private static class PipeConsensusClientMgrContainerHolder {
+ public PipeSubtaskExecutor getConsensusExecutor() {
+ return consensusExecutor;
+ }
+
+ public void setConsensusExecutor(PipeSubtaskExecutor consensusExecutor) {
+ this.consensusExecutor = consensusExecutor;
+ }
+
+ private static class IoTV2GlobalComponentContainerHolder {
private static IoTV2GlobalComponentContainer INSTANCE;
- private PipeConsensusClientMgrContainerHolder() {}
+ private IoTV2GlobalComponentContainerHolder() {}
public static void build() {
if (INSTANCE == null) {
@@ -109,12 +119,15 @@ public class IoTV2GlobalComponentContainer {
}
public static IoTV2GlobalComponentContainer getInstance() {
- return PipeConsensusClientMgrContainerHolder.INSTANCE;
+ if (IoTV2GlobalComponentContainerHolder.INSTANCE == null) {
+ IoTV2GlobalComponentContainer.build();
+ }
+ return IoTV2GlobalComponentContainerHolder.INSTANCE;
}
// Only when consensus protocol is PipeConsensus, this method will be called
once when construct
// consensus class.
public static void build() {
- PipeConsensusClientMgrContainerHolder.build();
+ IoTV2GlobalComponentContainerHolder.build();
}
}
