From da47424102072bc6b9d27a8ccac14b91a622700c Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Tue, 21 Jul 2026 11:10:15 +0800 Subject: [PATCH 1/4] [Pipe] Fairly limit concurrent TsFile parsers --- .../tsfile/PipeTsFileInsertionEvent.java | 28 ++- .../resource/memory/PipeMemoryManager.java | 156 ++++++++++++++-- .../memory/PipeMemoryManagerTest.java | 173 ++++++++++++++++++ .../iotdb/commons/i18n/PipeMessages.java | 4 + .../iotdb/commons/i18n/PipeMessages.java | 4 + .../iotdb/commons/conf/CommonConfig.java | 35 ++++ .../iotdb/commons/pipe/config/PipeConfig.java | 14 ++ .../commons/pipe/config/PipeDescriptor.java | 10 + 8 files changed, 406 insertions(+), 18 deletions(-) create mode 100644 iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryManagerTest.java diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeTsFileInsertionEvent.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeTsFileInsertionEvent.java index 757411df9b976..73c0d3a920f6b 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeTsFileInsertionEvent.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeTsFileInsertionEvent.java @@ -98,6 +98,7 @@ public class PipeTsFileInsertionEvent extends PipeInsertionEvent private final AtomicBoolean isClosed; private final AtomicReference eventParser; private final AtomicBoolean isTsFileParserMemoryReserved = new AtomicBoolean(false); + private final Object tsFileParserMemoryReservationKey = new Object(); // The point count of the TsFile. Used for metrics on IoTConsensusV2' receiver side. // May be updated after it is flushed. Should be negative if not set. @@ -924,7 +925,8 @@ private boolean tryReserveTsFileParserMemory(final PipeMemoryManager memoryManag return true; } - if (!memoryManager.tryReserveTsFileParserMemory()) { + if (!memoryManager.tryReserveTsFileParserMemory( + pipeName, creationTime, tsFileParserMemoryReservationKey)) { return false; } @@ -936,11 +938,19 @@ private boolean tryReserveTsFileParserMemory(final PipeMemoryManager memoryManag private void releaseTsFileParserMemoryIfReserved() { synchronized (isTsFileParserMemoryReserved) { if (isTsFileParserMemoryReserved.compareAndSet(true, false)) { - PipeDataNodeResourceManager.memory().releaseTsFileParserMemory(); + PipeDataNodeResourceManager.memory().releaseTsFileParserMemory(pipeName, creationTime); } } } + private void cancelTsFileParserMemoryReservationIfPending() { + if (!isTsFileParserMemoryReserved.get()) { + PipeDataNodeResourceManager.memory() + .cancelTsFileParserMemoryReservation( + pipeName, creationTime, tsFileParserMemoryReservationKey); + } + } + /** The method is used to prevent circular replication in IoTConsensusV2 */ public boolean isGeneratedByIoTConsensusV2() { return isGeneratedByIoTConsensusV2; @@ -1009,6 +1019,7 @@ public long count(final boolean skipReportOnCommit) throws Exception { /** Release the resource of {@link TsFileInsertionEventParser}. */ @Override public void close() { + cancelTsFileParserMemoryReservationIfPending(); eventParser.getAndUpdate( parser -> { if (Objects.nonNull(parser)) { @@ -1058,7 +1069,8 @@ public PipeEventResource eventResourceBuilder() { this.modFile, this.sharedModFile, this.eventParser, - this.isTsFileParserMemoryReserved); + this.isTsFileParserMemoryReserved, + this.tsFileParserMemoryReservationKey); } private static class PipeTsFileInsertionEventResource extends PipeEventResource { @@ -1071,6 +1083,7 @@ private static class PipeTsFileInsertionEventResource extends PipeEventResource private final String pipeName; private final long creationTime; private final AtomicBoolean isTsFileParserMemoryReserved; + private final Object tsFileParserMemoryReservationKey; private PipeTsFileInsertionEventResource( final AtomicBoolean isReleased, @@ -1082,7 +1095,8 @@ private PipeTsFileInsertionEventResource( final File modFile, final File sharedModFile, final AtomicReference eventParser, - final AtomicBoolean isTsFileParserMemoryReserved) { + final AtomicBoolean isTsFileParserMemoryReserved, + final Object tsFileParserMemoryReservationKey) { super(isReleased, referenceCount); this.pipeName = pipeName; this.creationTime = creationTime; @@ -1092,11 +1106,15 @@ private PipeTsFileInsertionEventResource( this.sharedModFile = sharedModFile; this.eventParser = eventParser; this.isTsFileParserMemoryReserved = isTsFileParserMemoryReserved; + this.tsFileParserMemoryReservationKey = tsFileParserMemoryReservationKey; } @Override protected void finalizeResource() { try { + PipeDataNodeResourceManager.memory() + .cancelTsFileParserMemoryReservation( + pipeName, creationTime, tsFileParserMemoryReservationKey); final String pipeTsFileResourcePipeName = PipeTsFileResourceManager.getPipeTsFileResourcePipeName(pipeName, creationTime); // decrease reference count @@ -1117,7 +1135,7 @@ protected void finalizeResource() { }); synchronized (isTsFileParserMemoryReserved) { if (isTsFileParserMemoryReserved.compareAndSet(true, false)) { - PipeDataNodeResourceManager.memory().releaseTsFileParserMemory(); + PipeDataNodeResourceManager.memory().releaseTsFileParserMemory(pipeName, creationTime); } } } catch (final Exception e) { diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryManager.java index 6360efcdedd19..dc1ca9f05938a 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryManager.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryManager.java @@ -31,10 +31,15 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import java.util.ArrayDeque; import java.util.ArrayList; import java.util.Collections; +import java.util.HashMap; import java.util.HashSet; +import java.util.LinkedHashSet; import java.util.List; +import java.util.Map; +import java.util.Objects; import java.util.Set; import java.util.function.LongUnaryOperator; @@ -62,6 +67,11 @@ public class PipeMemoryManager { private volatile long reservedTsFileParserCount; + private final Map reservedTsFileParserCountByPipe = new HashMap<>(); + private final Map> waitingTsFileParserRequestsByPipe = + new HashMap<>(); + private final ArrayDeque waitingTsFileParserPipeOrder = new ArrayDeque<>(); + // Only non-zero memory blocks will be added to this set. private final Set allocatedBlocks = new HashSet<>(); private final Set shrinkableBlocks = new HashSet<>(); @@ -123,9 +133,12 @@ private boolean isEnough4TabletParsingWithReservedParserMemory(final long extraM < EXCEED_PROTECT_THRESHOLD * allowedMaxMemorySizeInBytesOfTablets(); } - private boolean isHardEnough4TabletParsingWithReservedParserMemory() { + private boolean isHardEnough4TabletParsingWithReservedParserMemory( + final long extraMemoryInBytes) { final double tabletMemoryWithParserMemory = - (double) usedMemorySizeInBytesOfTablets + getReservedTsFileParserMemorySizeInBytes(); + (double) usedMemorySizeInBytesOfTablets + + getReservedTsFileParserMemorySizeInBytes() + + extraMemoryInBytes; return tabletMemoryWithParserMemory + (double) usedMemorySizeInBytesOfTsFiles < allowedMaxMemorySizeInBytesOfTabletsAndTsFiles() && tabletMemoryWithParserMemory < allowedMaxMemorySizeInBytesOfTablets(); @@ -144,29 +157,118 @@ < allowedMaxMemorySizeInBytesOfTabletsAndTsFiles() && (double) usedMemorySizeInBytesOfTablets < allowedMaxMemorySizeInBytesOfTablets(); } - public synchronized boolean tryReserveTsFileParserMemory() { - if (!PIPE_MEMORY_MANAGEMENT_ENABLED) { - return true; + public synchronized boolean tryReserveTsFileParserMemory( + final String pipeName, final long creationTime, final Object reservationKey) { + if (reservationKey == null) { + return false; + } + + final PipeIdentity pipeIdentity = new PipeIdentity(pipeName, creationTime); + enqueueTsFileParserReservationRequest(pipeIdentity, reservationKey); + + final int globalLimit = Math.max(1, PIPE_CONFIG.getPipeTsFileParserInFlightMaxNum()); + final int perPipeLimit = + Math.max(1, Math.min(globalLimit, PIPE_CONFIG.getPipeTsFileParserInFlightMaxNumPerPipe())); + final int reservedCountOfPipe = reservedTsFileParserCountByPipe.getOrDefault(pipeIdentity, 0); + if (reservedTsFileParserCount >= globalLimit || reservedCountOfPipe >= perPipeLimit) { + return false; } final long parserMemorySizeInBytes = getTsFileParserMemorySizeInBytes(); - if (isEnough4TabletParsingWithReservedParserMemory(parserMemorySizeInBytes)) { - reservedTsFileParserCount++; - return true; + final boolean isSoftMemoryEnough = + !PIPE_MEMORY_MANAGEMENT_ENABLED + || isEnough4TabletParsingWithReservedParserMemory(parserMemorySizeInBytes); + if (!isSoftMemoryEnough + && !isHardEnough4TabletParsingWithReservedParserMemory(parserMemorySizeInBytes)) { + return false; } - return false; + final PipeIdentity nextPipe = + getNextEligibleTsFileParserPipe(perPipeLimit, !isSoftMemoryEnough); + final LinkedHashSet requestsOfPipe = + waitingTsFileParserRequestsByPipe.get(pipeIdentity); + if (!pipeIdentity.equals(nextPipe) + || requestsOfPipe == null + || !reservationKey.equals(requestsOfPipe.iterator().next())) { + return false; + } + + removeTsFileParserReservationRequest(pipeIdentity, reservationKey, true); + reservedTsFileParserCount++; + reservedTsFileParserCountByPipe.put(pipeIdentity, reservedCountOfPipe + 1); + return true; } - public synchronized void releaseTsFileParserMemory() { - if (!PIPE_MEMORY_MANAGEMENT_ENABLED) { + public synchronized void cancelTsFileParserMemoryReservation( + final String pipeName, final long creationTime, final Object reservationKey) { + if (reservationKey == null) { + return; + } + removeTsFileParserReservationRequest( + new PipeIdentity(pipeName, creationTime), reservationKey, false); + this.notifyAll(); + } + + public synchronized void releaseTsFileParserMemory( + final String pipeName, final long creationTime) { + final PipeIdentity pipeIdentity = new PipeIdentity(pipeName, creationTime); + final int reservedCountOfPipe = reservedTsFileParserCountByPipe.getOrDefault(pipeIdentity, 0); + if (reservedCountOfPipe <= 0) { return; } - reservedTsFileParserCount = Math.max(0, reservedTsFileParserCount - 1); + if (reservedCountOfPipe == 1) { + reservedTsFileParserCountByPipe.remove(pipeIdentity); + } else { + reservedTsFileParserCountByPipe.put(pipeIdentity, reservedCountOfPipe - 1); + } + reservedTsFileParserCount--; this.notifyAll(); } + private void enqueueTsFileParserReservationRequest( + final PipeIdentity pipeIdentity, final Object reservationKey) { + final LinkedHashSet requestsOfPipe = + waitingTsFileParserRequestsByPipe.computeIfAbsent( + pipeIdentity, + key -> { + waitingTsFileParserPipeOrder.addLast(key); + return new LinkedHashSet<>(); + }); + requestsOfPipe.add(reservationKey); + } + + private PipeIdentity getNextEligibleTsFileParserPipe( + final int perPipeLimit, final boolean requirePipeWithoutReservedParser) { + for (final PipeIdentity pipeIdentity : waitingTsFileParserPipeOrder) { + final int reservedCount = reservedTsFileParserCountByPipe.getOrDefault(pipeIdentity, 0); + if (reservedCount < perPipeLimit + // Under soft memory pressure, reserve the hard-threshold headroom for a pipe that has no + // parser yet. Otherwise a busy pipe at the queue head can block every pipe behind it. + && (!requirePipeWithoutReservedParser || reservedCount == 0)) { + return pipeIdentity; + } + } + return null; + } + + private void removeTsFileParserReservationRequest( + final PipeIdentity pipeIdentity, final Object reservationKey, final boolean rotatePipe) { + final LinkedHashSet requestsOfPipe = + waitingTsFileParserRequestsByPipe.get(pipeIdentity); + if (requestsOfPipe == null || !requestsOfPipe.remove(reservationKey)) { + return; + } + + if (requestsOfPipe.isEmpty()) { + waitingTsFileParserPipeOrder.remove(pipeIdentity); + waitingTsFileParserRequestsByPipe.remove(pipeIdentity); + } else if (rotatePipe) { + waitingTsFileParserPipeOrder.remove(pipeIdentity); + waitingTsFileParserPipeOrder.addLast(pipeIdentity); + } + } + public boolean shouldReleaseTsFileParserOnOutOfMemory( final long firstOutOfMemoryTimeInMs, final int retryCount) { final long retryIntervalInMs = PIPE_CONFIG.getPipeMemoryAllocateRetryIntervalInMs(); @@ -184,7 +286,7 @@ public boolean shouldReleaseTsFileParserOnOutOfMemory( return elapsedTimeInMs >= maxRetryTimeInMs; } - if (!isHardEnough4TabletParsingWithReservedParserMemory()) { + if (!isHardEnough4TabletParsingWithReservedParserMemory(0)) { return true; } @@ -761,4 +863,32 @@ public long getTotalFloatingMemorySizeInBytes() { public long getTotalMemorySizeInBytes() { return memoryBlock.getTotalMemorySizeInBytes(); } + + private static class PipeIdentity { + + private final String pipeName; + private final long creationTime; + + private PipeIdentity(final String pipeName, final long creationTime) { + this.pipeName = pipeName; + this.creationTime = creationTime; + } + + @Override + public boolean equals(final Object object) { + if (this == object) { + return true; + } + if (!(object instanceof PipeIdentity)) { + return false; + } + final PipeIdentity that = (PipeIdentity) object; + return creationTime == that.creationTime && Objects.equals(pipeName, that.pipeName); + } + + @Override + public int hashCode() { + return Objects.hash(pipeName, creationTime); + } + } } diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryManagerTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryManagerTest.java new file mode 100644 index 0000000000000..d9d9ffb38b52f --- /dev/null +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryManagerTest.java @@ -0,0 +1,173 @@ +/* + * 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.iotdb.db.pipe.resource.memory; + +import org.apache.iotdb.commons.conf.CommonConfig; +import org.apache.iotdb.commons.conf.CommonDescriptor; +import org.apache.iotdb.db.pipe.resource.PipeDataNodeResourceManager; + +import org.junit.After; +import org.junit.Assert; +import org.junit.Before; +import org.junit.Test; + +import java.util.ArrayList; +import java.util.List; + +public class PipeMemoryManagerTest { + + private final PipeMemoryManager memoryManager = PipeDataNodeResourceManager.memory(); + private final CommonConfig commonConfig = CommonDescriptor.getInstance().getConfig(); + private final List reservations = new ArrayList<>(); + private int originalGlobalLimit; + private int originalPerPipeLimit; + private long originalParserMemoryInBytes; + + @Before + public void setUp() { + originalGlobalLimit = commonConfig.getPipeTsFileParserInFlightMaxNum(); + originalPerPipeLimit = commonConfig.getPipeTsFileParserInFlightMaxNumPerPipe(); + originalParserMemoryInBytes = commonConfig.getPipeTsFileParserMemory(); + commonConfig.setPipeTsFileParserMemory(1); + } + + @After + public void tearDown() { + for (final Reservation reservation : reservations) { + memoryManager.cancelTsFileParserMemoryReservation( + reservation.pipeName, reservation.creationTime, reservation.key); + if (reservation.acquired) { + memoryManager.releaseTsFileParserMemory(reservation.pipeName, reservation.creationTime); + } + } + commonConfig.setPipeTsFileParserInFlightMaxNum(originalGlobalLimit); + commonConfig.setPipeTsFileParserInFlightMaxNumPerPipe(originalPerPipeLimit); + commonConfig.setPipeTsFileParserMemory(originalParserMemoryInBytes); + } + + @Test + public void testWaitingPipesAreAdmittedInRoundRobinOrder() { + commonConfig.setPipeTsFileParserInFlightMaxNum(1); + commonConfig.setPipeTsFileParserInFlightMaxNumPerPipe(1); + + final Reservation pipeAActive = new Reservation("pipeA", 1); + final Reservation pipeAFirstWaiting = new Reservation("pipeA", 1); + final Reservation pipeASecondWaiting = new Reservation("pipeA", 1); + final Reservation pipeBWaiting = new Reservation("pipeB", 2); + + Assert.assertTrue(tryAcquire(pipeAActive)); + Assert.assertFalse(tryAcquire(pipeAFirstWaiting)); + Assert.assertFalse(tryAcquire(pipeBWaiting)); + Assert.assertFalse(tryAcquire(pipeASecondWaiting)); + + release(pipeAActive); + Assert.assertTrue(tryAcquire(pipeAFirstWaiting)); + release(pipeAFirstWaiting); + + // Pipe A still has another waiting TsFile, but it was rotated behind pipe B after admission. + Assert.assertFalse(tryAcquire(pipeASecondWaiting)); + Assert.assertTrue(tryAcquire(pipeBWaiting)); + release(pipeBWaiting); + + Assert.assertTrue(tryAcquire(pipeASecondWaiting)); + } + + @Test + public void testGlobalAndPerPipeLimitsAreBothEnforced() { + commonConfig.setPipeTsFileParserInFlightMaxNum(2); + commonConfig.setPipeTsFileParserInFlightMaxNumPerPipe(1); + + final Reservation pipeAFirst = new Reservation("pipeA", 1); + final Reservation pipeASecond = new Reservation("pipeA", 1); + final Reservation pipeB = new Reservation("pipeB", 2); + final Reservation pipeC = new Reservation("pipeC", 3); + + Assert.assertTrue(tryAcquire(pipeAFirst)); + Assert.assertFalse(tryAcquire(pipeASecond)); + Assert.assertTrue(tryAcquire(pipeB)); + Assert.assertFalse(tryAcquire(pipeC)); + + release(pipeAFirst); + Assert.assertTrue(tryAcquire(pipeASecond)); + Assert.assertFalse(tryAcquire(pipeC)); + + release(pipeB); + Assert.assertTrue(tryAcquire(pipeC)); + } + + @Test + public void testSoftMemoryHeadroomIsReservedForPipeWithoutParser() { + commonConfig.setPipeTsFileParserInFlightMaxNum(2); + commonConfig.setPipeTsFileParserInFlightMaxNumPerPipe(2); + + final double tabletMemoryLimit = + (commonConfig.getPipeDataStructureTabletMemoryBlockAllocationRejectThreshold() + + commonConfig.getPipeDataStructureTsFileMemoryBlockAllocationRejectThreshold() / 2) + * memoryManager.getTotalNonFloatingMemorySizeInBytes(); + final double tabletAndTsFileMemoryLimit = + (commonConfig.getPipeDataStructureTabletMemoryBlockAllocationRejectThreshold() + + commonConfig.getPipeDataStructureTsFileMemoryBlockAllocationRejectThreshold()) + * memoryManager.getTotalNonFloatingMemorySizeInBytes(); + commonConfig.setPipeTsFileParserMemory( + Math.max(1, (long) (Math.min(tabletMemoryLimit, tabletAndTsFileMemoryLimit) * 0.49))); + + final Reservation pipeAActive = new Reservation("pipeA", 1); + final Reservation pipeAWaiting = new Reservation("pipeA", 1); + final Reservation pipeBWaiting = new Reservation("pipeB", 2); + + Assert.assertTrue(tryAcquire(pipeAActive)); + Assert.assertFalse(tryAcquire(pipeAWaiting)); + + // The second parser would fit only below the hard threshold. Pipe A already has a parser, so + // the headroom must go to pipe B even though pipe A is ahead in the waiting queue. + Assert.assertTrue(tryAcquire(pipeBWaiting)); + } + + private boolean tryAcquire(final Reservation reservation) { + if (!reservations.contains(reservation)) { + reservations.add(reservation); + } + reservation.acquired = + memoryManager.tryReserveTsFileParserMemory( + reservation.pipeName, reservation.creationTime, reservation.key); + return reservation.acquired; + } + + private void release(final Reservation reservation) { + if (!reservation.acquired) { + return; + } + memoryManager.releaseTsFileParserMemory(reservation.pipeName, reservation.creationTime); + reservation.acquired = false; + } + + private static class Reservation { + + private final String pipeName; + private final long creationTime; + private final Object key = new Object(); + private boolean acquired; + + private Reservation(final String pipeName, final long creationTime) { + this.pipeName = pipeName; + this.creationTime = creationTime; + } + } +} diff --git a/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/PipeMessages.java b/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/PipeMessages.java index e69dc15c1ddbd..92fc2975c011f 100644 --- a/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/PipeMessages.java +++ b/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/PipeMessages.java @@ -48,6 +48,10 @@ public final class PipeMessages { public static final String CONFIG_IS_PIPE_ENABLE_MEMORY_CHECK = "IsPipeEnableMemoryCheck: {}"; public static final String CONFIG_PIPE_TSFILE_PARSER_MEMORY = "PipeTsFileParserMemory: {}"; + public static final String CONFIG_PIPE_TSFILE_PARSER_IN_FLIGHT_MAX_NUM = + "PipeTsFileParserInFlightMaxNum: {}"; + public static final String CONFIG_PIPE_TSFILE_PARSER_IN_FLIGHT_MAX_NUM_PER_PIPE = + "PipeTsFileParserInFlightMaxNumPerPipe: {}"; public static final String CONFIG_SINK_BATCH_MEMORY_INSERT_NODE = "SinkBatchMemoryInsertNode: {}"; public static final String CONFIG_SINK_BATCH_MEMORY_TSFILE = "SinkBatchMemoryTsFile: {}"; diff --git a/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/PipeMessages.java b/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/PipeMessages.java index ca3243cdc01d7..7d0eb8752187e 100644 --- a/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/PipeMessages.java +++ b/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/PipeMessages.java @@ -47,6 +47,10 @@ public final class PipeMessages { public static final String CONFIG_IS_PIPE_ENABLE_MEMORY_CHECK = "IsPipeEnableMemoryCheck: {}"; public static final String CONFIG_PIPE_TSFILE_PARSER_MEMORY = "PipeTsFileParserMemory: {}"; + public static final String CONFIG_PIPE_TSFILE_PARSER_IN_FLIGHT_MAX_NUM = + "PipeTsFileParserInFlightMaxNum: {}"; + public static final String CONFIG_PIPE_TSFILE_PARSER_IN_FLIGHT_MAX_NUM_PER_PIPE = + "PipeTsFileParserInFlightMaxNumPerPipe: {}"; public static final String CONFIG_SINK_BATCH_MEMORY_INSERT_NODE = "SinkBatchMemoryInsertNode: {}"; public static final String CONFIG_SINK_BATCH_MEMORY_TSFILE = "SinkBatchMemoryTsFile: {}"; diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonConfig.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonConfig.java index 91393bf9fd2b4..6fb2714c1ea85 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonConfig.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonConfig.java @@ -248,6 +248,13 @@ public class CommonConfig { // Note: Pipes that do not decompose pattern/time do not need this part of memory private long pipeTsFileParserMemory = 17 * MB; + // Limit concurrently active TsFile parsers globally and for each pipe. The per-pipe limit also + // serves as an approximate parser memory quota because every admitted parser reserves + // pipeTsFileParserMemory bytes. + private int pipeTsFileParserInFlightMaxNum = + Math.max(1, Runtime.getRuntime().availableProcessors() / 2); + private int pipeTsFileParserInFlightMaxNumPerPipe = 1; + // Memory for Sink batch sending (InsertNode/TsFile, choose one) // 1. InsertNode: 15MB, used for batch sending data to the downstream system private long pipeSinkBatchMemoryInsertNode = 15 * MB; @@ -1039,6 +1046,34 @@ public void setPipeTsFileParserMemory(long pipeTsFileParserMemory) { logger.info(ConfigMessages.CONFIG_SET_TO, "pipeTsFileParserMemory", pipeTsFileParserMemory); } + public int getPipeTsFileParserInFlightMaxNum() { + return pipeTsFileParserInFlightMaxNum; + } + + public void setPipeTsFileParserInFlightMaxNum(final int pipeTsFileParserInFlightMaxNum) { + final int validatedValue = Math.max(1, pipeTsFileParserInFlightMaxNum); + if (this.pipeTsFileParserInFlightMaxNum == validatedValue) { + return; + } + this.pipeTsFileParserInFlightMaxNum = validatedValue; + logger.info(ConfigMessages.CONFIG_SET_TO, "pipeTsFileParserInFlightMaxNum", validatedValue); + } + + public int getPipeTsFileParserInFlightMaxNumPerPipe() { + return pipeTsFileParserInFlightMaxNumPerPipe; + } + + public void setPipeTsFileParserInFlightMaxNumPerPipe( + final int pipeTsFileParserInFlightMaxNumPerPipe) { + final int validatedValue = Math.max(1, pipeTsFileParserInFlightMaxNumPerPipe); + if (this.pipeTsFileParserInFlightMaxNumPerPipe == validatedValue) { + return; + } + this.pipeTsFileParserInFlightMaxNumPerPipe = validatedValue; + logger.info( + ConfigMessages.CONFIG_SET_TO, "pipeTsFileParserInFlightMaxNumPerPipe", validatedValue); + } + public long getPipeSinkBatchMemoryInsertNode() { return pipeSinkBatchMemoryInsertNode; } diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/PipeConfig.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/PipeConfig.java index 6e85519a85f68..c3ff65c0ab47a 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/PipeConfig.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/PipeConfig.java @@ -84,6 +84,14 @@ public long getTsFileParserMemory() { return COMMON_CONFIG.getPipeTsFileParserMemory(); } + public int getPipeTsFileParserInFlightMaxNum() { + return COMMON_CONFIG.getPipeTsFileParserInFlightMaxNum(); + } + + public int getPipeTsFileParserInFlightMaxNumPerPipe() { + return COMMON_CONFIG.getPipeTsFileParserInFlightMaxNumPerPipe(); + } + public long getSinkBatchMemoryInsertNode() { return COMMON_CONFIG.getPipeSinkBatchMemoryInsertNode(); } @@ -506,6 +514,12 @@ public void printAllConfigs() { LOGGER.info(PipeMessages.CONFIG_IS_PIPE_ENABLE_MEMORY_CHECK, isPipeEnableMemoryCheck()); LOGGER.info(PipeMessages.CONFIG_PIPE_TSFILE_PARSER_MEMORY, getTsFileParserMemory()); + LOGGER.info( + PipeMessages.CONFIG_PIPE_TSFILE_PARSER_IN_FLIGHT_MAX_NUM, + getPipeTsFileParserInFlightMaxNum()); + LOGGER.info( + PipeMessages.CONFIG_PIPE_TSFILE_PARSER_IN_FLIGHT_MAX_NUM_PER_PIPE, + getPipeTsFileParserInFlightMaxNumPerPipe()); LOGGER.info(PipeMessages.CONFIG_SINK_BATCH_MEMORY_INSERT_NODE, getSinkBatchMemoryInsertNode()); LOGGER.info(PipeMessages.CONFIG_SINK_BATCH_MEMORY_TSFILE, getSinkBatchMemoryTsFile()); LOGGER.info(PipeMessages.CONFIG_SEND_TSFILE_READ_BUFFER, getSendTsFileReadBuffer()); diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/PipeDescriptor.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/PipeDescriptor.java index 7909ef89e80d9..5586a51d3939e 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/PipeDescriptor.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/PipeDescriptor.java @@ -238,6 +238,16 @@ public static void loadPipeInternalConfig(CommonConfig config, TrimProperties pr Long.parseLong( properties.getProperty( "pipe_tsfile_parser_memory", String.valueOf(config.getPipeTsFileParserMemory())))); + config.setPipeTsFileParserInFlightMaxNum( + Integer.parseInt( + properties.getProperty( + "pipe_tsfile_parser_in_flight_max_num", + String.valueOf(config.getPipeTsFileParserInFlightMaxNum())))); + config.setPipeTsFileParserInFlightMaxNumPerPipe( + Integer.parseInt( + properties.getProperty( + "pipe_tsfile_parser_in_flight_max_num_per_pipe", + String.valueOf(config.getPipeTsFileParserInFlightMaxNumPerPipe())))); config.setPipeSinkBatchMemoryInsertNode( Long.parseLong( properties.getProperty( From 8ba32cd11d699cd33da92be536c26e16a99059cd Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Tue, 21 Jul 2026 12:05:00 +0800 Subject: [PATCH 2/4] [Pipe] Test concurrent TsFile parser fairness --- .../memory/PipeMemoryManagerTest.java | 86 ++++++++++++++++++- 1 file changed, 85 insertions(+), 1 deletion(-) diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryManagerTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryManagerTest.java index d9d9ffb38b52f..5c08407b24cb0 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryManagerTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryManagerTest.java @@ -29,7 +29,14 @@ import org.junit.Test; import java.util.ArrayList; +import java.util.Collections; +import java.util.HashSet; import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; public class PipeMemoryManagerTest { @@ -140,6 +147,76 @@ public void testSoftMemoryHeadroomIsReservedForPipeWithoutParser() { Assert.assertTrue(tryAcquire(pipeBWaiting)); } + @Test + public void testConcurrentTsFilesFromMultiplePipesAreNotStarved() throws Exception { + commonConfig.setPipeTsFileParserInFlightMaxNum(1); + commonConfig.setPipeTsFileParserInFlightMaxNumPerPipe(1); + + final Reservation blocker = new Reservation("blocker", 0); + Assert.assertTrue(tryAcquire(blocker)); + + // Each reservation represents a distinct TsFile event. Pipe A deliberately has more waiting + // TsFiles so the test can detect whether it monopolizes the single parser slot. + final List waitingTsFiles = new ArrayList<>(); + for (int i = 0; i < 5; i++) { + waitingTsFiles.add(new Reservation("pipeA", 1)); + } + for (int i = 0; i < 2; i++) { + waitingTsFiles.add(new Reservation("pipeB", 2)); + waitingTsFiles.add(new Reservation("pipeC", 3)); + } + reservations.addAll(waitingTsFiles); + + final List acquisitionOrder = Collections.synchronizedList(new ArrayList<>()); + final CountDownLatch ready = new CountDownLatch(waitingTsFiles.size()); + final CountDownLatch start = new CountDownLatch(1); + final CountDownLatch enqueued = new CountDownLatch(waitingTsFiles.size()); + final ExecutorService executor = Executors.newFixedThreadPool(waitingTsFiles.size()); + final List> futures = new ArrayList<>(); + + try { + for (final Reservation reservation : waitingTsFiles) { + futures.add( + executor.submit( + () -> { + ready.countDown(); + start.await(); + + boolean acquired = tryAcquireWithoutTracking(reservation); + enqueued.countDown(); + final long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(10); + while (!acquired && System.nanoTime() < deadline) { + Thread.sleep(1); + acquired = tryAcquireWithoutTracking(reservation); + } + if (!acquired) { + return false; + } + + acquisitionOrder.add(reservation.pipeName); + Thread.sleep(5); + release(reservation); + return true; + })); + } + + Assert.assertTrue(ready.await(5, TimeUnit.SECONDS)); + start.countDown(); + Assert.assertTrue(enqueued.await(5, TimeUnit.SECONDS)); + release(blocker); + + for (final Future future : futures) { + Assert.assertTrue(future.get(15, TimeUnit.SECONDS)); + } + Assert.assertEquals(waitingTsFiles.size(), acquisitionOrder.size()); + Assert.assertEquals(3, new HashSet<>(acquisitionOrder.subList(0, 3)).size()); + } finally { + release(blocker); + executor.shutdownNow(); + Assert.assertTrue(executor.awaitTermination(5, TimeUnit.SECONDS)); + } + } + private boolean tryAcquire(final Reservation reservation) { if (!reservations.contains(reservation)) { reservations.add(reservation); @@ -150,6 +227,13 @@ private boolean tryAcquire(final Reservation reservation) { return reservation.acquired; } + private boolean tryAcquireWithoutTracking(final Reservation reservation) { + reservation.acquired = + memoryManager.tryReserveTsFileParserMemory( + reservation.pipeName, reservation.creationTime, reservation.key); + return reservation.acquired; + } + private void release(final Reservation reservation) { if (!reservation.acquired) { return; @@ -163,7 +247,7 @@ private static class Reservation { private final String pipeName; private final long creationTime; private final Object key = new Object(); - private boolean acquired; + private volatile boolean acquired; private Reservation(final String pipeName, final long creationTime) { this.pipeName = pipeName; From f75b6c6da1aecd00221ac0a5f11feb301db136db Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Wed, 22 Jul 2026 11:02:03 +0800 Subject: [PATCH 3/4] [Pipe] Scope TsFile parser limits by region --- .../tsfile/PipeTsFileInsertionEvent.java | 18 +- .../resource/memory/PipeMemoryManager.java | 177 +++++++++++++----- .../memory/PipeMemoryManagerTest.java | 117 ++++++++++-- .../iotdb/commons/i18n/PipeMessages.java | 4 +- .../iotdb/commons/i18n/PipeMessages.java | 4 +- .../iotdb/commons/conf/CommonConfig.java | 26 +-- .../iotdb/commons/pipe/config/PipeConfig.java | 8 +- .../commons/pipe/config/PipeDescriptor.java | 6 +- 8 files changed, 273 insertions(+), 87 deletions(-) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeTsFileInsertionEvent.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeTsFileInsertionEvent.java index 73c0d3a920f6b..4f6589a173ddb 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeTsFileInsertionEvent.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeTsFileInsertionEvent.java @@ -80,6 +80,7 @@ public class PipeTsFileInsertionEvent extends PipeInsertionEvent private static final Logger LOGGER = LoggerFactory.getLogger(PipeTsFileInsertionEvent.class); private final TsFileResource resource; + private final String dataRegionId; private File tsFile; private long extractTime = 0; @@ -222,6 +223,7 @@ private PipeTsFileInsertionEvent( databaseNameFromDataRegion); this.resource = resource; + this.dataRegionId = resource.getDataRegionId(); // For events created at assigner or historical extractor, the tsFile is get from the resource // For events created for source, the tsFile is inherited from the assigner, because the @@ -926,7 +928,7 @@ private boolean tryReserveTsFileParserMemory(final PipeMemoryManager memoryManag } if (!memoryManager.tryReserveTsFileParserMemory( - pipeName, creationTime, tsFileParserMemoryReservationKey)) { + pipeName, creationTime, dataRegionId, tsFileParserMemoryReservationKey)) { return false; } @@ -938,7 +940,8 @@ private boolean tryReserveTsFileParserMemory(final PipeMemoryManager memoryManag private void releaseTsFileParserMemoryIfReserved() { synchronized (isTsFileParserMemoryReserved) { if (isTsFileParserMemoryReserved.compareAndSet(true, false)) { - PipeDataNodeResourceManager.memory().releaseTsFileParserMemory(pipeName, creationTime); + PipeDataNodeResourceManager.memory() + .releaseTsFileParserMemory(pipeName, creationTime, dataRegionId); } } } @@ -947,7 +950,7 @@ private void cancelTsFileParserMemoryReservationIfPending() { if (!isTsFileParserMemoryReserved.get()) { PipeDataNodeResourceManager.memory() .cancelTsFileParserMemoryReservation( - pipeName, creationTime, tsFileParserMemoryReservationKey); + pipeName, creationTime, dataRegionId, tsFileParserMemoryReservationKey); } } @@ -1064,6 +1067,7 @@ public PipeEventResource eventResourceBuilder() { this.referenceCount, this.pipeName, this.creationTime, + this.dataRegionId, this.tsFile, this.isWithMod, this.modFile, @@ -1082,6 +1086,7 @@ private static class PipeTsFileInsertionEventResource extends PipeEventResource private final AtomicReference eventParser; private final String pipeName; private final long creationTime; + private final String dataRegionId; private final AtomicBoolean isTsFileParserMemoryReserved; private final Object tsFileParserMemoryReservationKey; @@ -1090,6 +1095,7 @@ private PipeTsFileInsertionEventResource( final AtomicInteger referenceCount, final String pipeName, final long creationTime, + final String dataRegionId, final File tsFile, final boolean isWithMod, final File modFile, @@ -1100,6 +1106,7 @@ private PipeTsFileInsertionEventResource( super(isReleased, referenceCount); this.pipeName = pipeName; this.creationTime = creationTime; + this.dataRegionId = dataRegionId; this.tsFile = tsFile; this.isWithMod = isWithMod; this.modFile = modFile; @@ -1114,7 +1121,7 @@ protected void finalizeResource() { try { PipeDataNodeResourceManager.memory() .cancelTsFileParserMemoryReservation( - pipeName, creationTime, tsFileParserMemoryReservationKey); + pipeName, creationTime, dataRegionId, tsFileParserMemoryReservationKey); final String pipeTsFileResourcePipeName = PipeTsFileResourceManager.getPipeTsFileResourcePipeName(pipeName, creationTime); // decrease reference count @@ -1135,7 +1142,8 @@ protected void finalizeResource() { }); synchronized (isTsFileParserMemoryReserved) { if (isTsFileParserMemoryReserved.compareAndSet(true, false)) { - PipeDataNodeResourceManager.memory().releaseTsFileParserMemory(pipeName, creationTime); + PipeDataNodeResourceManager.memory() + .releaseTsFileParserMemory(pipeName, creationTime, dataRegionId); } } } catch (final Exception e) { diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryManager.java index dc1ca9f05938a..53e8013d00cf0 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryManager.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryManager.java @@ -68,8 +68,12 @@ public class PipeMemoryManager { private volatile long reservedTsFileParserCount; private final Map reservedTsFileParserCountByPipe = new HashMap<>(); - private final Map> waitingTsFileParserRequestsByPipe = + private final Map reservedTsFileParserCountByPipeRegion = new HashMap<>(); + private final Map> + waitingTsFileParserRequestsByPipeRegion = new HashMap<>(); + private final Map> + waitingTsFileParserRegionOrderByPipe = new HashMap<>(); private final ArrayDeque waitingTsFileParserPipeOrder = new ArrayDeque<>(); // Only non-zero memory blocks will be added to this set. @@ -158,19 +162,27 @@ < allowedMaxMemorySizeInBytesOfTabletsAndTsFiles() } public synchronized boolean tryReserveTsFileParserMemory( - final String pipeName, final long creationTime, final Object reservationKey) { + final String pipeName, + final long creationTime, + final String dataRegionId, + final Object reservationKey) { if (reservationKey == null) { return false; } final PipeIdentity pipeIdentity = new PipeIdentity(pipeName, creationTime); - enqueueTsFileParserReservationRequest(pipeIdentity, reservationKey); + final PipeRegionIdentity pipeRegionIdentity = + new PipeRegionIdentity(pipeIdentity, dataRegionId); + enqueueTsFileParserReservationRequest(pipeRegionIdentity, reservationKey); final int globalLimit = Math.max(1, PIPE_CONFIG.getPipeTsFileParserInFlightMaxNum()); - final int perPipeLimit = - Math.max(1, Math.min(globalLimit, PIPE_CONFIG.getPipeTsFileParserInFlightMaxNumPerPipe())); - final int reservedCountOfPipe = reservedTsFileParserCountByPipe.getOrDefault(pipeIdentity, 0); - if (reservedTsFileParserCount >= globalLimit || reservedCountOfPipe >= perPipeLimit) { + final int perPipeRegionLimit = + Math.max( + 1, Math.min(globalLimit, PIPE_CONFIG.getPipeTsFileParserInFlightMaxNumPerPipeRegion())); + final int reservedCountOfPipeRegion = + reservedTsFileParserCountByPipeRegion.getOrDefault(pipeRegionIdentity, 0); + if (reservedTsFileParserCount >= globalLimit + || reservedCountOfPipeRegion >= perPipeRegionLimit) { return false; } @@ -183,40 +195,55 @@ public synchronized boolean tryReserveTsFileParserMemory( return false; } - final PipeIdentity nextPipe = - getNextEligibleTsFileParserPipe(perPipeLimit, !isSoftMemoryEnough); - final LinkedHashSet requestsOfPipe = - waitingTsFileParserRequestsByPipe.get(pipeIdentity); - if (!pipeIdentity.equals(nextPipe) - || requestsOfPipe == null - || !reservationKey.equals(requestsOfPipe.iterator().next())) { + final PipeRegionIdentity nextPipeRegion = + getNextEligibleTsFileParserPipeRegion(perPipeRegionLimit, !isSoftMemoryEnough); + final LinkedHashSet requestsOfPipeRegion = + waitingTsFileParserRequestsByPipeRegion.get(pipeRegionIdentity); + if (!pipeRegionIdentity.equals(nextPipeRegion) + || requestsOfPipeRegion == null + || !reservationKey.equals(requestsOfPipeRegion.iterator().next())) { return false; } - removeTsFileParserReservationRequest(pipeIdentity, reservationKey, true); + removeTsFileParserReservationRequest(pipeRegionIdentity, reservationKey, true); reservedTsFileParserCount++; - reservedTsFileParserCountByPipe.put(pipeIdentity, reservedCountOfPipe + 1); + reservedTsFileParserCountByPipe.merge(pipeIdentity, 1, Integer::sum); + reservedTsFileParserCountByPipeRegion.put(pipeRegionIdentity, reservedCountOfPipeRegion + 1); return true; } public synchronized void cancelTsFileParserMemoryReservation( - final String pipeName, final long creationTime, final Object reservationKey) { + final String pipeName, + final long creationTime, + final String dataRegionId, + final Object reservationKey) { if (reservationKey == null) { return; } removeTsFileParserReservationRequest( - new PipeIdentity(pipeName, creationTime), reservationKey, false); + new PipeRegionIdentity(new PipeIdentity(pipeName, creationTime), dataRegionId), + reservationKey, + false); this.notifyAll(); } public synchronized void releaseTsFileParserMemory( - final String pipeName, final long creationTime) { + final String pipeName, final long creationTime, final String dataRegionId) { final PipeIdentity pipeIdentity = new PipeIdentity(pipeName, creationTime); - final int reservedCountOfPipe = reservedTsFileParserCountByPipe.getOrDefault(pipeIdentity, 0); - if (reservedCountOfPipe <= 0) { + final PipeRegionIdentity pipeRegionIdentity = + new PipeRegionIdentity(pipeIdentity, dataRegionId); + final int reservedCountOfPipeRegion = + reservedTsFileParserCountByPipeRegion.getOrDefault(pipeRegionIdentity, 0); + if (reservedCountOfPipeRegion <= 0) { return; } + if (reservedCountOfPipeRegion == 1) { + reservedTsFileParserCountByPipeRegion.remove(pipeRegionIdentity); + } else { + reservedTsFileParserCountByPipeRegion.put(pipeRegionIdentity, reservedCountOfPipeRegion - 1); + } + final int reservedCountOfPipe = reservedTsFileParserCountByPipe.getOrDefault(pipeIdentity, 0); if (reservedCountOfPipe == 1) { reservedTsFileParserCountByPipe.remove(pipeIdentity); } else { @@ -227,43 +254,78 @@ public synchronized void releaseTsFileParserMemory( } private void enqueueTsFileParserReservationRequest( - final PipeIdentity pipeIdentity, final Object reservationKey) { - final LinkedHashSet requestsOfPipe = - waitingTsFileParserRequestsByPipe.computeIfAbsent( - pipeIdentity, + final PipeRegionIdentity pipeRegionIdentity, final Object reservationKey) { + final LinkedHashSet requestsOfPipeRegion = + waitingTsFileParserRequestsByPipeRegion.computeIfAbsent( + pipeRegionIdentity, key -> { - waitingTsFileParserPipeOrder.addLast(key); + final ArrayDeque regionOrder = + waitingTsFileParserRegionOrderByPipe.computeIfAbsent( + key.pipeIdentity, + pipe -> { + waitingTsFileParserPipeOrder.addLast(pipe); + return new ArrayDeque<>(); + }); + regionOrder.addLast(key); return new LinkedHashSet<>(); }); - requestsOfPipe.add(reservationKey); + requestsOfPipeRegion.add(reservationKey); } - private PipeIdentity getNextEligibleTsFileParserPipe( - final int perPipeLimit, final boolean requirePipeWithoutReservedParser) { + private PipeRegionIdentity getNextEligibleTsFileParserPipeRegion( + final int perPipeRegionLimit, final boolean requirePipeWithoutReservedParser) { for (final PipeIdentity pipeIdentity : waitingTsFileParserPipeOrder) { - final int reservedCount = reservedTsFileParserCountByPipe.getOrDefault(pipeIdentity, 0); - if (reservedCount < perPipeLimit - // Under soft memory pressure, reserve the hard-threshold headroom for a pipe that has no - // parser yet. Otherwise a busy pipe at the queue head can block every pipe behind it. - && (!requirePipeWithoutReservedParser || reservedCount == 0)) { - return pipeIdentity; + // Under soft memory pressure, reserve the hard-threshold headroom for a pipe that has no + // parser yet. Otherwise a busy pipe at the queue head can block every pipe behind it. + if (requirePipeWithoutReservedParser + && reservedTsFileParserCountByPipe.getOrDefault(pipeIdentity, 0) > 0) { + continue; + } + + final ArrayDeque regionOrder = + waitingTsFileParserRegionOrderByPipe.get(pipeIdentity); + if (regionOrder == null) { + continue; + } + for (final PipeRegionIdentity pipeRegionIdentity : regionOrder) { + if (reservedTsFileParserCountByPipeRegion.getOrDefault(pipeRegionIdentity, 0) + < perPipeRegionLimit) { + return pipeRegionIdentity; + } } } return null; } private void removeTsFileParserReservationRequest( - final PipeIdentity pipeIdentity, final Object reservationKey, final boolean rotatePipe) { - final LinkedHashSet requestsOfPipe = - waitingTsFileParserRequestsByPipe.get(pipeIdentity); - if (requestsOfPipe == null || !requestsOfPipe.remove(reservationKey)) { + final PipeRegionIdentity pipeRegionIdentity, + final Object reservationKey, + final boolean rotateAfterAdmission) { + final LinkedHashSet requestsOfPipeRegion = + waitingTsFileParserRequestsByPipeRegion.get(pipeRegionIdentity); + if (requestsOfPipeRegion == null || !requestsOfPipeRegion.remove(reservationKey)) { return; } - if (requestsOfPipe.isEmpty()) { - waitingTsFileParserPipeOrder.remove(pipeIdentity); - waitingTsFileParserRequestsByPipe.remove(pipeIdentity); - } else if (rotatePipe) { + final PipeIdentity pipeIdentity = pipeRegionIdentity.pipeIdentity; + final ArrayDeque regionOrder = + waitingTsFileParserRegionOrderByPipe.get(pipeIdentity); + if (requestsOfPipeRegion.isEmpty()) { + waitingTsFileParserRequestsByPipeRegion.remove(pipeRegionIdentity); + if (regionOrder != null) { + regionOrder.remove(pipeRegionIdentity); + if (regionOrder.isEmpty()) { + waitingTsFileParserRegionOrderByPipe.remove(pipeIdentity); + waitingTsFileParserPipeOrder.remove(pipeIdentity); + return; + } + } + } else if (rotateAfterAdmission && regionOrder != null) { + regionOrder.remove(pipeRegionIdentity); + regionOrder.addLast(pipeRegionIdentity); + } + + if (rotateAfterAdmission) { waitingTsFileParserPipeOrder.remove(pipeIdentity); waitingTsFileParserPipeOrder.addLast(pipeIdentity); } @@ -891,4 +953,33 @@ public int hashCode() { return Objects.hash(pipeName, creationTime); } } + + private static class PipeRegionIdentity { + + private final PipeIdentity pipeIdentity; + private final String dataRegionId; + + private PipeRegionIdentity(final PipeIdentity pipeIdentity, final String dataRegionId) { + this.pipeIdentity = pipeIdentity; + this.dataRegionId = dataRegionId; + } + + @Override + public boolean equals(final Object object) { + if (this == object) { + return true; + } + if (!(object instanceof PipeRegionIdentity)) { + return false; + } + final PipeRegionIdentity that = (PipeRegionIdentity) object; + return Objects.equals(pipeIdentity, that.pipeIdentity) + && Objects.equals(dataRegionId, that.dataRegionId); + } + + @Override + public int hashCode() { + return Objects.hash(pipeIdentity, dataRegionId); + } + } } diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryManagerTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryManagerTest.java index 5c08407b24cb0..6fc5e72968997 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryManagerTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryManagerTest.java @@ -44,13 +44,13 @@ public class PipeMemoryManagerTest { private final CommonConfig commonConfig = CommonDescriptor.getInstance().getConfig(); private final List reservations = new ArrayList<>(); private int originalGlobalLimit; - private int originalPerPipeLimit; + private int originalPerPipeRegionLimit; private long originalParserMemoryInBytes; @Before public void setUp() { originalGlobalLimit = commonConfig.getPipeTsFileParserInFlightMaxNum(); - originalPerPipeLimit = commonConfig.getPipeTsFileParserInFlightMaxNumPerPipe(); + originalPerPipeRegionLimit = commonConfig.getPipeTsFileParserInFlightMaxNumPerPipeRegion(); originalParserMemoryInBytes = commonConfig.getPipeTsFileParserMemory(); commonConfig.setPipeTsFileParserMemory(1); } @@ -59,20 +59,24 @@ public void setUp() { public void tearDown() { for (final Reservation reservation : reservations) { memoryManager.cancelTsFileParserMemoryReservation( - reservation.pipeName, reservation.creationTime, reservation.key); + reservation.pipeName, + reservation.creationTime, + reservation.dataRegionId, + reservation.key); if (reservation.acquired) { - memoryManager.releaseTsFileParserMemory(reservation.pipeName, reservation.creationTime); + memoryManager.releaseTsFileParserMemory( + reservation.pipeName, reservation.creationTime, reservation.dataRegionId); } } commonConfig.setPipeTsFileParserInFlightMaxNum(originalGlobalLimit); - commonConfig.setPipeTsFileParserInFlightMaxNumPerPipe(originalPerPipeLimit); + commonConfig.setPipeTsFileParserInFlightMaxNumPerPipeRegion(originalPerPipeRegionLimit); commonConfig.setPipeTsFileParserMemory(originalParserMemoryInBytes); } @Test public void testWaitingPipesAreAdmittedInRoundRobinOrder() { commonConfig.setPipeTsFileParserInFlightMaxNum(1); - commonConfig.setPipeTsFileParserInFlightMaxNumPerPipe(1); + commonConfig.setPipeTsFileParserInFlightMaxNumPerPipeRegion(1); final Reservation pipeAActive = new Reservation("pipeA", 1); final Reservation pipeAFirstWaiting = new Reservation("pipeA", 1); @@ -97,9 +101,9 @@ public void testWaitingPipesAreAdmittedInRoundRobinOrder() { } @Test - public void testGlobalAndPerPipeLimitsAreBothEnforced() { + public void testGlobalAndPerPipeRegionLimitsAreBothEnforced() { commonConfig.setPipeTsFileParserInFlightMaxNum(2); - commonConfig.setPipeTsFileParserInFlightMaxNumPerPipe(1); + commonConfig.setPipeTsFileParserInFlightMaxNumPerPipeRegion(1); final Reservation pipeAFirst = new Reservation("pipeA", 1); final Reservation pipeASecond = new Reservation("pipeA", 1); @@ -119,10 +123,78 @@ public void testGlobalAndPerPipeLimitsAreBothEnforced() { Assert.assertTrue(tryAcquire(pipeC)); } + @Test + public void testDifferentRegionsOfSamePipeCanRunConcurrently() { + commonConfig.setPipeTsFileParserInFlightMaxNum(2); + commonConfig.setPipeTsFileParserInFlightMaxNumPerPipeRegion(1); + + final Reservation pipeARegion1First = new Reservation("pipeA", 1, "1"); + final Reservation pipeARegion1Second = new Reservation("pipeA", 1, "1"); + final Reservation pipeARegion2 = new Reservation("pipeA", 1, "2"); + + Assert.assertTrue(tryAcquire(pipeARegion1First)); + Assert.assertFalse(tryAcquire(pipeARegion1Second)); + Assert.assertTrue(tryAcquire(pipeARegion2)); + } + + @Test + public void testWaitingRegionsWithinPipeAreAdmittedInRoundRobinOrder() { + commonConfig.setPipeTsFileParserInFlightMaxNum(1); + commonConfig.setPipeTsFileParserInFlightMaxNumPerPipeRegion(1); + + final Reservation blocker = new Reservation("blocker", 0); + final Reservation pipeARegion1First = new Reservation("pipeA", 1, "1"); + final Reservation pipeARegion1Second = new Reservation("pipeA", 1, "1"); + final Reservation pipeARegion2 = new Reservation("pipeA", 1, "2"); + + Assert.assertTrue(tryAcquire(blocker)); + Assert.assertFalse(tryAcquire(pipeARegion1First)); + Assert.assertFalse(tryAcquire(pipeARegion1Second)); + Assert.assertFalse(tryAcquire(pipeARegion2)); + + release(blocker); + Assert.assertTrue(tryAcquire(pipeARegion1First)); + release(pipeARegion1First); + + Assert.assertFalse(tryAcquire(pipeARegion1Second)); + Assert.assertTrue(tryAcquire(pipeARegion2)); + release(pipeARegion2); + + Assert.assertTrue(tryAcquire(pipeARegion1Second)); + } + + @Test + public void testPipeFairnessIsNotWeightedByRegionCount() { + commonConfig.setPipeTsFileParserInFlightMaxNum(1); + commonConfig.setPipeTsFileParserInFlightMaxNumPerPipeRegion(1); + + final Reservation blocker = new Reservation("blocker", 0); + final Reservation pipeARegion1 = new Reservation("pipeA", 1, "1"); + final Reservation pipeARegion2 = new Reservation("pipeA", 1, "2"); + final Reservation pipeARegion3 = new Reservation("pipeA", 1, "3"); + final Reservation pipeBRegion1 = new Reservation("pipeB", 2, "1"); + + Assert.assertTrue(tryAcquire(blocker)); + Assert.assertFalse(tryAcquire(pipeARegion1)); + Assert.assertFalse(tryAcquire(pipeARegion2)); + Assert.assertFalse(tryAcquire(pipeARegion3)); + Assert.assertFalse(tryAcquire(pipeBRegion1)); + + release(blocker); + Assert.assertTrue(tryAcquire(pipeARegion1)); + release(pipeARegion1); + + Assert.assertFalse(tryAcquire(pipeARegion2)); + Assert.assertTrue(tryAcquire(pipeBRegion1)); + release(pipeBRegion1); + + Assert.assertTrue(tryAcquire(pipeARegion2)); + } + @Test public void testSoftMemoryHeadroomIsReservedForPipeWithoutParser() { commonConfig.setPipeTsFileParserInFlightMaxNum(2); - commonConfig.setPipeTsFileParserInFlightMaxNumPerPipe(2); + commonConfig.setPipeTsFileParserInFlightMaxNumPerPipeRegion(2); final double tabletMemoryLimit = (commonConfig.getPipeDataStructureTabletMemoryBlockAllocationRejectThreshold() @@ -135,9 +207,9 @@ public void testSoftMemoryHeadroomIsReservedForPipeWithoutParser() { commonConfig.setPipeTsFileParserMemory( Math.max(1, (long) (Math.min(tabletMemoryLimit, tabletAndTsFileMemoryLimit) * 0.49))); - final Reservation pipeAActive = new Reservation("pipeA", 1); - final Reservation pipeAWaiting = new Reservation("pipeA", 1); - final Reservation pipeBWaiting = new Reservation("pipeB", 2); + final Reservation pipeAActive = new Reservation("pipeA", 1, "1"); + final Reservation pipeAWaiting = new Reservation("pipeA", 1, "2"); + final Reservation pipeBWaiting = new Reservation("pipeB", 2, "1"); Assert.assertTrue(tryAcquire(pipeAActive)); Assert.assertFalse(tryAcquire(pipeAWaiting)); @@ -150,7 +222,7 @@ public void testSoftMemoryHeadroomIsReservedForPipeWithoutParser() { @Test public void testConcurrentTsFilesFromMultiplePipesAreNotStarved() throws Exception { commonConfig.setPipeTsFileParserInFlightMaxNum(1); - commonConfig.setPipeTsFileParserInFlightMaxNumPerPipe(1); + commonConfig.setPipeTsFileParserInFlightMaxNumPerPipeRegion(1); final Reservation blocker = new Reservation("blocker", 0); Assert.assertTrue(tryAcquire(blocker)); @@ -223,14 +295,20 @@ private boolean tryAcquire(final Reservation reservation) { } reservation.acquired = memoryManager.tryReserveTsFileParserMemory( - reservation.pipeName, reservation.creationTime, reservation.key); + reservation.pipeName, + reservation.creationTime, + reservation.dataRegionId, + reservation.key); return reservation.acquired; } private boolean tryAcquireWithoutTracking(final Reservation reservation) { reservation.acquired = memoryManager.tryReserveTsFileParserMemory( - reservation.pipeName, reservation.creationTime, reservation.key); + reservation.pipeName, + reservation.creationTime, + reservation.dataRegionId, + reservation.key); return reservation.acquired; } @@ -238,7 +316,8 @@ private void release(final Reservation reservation) { if (!reservation.acquired) { return; } - memoryManager.releaseTsFileParserMemory(reservation.pipeName, reservation.creationTime); + memoryManager.releaseTsFileParserMemory( + reservation.pipeName, reservation.creationTime, reservation.dataRegionId); reservation.acquired = false; } @@ -246,12 +325,18 @@ private static class Reservation { private final String pipeName; private final long creationTime; + private final String dataRegionId; private final Object key = new Object(); private volatile boolean acquired; private Reservation(final String pipeName, final long creationTime) { + this(pipeName, creationTime, "0"); + } + + private Reservation(final String pipeName, final long creationTime, final String dataRegionId) { this.pipeName = pipeName; this.creationTime = creationTime; + this.dataRegionId = dataRegionId; } } } diff --git a/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/PipeMessages.java b/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/PipeMessages.java index 92fc2975c011f..742239bc1ca75 100644 --- a/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/PipeMessages.java +++ b/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/PipeMessages.java @@ -50,8 +50,8 @@ public final class PipeMessages { public static final String CONFIG_PIPE_TSFILE_PARSER_MEMORY = "PipeTsFileParserMemory: {}"; public static final String CONFIG_PIPE_TSFILE_PARSER_IN_FLIGHT_MAX_NUM = "PipeTsFileParserInFlightMaxNum: {}"; - public static final String CONFIG_PIPE_TSFILE_PARSER_IN_FLIGHT_MAX_NUM_PER_PIPE = - "PipeTsFileParserInFlightMaxNumPerPipe: {}"; + public static final String CONFIG_PIPE_TSFILE_PARSER_IN_FLIGHT_MAX_NUM_PER_PIPE_REGION = + "PipeTsFileParserInFlightMaxNumPerPipeRegion: {}"; public static final String CONFIG_SINK_BATCH_MEMORY_INSERT_NODE = "SinkBatchMemoryInsertNode: {}"; public static final String CONFIG_SINK_BATCH_MEMORY_TSFILE = "SinkBatchMemoryTsFile: {}"; diff --git a/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/PipeMessages.java b/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/PipeMessages.java index 7d0eb8752187e..0a8c0badb3544 100644 --- a/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/PipeMessages.java +++ b/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/PipeMessages.java @@ -49,8 +49,8 @@ public final class PipeMessages { public static final String CONFIG_PIPE_TSFILE_PARSER_MEMORY = "PipeTsFileParserMemory: {}"; public static final String CONFIG_PIPE_TSFILE_PARSER_IN_FLIGHT_MAX_NUM = "PipeTsFileParserInFlightMaxNum: {}"; - public static final String CONFIG_PIPE_TSFILE_PARSER_IN_FLIGHT_MAX_NUM_PER_PIPE = - "PipeTsFileParserInFlightMaxNumPerPipe: {}"; + public static final String CONFIG_PIPE_TSFILE_PARSER_IN_FLIGHT_MAX_NUM_PER_PIPE_REGION = + "PipeTsFileParserInFlightMaxNumPerPipeRegion: {}"; public static final String CONFIG_SINK_BATCH_MEMORY_INSERT_NODE = "SinkBatchMemoryInsertNode: {}"; public static final String CONFIG_SINK_BATCH_MEMORY_TSFILE = "SinkBatchMemoryTsFile: {}"; diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonConfig.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonConfig.java index 6fb2714c1ea85..c127540efbbd9 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonConfig.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonConfig.java @@ -248,12 +248,12 @@ public class CommonConfig { // Note: Pipes that do not decompose pattern/time do not need this part of memory private long pipeTsFileParserMemory = 17 * MB; - // Limit concurrently active TsFile parsers globally and for each pipe. The per-pipe limit also - // serves as an approximate parser memory quota because every admitted parser reserves - // pipeTsFileParserMemory bytes. + // Limit concurrently active TsFile parsers globally and for each region task of a pipe. The + // per-pipe-region limit also serves as an approximate parser memory quota because every admitted + // parser reserves pipeTsFileParserMemory bytes. private int pipeTsFileParserInFlightMaxNum = Math.max(1, Runtime.getRuntime().availableProcessors() / 2); - private int pipeTsFileParserInFlightMaxNumPerPipe = 1; + private int pipeTsFileParserInFlightMaxNumPerPipeRegion = 1; // Memory for Sink batch sending (InsertNode/TsFile, choose one) // 1. InsertNode: 15MB, used for batch sending data to the downstream system @@ -1059,19 +1059,21 @@ public void setPipeTsFileParserInFlightMaxNum(final int pipeTsFileParserInFlight logger.info(ConfigMessages.CONFIG_SET_TO, "pipeTsFileParserInFlightMaxNum", validatedValue); } - public int getPipeTsFileParserInFlightMaxNumPerPipe() { - return pipeTsFileParserInFlightMaxNumPerPipe; + public int getPipeTsFileParserInFlightMaxNumPerPipeRegion() { + return pipeTsFileParserInFlightMaxNumPerPipeRegion; } - public void setPipeTsFileParserInFlightMaxNumPerPipe( - final int pipeTsFileParserInFlightMaxNumPerPipe) { - final int validatedValue = Math.max(1, pipeTsFileParserInFlightMaxNumPerPipe); - if (this.pipeTsFileParserInFlightMaxNumPerPipe == validatedValue) { + public void setPipeTsFileParserInFlightMaxNumPerPipeRegion( + final int pipeTsFileParserInFlightMaxNumPerPipeRegion) { + final int validatedValue = Math.max(1, pipeTsFileParserInFlightMaxNumPerPipeRegion); + if (this.pipeTsFileParserInFlightMaxNumPerPipeRegion == validatedValue) { return; } - this.pipeTsFileParserInFlightMaxNumPerPipe = validatedValue; + this.pipeTsFileParserInFlightMaxNumPerPipeRegion = validatedValue; logger.info( - ConfigMessages.CONFIG_SET_TO, "pipeTsFileParserInFlightMaxNumPerPipe", validatedValue); + ConfigMessages.CONFIG_SET_TO, + "pipeTsFileParserInFlightMaxNumPerPipeRegion", + validatedValue); } public long getPipeSinkBatchMemoryInsertNode() { diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/PipeConfig.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/PipeConfig.java index c3ff65c0ab47a..561d2923c07e7 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/PipeConfig.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/PipeConfig.java @@ -88,8 +88,8 @@ public int getPipeTsFileParserInFlightMaxNum() { return COMMON_CONFIG.getPipeTsFileParserInFlightMaxNum(); } - public int getPipeTsFileParserInFlightMaxNumPerPipe() { - return COMMON_CONFIG.getPipeTsFileParserInFlightMaxNumPerPipe(); + public int getPipeTsFileParserInFlightMaxNumPerPipeRegion() { + return COMMON_CONFIG.getPipeTsFileParserInFlightMaxNumPerPipeRegion(); } public long getSinkBatchMemoryInsertNode() { @@ -518,8 +518,8 @@ public void printAllConfigs() { PipeMessages.CONFIG_PIPE_TSFILE_PARSER_IN_FLIGHT_MAX_NUM, getPipeTsFileParserInFlightMaxNum()); LOGGER.info( - PipeMessages.CONFIG_PIPE_TSFILE_PARSER_IN_FLIGHT_MAX_NUM_PER_PIPE, - getPipeTsFileParserInFlightMaxNumPerPipe()); + PipeMessages.CONFIG_PIPE_TSFILE_PARSER_IN_FLIGHT_MAX_NUM_PER_PIPE_REGION, + getPipeTsFileParserInFlightMaxNumPerPipeRegion()); LOGGER.info(PipeMessages.CONFIG_SINK_BATCH_MEMORY_INSERT_NODE, getSinkBatchMemoryInsertNode()); LOGGER.info(PipeMessages.CONFIG_SINK_BATCH_MEMORY_TSFILE, getSinkBatchMemoryTsFile()); LOGGER.info(PipeMessages.CONFIG_SEND_TSFILE_READ_BUFFER, getSendTsFileReadBuffer()); diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/PipeDescriptor.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/PipeDescriptor.java index 5586a51d3939e..3a1da5f18ba46 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/PipeDescriptor.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/PipeDescriptor.java @@ -243,11 +243,11 @@ public static void loadPipeInternalConfig(CommonConfig config, TrimProperties pr properties.getProperty( "pipe_tsfile_parser_in_flight_max_num", String.valueOf(config.getPipeTsFileParserInFlightMaxNum())))); - config.setPipeTsFileParserInFlightMaxNumPerPipe( + config.setPipeTsFileParserInFlightMaxNumPerPipeRegion( Integer.parseInt( properties.getProperty( - "pipe_tsfile_parser_in_flight_max_num_per_pipe", - String.valueOf(config.getPipeTsFileParserInFlightMaxNumPerPipe())))); + "pipe_tsfile_parser_in_flight_max_num_per_pipe_region", + String.valueOf(config.getPipeTsFileParserInFlightMaxNumPerPipeRegion())))); config.setPipeSinkBatchMemoryInsertNode( Long.parseLong( properties.getProperty( From 043f79892f72cf6a6efb82402d1d8fab3040f82d Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Wed, 22 Jul 2026 11:27:58 +0800 Subject: [PATCH 4/4] [Pipe] Address TsFile parser admission review --- .../iotdb/db/i18n/DataNodePipeMessages.java | 2 + .../iotdb/db/i18n/DataNodePipeMessages.java | 2 + .../apache/iotdb/db/conf/IoTDBDescriptor.java | 2 + .../tsfile/PipeTsFileInsertionEvent.java | 27 +++--- .../resource/memory/PipeMemoryManager.java | 85 ++++++++++++++++--- .../apache/iotdb/db/conf/PropertiesTest.java | 29 +++++++ .../memory/PipeMemoryManagerTest.java | 9 +- .../conf/iotdb-system.properties.template | 13 +++ .../iotdb/commons/conf/CommonConfig.java | 5 +- 9 files changed, 150 insertions(+), 24 deletions(-) diff --git a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java index 40b499410aaec..128c7af3c8c80 100644 --- a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java +++ b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java @@ -2574,4 +2574,6 @@ private DataNodePipeMessages() {} "Topic metadata for %s is unavailable during consensus subscription setup"; public static final String EXCEPTION_TOPIC_CONFIG_FOR_ARG_IS_UNAVAILABLE_DURING_CONSENSUS_SUBSCRIPTION_SETUP_B94404EE = "Topic config for %s is unavailable during consensus subscription setup"; + public static final String LOG_FAILED_TO_RELEASE_TSFILE_PARSER_MEMORY_FOR_PIPE_ARG_CREATION_TIME_ARG_IN_DATAREGION_ARG_BECAUSE_NO_RESERVATION_EXISTS_BB8321C0 = + "Failed to release TsFile parser memory for Pipe {} (creation time {}) in DataRegion {} because no reservation exists."; } diff --git a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java index 85a8ae289cf4f..b119dce0f709a 100644 --- a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java +++ b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java @@ -2403,4 +2403,6 @@ private DataNodePipeMessages() {} "共识订阅设置期间 topic %s 的元数据不可用"; public static final String EXCEPTION_TOPIC_CONFIG_FOR_ARG_IS_UNAVAILABLE_DURING_CONSENSUS_SUBSCRIPTION_SETUP_B94404EE = "共识订阅设置期间 topic %s 的配置不可用"; + public static final String LOG_FAILED_TO_RELEASE_TSFILE_PARSER_MEMORY_FOR_PIPE_ARG_CREATION_TIME_ARG_IN_DATAREGION_ARG_BECAUSE_NO_RESERVATION_EXISTS_BB8321C0 = + "无法释放 Pipe {}（创建时间 {}）在 DataRegion {} 中的 TsFile 解析器内存，因为不存在对应的预留。"; } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java index 1ab0c22bbf622..2874061e96539 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java @@ -41,6 +41,7 @@ import org.apache.iotdb.consensus.config.IoTConsensusV2Config; import org.apache.iotdb.db.consensus.DataRegionConsensusImpl; import org.apache.iotdb.db.i18n.DataNodeMiscMessages; +import org.apache.iotdb.db.pipe.resource.PipeDataNodeResourceManager; import org.apache.iotdb.db.queryengine.plan.relational.metadata.fetcher.cache.LastCacheLoadStrategy; import org.apache.iotdb.db.service.metrics.IoTDBInternalLocalReporter; import org.apache.iotdb.db.storageengine.StorageEngine; @@ -2761,6 +2762,7 @@ private void loadLoadTsFileHotModifiedProp(TrimProperties properties) throws IOE private void loadPipeHotModifiedProp(TrimProperties properties) throws IOException { PipeDescriptor.loadPipeProps(commonDescriptor.getConfig(), properties, true); + PipeDataNodeResourceManager.memory().notifyNextTsFileParserMemoryReservation(); LoggerPeriodicalLogReducer.update(); } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeTsFileInsertionEvent.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeTsFileInsertionEvent.java index 4f6589a173ddb..3ca527c30fb74 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeTsFileInsertionEvent.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeTsFileInsertionEvent.java @@ -47,6 +47,7 @@ import org.apache.iotdb.db.pipe.metric.overview.PipeDataNodeSinglePipeMetrics; import org.apache.iotdb.db.pipe.resource.PipeDataNodeResourceManager; import org.apache.iotdb.db.pipe.resource.memory.PipeMemoryManager; +import org.apache.iotdb.db.pipe.resource.memory.PipeMemoryManager.TsFileParserMemoryReservation; import org.apache.iotdb.db.pipe.resource.tsfile.PipeTsFileResourceManager; import org.apache.iotdb.db.pipe.source.dataregion.realtime.assigner.PipeTsFileEpochProgressIndexKeeper; import org.apache.iotdb.db.storageengine.dataregion.memtable.TsFileProcessor; @@ -99,7 +100,8 @@ public class PipeTsFileInsertionEvent extends PipeInsertionEvent private final AtomicBoolean isClosed; private final AtomicReference eventParser; private final AtomicBoolean isTsFileParserMemoryReserved = new AtomicBoolean(false); - private final Object tsFileParserMemoryReservationKey = new Object(); + private final TsFileParserMemoryReservation tsFileParserMemoryReservationKey = + new TsFileParserMemoryReservation(); // The point count of the TsFile. Used for metrics on IoTConsensusV2' receiver side. // May be updated after it is flushed. Should be negative if not set. @@ -882,15 +884,12 @@ private void waitForResourceEnough4Parsing(final long timeoutMs) throws Interrup final long startTime = System.currentTimeMillis(); long lastRecordTime = startTime; - final long memoryCheckIntervalMs = - PipeConfig.getInstance().getPipeCheckMemoryEnoughIntervalMs(); while (!tryReserveTsFileParserMemory(memoryManager)) { - Thread.sleep(memoryCheckIntervalMs); - final long currentTime = System.currentTimeMillis(); - final double elapsedRecordTimeSeconds = (currentTime - lastRecordTime) / 1000.0; - final double waitTimeSeconds = (currentTime - startTime) / 1000.0; - if (elapsedRecordTimeSeconds > 10.0) { + final long elapsedRecordTimeInMs = currentTime - lastRecordTime; + final long waitTimeInMs = currentTime - startTime; + final double waitTimeSeconds = waitTimeInMs / 1000.0; + if (elapsedRecordTimeInMs > 10_000) { LOGGER.info( DataNodePipeMessages.WAIT_FOR_MEMORY_ENOUGH_FOR_PARSING_FOR, resource != null ? resource.getTsFilePath() : "tsfile", @@ -903,7 +902,7 @@ private void waitForResourceEnough4Parsing(final long timeoutMs) throws Interrup waitTimeSeconds); } - if (waitTimeSeconds * 1000 > timeoutMs) { + if (waitTimeInMs > timeoutMs) { // should contain 'TimeoutException' in exception message throw new PipeRuntimeOutOfMemoryCriticalException( String.format( @@ -911,6 +910,12 @@ private void waitForResourceEnough4Parsing(final long timeoutMs) throws Interrup .PIPE_EXCEPTION_TIMEOUTEXCEPTION_WAITED_S_SECONDS_FOR_MEMORY_TO_PARSE_TSFILE_0E4EF8FD, waitTimeSeconds)); } + + tsFileParserMemoryReservationKey.await( + Math.max( + 1, + Math.min( + timeoutMs - waitTimeInMs, 10_000 - Math.min(10_000, elapsedRecordTimeInMs)))); } final long currentTime = System.currentTimeMillis(); @@ -1088,7 +1093,7 @@ private static class PipeTsFileInsertionEventResource extends PipeEventResource private final long creationTime; private final String dataRegionId; private final AtomicBoolean isTsFileParserMemoryReserved; - private final Object tsFileParserMemoryReservationKey; + private final TsFileParserMemoryReservation tsFileParserMemoryReservationKey; private PipeTsFileInsertionEventResource( final AtomicBoolean isReleased, @@ -1102,7 +1107,7 @@ private PipeTsFileInsertionEventResource( final File sharedModFile, final AtomicReference eventParser, final AtomicBoolean isTsFileParserMemoryReserved, - final Object tsFileParserMemoryReservationKey) { + final TsFileParserMemoryReservation tsFileParserMemoryReservationKey) { super(isReleased, referenceCount); this.pipeName = pipeName; this.creationTime = creationTime; diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryManager.java index 53e8013d00cf0..086cb869115c5 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryManager.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryManager.java @@ -70,7 +70,7 @@ public class PipeMemoryManager { private final Map reservedTsFileParserCountByPipe = new HashMap<>(); private final Map reservedTsFileParserCountByPipeRegion = new HashMap<>(); - private final Map> + private final Map> waitingTsFileParserRequestsByPipeRegion = new HashMap<>(); private final Map> waitingTsFileParserRegionOrderByPipe = new HashMap<>(); @@ -165,7 +165,7 @@ public synchronized boolean tryReserveTsFileParserMemory( final String pipeName, final long creationTime, final String dataRegionId, - final Object reservationKey) { + final TsFileParserMemoryReservation reservationKey) { if (reservationKey == null) { return false; } @@ -183,6 +183,7 @@ public synchronized boolean tryReserveTsFileParserMemory( reservedTsFileParserCountByPipeRegion.getOrDefault(pipeRegionIdentity, 0); if (reservedTsFileParserCount >= globalLimit || reservedCountOfPipeRegion >= perPipeRegionLimit) { + notifyNextTsFileParserMemoryReservationInternal(); return false; } @@ -197,11 +198,12 @@ public synchronized boolean tryReserveTsFileParserMemory( final PipeRegionIdentity nextPipeRegion = getNextEligibleTsFileParserPipeRegion(perPipeRegionLimit, !isSoftMemoryEnough); - final LinkedHashSet requestsOfPipeRegion = + final LinkedHashSet requestsOfPipeRegion = waitingTsFileParserRequestsByPipeRegion.get(pipeRegionIdentity); if (!pipeRegionIdentity.equals(nextPipeRegion) || requestsOfPipeRegion == null || !reservationKey.equals(requestsOfPipeRegion.iterator().next())) { + notifyNextTsFileParserMemoryReservationInternal(); return false; } @@ -209,6 +211,7 @@ public synchronized boolean tryReserveTsFileParserMemory( reservedTsFileParserCount++; reservedTsFileParserCountByPipe.merge(pipeIdentity, 1, Integer::sum); reservedTsFileParserCountByPipeRegion.put(pipeRegionIdentity, reservedCountOfPipeRegion + 1); + notifyNextTsFileParserMemoryReservationInternal(); return true; } @@ -216,7 +219,7 @@ public synchronized void cancelTsFileParserMemoryReservation( final String pipeName, final long creationTime, final String dataRegionId, - final Object reservationKey) { + final TsFileParserMemoryReservation reservationKey) { if (reservationKey == null) { return; } @@ -224,7 +227,7 @@ public synchronized void cancelTsFileParserMemoryReservation( new PipeRegionIdentity(new PipeIdentity(pipeName, creationTime), dataRegionId), reservationKey, false); - this.notifyAll(); + notifyNextTsFileParserMemoryReservationInternal(); } public synchronized void releaseTsFileParserMemory( @@ -235,6 +238,12 @@ public synchronized void releaseTsFileParserMemory( final int reservedCountOfPipeRegion = reservedTsFileParserCountByPipeRegion.getOrDefault(pipeRegionIdentity, 0); if (reservedCountOfPipeRegion <= 0) { + LOGGER.warn( + DataNodePipeMessages + .LOG_FAILED_TO_RELEASE_TSFILE_PARSER_MEMORY_FOR_PIPE_ARG_CREATION_TIME_ARG_IN_DATAREGION_ARG_BECAUSE_NO_RESERVATION_EXISTS_BB8321C0, + pipeName, + creationTime, + dataRegionId); return; } @@ -250,12 +259,13 @@ public synchronized void releaseTsFileParserMemory( reservedTsFileParserCountByPipe.put(pipeIdentity, reservedCountOfPipe - 1); } reservedTsFileParserCount--; - this.notifyAll(); + notifyNextTsFileParserMemoryReservationInternal(); } private void enqueueTsFileParserReservationRequest( - final PipeRegionIdentity pipeRegionIdentity, final Object reservationKey) { - final LinkedHashSet requestsOfPipeRegion = + final PipeRegionIdentity pipeRegionIdentity, + final TsFileParserMemoryReservation reservationKey) { + final LinkedHashSet requestsOfPipeRegion = waitingTsFileParserRequestsByPipeRegion.computeIfAbsent( pipeRegionIdentity, key -> { @@ -272,6 +282,41 @@ private void enqueueTsFileParserReservationRequest( requestsOfPipeRegion.add(reservationKey); } + public synchronized void notifyNextTsFileParserMemoryReservation() { + notifyNextTsFileParserMemoryReservationInternal(); + } + + private void notifyNextTsFileParserMemoryReservationInternal() { + final int globalLimit = Math.max(1, PIPE_CONFIG.getPipeTsFileParserInFlightMaxNum()); + if (reservedTsFileParserCount >= globalLimit) { + return; + } + + final long parserMemorySizeInBytes = getTsFileParserMemorySizeInBytes(); + final boolean isSoftMemoryEnough = + !PIPE_MEMORY_MANAGEMENT_ENABLED + || isEnough4TabletParsingWithReservedParserMemory(parserMemorySizeInBytes); + if (!isSoftMemoryEnough + && !isHardEnough4TabletParsingWithReservedParserMemory(parserMemorySizeInBytes)) { + return; + } + + final int perPipeRegionLimit = + Math.max( + 1, Math.min(globalLimit, PIPE_CONFIG.getPipeTsFileParserInFlightMaxNumPerPipeRegion())); + final PipeRegionIdentity nextPipeRegion = + getNextEligibleTsFileParserPipeRegion(perPipeRegionLimit, !isSoftMemoryEnough); + if (nextPipeRegion == null) { + return; + } + + final LinkedHashSet requestsOfPipeRegion = + waitingTsFileParserRequestsByPipeRegion.get(nextPipeRegion); + if (requestsOfPipeRegion != null && !requestsOfPipeRegion.isEmpty()) { + requestsOfPipeRegion.iterator().next().signal(); + } + } + private PipeRegionIdentity getNextEligibleTsFileParserPipeRegion( final int perPipeRegionLimit, final boolean requirePipeWithoutReservedParser) { for (final PipeIdentity pipeIdentity : waitingTsFileParserPipeOrder) { @@ -299,9 +344,9 @@ private PipeRegionIdentity getNextEligibleTsFileParserPipeRegion( private void removeTsFileParserReservationRequest( final PipeRegionIdentity pipeRegionIdentity, - final Object reservationKey, + final TsFileParserMemoryReservation reservationKey, final boolean rotateAfterAdmission) { - final LinkedHashSet requestsOfPipeRegion = + final LinkedHashSet requestsOfPipeRegion = waitingTsFileParserRequestsByPipeRegion.get(pipeRegionIdentity); if (requestsOfPipeRegion == null || !requestsOfPipeRegion.remove(reservationKey)) { return; @@ -331,6 +376,23 @@ private void removeTsFileParserReservationRequest( } } + public static final class TsFileParserMemoryReservation { + + private boolean isSignaled; + + public synchronized void await(final long timeoutInMs) throws InterruptedException { + if (!isSignaled) { + wait(timeoutInMs); + } + isSignaled = false; + } + + private synchronized void signal() { + isSignaled = true; + notify(); + } + } + public boolean shouldReleaseTsFileParserOnOutOfMemory( final long firstOutOfMemoryTimeInMs, final int retryCount) { final long retryIntervalInMs = PIPE_CONFIG.getPipeMemoryAllocateRetryIntervalInMs(); @@ -586,6 +648,7 @@ public synchronized void resize( allocatedBlocks.remove(block); } + notifyNextTsFileParserMemoryReservationInternal(); this.notifyAll(); return; } @@ -872,6 +935,7 @@ public synchronized void release(PipeMemoryBlock block) { } block.markAsReleased(); + notifyNextTsFileParserMemoryReservationInternal(); this.notifyAll(); } @@ -889,6 +953,7 @@ public synchronized boolean release(PipeMemoryBlock block, long sizeInBytes) { } block.setMemoryUsageInBytes(block.getMemoryUsageInBytes() - sizeInBytes); + notifyNextTsFileParserMemoryReservationInternal(); this.notifyAll(); return true; diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/conf/PropertiesTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/conf/PropertiesTest.java index f4daffbf303b9..05b27cdabec77 100755 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/conf/PropertiesTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/conf/PropertiesTest.java @@ -19,6 +19,8 @@ package org.apache.iotdb.db.conf; +import org.apache.iotdb.commons.conf.CommonConfig; +import org.apache.iotdb.commons.conf.CommonDescriptor; import org.apache.iotdb.commons.conf.TrimProperties; import org.apache.iotdb.commons.utils.RegionMigrationFileRemoveRateLimiter; @@ -36,6 +38,33 @@ import static com.tngtech.archunit.lang.syntax.ArchRuleDefinition.noClasses; public class PropertiesTest { + @Test + public void testHotReloadTsFileParserInFlightLimits() throws Exception { + final IoTDBDescriptor descriptor = IoTDBDescriptor.getInstance(); + final CommonConfig commonConfig = CommonDescriptor.getInstance().getConfig(); + final int originalGlobalLimit = commonConfig.getPipeTsFileParserInFlightMaxNum(); + final int originalPerPipeRegionLimit = + commonConfig.getPipeTsFileParserInFlightMaxNumPerPipeRegion(); + + try { + final TrimProperties properties = new TrimProperties(); + properties.setProperty("pipe_tsfile_parser_in_flight_max_num", "3"); + properties.setProperty("pipe_tsfile_parser_in_flight_max_num_per_pipe_region", "2"); + descriptor.loadHotModifiedProps(properties); + + Assert.assertEquals(3, commonConfig.getPipeTsFileParserInFlightMaxNum()); + Assert.assertEquals(2, commonConfig.getPipeTsFileParserInFlightMaxNumPerPipeRegion()); + } finally { + final TrimProperties properties = new TrimProperties(); + properties.setProperty( + "pipe_tsfile_parser_in_flight_max_num", Integer.toString(originalGlobalLimit)); + properties.setProperty( + "pipe_tsfile_parser_in_flight_max_num_per_pipe_region", + Integer.toString(originalPerPipeRegionLimit)); + descriptor.loadHotModifiedProps(properties); + } + } + @Test public void testHotReloadRegionMigrationFileRemoveSpeedLimit() throws Exception { IoTDBDescriptor descriptor = IoTDBDescriptor.getInstance(); diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryManagerTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryManagerTest.java index 6fc5e72968997..72bffbc2c610c 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryManagerTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryManagerTest.java @@ -22,6 +22,7 @@ import org.apache.iotdb.commons.conf.CommonConfig; import org.apache.iotdb.commons.conf.CommonDescriptor; import org.apache.iotdb.db.pipe.resource.PipeDataNodeResourceManager; +import org.apache.iotdb.db.pipe.resource.memory.PipeMemoryManager.TsFileParserMemoryReservation; import org.junit.After; import org.junit.Assert; @@ -258,7 +259,11 @@ public void testConcurrentTsFilesFromMultiplePipesAreNotStarved() throws Excepti enqueued.countDown(); final long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(10); while (!acquired && System.nanoTime() < deadline) { - Thread.sleep(1); + reservation.key.await( + Math.max( + 1, + TimeUnit.NANOSECONDS.toMillis( + Math.max(1, deadline - System.nanoTime())))); acquired = tryAcquireWithoutTracking(reservation); } if (!acquired) { @@ -326,7 +331,7 @@ private static class Reservation { private final String pipeName; private final long creationTime; private final String dataRegionId; - private final Object key = new Object(); + private final TsFileParserMemoryReservation key = new TsFileParserMemoryReservation(); private volatile boolean acquired; private Reservation(final String pipeName, final long creationTime) { diff --git a/iotdb-core/node-commons/src/assembly/resources/conf/iotdb-system.properties.template b/iotdb-core/node-commons/src/assembly/resources/conf/iotdb-system.properties.template index 0150f9db6fa0a..6b25bdd25e69b 100644 --- a/iotdb-core/node-commons/src/assembly/resources/conf/iotdb-system.properties.template +++ b/iotdb-core/node-commons/src/assembly/resources/conf/iotdb-system.properties.template @@ -2011,6 +2011,19 @@ pipe_lib_dir=ext/pipe # Datatype: int pipe_subtask_executor_max_thread_num=0 +# The maximum number of TsFile parsers that can run concurrently across all Pipes on this DataNode. +# When <= 0, use max(1, CPU core number / 2). +# effectiveMode: hot_reload +# Datatype: int +pipe_tsfile_parser_in_flight_max_num=0 + +# The maximum number of TsFile parsers that can run concurrently for one DataRegion of one Pipe. +# Different DataRegions of the same Pipe have independent limits. +# When <= 0, use 1. +# effectiveMode: hot_reload +# Datatype: int +pipe_tsfile_parser_in_flight_max_num_per_pipe_region=1 + # The connection timeout (in milliseconds) for the thrift client. # effectiveMode: restart # Datatype: int diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonConfig.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonConfig.java index c127540efbbd9..fea15d4f3056b 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonConfig.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonConfig.java @@ -1051,7 +1051,10 @@ public int getPipeTsFileParserInFlightMaxNum() { } public void setPipeTsFileParserInFlightMaxNum(final int pipeTsFileParserInFlightMaxNum) { - final int validatedValue = Math.max(1, pipeTsFileParserInFlightMaxNum); + final int validatedValue = + pipeTsFileParserInFlightMaxNum > 0 + ? pipeTsFileParserInFlightMaxNum + : Math.max(1, Runtime.getRuntime().availableProcessors() / 2); if (this.pipeTsFileParserInFlightMaxNum == validatedValue) { return; }