You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ro...@apache.org on 2025/02/17 08:58:04 UTC
(iotdb) 02/03: refactor
This is an automated email from the ASF dual-hosted git repository.
rong pushed a commit to branch flush-hook-after-watchdog
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit fefd6de8769ce775f696ca27adf2e81f25381f38
Author: Steve Yurong Su <ro...@apache.org>
AuthorDate: Mon Feb 17 16:32:04 2025 +0800
refactor
---
.../db/pipe/agent/task/PipeDataNodeTaskAgent.java | 59 +++++++++++++++++-----
.../apache/iotdb/commons/conf/CommonConfig.java | 4 +-
2 files changed, 48 insertions(+), 15 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 5ae9155cb66..39c20ecade0 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
@@ -476,7 +476,21 @@ public class PipeDataNodeTaskAgent extends PipeTaskAgent {
///////////////////////// Restart Logic /////////////////////////
public void restartAllStuckPipes() {
- removeOutdatedPipeInfoFromLastRestartTimeMap();
+ final List<String> removedPipeName = removeOutdatedPipeInfoFromLastRestartTimeMap();
+ if (!removedPipeName.isEmpty()) {
+ final long currentTime = System.currentTimeMillis();
+ LOGGER.info(
+ "Pipes {} now can dynamically adjust their extraction strategies. "
+ + "Start to flush storage engine to trigger the adjustment.",
+ removedPipeName);
+ StorageEngine.getInstance().syncCloseAllProcessor();
+ WALManager.getInstance().syncDeleteOutdatedFilesInWALNodes();
+ LOGGER.info(
+ "Finished flushing storage engine, time cost: {} ms.",
+ System.currentTimeMillis() - currentTime);
+ LOGGER.info("Skipping restarting pipes this round because of the dynamic flushing.");
+ return;
+ }
if (!tryWriteLockWithTimeOut(5)) {
return;
@@ -508,16 +522,23 @@ public class PipeDataNodeTaskAgent extends PipeTaskAgent {
stuckPipes.forEach(this::restartStuckPipe);
}
- private void removeOutdatedPipeInfoFromLastRestartTimeMap() {
+ private List<String> removeOutdatedPipeInfoFromLastRestartTimeMap() {
+ final List<String> removedPipeName = new ArrayList<>();
PIPE_NAME_TO_LAST_RESTART_TIME_MAP
.entrySet()
.removeIf(
entry -> {
final AtomicLong lastRestartTime = entry.getValue();
- return lastRestartTime == null
- || PipeConfig.getInstance().getPipeStuckRestartMinIntervalMs()
- <= System.currentTimeMillis() - lastRestartTime.get();
+ final boolean shouldRemove =
+ lastRestartTime == null
+ || PipeConfig.getInstance().getPipeStuckRestartMinIntervalMs()
+ <= System.currentTimeMillis() - lastRestartTime.get();
+ if (shouldRemove) {
+ removedPipeName.add(entry.getKey());
+ }
+ return shouldRemove;
});
+ return removedPipeName;
}
private Set<PipeMeta> findAllStuckPipes() {
@@ -537,15 +558,20 @@ public class PipeDataNodeTaskAgent extends PipeTaskAgent {
return stuckPipes;
}
- if (3 * PipeDataNodeResourceManager.tsfile().getTotalLinkedButDeletedTsfileResourceRamSize()
- >= 2 * PipeDataNodeResourceManager.memory().getFreeMemorySizeInBytes()) {
+ final long totalLinkedButDeletedTsFileResourceRamSize =
+ PipeDataNodeResourceManager.tsfile().getTotalLinkedButDeletedTsfileResourceRamSize();
+ final long freeMemorySizeInBytes =
+ PipeDataNodeResourceManager.memory().getFreeMemorySizeInBytes();
+ if (3 * totalLinkedButDeletedTsFileResourceRamSize >= 2 * freeMemorySizeInBytes) {
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 memory limit.",
- stuckPipes.size());
+ "All {} pipe(s) will be restarted because linked tsfiles' resource size {} exceeds limit {}.",
+ stuckPipes.size(),
+ totalLinkedButDeletedTsFileResourceRamSize,
+ freeMemorySizeInBytes * 2.0 / 3);
}
return stuckPipes;
}
@@ -584,16 +610,23 @@ public class PipeDataNodeTaskAgent extends PipeTaskAgent {
|| mayWalSizeReachThrottleThreshold())) {
// Extractors of this pipe may be stuck and is pinning too many MemTables.
LOGGER.warn(
- "Pipe {} needs to restart because too many memTables are pinned.",
- pipeMeta.getStaticMeta());
+ "Pipe {} needs to restart because too many memtables are pinned. mayMemTablePinnedCountReachDangerousThreshold: {}, mayWalSizeReachThrottleThreshold: {}",
+ pipeMeta.getStaticMeta(),
+ 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.",
- pipeMeta.getStaticMeta());
+ "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/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 5a08596d62c..3e8698af3d5 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
@@ -256,8 +256,8 @@ public class CommonConfig {
private long pipeReceiverLoginPeriodicVerificationIntervalMs = 300000;
private int pipeMaxAllowedHistoricalTsFilePerDataRegion = 100;
- private int pipeMaxAllowedPendingTsFileEpochPerDataRegion = 2;
- private int pipeMaxAllowedPinnedMemTableCount = 10000;
+ private int pipeMaxAllowedPendingTsFileEpochPerDataRegion = 10;
+ private int pipeMaxAllowedPinnedMemTableCount = 1000;
private long pipeMaxAllowedLinkedTsFileCount = 100;
private float pipeMaxAllowedLinkedDeletedTsFileDiskUsagePercentage = 0.1F;
private long pipeStuckRestartIntervalSeconds = 120;