From d88f9f9c28f30d85c4b28861fcb038c051ed3b31 Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Mon, 10 Aug 2026 10:10:23 +0800 Subject: [PATCH] Pipe: dynamically share idle floating memory (#18422) --- .../TsFileInsertionDataContainerProvider.java | 3 +- .../metric/overview/PipeResourceMetrics.java | 4 +- .../resource/memory/PipeMemoryManager.java | 48 ++++++++++++++----- .../PipeRealtimeDataRegionHybridSource.java | 5 +- .../PipeProcessorSubtaskExecutorTest.java | 2 +- .../memory/PipeMemoryManagerResizeTest.java | 36 ++++++++++++-- .../iotdb/commons/conf/CommonConfig.java | 3 ++ .../iotdb/commons/pipe/config/PipeConfig.java | 1 + 8 files changed, 80 insertions(+), 22 deletions(-) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/container/TsFileInsertionDataContainerProvider.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/container/TsFileInsertionDataContainerProvider.java index 59d630a90ce5..00307814942d 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/container/TsFileInsertionDataContainerProvider.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/container/TsFileInsertionDataContainerProvider.java @@ -28,7 +28,6 @@ import org.apache.iotdb.db.pipe.event.common.tsfile.container.scan.TsFileInsertionScanDataContainer; import org.apache.iotdb.db.pipe.metric.overview.PipeTsFileToTabletsMetrics; import org.apache.iotdb.db.pipe.resource.PipeDataNodeResourceManager; -import org.apache.iotdb.db.pipe.resource.memory.PipeMemoryManager; import org.apache.iotdb.db.pipe.resource.tsfile.PipeTsFilePublicResource; import org.apache.tsfile.file.metadata.IDeviceID; @@ -80,7 +79,7 @@ public TsFileInsertionDataContainer provide(final boolean isWithMod) throws IOEx // Use scan container to save memory if ((double) PipeDataNodeResourceManager.memory().getUsedMemorySizeInBytes() - / PipeMemoryManager.getTotalNonFloatingMemorySizeInBytes() + / PipeDataNodeResourceManager.memory().getTotalNonFloatingMemorySizeInBytes() > PipeTsFilePublicResource.MEMORY_SUFFICIENT_THRESHOLD) { return new TsFileInsertionScanDataContainer( pipeName, diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/metric/overview/PipeResourceMetrics.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/metric/overview/PipeResourceMetrics.java index 48c55fba59d0..dd41d032f80e 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/metric/overview/PipeResourceMetrics.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/metric/overview/PipeResourceMetrics.java @@ -75,14 +75,14 @@ public void bindTo(final AbstractMetricService metricService) { Metric.PIPE_MEM.toString(), MetricLevel.IMPORTANT, PipeDataNodeResourceManager.memory(), - o -> PipeMemoryManager.getTotalNonFloatingMemorySizeInBytes(), + PipeMemoryManager::getTotalNonFloatingMemorySizeInBytes, Tag.NAME.toString(), PIPE_TOTAL_MEMORY); metricService.createAutoGauge( Metric.PIPE_MEM.toString(), MetricLevel.IMPORTANT, PipeDataNodeResourceManager.memory(), - o -> PipeMemoryManager.getTotalFloatingMemorySizeInBytes(), + PipeMemoryManager::getTotalFloatingMemorySizeInBytes, Tag.NAME.toString(), PIPE_FLOATING_MEMORY); metricService.createAutoGauge( 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 5dfff56c52e1..9c36c8dbf7ea 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 @@ -38,6 +38,7 @@ import java.util.Map; import java.util.Objects; import java.util.Set; +import java.util.function.LongSupplier; import java.util.function.LongUnaryOperator; public class PipeMemoryManager { @@ -53,6 +54,9 @@ public class PipeMemoryManager { private static final long MEMORY_ALLOCATE_MIN_SIZE_IN_BYTES = PipeConfig.getInstance().getPipeMemoryAllocateMinSizeInBytes(); + private final long totalMemorySizeInBytes; + private final LongSupplier floatingMemoryUsageSupplier; + private long usedMemorySizeInBytes; private static final double EXCEED_PROTECT_THRESHOLD = 0.95; @@ -79,6 +83,9 @@ public class PipeMemoryManager { private final Set expandableBlocks = new HashSet<>(); public PipeMemoryManager() { + this( + TOTAL_MEMORY_SIZE_IN_BYTES, + () -> PipeDataNodeAgent.task().getAllFloatingMemoryUsageInByte()); PipeDataNodeAgent.runtime() .registerPeriodicalJob( "PipeMemoryManager#tryExpandAll()", @@ -86,6 +93,12 @@ public PipeMemoryManager() { PipeConfig.getInstance().getPipeMemoryExpanderIntervalSeconds()); } + PipeMemoryManager( + final long totalMemorySizeInBytes, final LongSupplier floatingMemoryUsageSupplier) { + this.totalMemorySizeInBytes = totalMemorySizeInBytes; + this.floatingMemoryUsageSupplier = floatingMemoryUsageSupplier; + } + // NOTE: Here we unify the memory threshold judgment for tablet and tsfile memory block, because // introducing too many heuristic rules not conducive to flexible dynamic adjustment of memory // configuration: @@ -96,7 +109,7 @@ public PipeMemoryManager() { // 3. The sum of the memory proportion occupied by the tablet memory block and the tsfile memory // block does not exceed TABLET_MEMORY_REJECT_THRESHOLD + TS_FILE_MEMORY_REJECT_THRESHOLD - private static double allowedMaxMemorySizeInBytesOfTabletsAndTsFiles() { + private double allowedMaxMemorySizeInBytesOfTabletsAndTsFiles() { return (PipeConfig.getInstance() .getPipeDataStructureTabletMemoryBlockAllocationRejectThreshold() + PipeConfig.getInstance() @@ -104,7 +117,7 @@ private static double allowedMaxMemorySizeInBytesOfTabletsAndTsFiles() { * getTotalNonFloatingMemorySizeInBytes(); } - private static double allowedMaxMemorySizeInBytesOfTablets() { + private double allowedMaxMemorySizeInBytesOfTablets() { return (PipeConfig.getInstance() .getPipeDataStructureTabletMemoryBlockAllocationRejectThreshold() + PipeConfig.getInstance() @@ -113,7 +126,7 @@ private static double allowedMaxMemorySizeInBytesOfTablets() { * getTotalNonFloatingMemorySizeInBytes(); } - private static double allowedMaxMemorySizeInBytesOfTsTiles() { + private double allowedMaxMemorySizeInBytesOfTsTiles() { return (PipeConfig.getInstance() .getPipeDataStructureTsFileMemoryBlockAllocationRejectThreshold() + PipeConfig.getInstance() @@ -1037,19 +1050,30 @@ public long getUsedMemorySizeInBytesOfTsFiles() { } public long getFreeMemorySizeInBytes() { - return TOTAL_MEMORY_SIZE_IN_BYTES - usedMemorySizeInBytes; + return Math.max(0, getTotalNonFloatingMemorySizeInBytes() - usedMemorySizeInBytes); } - public static long getTotalNonFloatingMemorySizeInBytes() { - return (long) - (TOTAL_MEMORY_SIZE_IN_BYTES - * (1 - PipeConfig.getInstance().getPipeTotalFloatingMemoryProportion())); + public long getTotalNonFloatingMemorySizeInBytes() { + // Floating memory is an upper limit for retained InsertNodes instead of a statically reserved + // partition. Non-floating allocations can borrow all floating memory that is not actually in + // use, which is especially important for TsFile-only pipes. + return Math.max(0, totalMemorySizeInBytes - getUsedFloatingMemorySizeInBytes()); + } + + public long getTotalFloatingMemorySizeInBytes() { + final long configuredUpperLimit = + Math.max( + 0, + (long) + (totalMemorySizeInBytes + * PipeConfig.getInstance().getPipeTotalFloatingMemoryProportion())); + final long memoryNotUsedByNonFloatingAllocations = + Math.max(0, totalMemorySizeInBytes - usedMemorySizeInBytes); + return Math.min(configuredUpperLimit, memoryNotUsedByNonFloatingAllocations); } - public static long getTotalFloatingMemorySizeInBytes() { - return (long) - (TOTAL_MEMORY_SIZE_IN_BYTES - * PipeConfig.getInstance().getPipeTotalFloatingMemoryProportion()); + private long getUsedFloatingMemorySizeInBytes() { + return Math.max(0, floatingMemoryUsageSupplier.getAsLong()); } public static long getTotalMemorySizeInBytes() { diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/PipeRealtimeDataRegionHybridSource.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/PipeRealtimeDataRegionHybridSource.java index faabf8b68f67..0efa1aaf5f35 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/PipeRealtimeDataRegionHybridSource.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/PipeRealtimeDataRegionHybridSource.java @@ -29,7 +29,7 @@ import org.apache.iotdb.db.pipe.event.realtime.PipeRealtimeEvent; import org.apache.iotdb.db.pipe.metric.overview.PipeDataNodeRemainingEventAndTimeOperator; import org.apache.iotdb.db.pipe.metric.overview.PipeDataNodeSinglePipeMetrics; -import org.apache.iotdb.db.pipe.resource.memory.PipeMemoryManager; +import org.apache.iotdb.db.pipe.resource.PipeDataNodeResourceManager; import org.apache.iotdb.db.pipe.source.dataregion.realtime.assigner.PipeTsFileEpochProgressIndexKeeper; import org.apache.iotdb.db.pipe.source.dataregion.realtime.epoch.TsFileEpoch; import org.apache.iotdb.pipe.api.event.Event; @@ -178,7 +178,8 @@ private boolean canNotUseTabletAnymore(final PipeRealtimeEvent event) { final long floatingMemoryUsageInByte = PipeDataNodeAgent.task().getFloatingMemoryUsageInByte(pipeName); final long pipeCount = PipeDataNodeAgent.task().getPipeCount(); - long totalFloatingMemorySizeInBytes = PipeMemoryManager.getTotalFloatingMemorySizeInBytes(); + long totalFloatingMemorySizeInBytes = + PipeDataNodeResourceManager.memory().getTotalFloatingMemorySizeInBytes(); // If the occupied memory has reached the max, it may cause a large latency to the receiver due // to queuing. To reduce the latency, we lower the memory limit forcibly in the single tsFile // since the tsFile is doomed to be transferred, then more downgrading will just cause more diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/PipeProcessorSubtaskExecutorTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/PipeProcessorSubtaskExecutorTest.java index a403e83329ef..267da1356c86 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/PipeProcessorSubtaskExecutorTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/PipeProcessorSubtaskExecutorTest.java @@ -70,7 +70,7 @@ public void testTsFileInsertionEventPreservesOutOfMemoryCause() { try { memoryBlock = memoryManager.forceAllocateForTabletWithRetry( - PipeMemoryManager.getTotalNonFloatingMemorySizeInBytes()); + memoryManager.getTotalNonFloatingMemorySizeInBytes()); Assert.assertFalse(memoryManager.isEnough4TabletParsing()); final File tsFile = diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryManagerResizeTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryManagerResizeTest.java index ad183f5e37f8..08b1d8960042 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryManagerResizeTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryManagerResizeTest.java @@ -28,8 +28,11 @@ import org.junit.Before; import org.junit.Test; +import java.util.concurrent.atomic.AtomicLong; + public class PipeMemoryManagerResizeTest { + private static final long TOTAL_MEMORY_SIZE_IN_BYTES = 2000; private final CommonConfig config = CommonDescriptor.getInstance().getConfig(); private boolean originalMemoryManagementEnabled; @@ -76,7 +79,7 @@ public void testTabletResizeCannotCrossTabletHardLimit() { final PipeTabletMemoryBlock tablet = manager.forceAllocateForTabletWithRetry(0); final long tabletMemorySizeInBytes = (long) - (PipeMemoryManager.getTotalNonFloatingMemorySizeInBytes() + (manager.getTotalNonFloatingMemorySizeInBytes() * (config.getPipeDataStructureTabletMemoryBlockAllocationRejectThreshold() + config.getPipeDataStructureTsFileMemoryBlockAllocationRejectThreshold() / 2)) @@ -97,8 +100,7 @@ public void testTabletResizeCannotCrossTabletHardLimit() { @Test public void testTabletResizeLeavesMemoryForSinkForwardProgress() { final PipeMemoryManager manager = new PipeMemoryManager(); - final long totalNonFloatingMemorySizeInBytes = - PipeMemoryManager.getTotalNonFloatingMemorySizeInBytes(); + final long totalNonFloatingMemorySizeInBytes = manager.getTotalNonFloatingMemorySizeInBytes(); final long tabletMemorySizeInBytes = (long) (totalNonFloatingMemorySizeInBytes @@ -134,4 +136,32 @@ public void testTabletResizeLeavesMemoryForSinkForwardProgress() { Assert.assertEquals(0, manager.getUsedMemorySizeInBytes()); } + + @Test + public void testFloatingAndNonFloatingMemoryShareTheSamePool() { + final AtomicLong floatingMemoryUsageInBytes = new AtomicLong(0); + final PipeMemoryManager manager = + new PipeMemoryManager(TOTAL_MEMORY_SIZE_IN_BYTES, floatingMemoryUsageInBytes::get); + + Assert.assertEquals(TOTAL_MEMORY_SIZE_IN_BYTES, manager.getTotalNonFloatingMemorySizeInBytes()); + Assert.assertEquals( + TOTAL_MEMORY_SIZE_IN_BYTES / 2, manager.getTotalFloatingMemorySizeInBytes()); + + final PipeTsFileMemoryBlock nonFloatingMemory = manager.forceAllocateForTsFileWithRetry(1200); + try { + // Non-floating memory can borrow the unused half that was previously reserved for InsertNode + // queues. Its usage also reduces the current floating-memory limit symmetrically. + Assert.assertEquals(1200, manager.getUsedMemorySizeInBytes()); + Assert.assertEquals(800, manager.getTotalFloatingMemorySizeInBytes()); + + floatingMemoryUsageInBytes.set(500); + Assert.assertEquals(1500, manager.getTotalNonFloatingMemorySizeInBytes()); + Assert.assertEquals(300, manager.getFreeMemorySizeInBytes()); + + Assert.assertThrows( + PipeRuntimeOutOfMemoryCriticalException.class, () -> manager.forceAllocate(301)); + } finally { + manager.release(nonFloatingMemory); + } + } } 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 6bc02fbef2c4..de7e476c9356 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 @@ -224,6 +224,9 @@ public class CommonConfig { private int pipeDataStructureTabletSizeInBytes = 16 * 1024 * 1024; private double pipeDataStructureTabletMemoryBlockAllocationRejectThreshold = 0.3; private double pipeDataStructureTsFileMemoryBlockAllocationRejectThreshold = 0.3; + + // Maximum proportion for floating memory retained by InsertNode queues. Unused floating memory + // can be borrowed by non-floating Pipe allocations. private volatile double pipeTotalFloatingMemoryProportion = 0.5; // Check if memory check is enabled for Pipe 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 fe81318f5ddd..1a06e8220305 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 @@ -66,6 +66,7 @@ public double getPipeDataStructureTsFileMemoryBlockAllocationRejectThreshold() { } public double getPipeTotalFloatingMemoryProportion() { + // This is the upper limit of floating memory, not a statically reserved partition. return COMMON_CONFIG.getPipeTotalFloatingMemoryProportion(); }