This is an automated email from the ASF dual-hosted git repository.
jt2594838 pushed a commit to branch dev/1.3
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/dev/1.3 by this push:
new 6527fb0e8ca [Pipe] Wake parser workers when admission is available
(#18843) (#18865)
6527fb0e8ca is described below
commit 6527fb0e8ca471f49747c38acfed96d0a20e1163
Author: Caideyipi <[email protected]>
AuthorDate: Wed Oct 14 16:12:37 2026 +0800
[Pipe] Wake parser workers when admission is available (#18843) (#18865)
* Fix pipe parser admission wake-ups
* Avoid reflective test dependency
(cherry picked from commit aec1ad6139cb4dd1450725f53418cf40c26efb4e)
---
.../subtask/processor/PipeProcessorSubtask.java | 4 +
.../PipeProcessorSubtaskExecutionGuard.java | 16 +++
.../processor/PipeProcessorSubtaskWorker.java | 29 +++-
.../common/tsfile/PipeTsFileInsertionEvent.java | 4 +
.../pipe/metric/overview/PipeResourceMetrics.java | 47 +++++++
.../db/pipe/resource/memory/PipeMemoryManager.java | 154 +++++++++++++++------
.../PipeProcessorSubtaskExecutionGuardTest.java | 10 +-
.../processor/PipeProcessorSubtaskWorkerTest.java | 154 +++++++++++++++++++++
.../resource/memory/PipeMemoryManagerTest.java | 30 ++++
.../iotdb/commons/service/metric/enums/Metric.java | 5 +
10 files changed, 403 insertions(+), 50 deletions(-)
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtask.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtask.java
index 84485bc0c04..6e20498b784 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtask.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtask.java
@@ -271,6 +271,10 @@ public class PipeProcessorSubtask extends
PipeReportableSubtask {
// and the worker will be submitted to the executor
}
+ void setWorkerWakeUpCallback(final Runnable wakeUpCallback) {
+ executionGuard.setWorkerWakeUpCallback(wakeUpCallback);
+ }
+
@Override
protected void onAllowSubmittingSelf() {
executionGuard.start();
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtaskExecutionGuard.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtaskExecutionGuard.java
index a4eeab3482d..8510f58ab08 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtaskExecutionGuard.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtaskExecutionGuard.java
@@ -19,6 +19,8 @@
package org.apache.iotdb.db.pipe.agent.task.subtask.processor;
+import org.apache.iotdb.db.pipe.metric.overview.PipeResourceMetrics;
+
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
@@ -38,6 +40,7 @@ public class PipeProcessorSubtaskExecutionGuard {
private final AtomicBoolean isRunning = new AtomicBoolean(false);
private final AtomicLong executionEpoch = new AtomicLong(0);
private final ThreadLocal<Long> invocationEpoch = new ThreadLocal<>();
+ private volatile Runnable workerWakeUpCallback;
public PipeProcessorSubtaskExecutionGuard() {
this(true);
@@ -55,9 +58,21 @@ public class PipeProcessorSubtaskExecutionGuard {
return enabled;
}
+ void setWorkerWakeUpCallback(final Runnable workerWakeUpCallback) {
+ this.workerWakeUpCallback = workerWakeUpCallback;
+ }
+
+ public Runnable getWorkerWakeUpCallback() {
+ return workerWakeUpCallback;
+ }
+
void start() {
if (enabled) {
isRunning.set(true);
+ final Runnable wakeUpCallback = workerWakeUpCallback;
+ if (wakeUpCallback != null) {
+ wakeUpCallback.run();
+ }
}
}
@@ -105,6 +120,7 @@ public class PipeProcessorSubtaskExecutionGuard {
}
public void yieldIfParserNotAdmitted() {
+ PipeResourceMetrics.getInstance().recordTsFileParserAdmissionYield();
throw PipeProcessorSubtaskYieldException.parserNotAdmitted();
}
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtaskWorker.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtaskWorker.java
index ad1548d9432..4572d217e91 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtaskWorker.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtaskWorker.java
@@ -49,6 +49,9 @@ public class PipeProcessorSubtaskWorker extends
WrappedRunnable {
private long sleepingTimeInMilliSecond = 50;
private final Set<PipeProcessorSubtask> subtasks;
+ private final Object wakeUpMonitor = new Object();
+ private final Runnable wakeUpCallback = this::wakeUp;
+ private boolean isWakeUpRequested;
private volatile Thread workerThread;
private volatile PipeProcessorSubtask currentSubtask;
@@ -115,10 +118,18 @@ public class PipeProcessorSubtaskWorker extends
WrappedRunnable {
return canSleepBeforeNextRound;
}
- private void sleepIfNecessary(final boolean canSleepBeforeNextRound) {
+ @VisibleForTesting
+ void sleepIfNecessary(final boolean canSleepBeforeNextRound) {
if (canSleepBeforeNextRound) {
try {
- Thread.sleep(sleepingTimeInMilliSecond);
+ synchronized (wakeUpMonitor) {
+ // A parser can become eligible before this worker starts waiting.
Keep that notification
+ // until the idle round consumes it, instead of sleeping with the
fair queue head ready.
+ if (!isWakeUpRequested) {
+ wakeUpMonitor.wait(sleepingTimeInMilliSecond);
+ }
+ isWakeUpRequested = false;
+ }
} catch (final InterruptedException e) {
LOGGER.warn("subtask worker is interrupted", e);
Thread.currentThread().interrupt();
@@ -128,6 +139,11 @@ public class PipeProcessorSubtaskWorker extends
WrappedRunnable {
}
}
+ @VisibleForTesting
+ void setSleepingTimeInMilliSecond(final long sleepingTimeInMilliSecond) {
+ this.sleepingTimeInMilliSecond = sleepingTimeInMilliSecond;
+ }
+
private void adjustSleepingTimeIfNecessary() {
if (totalRoundInAdjustmentInterval %
SLEEP_INTERVAL_ADJUSTMENT_ROUND_INTERVAL == 0) {
final double workingRatioInAdjustmentInterval =
@@ -147,7 +163,16 @@ public class PipeProcessorSubtaskWorker extends
WrappedRunnable {
}
public void schedule(final PipeProcessorSubtask pipeProcessorSubtask) {
+ pipeProcessorSubtask.setWorkerWakeUpCallback(wakeUpCallback);
subtasks.add(pipeProcessorSubtask);
+ wakeUp();
+ }
+
+ private void wakeUp() {
+ synchronized (wakeUpMonitor) {
+ isWakeUpRequested = true;
+ wakeUpMonitor.notifyAll();
+ }
}
void watchLongRunningEvent() {
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 6258dfee1fd..0d379db7831 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
@@ -923,6 +923,10 @@ public class PipeTsFileInsertionEvent extends EnrichedEvent
final PipeProcessorSubtaskExecutionGuard processorExecutionGuard,
final long expectedParserStateGeneration)
throws InterruptedException {
+ // Register before enqueueing: memory may be released as soon as admission
fails, before this
+ // invocation yields or its worker enters the idle wait.
+ tsFileParserMemoryReservationKey.setWakeUpCallback(
+ processorExecutionGuard.getWorkerWakeUpCallback());
if (!processorExecutionGuard.isEnabled()) {
return waitForResourceEnough4Parsing(
(long) ((1 + Math.random()) * 20 * 1000),
expectedParserStateGeneration);
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 dd41d032f80..85ce5047b85 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
@@ -29,9 +29,12 @@ import org.apache.iotdb.metrics.AbstractMetricService;
import org.apache.iotdb.metrics.impl.DoNothingMetricManager;
import org.apache.iotdb.metrics.metricsets.IMetricSet;
import org.apache.iotdb.metrics.type.Counter;
+import org.apache.iotdb.metrics.type.Timer;
import org.apache.iotdb.metrics.utils.MetricLevel;
import org.apache.iotdb.metrics.utils.MetricType;
+import java.util.concurrent.TimeUnit;
+
public class PipeResourceMetrics implements IMetricSet {
private static final String PIPE_USED_MEMORY = "PipeUsedMemory";
@@ -44,6 +47,11 @@ public class PipeResourceMetrics implements IMetricSet {
private static final String PIPE_TOTAL_MEMORY = "PipeTotalMemory";
private static final String PIPE_FLOATING_MEMORY = "PipeFloatingMemory";
private Counter diskIOCounter = DoNothingMetricManager.DO_NOTHING_COUNTER;
+ private volatile Counter tsFileParserAdmissionRejectCounter =
+ DoNothingMetricManager.DO_NOTHING_COUNTER;
+ private volatile Counter tsFileParserAdmissionYieldCounter =
+ DoNothingMetricManager.DO_NOTHING_COUNTER;
+ private volatile Timer tsFileParserAdmissionWaitTimer =
DoNothingMetricManager.DO_NOTHING_TIMER;
//////////////////////////// bindTo & unbindFrom (metric framework)
////////////////////////////
@@ -102,6 +110,25 @@ public class PipeResourceMetrics implements IMetricSet {
diskIOCounter =
metricService.getOrCreateCounter(
Metric.PIPE_TSFILE_SEND_DISK_IO.toString(), MetricLevel.IMPORTANT);
+ tsFileParserAdmissionRejectCounter =
+ metricService.getOrCreateCounter(
+ Metric.PIPE_TSFILE_PARSER_ADMISSION_REJECT.toString(),
MetricLevel.IMPORTANT);
+ tsFileParserAdmissionYieldCounter =
+ metricService.getOrCreateCounter(
+ Metric.PIPE_TSFILE_PARSER_ADMISSION_YIELD.toString(),
MetricLevel.IMPORTANT);
+ tsFileParserAdmissionWaitTimer =
+ metricService.getOrCreateTimer(
+ Metric.PIPE_TSFILE_PARSER_ADMISSION_WAIT.toString(),
MetricLevel.IMPORTANT);
+ metricService.createAutoGauge(
+ Metric.PIPE_TSFILE_PARSER_IN_FLIGHT.toString(),
+ MetricLevel.IMPORTANT,
+ PipeDataNodeResourceManager.memory(),
+ PipeMemoryManager::getReservedTsFileParserCount);
+ metricService.createAutoGauge(
+ Metric.PIPE_TSFILE_PARSER_WAITING.toString(),
+ MetricLevel.IMPORTANT,
+ PipeDataNodeResourceManager.memory(),
+ PipeMemoryManager::getWaitingTsFileParserCount);
}
@Override
@@ -138,12 +165,32 @@ public class PipeResourceMetrics implements IMetricSet {
metricService.remove(MetricType.AUTO_GAUGE,
Metric.PIPE_PHANTOM_REFERENCE_COUNT.toString());
metricService.remove(MetricType.COUNTER,
Metric.PIPE_TSFILE_SEND_DISK_IO.toString());
+ metricService.remove(MetricType.COUNTER,
Metric.PIPE_TSFILE_PARSER_ADMISSION_REJECT.toString());
+ metricService.remove(MetricType.COUNTER,
Metric.PIPE_TSFILE_PARSER_ADMISSION_YIELD.toString());
+ metricService.remove(MetricType.TIMER,
Metric.PIPE_TSFILE_PARSER_ADMISSION_WAIT.toString());
+ metricService.remove(MetricType.AUTO_GAUGE,
Metric.PIPE_TSFILE_PARSER_IN_FLIGHT.toString());
+ metricService.remove(MetricType.AUTO_GAUGE,
Metric.PIPE_TSFILE_PARSER_WAITING.toString());
+ tsFileParserAdmissionRejectCounter =
DoNothingMetricManager.DO_NOTHING_COUNTER;
+ tsFileParserAdmissionYieldCounter =
DoNothingMetricManager.DO_NOTHING_COUNTER;
+ tsFileParserAdmissionWaitTimer = DoNothingMetricManager.DO_NOTHING_TIMER;
}
public void recordDiskIO(final long bytes) {
diskIOCounter.inc(bytes);
}
+ public void recordTsFileParserAdmissionReject() {
+ tsFileParserAdmissionRejectCounter.inc();
+ }
+
+ public void recordTsFileParserAdmissionYield() {
+ tsFileParserAdmissionYieldCounter.inc();
+ }
+
+ public void recordTsFileParserAdmissionWait(final long waitTimeInNanos) {
+ tsFileParserAdmissionWaitTimer.update(waitTimeInNanos,
TimeUnit.NANOSECONDS);
+ }
+
//////////////////////////// singleton ////////////////////////////
private static class PipeResourceMetricsHolder {
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 5f03991d350..8fc3d49536a 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
@@ -23,6 +23,7 @@ import
org.apache.iotdb.commons.exception.pipe.PipeRuntimeOutOfMemoryCriticalExc
import org.apache.iotdb.commons.pipe.config.PipeConfig;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.pipe.agent.PipeDataNodeAgent;
+import org.apache.iotdb.db.pipe.metric.overview.PipeResourceMetrics;
import
org.apache.iotdb.db.pipe.resource.memory.strategy.ThresholdAllocationStrategy;
import org.slf4j.Logger;
@@ -66,6 +67,7 @@ public class PipeMemoryManager {
private volatile long usedMemorySizeInBytesOfTsFiles;
private volatile long reservedTsFileParserCount;
+ private volatile long waitingTsFileParserCount;
private final Map<PipeIdentity, Integer> reservedTsFileParserCountByPipe =
new HashMap<>();
private final Map<PipeRegionIdentity, Integer>
reservedTsFileParserCountByPipeRegion =
@@ -109,30 +111,25 @@ public class 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 double allowedMaxMemorySizeInBytesOfTabletsAndTsFiles() {
- return (PipeConfig.getInstance()
-
.getPipeDataStructureTabletMemoryBlockAllocationRejectThreshold()
- + PipeConfig.getInstance()
-
.getPipeDataStructureTsFileMemoryBlockAllocationRejectThreshold())
- * getTotalNonFloatingMemorySizeInBytes();
+ private double allowedMaxMemorySizeInBytesOfTabletsAndTsFiles(
+ final long totalNonFloatingMemorySizeInBytes) {
+ return
(PIPE_CONFIG.getPipeDataStructureTabletMemoryBlockAllocationRejectThreshold()
+ +
PIPE_CONFIG.getPipeDataStructureTsFileMemoryBlockAllocationRejectThreshold())
+ * totalNonFloatingMemorySizeInBytes;
}
- private double allowedMaxMemorySizeInBytesOfTablets() {
- return (PipeConfig.getInstance()
-
.getPipeDataStructureTabletMemoryBlockAllocationRejectThreshold()
- + PipeConfig.getInstance()
-
.getPipeDataStructureTsFileMemoryBlockAllocationRejectThreshold()
- / 2)
- * getTotalNonFloatingMemorySizeInBytes();
+ private double allowedMaxMemorySizeInBytesOfTablets(
+ final long totalNonFloatingMemorySizeInBytes) {
+ return
(PIPE_CONFIG.getPipeDataStructureTabletMemoryBlockAllocationRejectThreshold()
+ +
PIPE_CONFIG.getPipeDataStructureTsFileMemoryBlockAllocationRejectThreshold() /
2)
+ * totalNonFloatingMemorySizeInBytes;
}
- private double allowedMaxMemorySizeInBytesOfTsTiles() {
- return (PipeConfig.getInstance()
-
.getPipeDataStructureTsFileMemoryBlockAllocationRejectThreshold()
- + PipeConfig.getInstance()
-
.getPipeDataStructureTabletMemoryBlockAllocationRejectThreshold()
- / 2)
- * getTotalNonFloatingMemorySizeInBytes();
+ private double allowedMaxMemorySizeInBytesOfTsTiles(
+ final long totalNonFloatingMemorySizeInBytes) {
+ return
(PIPE_CONFIG.getPipeDataStructureTsFileMemoryBlockAllocationRejectThreshold()
+ +
PIPE_CONFIG.getPipeDataStructureTabletMemoryBlockAllocationRejectThreshold() /
2)
+ * totalNonFloatingMemorySizeInBytes;
}
private static long getTsFileParserMemorySizeInBytes() {
@@ -144,39 +141,48 @@ public class PipeMemoryManager {
return reservedTsFileParserCount * getTsFileParserMemorySizeInBytes();
}
- private boolean isEnough4TabletParsingWithReservedParserMemory(final long
extraMemoryInBytes) {
+ private boolean isEnough4TabletParsingWithReservedParserMemory(
+ final long extraMemoryInBytes, final long
totalNonFloatingMemorySizeInBytes) {
final double tabletMemoryWithParserMemory =
(double) usedMemorySizeInBytesOfTablets
+ getReservedTsFileParserMemorySizeInBytes()
+ extraMemoryInBytes;
return tabletMemoryWithParserMemory + (double)
usedMemorySizeInBytesOfTsFiles
- < EXCEED_PROTECT_THRESHOLD *
allowedMaxMemorySizeInBytesOfTabletsAndTsFiles()
+ < EXCEED_PROTECT_THRESHOLD
+ *
allowedMaxMemorySizeInBytesOfTabletsAndTsFiles(totalNonFloatingMemorySizeInBytes)
&& tabletMemoryWithParserMemory
- < EXCEED_PROTECT_THRESHOLD *
allowedMaxMemorySizeInBytesOfTablets();
+ < EXCEED_PROTECT_THRESHOLD
+ *
allowedMaxMemorySizeInBytesOfTablets(totalNonFloatingMemorySizeInBytes);
}
private boolean isHardEnough4TabletParsingWithReservedParserMemory(
- final long extraMemoryInBytes) {
+ final long extraMemoryInBytes, final long
totalNonFloatingMemorySizeInBytes) {
final double tabletMemoryWithParserMemory =
(double) usedMemorySizeInBytesOfTablets
+ getReservedTsFileParserMemorySizeInBytes()
+ extraMemoryInBytes;
return tabletMemoryWithParserMemory + (double)
usedMemorySizeInBytesOfTsFiles
- < allowedMaxMemorySizeInBytesOfTabletsAndTsFiles()
- && tabletMemoryWithParserMemory <
allowedMaxMemorySizeInBytesOfTablets();
+ <
allowedMaxMemorySizeInBytesOfTabletsAndTsFiles(totalNonFloatingMemorySizeInBytes)
+ && tabletMemoryWithParserMemory
+ <
allowedMaxMemorySizeInBytesOfTablets(totalNonFloatingMemorySizeInBytes);
}
public boolean isEnough4TabletParsing() {
+ final long totalNonFloatingMemorySizeInBytes =
getTotalNonFloatingMemorySizeInBytes();
return (double) usedMemorySizeInBytesOfTablets + (double)
usedMemorySizeInBytesOfTsFiles
- < EXCEED_PROTECT_THRESHOLD *
allowedMaxMemorySizeInBytesOfTabletsAndTsFiles()
+ < EXCEED_PROTECT_THRESHOLD
+ *
allowedMaxMemorySizeInBytesOfTabletsAndTsFiles(totalNonFloatingMemorySizeInBytes)
&& (double) usedMemorySizeInBytesOfTablets
- < EXCEED_PROTECT_THRESHOLD *
allowedMaxMemorySizeInBytesOfTablets();
+ < EXCEED_PROTECT_THRESHOLD
+ *
allowedMaxMemorySizeInBytesOfTablets(totalNonFloatingMemorySizeInBytes);
}
private boolean isHardEnough4TabletParsing() {
+ final long totalNonFloatingMemorySizeInBytes =
getTotalNonFloatingMemorySizeInBytes();
return (double) usedMemorySizeInBytesOfTablets + (double)
usedMemorySizeInBytesOfTsFiles
- < allowedMaxMemorySizeInBytesOfTabletsAndTsFiles()
- && (double) usedMemorySizeInBytesOfTablets <
allowedMaxMemorySizeInBytesOfTablets();
+ <
allowedMaxMemorySizeInBytesOfTabletsAndTsFiles(totalNonFloatingMemorySizeInBytes)
+ && (double) usedMemorySizeInBytesOfTablets
+ <
allowedMaxMemorySizeInBytesOfTablets(totalNonFloatingMemorySizeInBytes);
}
public synchronized boolean tryReserveTsFileParserMemory(
@@ -202,16 +208,24 @@ public class PipeMemoryManager {
reservedTsFileParserCountByPipeRegion.getOrDefault(pipeRegionIdentity,
0);
if (reservedTsFileParserCount >= globalLimit
|| reservedCountOfPipeRegion >= perPipeRegionLimit) {
+ PipeResourceMetrics.getInstance().recordTsFileParserAdmissionReject();
notifyNextTsFileParserMemoryReservationInternal();
return false;
}
final long parserMemorySizeInBytes = getTsFileParserMemorySizeInBytes();
+ // Reuse one live snapshot for both thresholds. The supplier traverses all
pipe metadata, so
+ // repeatedly sampling it while holding this monitor amplifies contention
with many pipes.
+ final long totalNonFloatingMemorySizeInBytes =
+ PIPE_MEMORY_MANAGEMENT_ENABLED ?
getTotalNonFloatingMemorySizeInBytes() : 0;
final boolean isSoftMemoryEnough =
!PIPE_MEMORY_MANAGEMENT_ENABLED
- ||
isEnough4TabletParsingWithReservedParserMemory(parserMemorySizeInBytes);
+ || isEnough4TabletParsingWithReservedParserMemory(
+ parserMemorySizeInBytes, totalNonFloatingMemorySizeInBytes);
if (!isSoftMemoryEnough
- &&
!isHardEnough4TabletParsingWithReservedParserMemory(parserMemorySizeInBytes)) {
+ && !isHardEnough4TabletParsingWithReservedParserMemory(
+ parserMemorySizeInBytes, totalNonFloatingMemorySizeInBytes)) {
+ PipeResourceMetrics.getInstance().recordTsFileParserAdmissionReject();
return false;
}
@@ -222,13 +236,17 @@ public class PipeMemoryManager {
if (!pipeRegionIdentity.equals(nextPipeRegion)
|| requestsOfPipeRegion == null
|| !reservationKey.equals(requestsOfPipeRegion.iterator().next())) {
- notifyNextTsFileParserMemoryReservationInternal();
+ PipeResourceMetrics.getInstance().recordTsFileParserAdmissionReject();
+ signalTsFileParserMemoryReservation(nextPipeRegion);
return false;
}
removeTsFileParserReservationRequest(pipeRegionIdentity, reservationKey,
true);
if (wasRequestAlreadyWaiting) {
lastAdmittedWaitingTsFileParserPipe = pipeIdentity;
+ PipeResourceMetrics.getInstance()
+ .recordTsFileParserAdmissionWait(
+ System.nanoTime() - reservationKey.waitingStartTimeInNanos);
}
reservedTsFileParserCount++;
reservedTsFileParserCountByPipe.merge(pipeIdentity, 1, Integer::sum);
@@ -301,7 +319,12 @@ public class PipeMemoryManager {
regionOrder.addLast(key);
return new LinkedHashSet<>();
});
- return !requestsOfPipeRegion.add(reservationKey);
+ if (requestsOfPipeRegion.add(reservationKey)) {
+ waitingTsFileParserCount++;
+ reservationKey.waitingStartTimeInNanos = System.nanoTime();
+ return false;
+ }
+ return true;
}
public synchronized void notifyNextTsFileParserMemoryReservation() {
@@ -310,16 +333,20 @@ public class PipeMemoryManager {
private void notifyNextTsFileParserMemoryReservationInternal() {
final int globalLimit = Math.max(1,
PIPE_CONFIG.getPipeTsFileParserInFlightMaxNum());
- if (reservedTsFileParserCount >= globalLimit) {
+ if (waitingTsFileParserPipeOrder.isEmpty() || reservedTsFileParserCount >=
globalLimit) {
return;
}
final long parserMemorySizeInBytes = getTsFileParserMemorySizeInBytes();
+ final long totalNonFloatingMemorySizeInBytes =
+ PIPE_MEMORY_MANAGEMENT_ENABLED ?
getTotalNonFloatingMemorySizeInBytes() : 0;
final boolean isSoftMemoryEnough =
!PIPE_MEMORY_MANAGEMENT_ENABLED
- ||
isEnough4TabletParsingWithReservedParserMemory(parserMemorySizeInBytes);
+ || isEnough4TabletParsingWithReservedParserMemory(
+ parserMemorySizeInBytes, totalNonFloatingMemorySizeInBytes);
if (!isSoftMemoryEnough
- &&
!isHardEnough4TabletParsingWithReservedParserMemory(parserMemorySizeInBytes)) {
+ && !isHardEnough4TabletParsingWithReservedParserMemory(
+ parserMemorySizeInBytes, totalNonFloatingMemorySizeInBytes)) {
return;
}
@@ -328,6 +355,10 @@ public class PipeMemoryManager {
1, Math.min(globalLimit,
PIPE_CONFIG.getPipeTsFileParserInFlightMaxNumPerPipeRegion()));
final PipeRegionIdentity nextPipeRegion =
getNextEligibleTsFileParserPipeRegion(perPipeRegionLimit,
!isSoftMemoryEnough);
+ signalTsFileParserMemoryReservation(nextPipeRegion);
+ }
+
+ private void signalTsFileParserMemoryReservation(final PipeRegionIdentity
nextPipeRegion) {
if (nextPipeRegion == null) {
return;
}
@@ -398,6 +429,7 @@ public class PipeMemoryManager {
if (requestsOfPipeRegion == null ||
!requestsOfPipeRegion.remove(reservationKey)) {
return;
}
+ waitingTsFileParserCount--;
final PipeIdentity pipeIdentity = pipeRegionIdentity.pipeIdentity;
final ArrayDeque<PipeRegionIdentity> regionOrder =
@@ -431,6 +463,14 @@ public class PipeMemoryManager {
public static final class TsFileParserMemoryReservation {
private boolean isSignaled;
+ private volatile Runnable wakeUpCallback;
+ // Accessed only under the PipeMemoryManager monitor while this request is
queued.
+ private long waitingStartTimeInNanos;
+
+ /** The callback must only wake the worker; it runs while the memory
manager monitor is held. */
+ public void setWakeUpCallback(final Runnable wakeUpCallback) {
+ this.wakeUpCallback = wakeUpCallback;
+ }
public synchronized void await(final long timeoutInMs) throws
InterruptedException {
if (!isSignaled) {
@@ -442,6 +482,10 @@ public class PipeMemoryManager {
private synchronized void signal() {
isSignaled = true;
notify();
+ final Runnable callback = wakeUpCallback;
+ if (callback != null) {
+ callback.run();
+ }
}
}
@@ -464,7 +508,8 @@ public class PipeMemoryManager {
return elapsedTimeInMs >= maxRetryTimeInMs;
}
- if (!isHardEnough4TabletParsingWithReservedParserMemory(0)) {
+ if (!isHardEnough4TabletParsingWithReservedParserMemory(
+ 0, getTotalNonFloatingMemorySizeInBytes())) {
return true;
}
@@ -473,35 +518,44 @@ public class PipeMemoryManager {
}
public boolean isEnough4TsFileSlicing() {
+ final long totalNonFloatingMemorySizeInBytes =
getTotalNonFloatingMemorySizeInBytes();
return (double) usedMemorySizeInBytesOfTablets + (double)
usedMemorySizeInBytesOfTsFiles
- < EXCEED_PROTECT_THRESHOLD *
allowedMaxMemorySizeInBytesOfTabletsAndTsFiles()
+ < EXCEED_PROTECT_THRESHOLD
+ *
allowedMaxMemorySizeInBytesOfTabletsAndTsFiles(totalNonFloatingMemorySizeInBytes)
&& (double) usedMemorySizeInBytesOfTsFiles
- < EXCEED_PROTECT_THRESHOLD *
allowedMaxMemorySizeInBytesOfTsTiles();
+ < EXCEED_PROTECT_THRESHOLD
+ *
allowedMaxMemorySizeInBytesOfTsTiles(totalNonFloatingMemorySizeInBytes);
}
private boolean isHardEnough4TsFileSlicing() {
+ final long totalNonFloatingMemorySizeInBytes =
getTotalNonFloatingMemorySizeInBytes();
return (double) usedMemorySizeInBytesOfTablets + (double)
usedMemorySizeInBytesOfTsFiles
- < allowedMaxMemorySizeInBytesOfTabletsAndTsFiles()
- && (double) usedMemorySizeInBytesOfTsFiles <
allowedMaxMemorySizeInBytesOfTsTiles();
+ <
allowedMaxMemorySizeInBytesOfTabletsAndTsFiles(totalNonFloatingMemorySizeInBytes)
+ && (double) usedMemorySizeInBytesOfTsFiles
+ <
allowedMaxMemorySizeInBytesOfTsTiles(totalNonFloatingMemorySizeInBytes);
}
private boolean isHardEnoughForResizing(
final PipeMemoryBlock block, final long extraMemoryInBytes) {
+ if (!(block instanceof PipeTabletMemoryBlock) && !(block instanceof
PipeTsFileMemoryBlock)) {
+ return true;
+ }
+ final long totalNonFloatingMemorySizeInBytes =
getTotalNonFloatingMemorySizeInBytes();
if (block instanceof PipeTabletMemoryBlock) {
return (double) usedMemorySizeInBytesOfTablets
+ (double) extraMemoryInBytes
+ (double) usedMemorySizeInBytesOfTsFiles
- < allowedMaxMemorySizeInBytesOfTabletsAndTsFiles()
+ <
allowedMaxMemorySizeInBytesOfTabletsAndTsFiles(totalNonFloatingMemorySizeInBytes)
&& (double) usedMemorySizeInBytesOfTablets + (double)
extraMemoryInBytes
- < allowedMaxMemorySizeInBytesOfTablets();
+ <
allowedMaxMemorySizeInBytesOfTablets(totalNonFloatingMemorySizeInBytes);
}
if (block instanceof PipeTsFileMemoryBlock) {
return (double) usedMemorySizeInBytesOfTablets
+ (double) usedMemorySizeInBytesOfTsFiles
+ (double) extraMemoryInBytes
- < allowedMaxMemorySizeInBytesOfTabletsAndTsFiles()
+ <
allowedMaxMemorySizeInBytesOfTabletsAndTsFiles(totalNonFloatingMemorySizeInBytes)
&& (double) usedMemorySizeInBytesOfTsFiles + (double)
extraMemoryInBytes
- < allowedMaxMemorySizeInBytesOfTsTiles();
+ <
allowedMaxMemorySizeInBytesOfTsTiles(totalNonFloatingMemorySizeInBytes);
}
return true;
}
@@ -1077,6 +1131,14 @@ public class PipeMemoryManager {
return usedMemorySizeInBytesOfTsFiles;
}
+ public long getReservedTsFileParserCount() {
+ return reservedTsFileParserCount;
+ }
+
+ public long getWaitingTsFileParserCount() {
+ return waitingTsFileParserCount;
+ }
+
public long getFreeMemorySizeInBytes() {
return Math.max(0, getTotalNonFloatingMemorySizeInBytes() -
usedMemorySizeInBytes);
}
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtaskExecutionGuardTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtaskExecutionGuardTest.java
index b7c5fdc5542..a9e49892e83 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtaskExecutionGuardTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtaskExecutionGuardTest.java
@@ -90,7 +90,10 @@ public class PipeProcessorSubtaskExecutionGuardTest {
final PipeTsFileInsertionEvent eventToConsume = event;
final PipeProcessorSubtaskExecutionGuard executionGuard =
new PipeProcessorSubtaskExecutionGuard();
+ final AtomicInteger wakeUpCount = new AtomicInteger();
+ executionGuard.setWorkerWakeUpCallback(wakeUpCount::incrementAndGet);
executionGuard.start();
+ wakeUpCount.set(0);
executionGuard.enter();
final long startTimeInNanos = System.nanoTime();
@@ -108,10 +111,12 @@ public class PipeProcessorSubtaskExecutionGuardTest {
Assert.assertTrue(TimeUnit.NANOSECONDS.toMillis(System.nanoTime() -
startTimeInNanos) < 1000);
executionGuard.exit();
- executionGuard.stop();
- event.cancelTsFileParserMemoryReservationIfPending();
+ // A yielded invocation must keep its fair queue position and receive
the release signal.
memoryManager.releaseTsFileParserMemory("blocker", 0, "0");
isBlockerReserved = false;
+ Assert.assertEquals(1, wakeUpCount.get());
+ executionGuard.stop();
+ event.cancelTsFileParserMemoryReservationIfPending();
isCompetitorReserved =
memoryManager.tryReserveTsFileParserMemory("competitor", 0, "0",
competitorReservation);
Assert.assertTrue(isCompetitorReserved);
@@ -120,6 +125,7 @@ public class PipeProcessorSubtaskExecutionGuardTest {
final AtomicInteger consumedTabletCount = new AtomicInteger(0);
executionGuard.start();
+ Assert.assertEquals(2, wakeUpCount.get());
executionGuard.enter();
event.consumeTabletInsertionEventsWithRetry(
parsedEvent -> {
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtaskWorkerTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtaskWorkerTest.java
index d1c34ccf471..d1d24c11837 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtaskWorkerTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtaskWorkerTest.java
@@ -19,7 +19,12 @@
package org.apache.iotdb.db.pipe.agent.task.subtask.processor;
+import org.apache.iotdb.commons.conf.CommonConfig;
+import org.apache.iotdb.commons.conf.CommonDescriptor;
import org.apache.iotdb.commons.pipe.event.EnrichedEvent;
+import org.apache.iotdb.db.pipe.resource.PipeDataNodeResourceManager;
+import org.apache.iotdb.db.pipe.resource.memory.PipeMemoryManager;
+import
org.apache.iotdb.db.pipe.resource.memory.PipeMemoryManager.TsFileParserMemoryReservation;
import com.google.common.util.concurrent.ListeningExecutorService;
import com.google.common.util.concurrent.ListeningScheduledExecutorService;
@@ -28,12 +33,22 @@ import org.junit.Assert;
import org.junit.Test;
import org.mockito.InOrder;
+import java.util.ArrayList;
import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
+import static org.awaitility.Awaitility.await;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.atLeastOnce;
+import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.inOrder;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
@@ -42,6 +57,145 @@ import static org.mockito.Mockito.when;
public class PipeProcessorSubtaskWorkerTest {
+ @Test(timeout = 30000)
+ public void testOneHundredWaitingPipesDoNotWaitForWorkerPolling() throws
Exception {
+ final CommonConfig config = CommonDescriptor.getInstance().getConfig();
+ final PipeMemoryManager memoryManager =
PipeDataNodeResourceManager.memory();
+ final long originalParserMemory = config.getPipeTsFileParserMemory();
+ final int originalGlobalLimit = config.getPipeTsFileParserInFlightMaxNum();
+ final int originalRegionLimit =
config.getPipeTsFileParserInFlightMaxNumPerPipeRegion();
+ final List<PipeProcessorSubtaskWorker> workers = new ArrayList<>();
+ final List<AtomicReference<Thread>> workerThreads = new ArrayList<>();
+ final List<TsFileParserMemoryReservation> requests = new ArrayList<>();
+ final List<Future<?>> futures = new ArrayList<>();
+ final CountDownLatch processed = new CountDownLatch(100);
+ final ExecutorService executor = Executors.newFixedThreadPool(5);
+ final boolean[] reservedBlockers = new boolean[4];
+ try {
+ config.setPipeTsFileParserMemory(1);
+ config.setPipeTsFileParserInFlightMaxNum(4);
+ config.setPipeTsFileParserInFlightMaxNumPerPipeRegion(1);
+ for (int i = 0; i < 4; i++) {
+ Assert.assertTrue(
+ memoryManager.tryReserveTsFileParserMemory(
+ "blocker" + i, 0, "0", new TsFileParserMemoryReservation()));
+ reservedBlockers[i] = true;
+ }
+ for (int i = 0; i < 5; i++) {
+ final PipeProcessorSubtaskWorker worker =
+ new PipeProcessorSubtaskWorker(new LinkedHashSet<>());
+ workers.add(worker);
+ workerThreads.add(new AtomicReference<>());
+ }
+ for (int i = 0; i < 100; i++) {
+ final String pipeName = "waitingPipe" + i;
+ final TsFileParserMemoryReservation request = new
TsFileParserMemoryReservation();
+ requests.add(request);
+ final PipeProcessorSubtask subtask = createRunnableSubtask(pipeName);
+ doAnswer(
+ invocation -> {
+ request.setWakeUpCallback(invocation.getArgument(0));
+ return null;
+ })
+ .when(subtask)
+ .setWorkerWakeUpCallback(any());
+ final AtomicBoolean hasProcessed = new AtomicBoolean();
+ when(subtask.call())
+ .thenAnswer(
+ invocation -> {
+ if (hasProcessed.get()) {
+ return false;
+ }
+ if (!memoryManager.tryReserveTsFileParserMemory(pipeName, 0,
"0", request)) {
+ throw
PipeProcessorSubtaskYieldException.parserNotAdmitted();
+ }
+ hasProcessed.set(true);
+ memoryManager.releaseTsFileParserMemory(pipeName, 0, "0");
+ processed.countDown();
+ return true;
+ });
+ workers.get(i % workers.size()).schedule(subtask);
+
Assert.assertFalse(memoryManager.tryReserveTsFileParserMemory(pipeName, 0, "0",
request));
+ }
+ Assert.assertEquals(100, memoryManager.getWaitingTsFileParserCount());
+ for (int i = 0; i < workers.size(); i++) {
+ final PipeProcessorSubtaskWorker worker = workers.get(i);
+ final AtomicReference<Thread> workerThread = workerThreads.get(i);
+ // Consume the scheduling notification before all workers become idle
on admission.
+ worker.sleepIfNecessary(true);
+ // Make a missed wake-up observable without relying on a tight
performance assertion.
+ worker.setSleepingTimeInMilliSecond(60000L);
+ futures.add(
+ executor.submit(
+ () -> {
+ workerThread.set(Thread.currentThread());
+ while (!Thread.currentThread().isInterrupted()) {
+ worker.sleepIfNecessary(worker.runSubtasks());
+ }
+ }));
+ }
+ await()
+ .atMost(5, TimeUnit.SECONDS)
+ .until(
+ () ->
+ workerThreads.stream()
+ .allMatch(
+ thread ->
+ thread.get() != null
+ && thread.get().getState() ==
Thread.State.TIMED_WAITING));
+ for (int i = 0; i < 4; i++) {
+ memoryManager.releaseTsFileParserMemory("blocker" + i, 0, "0");
+ reservedBlockers[i] = false;
+ }
+ Assert.assertTrue(
+ "Admission notifications must wake every queue head",
+ processed.await(5, TimeUnit.SECONDS));
+ Assert.assertEquals(0, memoryManager.getWaitingTsFileParserCount());
+ Assert.assertEquals(0, memoryManager.getReservedTsFileParserCount());
+ } finally {
+ futures.forEach(future -> future.cancel(true));
+ executor.shutdownNow();
+ Assert.assertTrue(executor.awaitTermination(5, TimeUnit.SECONDS));
+ for (int i = 0; i < requests.size(); i++) {
+ memoryManager.cancelTsFileParserMemoryReservation(
+ "waitingPipe" + i, 0, "0", requests.get(i));
+ }
+ for (int i = 0; i < reservedBlockers.length; i++) {
+ if (reservedBlockers[i]) {
+ memoryManager.releaseTsFileParserMemory("blocker" + i, 0, "0");
+ }
+ }
+ config.setPipeTsFileParserMemory(originalParserMemory);
+ config.setPipeTsFileParserInFlightMaxNum(originalGlobalLimit);
+
config.setPipeTsFileParserInFlightMaxNumPerPipeRegion(originalRegionLimit);
+ }
+ }
+
+ @Test(timeout = 10000)
+ public void testNotificationBeforeIdleWaitIsNotLost() throws Exception {
+ final PipeProcessorSubtaskWorker worker = new
PipeProcessorSubtaskWorker(new LinkedHashSet<>());
+ worker.setSleepingTimeInMilliSecond(60000L);
+ final PipeProcessorSubtask subtask = createRunnableSubtask("waitingPipe");
+ final AtomicReference<Runnable> wakeUpCallback = new AtomicReference<>();
+ doAnswer(
+ invocation -> {
+ wakeUpCallback.set(invocation.getArgument(0));
+ return null;
+ })
+ .when(subtask)
+ .setWorkerWakeUpCallback(any());
+ worker.schedule(subtask);
+ worker.sleepIfNecessary(true);
+ final ExecutorService executor = Executors.newSingleThreadExecutor();
+ try {
+ wakeUpCallback.get().run();
+ executor.submit(() -> worker.sleepIfNecessary(true)).get(3,
TimeUnit.SECONDS);
+ } finally {
+ executor.shutdownNow();
+ Assert.assertTrue(executor.awaitTermination(3, TimeUnit.SECONDS));
+ }
+ }
+
@Test
public void testYieldingPipesDoNotBlockAnotherPipeOnSameWorker() throws
Exception {
final PipeProcessorSubtaskWorker worker = new
PipeProcessorSubtaskWorker(new LinkedHashSet<>());
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryManagerTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryManagerTest.java
index 3a50002667e..6b681c5dd49 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryManagerTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryManagerTest.java
@@ -38,6 +38,7 @@ import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
public class PipeMemoryManagerTest {
@@ -74,6 +75,35 @@ public class PipeMemoryManagerTest {
commonConfig.setPipeTsFileParserMemory(originalParserMemoryInBytes);
}
+ @Test
+ public void testAdmissionNotificationOnlyWakesEligibleWaitingPipe() {
+ commonConfig.setPipeTsFileParserInFlightMaxNum(1);
+ commonConfig.setPipeTsFileParserInFlightMaxNumPerPipeRegion(1);
+ final Reservation blocker = new Reservation("blocker", 0);
+ final Reservation pipeA = new Reservation("pipeA", 1);
+ final Reservation pipeB = new Reservation("pipeB", 2);
+ final AtomicInteger pipeAWakeUps = new AtomicInteger();
+ final AtomicInteger pipeBWakeUps = new AtomicInteger();
+ pipeA.key.setWakeUpCallback(pipeAWakeUps::incrementAndGet);
+ pipeB.key.setWakeUpCallback(pipeBWakeUps::incrementAndGet);
+ Assert.assertTrue(tryAcquire(blocker));
+ Assert.assertFalse(tryAcquire(pipeA));
+ Assert.assertFalse(tryAcquire(pipeB));
+ Assert.assertEquals(2, memoryManager.getWaitingTsFileParserCount());
+ Assert.assertEquals(0, pipeAWakeUps.get());
+ Assert.assertEquals(0, pipeBWakeUps.get());
+
+ release(blocker);
+ Assert.assertEquals(1, pipeAWakeUps.get());
+ Assert.assertEquals(0, pipeBWakeUps.get());
+ Assert.assertTrue(tryAcquire(pipeA));
+ Assert.assertEquals(1, memoryManager.getWaitingTsFileParserCount());
+ release(pipeA);
+ Assert.assertEquals(1, pipeBWakeUps.get());
+ Assert.assertTrue(tryAcquire(pipeB));
+ Assert.assertEquals(0, memoryManager.getWaitingTsFileParserCount());
+ }
+
@Test
public void testWaitingPipesAreAdmittedInRoundRobinOrder() {
commonConfig.setPipeTsFileParserInFlightMaxNum(1);
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/service/metric/enums/Metric.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/service/metric/enums/Metric.java
index 8a01eb3a1bf..2a53b1f940b 100644
---
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/service/metric/enums/Metric.java
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/service/metric/enums/Metric.java
@@ -165,6 +165,11 @@ public enum Metric {
PIPE_HEARTBEAT_EVENT("pipe_heartbeat_event"),
PIPE_EXTRACTOR_TSFILE_EPOCH_STATE("pipe_extractor_tsfile_epoch_state"),
PIPE_MEM("pipe_mem"),
+ PIPE_TSFILE_PARSER_ADMISSION_REJECT("pipe_tsfile_parser_admission_reject"),
+ PIPE_TSFILE_PARSER_ADMISSION_YIELD("pipe_tsfile_parser_admission_yield"),
+ PIPE_TSFILE_PARSER_ADMISSION_WAIT("pipe_tsfile_parser_admission_wait"),
+ PIPE_TSFILE_PARSER_IN_FLIGHT("pipe_tsfile_parser_in_flight"),
+ PIPE_TSFILE_PARSER_WAITING("pipe_tsfile_parser_waiting"),
PIPE_LINKED_TSFILE_COUNT("pipe_linked_tsfile_count"),
PIPE_LINKED_TSFILE_SIZE("pipe_linked_tsfile_size"),
PIPE_PHANTOM_REFERENCE_COUNT("pipe_phantom_reference_count"),
