This is an automated email from the ASF dual-hosted git repository.
rong 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 82be8be5844 Pipe: Added total floating memory usage for mem-ctrl of
linked but deleted tsFile resource and insertNode memory (#15205)
82be8be5844 is described below
commit 82be8be58444b14bb6b9e688cab428effe8b578b
Author: Caideyipi <[email protected]>
AuthorDate: Thu Apr 3 14:47:56 2025 +0800
Pipe: Added total floating memory usage for mem-ctrl of linked but deleted
tsFile resource and insertNode memory (#15205)
---
.../db/pipe/agent/task/PipeDataNodeTaskAgent.java | 28 +++++---------
.../IoTDBDataNodeCacheLeaderClientManager.java | 4 +-
.../parser/TsFileInsertionEventParserProvider.java | 2 +-
.../PipeRealtimeDataRegionHybridExtractor.java | 8 ++--
.../pipe/metric/overview/PipeResourceMetrics.java | 2 +-
.../db/pipe/resource/memory/PipeMemoryManager.java | 45 +++++++++++++---------
.../resource/tsfile/PipeTsFileResourceManager.java | 2 +-
.../broker/SubscriptionPrefetchingQueueStates.java | 2 +-
.../event/cache/SubscriptionPollResponseCache.java | 4 +-
.../apache/iotdb/commons/conf/CommonConfig.java | 9 +++++
.../iotdb/commons/conf/CommonDescriptor.java | 5 +++
.../commons/pipe/agent/task/PipeTaskAgent.java | 13 +++++++
.../iotdb/commons/pipe/config/PipeConfig.java | 5 +++
13 files changed, 79 insertions(+), 50 deletions(-)
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/PipeDataNodeTaskAgent.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/PipeDataNodeTaskAgent.java
index b6f3cffce33..0af0be13515 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/PipeDataNodeTaskAgent.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/PipeDataNodeTaskAgent.java
@@ -624,19 +624,22 @@ public class PipeDataNodeTaskAgent extends PipeTaskAgent {
}
final long totalLinkedButDeletedTsFileResourceRamSize =
-
PipeDataNodeResourceManager.tsfile().getTotalLinkedButDeletedTsfileResourceRamSize();
- final long freeMemorySizeInBytes =
- PipeDataNodeResourceManager.memory().getFreeMemorySizeInBytes();
- if (3 * totalLinkedButDeletedTsFileResourceRamSize >= 2 *
freeMemorySizeInBytes) {
+
PipeDataNodeResourceManager.tsfile().getTotalLinkedButDeletedTsFileResourceRamSize();
+ final long totalInsertNodeFloatingMemoryUsageInBytes =
getAllFloatingMemoryUsageInByte();
+ final long totalFloatingMemorySizeInBytes =
+
PipeDataNodeResourceManager.memory().getTotalFloatingMemorySizeInBytes();
+ if (totalInsertNodeFloatingMemoryUsageInBytes +
totalLinkedButDeletedTsFileResourceRamSize
+ >= totalFloatingMemorySizeInBytes) {
for (final PipeMeta pipeMeta : pipeMetaKeeper.getPipeMetaList()) {
stuckPipes.add(pipeMeta);
}
if (!stuckPipes.isEmpty()) {
LOGGER.warn(
- "All {} pipe(s) will be restarted because linked tsfiles' resource
size {} exceeds limit {}.",
+ "All {} pipe(s) will be restarted because linked but deleted
tsFiles' resource size {} and all insertNode's size {} exceeds limit {}.",
stuckPipes.size(),
totalLinkedButDeletedTsFileResourceRamSize,
- freeMemorySizeInBytes * 2.0 / 3);
+ totalInsertNodeFloatingMemoryUsageInBytes,
+ totalFloatingMemorySizeInBytes);
}
return stuckPipes;
}
@@ -680,19 +683,6 @@ public class PipeDataNodeTaskAgent extends PipeTaskAgent {
mayMemTablePinnedCountReachDangerousThreshold(),
mayWalSizeReachThrottleThreshold());
stuckPipes.add(pipeMeta);
- } else if (getFloatingMemoryUsageInByte(pipeName)
- >= PipeDataNodeResourceManager.memory().getFreeMemorySizeInBytes()
- / pipeMetaKeeper.getPipeMetaCount()) {
- // Extractors of this pipe may have too many insert nodes
- LOGGER.warn(
- "Pipe {} needs to restart because too many insertNodes are
extracted. "
- + "Floating memory usage for this pipe: {}, free memory
size: {}, allowed free memory size for floating memory usage: {}",
- pipeMeta.getStaticMeta(),
- getFloatingMemoryUsageInByte(pipeName),
- PipeDataNodeResourceManager.memory().getFreeMemorySizeInBytes(),
- PipeDataNodeResourceManager.memory().getFreeMemorySizeInBytes()
- / pipeMetaKeeper.getPipeMetaCount());
- stuckPipes.add(pipeMeta);
}
}
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/client/IoTDBDataNodeCacheLeaderClientManager.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/client/IoTDBDataNodeCacheLeaderClientManager.java
index e45954e7e4d..8b88a1c36ee 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/client/IoTDBDataNodeCacheLeaderClientManager.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/client/IoTDBDataNodeCacheLeaderClientManager.java
@@ -51,10 +51,10 @@ public interface IoTDBDataNodeCacheLeaderClientManager {
public LeaderCacheManager() {
final long initMemorySizeInBytes =
- PipeDataNodeResourceManager.memory().getTotalMemorySizeInBytes() /
10;
+
PipeDataNodeResourceManager.memory().getTotalNonFloatingMemorySizeInBytes() /
10;
final long maxMemorySizeInBytes =
(long)
- (PipeDataNodeResourceManager.memory().getTotalMemorySizeInBytes()
+
(PipeDataNodeResourceManager.memory().getTotalNonFloatingMemorySizeInBytes()
* CONFIG.getPipeLeaderCacheMemoryUsagePercentage());
// properties required by pipe memory control framework
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/parser/TsFileInsertionEventParserProvider.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/parser/TsFileInsertionEventParserProvider.java
index 1dee3f79df9..72c857c5c51 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/parser/TsFileInsertionEventParserProvider.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/parser/TsFileInsertionEventParserProvider.java
@@ -99,7 +99,7 @@ public class TsFileInsertionEventParserProvider {
// Use scan container to save memory
if ((double)
PipeDataNodeResourceManager.memory().getUsedMemorySizeInBytes()
- / PipeDataNodeResourceManager.memory().getTotalMemorySizeInBytes()
+ /
PipeDataNodeResourceManager.memory().getTotalNonFloatingMemorySizeInBytes()
> PipeTsFileResource.MEMORY_SUFFICIENT_THRESHOLD) {
return new TsFileInsertionEventScanParser(
pipeName,
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/realtime/PipeRealtimeDataRegionHybridExtractor.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/realtime/PipeRealtimeDataRegionHybridExtractor.java
index 38dcd652e87..8a3826b66a8 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/realtime/PipeRealtimeDataRegionHybridExtractor.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/realtime/PipeRealtimeDataRegionHybridExtractor.java
@@ -329,17 +329,17 @@ public class PipeRealtimeDataRegionHybridExtractor
extends PipeRealtimeDataRegio
final long floatingMemoryUsageInByte =
PipeDataNodeAgent.task().getFloatingMemoryUsageInByte(pipeName);
final long pipeCount = PipeDataNodeAgent.task().getPipeCount();
- final long freeMemorySizeInBytes =
- PipeDataNodeResourceManager.memory().getFreeMemorySizeInBytes();
+ final long totalFloatingMemorySizeInBytes =
+
PipeDataNodeResourceManager.memory().getTotalFloatingMemorySizeInBytes();
final boolean mayInsertNodeMemoryReachDangerousThreshold =
- 3 * floatingMemoryUsageInByte * pipeCount >= 2 * freeMemorySizeInBytes;
+ 3 * floatingMemoryUsageInByte * pipeCount >= 2 *
totalFloatingMemorySizeInBytes;
if (mayInsertNodeMemoryReachDangerousThreshold &&
event.mayExtractorUseTablets(this)) {
LOGGER.info(
"Pipe task {}@{} canNotUseTabletAnyMore7: The shallow memory usage
of the insert node {} has reached the dangerous threshold {}",
pipeName,
dataRegionId,
floatingMemoryUsageInByte * pipeCount,
- 2 * freeMemorySizeInBytes / 3.0d);
+ 2 * totalFloatingMemorySizeInBytes / 3.0d);
}
return mayInsertNodeMemoryReachDangerousThreshold;
}
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 7b72ea70a71..9eb71b37124 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
@@ -71,7 +71,7 @@ public class PipeResourceMetrics implements IMetricSet {
Metric.PIPE_MEM.toString(),
MetricLevel.IMPORTANT,
PipeDataNodeResourceManager.memory(),
- PipeMemoryManager::getTotalMemorySizeInBytes,
+ PipeMemoryManager::getTotalNonFloatingMemorySizeInBytes,
Tag.NAME.toString(),
PIPE_TOTAL_MEMORY);
// resource reference count
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 0fc888a7cd2..0c7a473950d 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
@@ -71,6 +71,9 @@ public class PipeMemoryManager {
PipeConfig.getInstance().getPipeDataStructureTsFileMemoryBlockAllocationRejectThreshold();
private volatile long usedMemorySizeInBytesOfTsFiles;
+ private static final double FLOATING_MEMORY_RATIO =
+ PipeConfig.getInstance().getPipeTotalFloatingMemoryProportion();
+
// Only non-zero memory blocks will be added to this set.
private final Set<PipeMemoryBlock> allocatedBlocks = new HashSet<>();
@@ -94,17 +97,17 @@ public class PipeMemoryManager {
private double allowedMaxMemorySizeInBytesOfTabletsAndTsFiles() {
return (TABLET_MEMORY_REJECT_THRESHOLD + TS_FILE_MEMORY_REJECT_THRESHOLD)
- * memoryBlock.getTotalMemorySizeInBytes();
+ * getTotalNonFloatingMemorySizeInBytes();
}
private double allowedMaxMemorySizeInBytesOfTablets() {
return (TABLET_MEMORY_REJECT_THRESHOLD + TS_FILE_MEMORY_REJECT_THRESHOLD /
2)
- * memoryBlock.getTotalMemorySizeInBytes();
+ * getTotalNonFloatingMemorySizeInBytes();
}
private double allowedMaxMemorySizeInBytesOfTsTiles() {
return (TS_FILE_MEMORY_REJECT_THRESHOLD + TABLET_MEMORY_REJECT_THRESHOLD /
2)
- * memoryBlock.getTotalMemorySizeInBytes();
+ * getTotalNonFloatingMemorySizeInBytes();
}
public boolean isEnough4TabletParsing() {
@@ -176,7 +179,7 @@ public class PipeMemoryManager {
String.format(
"forceAllocateForTablet: failed to allocate because there's too
much memory for tablets, "
+ "total memory size %d bytes, used memory for tablet size
%d bytes, requested memory size %d bytes",
- memoryBlock.getTotalMemorySizeInBytes(),
+ getTotalNonFloatingMemorySizeInBytes(),
usedMemorySizeInBytesOfTablets,
tabletSizeInBytes));
}
@@ -218,7 +221,7 @@ public class PipeMemoryManager {
String.format(
"forceAllocateForTsFile: failed to allocate because there's too
much memory for tsfiles, "
+ "total memory size %d bytes, used memory for tsfile size
%d bytes, requested memory size %d bytes",
- memoryBlock.getTotalMemorySizeInBytes(),
+ getTotalNonFloatingMemorySizeInBytes(),
usedMemorySizeInBytesOfTsFiles,
tsFileSizeInBytes));
}
@@ -246,7 +249,7 @@ public class PipeMemoryManager {
}
for (int i = 1; i <= MEMORY_ALLOCATE_MAX_RETRIES; i++) {
- if (memoryBlock.getTotalMemorySizeInBytes() -
memoryBlock.getUsedMemoryInBytes()
+ if (getTotalNonFloatingMemorySizeInBytes() -
memoryBlock.getUsedMemoryInBytes()
>= sizeInBytes) {
return registerMemoryBlock(sizeInBytes, type);
}
@@ -266,7 +269,7 @@ public class PipeMemoryManager {
+ "total memory size %d bytes, used memory size %d bytes, "
+ "requested memory size %d bytes",
MEMORY_ALLOCATE_MAX_RETRIES,
- memoryBlock.getTotalMemorySizeInBytes(),
+ getTotalNonFloatingMemorySizeInBytes(),
memoryBlock.getUsedMemoryInBytes(),
sizeInBytes));
}
@@ -309,7 +312,7 @@ public class PipeMemoryManager {
long sizeInBytes = targetSize - oldSize;
for (int i = 1; i <= MEMORY_ALLOCATE_MAX_RETRIES; i++) {
- if (memoryBlock.getTotalMemorySizeInBytes() -
memoryBlock.getUsedMemoryInBytes()
+ if (getTotalNonFloatingMemorySizeInBytes() -
memoryBlock.getUsedMemoryInBytes()
>= sizeInBytes) {
memoryBlock.forceAllocateWithoutLimitation(sizeInBytes);
if (block instanceof PipeTabletMemoryBlock) {
@@ -337,7 +340,7 @@ public class PipeMemoryManager {
+ "total memory size %d bytes, used memory size %d bytes, "
+ "requested memory size %d bytes",
MEMORY_ALLOCATE_MAX_RETRIES,
- memoryBlock.getTotalMemorySizeInBytes(),
+ getTotalNonFloatingMemorySizeInBytes(),
memoryBlock.getUsedMemoryInBytes(),
sizeInBytes));
}
@@ -366,7 +369,7 @@ public class PipeMemoryManager {
}
if ((float) (memoryBlock.getUsedMemoryInBytes() + sizeInBytes)
- <= memoryBlock.getTotalMemorySizeInBytes() * usedThreshold) {
+ <= getTotalNonFloatingMemorySizeInBytes() * usedThreshold) {
return forceAllocate(sizeInBytes);
}
@@ -384,21 +387,21 @@ public class PipeMemoryManager {
}
if (sizeInBytes == 0
- || memoryBlock.getTotalMemorySizeInBytes() -
memoryBlock.getUsedMemoryInBytes()
+ || getTotalNonFloatingMemorySizeInBytes() -
memoryBlock.getUsedMemoryInBytes()
>= sizeInBytes) {
return registerMemoryBlock(sizeInBytes);
}
long sizeToAllocateInBytes = sizeInBytes;
while (sizeToAllocateInBytes > MEMORY_ALLOCATE_MIN_SIZE_IN_BYTES) {
- if (memoryBlock.getTotalMemorySizeInBytes() -
memoryBlock.getUsedMemoryInBytes()
+ if (getTotalNonFloatingMemorySizeInBytes() -
memoryBlock.getUsedMemoryInBytes()
>= sizeToAllocateInBytes) {
LOGGER.info(
"tryAllocate: allocated memory, "
+ "total memory size {} bytes, used memory size {} bytes, "
+ "original requested memory size {} bytes, "
+ "actual requested memory size {} bytes",
- memoryBlock.getTotalMemorySizeInBytes(),
+ getTotalNonFloatingMemorySizeInBytes(),
memoryBlock.getUsedMemoryInBytes(),
sizeInBytes,
sizeToAllocateInBytes);
@@ -417,7 +420,7 @@ public class PipeMemoryManager {
+ "total memory size {} bytes, used memory size {} bytes, "
+ "original requested memory size {} bytes, "
+ "actual requested memory size {} bytes",
- memoryBlock.getTotalMemorySizeInBytes(),
+ getTotalNonFloatingMemorySizeInBytes(),
memoryBlock.getUsedMemoryInBytes(),
sizeInBytes,
sizeToAllocateInBytes);
@@ -427,7 +430,7 @@ public class PipeMemoryManager {
"tryAllocate: failed to allocate memory, "
+ "total memory size {} bytes, used memory size {} bytes, "
+ "requested memory size {} bytes",
- memoryBlock.getTotalMemorySizeInBytes(),
+ getTotalNonFloatingMemorySizeInBytes(),
memoryBlock.getUsedMemoryInBytes(),
sizeInBytes);
return registerMemoryBlock(0);
@@ -440,7 +443,7 @@ public class PipeMemoryManager {
return false;
}
- if (memoryBlock.getTotalMemorySizeInBytes() -
memoryBlock.getUsedMemoryInBytes()
+ if (getTotalNonFloatingMemorySizeInBytes() -
memoryBlock.getUsedMemoryInBytes()
>= memoryInBytesNeededToBeAllocated) {
memoryBlock.forceAllocateWithoutLimitation(memoryInBytesNeededToBeAllocated);
if (block instanceof PipeTabletMemoryBlock) {
@@ -494,7 +497,7 @@ public class PipeMemoryManager {
for (final PipeMemoryBlock block : shuffledBlocks) {
if (block.shrink()) {
hasAtLeastOneBlockShrinkable = true;
- if (memoryBlock.getTotalMemorySizeInBytes() -
memoryBlock.getUsedMemoryInBytes()
+ if (getTotalNonFloatingMemorySizeInBytes() -
memoryBlock.getUsedMemoryInBytes()
>= sizeInBytes) {
return true;
}
@@ -599,7 +602,11 @@ public class PipeMemoryManager {
return memoryBlock.getFreeMemoryInBytes();
}
- public long getTotalMemorySizeInBytes() {
- return memoryBlock.getTotalMemorySizeInBytes();
+ public long getTotalNonFloatingMemorySizeInBytes() {
+ return (long) (memoryBlock.getTotalMemorySizeInBytes() * (1 -
FLOATING_MEMORY_RATIO));
+ }
+
+ public long getTotalFloatingMemorySizeInBytes() {
+ return (long) (memoryBlock.getTotalMemorySizeInBytes() *
FLOATING_MEMORY_RATIO);
}
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/tsfile/PipeTsFileResourceManager.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/tsfile/PipeTsFileResourceManager.java
index e7d649917e1..d851b58985c 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/tsfile/PipeTsFileResourceManager.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/tsfile/PipeTsFileResourceManager.java
@@ -374,7 +374,7 @@ public class PipeTsFileResourceManager {
}
}
- public long getTotalLinkedButDeletedTsfileResourceRamSize() {
+ public long getTotalLinkedButDeletedTsFileResourceRamSize() {
long totalLinkedButDeletedTsfileResourceRamSize = 0;
try {
for (final Map.Entry<String, PipeTsFileResource> resourceEntry :
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/SubscriptionPrefetchingQueueStates.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/SubscriptionPrefetchingQueueStates.java
index 8b0eb256f42..70eacd2eaaa 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/SubscriptionPrefetchingQueueStates.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/SubscriptionPrefetchingQueueStates.java
@@ -133,7 +133,7 @@ public class SubscriptionPrefetchingQueueStates {
}
private boolean isMemoryEnough() {
- return PipeDataNodeResourceManager.memory().getTotalMemorySizeInBytes()
+ return
PipeDataNodeResourceManager.memory().getTotalNonFloatingMemorySizeInBytes()
* PREFETCH_MEMORY_THRESHOLD
> PipeDataNodeResourceManager.memory().getUsedMemorySizeInBytes();
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/event/cache/SubscriptionPollResponseCache.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/event/cache/SubscriptionPollResponseCache.java
index 4a49042b9ee..55848c47c01 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/event/cache/SubscriptionPollResponseCache.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/event/cache/SubscriptionPollResponseCache.java
@@ -105,10 +105,10 @@ public class SubscriptionPollResponseCache {
private SubscriptionPollResponseCache() {
final long initMemorySizeInBytes =
- PipeDataNodeResourceManager.memory().getTotalMemorySizeInBytes() / 5;
+
PipeDataNodeResourceManager.memory().getTotalNonFloatingMemorySizeInBytes() / 5;
final long maxMemorySizeInBytes =
(long)
- (PipeDataNodeResourceManager.memory().getTotalMemorySizeInBytes()
+
(PipeDataNodeResourceManager.memory().getTotalNonFloatingMemorySizeInBytes()
*
SubscriptionConfig.getInstance().getSubscriptionCacheMemoryUsagePercentage());
// properties required by pipe memory control framework
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 9da31b3cbd7..ef289d75c19 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
@@ -217,6 +217,7 @@ public class CommonConfig {
private int pipeDataStructureTabletSizeInBytes = 2097152;
private double pipeDataStructureTabletMemoryBlockAllocationRejectThreshold =
0.4;
private double pipeDataStructureTsFileMemoryBlockAllocationRejectThreshold =
0.4;
+ private double pipeTotalFloatingMemoryProportion = 0.2;
private int pipeSubtaskExecutorBasicCheckPointIntervalByConsumedEventCount =
10_000;
private long pipeSubtaskExecutorBasicCheckPointIntervalByTimeDuration = 10 *
1000L;
@@ -784,6 +785,14 @@ public class CommonConfig {
pipeDataStructureTsFileMemoryBlockAllocationRejectThreshold;
}
+ public double getPipeTotalFloatingMemoryProportion() {
+ return pipeTotalFloatingMemoryProportion;
+ }
+
+ public void setPipeTotalFloatingMemoryProportion(double
pipeTotalFloatingMemoryProportion) {
+ this.pipeTotalFloatingMemoryProportion = pipeTotalFloatingMemoryProportion;
+ }
+
public int getPipeExtractorAssignerDisruptorRingBufferSize() {
return pipeExtractorAssignerDisruptorRingBufferSize;
}
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonDescriptor.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonDescriptor.java
index ac087dc2afd..d57c129d0a6 100644
---
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonDescriptor.java
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonDescriptor.java
@@ -303,6 +303,11 @@ public class CommonDescriptor {
"pipe_data_structure_ts_file_memory_block_allocation_reject_threshold",
String.valueOf(
config.getPipeDataStructureTsFileMemoryBlockAllocationRejectThreshold()))));
+ config.setPipeTotalFloatingMemoryProportion(
+ Double.parseDouble(
+ properties.getProperty(
+ "pipe_total_floating_memory_proportion",
+
String.valueOf(config.getPipeTotalFloatingMemoryProportion()))));
config.setPipeRealTimeQueuePollTsFileThreshold(
Integer.parseInt(
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/PipeTaskAgent.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/PipeTaskAgent.java
index 9b099cd4cdc..2e3b9d0a868 100644
---
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/PipeTaskAgent.java
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/PipeTaskAgent.java
@@ -54,6 +54,7 @@ import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
import java.util.stream.Collectors;
@@ -1079,6 +1080,18 @@ public abstract class PipeTaskAgent {
.getCommitterKey(pipeName, creationTime, regionId, restartTime);
}
+ public long getAllFloatingMemoryUsageInByte() {
+ final AtomicLong bytes = new AtomicLong(0);
+ pipeMetaKeeper
+ .getPipeMetaList()
+ .forEach(
+ pipeMeta ->
+ bytes.addAndGet(
+ ((PipeTemporaryMetaInAgent) pipeMeta.getTemporaryMeta())
+ .getFloatingMemoryUsageInByte()));
+ return bytes.get();
+ }
+
public long getFloatingMemoryUsageInByte(final String pipeName) {
final PipeMeta pipeMeta = pipeMetaKeeper.getPipeMeta(pipeName);
return pipeMeta == null
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 03d8653b4dd..a54fd96426a 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
@@ -76,6 +76,10 @@ public class PipeConfig {
return
COMMON_CONFIG.getPipeDataStructureTsFileMemoryBlockAllocationRejectThreshold();
}
+ public double getPipeTotalFloatingMemoryProportion() {
+ return COMMON_CONFIG.getPipeTotalFloatingMemoryProportion();
+ }
+
/////////////////////////////// Subtask Connector
///////////////////////////////
public int getPipeRealTimeQueuePollTsFileThreshold() {
@@ -403,6 +407,7 @@ public class PipeConfig {
LOGGER.info(
"PipeDataStructureTsFileMemoryBlockAllocationRejectThreshold: {}",
getPipeDataStructureTsFileMemoryBlockAllocationRejectThreshold());
+ LOGGER.info("PipeTotalFloatingMemoryProportion: {}",
getPipeTotalFloatingMemoryProportion());
LOGGER.info(
"PipeRealTimeQueuePollTsFileThreshold: {}",
getPipeRealTimeQueuePollTsFileThreshold());
