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 10950e13ea9 Pipe: dynamically share idle floating memory (#18422)
10950e13ea9 is described below
commit 10950e13ea9cdc1937d4ff63254538e51a0fbf3a
Author: Caideyipi <[email protected]>
AuthorDate: Mon Aug 10 10:10:23 2026 +0800
Pipe: dynamically share idle floating memory (#18422)
---
.../db/pipe/resource/memory/PipeMemoryManager.java | 38 +++++++++++++++++-----
.../memory/PipeMemoryManagerResizeTest.java | 38 +++++++++++++++++++++-
.../apache/iotdb/commons/conf/CommonConfig.java | 3 ++
.../iotdb/commons/pipe/config/PipeConfig.java | 1 +
4 files changed, 71 insertions(+), 9 deletions(-)
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 dd0485992ee..90a45c1542e 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
@@ -41,6 +41,7 @@ import java.util.List;
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 {
@@ -55,6 +56,8 @@ public class PipeMemoryManager {
// TODO @spricoder: consider combine memory block and used MemorySizeInBytes
private final IMemoryBlock memoryBlock;
+ private final LongSupplier floatingMemoryUsageSupplier;
+
private static final double EXCEED_PROTECT_THRESHOLD = 0.95;
private volatile long usedMemorySizeInBytesOfTablets;
@@ -83,7 +86,8 @@ public class PipeMemoryManager {
IoTDBDescriptor.getInstance()
.getMemoryConfig()
.getPipeMemoryManager()
- .exactAllocate("Stream", MemoryBlockType.DYNAMIC));
+ .exactAllocate("Stream", MemoryBlockType.DYNAMIC),
+ () -> PipeDataNodeAgent.task().getAllFloatingMemoryUsageInByte());
PipeDataNodeAgent.runtime()
.registerPeriodicalJob(
"PipeMemoryManager#tryExpandAll()",
@@ -92,7 +96,13 @@ public class PipeMemoryManager {
}
PipeMemoryManager(final IMemoryBlock memoryBlock) {
+ this(memoryBlock, () ->
PipeDataNodeAgent.task().getAllFloatingMemoryUsageInByte());
+ }
+
+ PipeMemoryManager(
+ final IMemoryBlock memoryBlock, final LongSupplier
floatingMemoryUsageSupplier) {
this.memoryBlock = memoryBlock;
+ this.floatingMemoryUsageSupplier = floatingMemoryUsageSupplier;
}
// NOTE: Here we unify the memory threshold judgment for tablet and tsfile
memory block, because
@@ -1038,19 +1048,31 @@ public class PipeMemoryManager {
}
public long getFreeMemorySizeInBytes() {
- return memoryBlock.getFreeMemoryInBytes();
+ return Math.max(0, getTotalNonFloatingMemorySizeInBytes() -
memoryBlock.getUsedMemoryInBytes());
}
public long getTotalNonFloatingMemorySizeInBytes() {
- return (long)
- (memoryBlock.getTotalMemorySizeInBytes()
- * (1 -
PipeConfig.getInstance().getPipeTotalFloatingMemoryProportion()));
+ // 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, memoryBlock.getTotalMemorySizeInBytes() -
getUsedFloatingMemorySizeInBytes());
}
public long getTotalFloatingMemorySizeInBytes() {
- return (long)
- (memoryBlock.getTotalMemorySizeInBytes()
- * PipeConfig.getInstance().getPipeTotalFloatingMemoryProportion());
+ final long configuredUpperLimit =
+ Math.max(
+ 0,
+ (long)
+ (memoryBlock.getTotalMemorySizeInBytes()
+ *
PipeConfig.getInstance().getPipeTotalFloatingMemoryProportion()));
+ final long memoryNotUsedByNonFloatingAllocations =
+ Math.max(0, memoryBlock.getTotalMemorySizeInBytes() -
memoryBlock.getUsedMemoryInBytes());
+ return Math.min(configuredUpperLimit,
memoryNotUsedByNonFloatingAllocations);
+ }
+
+ private long getUsedFloatingMemorySizeInBytes() {
+ return Math.max(0, floatingMemoryUsageSupplier.getAsLong());
}
public long getTotalMemorySizeInBytes() {
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 c151857e8cb..d63fe1c6e83 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
@@ -30,10 +30,12 @@ import org.junit.Assert;
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 static final long TABLET_MEMORY_SIZE_IN_BYTES = 451;
+ private static final long TABLET_MEMORY_SIZE_IN_BYTES = 901;
private static final long SINK_MEMORY_SIZE_IN_BYTES = 100;
private final CommonConfig config =
CommonDescriptor.getInstance().getConfig();
@@ -136,4 +138,38 @@ public class PipeMemoryManagerResizeTest {
Assert.assertEquals(0, manager.getUsedMemorySizeInBytes());
}
+
+ @Test
+ public void testFloatingAndNonFloatingMemoryShareTheSamePool() {
+ final AtomicLong floatingMemoryUsageInBytes = new AtomicLong(0);
+ final PipeMemoryManager manager =
+ new PipeMemoryManager(
+ new AtomicLongMemoryBlock(
+ "PipeMemoryManagerResizeTest",
+ null,
+ TOTAL_MEMORY_SIZE_IN_BYTES,
+ MemoryBlockType.DYNAMIC),
+ 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 73f5bf52fc3..4adacb0d7aa 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
@@ -235,6 +235,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 561d2923c07..e04237129e8 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
@@ -67,6 +67,7 @@ public class PipeConfig {
}
public double getPipeTotalFloatingMemoryProportion() {
+ // This is the upper limit of floating memory, not a statically reserved
partition.
return COMMON_CONFIG.getPipeTotalFloatingMemoryProportion();
}
