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..db0d1b261ab09 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.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 @@ 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 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 @@ private void waitForResourceEnough4Parsing(final long timeoutMs) throws Interrup 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 6360efcdedd19..98b820925c742 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 synchronized void releaseTsFileParserMemory() { 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 0000000000000..d3d7ae9cf8e9f --- /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> 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> 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 91393bf9fd2b4..14675b8b19e7a 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 =