This is an automated email from the ASF dual-hosted git repository.
jt2594838 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 4300930cbc2 [Pipe] Back off TsFile parser admission retries (#18262)
4300930cbc2 is described below
commit 4300930cbc23faa61f4d7fd9e44c444bebccfeaa
Author: Caideyipi <[email protected]>
AuthorDate: Wed Jul 22 15:14:08 2026 +0800
[Pipe] Back off TsFile parser admission retries (#18262)
* [Pipe] Back off TsFile parser admission retries
* [Pipe] Document parser admission backoff config semantics
---
.../common/tsfile/PipeTsFileInsertionEvent.java | 65 ++++++---
.../db/pipe/resource/memory/PipeMemoryManager.java | 5 +
.../PipeTsFileInsertionEventAdmissionTest.java | 162 +++++++++++++++++++++
.../apache/iotdb/commons/conf/CommonConfig.java | 2 +
4 files changed, 217 insertions(+), 17 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 757411df9b9..db0d1b261ab 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
@@ -69,6 +69,7 @@ import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
+import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
@@ -879,10 +880,27 @@ public class PipeTsFileInsertionEvent extends
PipeInsertionEvent
final long startTime = System.currentTimeMillis();
long lastRecordTime = startTime;
- final long memoryCheckIntervalMs =
- PipeConfig.getInstance().getPipeCheckMemoryEnoughIntervalMs();
- while (!tryReserveTsFileParserMemory(memoryManager)) {
- Thread.sleep(memoryCheckIntervalMs);
+ final long initialMemoryCheckIntervalMs =
+ Math.max(1,
PipeConfig.getInstance().getPipeCheckMemoryEnoughIntervalMs());
+ final long maxMemoryCheckIntervalMs =
+ getMaxMemoryCheckIntervalMs(
+ initialMemoryCheckIntervalMs,
+ PipeConfig.getInstance().getPipeMemoryAllocateMaxRetries());
+ long memoryCheckIntervalMs = initialMemoryCheckIntervalMs;
+ while (true) {
+ final long elapsedTimeMs = Math.max(0, System.currentTimeMillis() -
startTime);
+ if (elapsedTimeMs >= timeoutMs) {
+ // should contain 'TimeoutException' in exception message
+ throw new PipeRuntimeOutOfMemoryCriticalException(
+ String.format(
+ DataNodePipeMessages
+
.PIPE_EXCEPTION_TIMEOUTEXCEPTION_WAITED_S_SECONDS_FOR_MEMORY_TO_PARSE_TSFILE_0E4EF8FD,
+ elapsedTimeMs / 1000.0));
+ }
+
+ memoryManager.waitForTsFileParserMemory(
+ Math.min(
+ getMemoryCheckIntervalWithJitter(memoryCheckIntervalMs),
timeoutMs - elapsedTimeMs));
final long currentTime = System.currentTimeMillis();
final double elapsedRecordTimeSeconds = (currentTime - lastRecordTime) /
1000.0;
@@ -900,22 +918,35 @@ public class PipeTsFileInsertionEvent extends
PipeInsertionEvent
waitTimeSeconds);
}
- if (waitTimeSeconds * 1000 > timeoutMs) {
- // should contain 'TimeoutException' in exception message
- throw new PipeRuntimeOutOfMemoryCriticalException(
- String.format(
- DataNodePipeMessages
-
.PIPE_EXCEPTION_TIMEOUTEXCEPTION_WAITED_S_SECONDS_FOR_MEMORY_TO_PARSE_TSFILE_0E4EF8FD,
- waitTimeSeconds));
+ if (tryReserveTsFileParserMemory(memoryManager)) {
+ LOGGER.info(
+ DataNodePipeMessages.WAIT_FOR_MEMORY_ENOUGH_FOR_PARSING_FOR,
+ resource != null ? resource.getTsFilePath() : "tsfile",
+ waitTimeSeconds);
+ return;
}
+
+ memoryCheckIntervalMs =
+ getNextMemoryCheckIntervalMs(memoryCheckIntervalMs,
maxMemoryCheckIntervalMs);
}
+ }
+
+ static long getMaxMemoryCheckIntervalMs(final long initialIntervalMs, final
int maxRetries) {
+ final long multiplier = Math.max(1, maxRetries);
+ return initialIntervalMs > Long.MAX_VALUE / multiplier
+ ? Long.MAX_VALUE
+ : initialIntervalMs * multiplier;
+ }
+
+ static long getNextMemoryCheckIntervalMs(final long currentIntervalMs, final
long maxIntervalMs) {
+ return currentIntervalMs >= maxIntervalMs - currentIntervalMs
+ ? maxIntervalMs
+ : currentIntervalMs << 1;
+ }
- final long currentTime = System.currentTimeMillis();
- final double waitTimeSeconds = (currentTime - startTime) / 1000.0;
- LOGGER.info(
- DataNodePipeMessages.WAIT_FOR_MEMORY_ENOUGH_FOR_PARSING_FOR,
- resource != null ? resource.getTsFilePath() : "tsfile",
- waitTimeSeconds);
+ static long getMemoryCheckIntervalWithJitter(final long intervalMs) {
+ return Math.max(
+ 1, (long) (intervalMs * (0.5 +
ThreadLocalRandom.current().nextDouble() * 0.5)));
}
private boolean tryReserveTsFileParserMemory(final PipeMemoryManager
memoryManager) {
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 6360efcdedd..98b820925c7 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
@@ -167,6 +167,11 @@ public class PipeMemoryManager {
this.notifyAll();
}
+ public synchronized void waitForTsFileParserMemory(final long timeoutInMs)
+ throws InterruptedException {
+ this.wait(Math.max(1, timeoutInMs));
+ }
+
public boolean shouldReleaseTsFileParserOnOutOfMemory(
final long firstOutOfMemoryTimeInMs, final int retryCount) {
final long retryIntervalInMs =
PIPE_CONFIG.getPipeMemoryAllocateRetryIntervalInMs();
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeTsFileInsertionEventAdmissionTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeTsFileInsertionEventAdmissionTest.java
new file mode 100644
index 00000000000..d3d7ae9cf8e
--- /dev/null
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeTsFileInsertionEventAdmissionTest.java
@@ -0,0 +1,162 @@
+/*
+ * 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.event.common.tsfile;
+
+import org.apache.iotdb.commons.conf.CommonConfig;
+import org.apache.iotdb.commons.conf.CommonDescriptor;
+import org.apache.iotdb.commons.pipe.datastructure.pattern.PrefixTreePattern;
+import org.apache.iotdb.db.pipe.resource.PipeDataNodeResourceManager;
+import org.apache.iotdb.db.pipe.resource.memory.PipeMemoryManager;
+import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource;
+import
org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResourceStatus;
+import org.apache.iotdb.pipe.api.event.dml.insertion.TabletInsertionEvent;
+
+import org.apache.tsfile.file.metadata.IDeviceID;
+import org.apache.tsfile.utils.TsFileGeneratorUtils;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.io.File;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+public class PipeTsFileInsertionEventAdmissionTest {
+
+ @Test
+ public void testParserAdmissionBackoffIsBoundedAndJittered() {
+ Assert.assertEquals(100,
PipeTsFileInsertionEvent.getMaxMemoryCheckIntervalMs(10, 10));
+ Assert.assertEquals(10,
PipeTsFileInsertionEvent.getMaxMemoryCheckIntervalMs(10, 0));
+ Assert.assertEquals(
+ Long.MAX_VALUE,
PipeTsFileInsertionEvent.getMaxMemoryCheckIntervalMs(Long.MAX_VALUE, 10));
+
+ Assert.assertEquals(20,
PipeTsFileInsertionEvent.getNextMemoryCheckIntervalMs(10, 100));
+ Assert.assertEquals(100,
PipeTsFileInsertionEvent.getNextMemoryCheckIntervalMs(80, 100));
+ Assert.assertEquals(100,
PipeTsFileInsertionEvent.getNextMemoryCheckIntervalMs(100, 100));
+
+ for (int i = 0; i < 100; i++) {
+ final long intervalWithJitter =
+ PipeTsFileInsertionEvent.getMemoryCheckIntervalWithJitter(100);
+ Assert.assertTrue(intervalWithJitter >= 50);
+ Assert.assertTrue(intervalWithJitter <= 100);
+ }
+ }
+
+ @Test(timeout = 10000)
+ public void testParserAdmissionIsWokenWhenMemoryIsReleased() throws
Exception {
+ final CommonConfig commonConfig =
CommonDescriptor.getInstance().getConfig();
+ final PipeMemoryManager memoryManager =
PipeDataNodeResourceManager.memory();
+ final long originalParserMemoryInBytes =
commonConfig.getPipeTsFileParserMemory();
+ final long originalMemoryCheckIntervalMs =
commonConfig.getPipeCheckMemoryEnoughIntervalMs();
+ final int originalMemoryAllocateMaxRetries =
commonConfig.getPipeMemoryAllocateMaxRetries();
+
+ File tsFile = null;
+ PipeTsFileInsertionEvent event = null;
+ ExecutorService executor = null;
+ Future<Iterable<TabletInsertionEvent>> parsingFuture = null;
+ int blockerReservationCount = 0;
+ try {
+ commonConfig.setPipeTsFileParserMemory(
+ Math.max(1, memoryManager.getTotalNonFloatingMemorySizeInBytes() /
8));
+ commonConfig.setPipeCheckMemoryEnoughIntervalMs(10000);
+ commonConfig.setPipeMemoryAllocateMaxRetries(10);
+
+ boolean parserMemoryExhausted = false;
+ for (int i = 0; i < 100; i++) {
+ if (!memoryManager.tryReserveTsFileParserMemory()) {
+ parserMemoryExhausted = true;
+ break;
+ }
+ blockerReservationCount++;
+ }
+ Assert.assertTrue(blockerReservationCount > 0);
+ Assert.assertTrue(parserMemoryExhausted);
+
+ tsFile =
+ TsFileGeneratorUtils.generateNonAlignedTsFile(
+ "parser-admission-backoff.tsfile", 1, 1, 10, 0, 100, 10, 10);
+ final TsFileResource resource = new TsFileResource(tsFile);
+ resource.setStatusForTest(TsFileResourceStatus.NORMAL);
+ final IDeviceID deviceID =
IDeviceID.Factory.DEFAULT_FACTORY.create("root.testsg.d0");
+ resource.updateStartTime(deviceID, 0);
+ resource.updateEndTime(deviceID, 9);
+ event =
+ new PipeTsFileInsertionEvent(
+ false,
+ "root",
+ resource,
+ null,
+ false,
+ false,
+ false,
+ null,
+ null,
+ 0,
+ null,
+ new PrefixTreePattern("root"),
+ null,
+ null,
+ null,
+ null,
+ true,
+ Long.MIN_VALUE,
+ Long.MAX_VALUE);
+
+ executor = Executors.newSingleThreadExecutor();
+ final PipeTsFileInsertionEvent eventToParse = event;
+ parsingFuture = executor.submit(() ->
eventToParse.toTabletInsertionEvents(30000));
+
+ final Future<Iterable<TabletInsertionEvent>> blockedParsingFuture =
parsingFuture;
+ Assert.assertThrows(
+ TimeoutException.class, () -> blockedParsingFuture.get(200,
TimeUnit.MILLISECONDS));
+
+ final long releaseTimeInNanos = System.nanoTime();
+ memoryManager.releaseTsFileParserMemory();
+ blockerReservationCount--;
+
+ Assert.assertNotNull(parsingFuture.get(3, TimeUnit.SECONDS));
+ Assert.assertTrue(
+ TimeUnit.NANOSECONDS.toMillis(System.nanoTime() -
releaseTimeInNanos) < 3000);
+ } finally {
+ if (parsingFuture != null) {
+ parsingFuture.cancel(true);
+ }
+ if (executor != null) {
+ executor.shutdownNow();
+ executor.awaitTermination(3, TimeUnit.SECONDS);
+ }
+ if (event != null) {
+ event.close();
+ }
+ while (blockerReservationCount > 0) {
+ memoryManager.releaseTsFileParserMemory();
+ blockerReservationCount--;
+ }
+ commonConfig.setPipeTsFileParserMemory(originalParserMemoryInBytes);
+
commonConfig.setPipeCheckMemoryEnoughIntervalMs(originalMemoryCheckIntervalMs);
+
commonConfig.setPipeMemoryAllocateMaxRetries(originalMemoryAllocateMaxRetries);
+ if (tsFile != null) {
+ tsFile.delete();
+ }
+ }
+ }
+}
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 91393bf9fd2..14675b8b19e 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
@@ -341,6 +341,8 @@ public class CommonConfig {
private volatile boolean pipeMemoryManagementEnabled = true;
private volatile long pipeMemoryAllocateRetryIntervalMs = 50;
+ // Besides limiting allocation retries, this value also caps the TsFile
parser memory admission
+ // backoff at pipeCheckMemoryEnoughIntervalMs * max(1,
pipeMemoryAllocateMaxRetries).
private volatile int pipeMemoryAllocateMaxRetries = 10;
private volatile long pipeMemoryAllocateMinSizeInBytes = 32;
private volatile long pipeMemoryAllocateForTsFileSequenceReaderInBytes =
