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 aec1ad6139c [Pipe] Wake parser workers when admission is available
(#18843)
aec1ad6139c is described below
commit aec1ad6139cb4dd1450725f53418cf40c26efb4e
Author: Caideyipi <[email protected]>
AuthorDate: Wed Oct 14 11:37:30 2026 +0800
[Pipe] Wake parser workers when admission is available (#18843)
* Fix pipe parser admission wake-ups
* Avoid reflective test dependency
---
.../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 | 134 +++++++++++++-----
.../PipeProcessorSubtaskExecutionGuardTest.java | 10 +-
.../processor/PipeProcessorSubtaskWorkerTest.java | 154 +++++++++++++++++++++
.../resource/memory/PipeMemoryManagerTest.java | 30 ++++
.../iotdb/commons/service/metric/enums/Metric.java | 5 +
10 files changed, 397 insertions(+), 36 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 0a95465b50e..ca144eca48c 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
@@ -325,6 +325,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 bc967cbc2e9..e5033a04c27 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
@@ -50,6 +50,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;
@@ -116,10 +119,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(DataNodePipeMessages.SUBTASK_WORKER_IS_INTERRUPTED, e);
Thread.currentThread().interrupt();
@@ -129,6 +140,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 =
@@ -148,7 +164,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 9a89c1d367c..2264f956f68 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
@@ -1376,6 +1376,10 @@ public class PipeTsFileInsertionEvent extends
PipeInsertionEvent
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 eeb840ce0c5..e285e2f2260 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 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;
private static final String PIPE_FLOATING_MEMORY = "PipeFloatingMemory";
//////////////////////////// bindTo & unbindFrom (metric framework)
////////////////////////////
@@ -103,6 +111,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
@@ -139,12 +166,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 4e43cc9c35c..6d89ac0d9d2 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
@@ -26,6 +26,7 @@ import org.apache.iotdb.commons.pipe.config.PipeConfig;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.i18n.DataNodePipeMessages;
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;
@@ -65,6 +66,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 =
@@ -115,22 +117,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() {
+ private double allowedMaxMemorySizeInBytesOfTabletsAndTsFiles(
+ final long totalNonFloatingMemorySizeInBytes) {
return
(PIPE_CONFIG.getPipeDataStructureTabletMemoryBlockAllocationRejectThreshold()
+
PIPE_CONFIG.getPipeDataStructureTsFileMemoryBlockAllocationRejectThreshold())
- * getTotalNonFloatingMemorySizeInBytes();
+ * totalNonFloatingMemorySizeInBytes;
}
- private double allowedMaxMemorySizeInBytesOfTablets() {
+ private double allowedMaxMemorySizeInBytesOfTablets(
+ final long totalNonFloatingMemorySizeInBytes) {
return
(PIPE_CONFIG.getPipeDataStructureTabletMemoryBlockAllocationRejectThreshold()
+
PIPE_CONFIG.getPipeDataStructureTsFileMemoryBlockAllocationRejectThreshold() /
2)
- * getTotalNonFloatingMemorySizeInBytes();
+ * totalNonFloatingMemorySizeInBytes;
}
- private double allowedMaxMemorySizeInBytesOfTsTiles() {
+ private double allowedMaxMemorySizeInBytesOfTsTiles(
+ final long totalNonFloatingMemorySizeInBytes) {
return
(PIPE_CONFIG.getPipeDataStructureTsFileMemoryBlockAllocationRejectThreshold()
+
PIPE_CONFIG.getPipeDataStructureTabletMemoryBlockAllocationRejectThreshold() /
2)
- * getTotalNonFloatingMemorySizeInBytes();
+ * totalNonFloatingMemorySizeInBytes;
}
private static long getTsFileParserMemorySizeInBytes() {
@@ -142,39 +147,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(
@@ -200,16 +214,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;
}
@@ -220,13 +242,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);
@@ -300,7 +326,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() {
@@ -309,16 +340,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;
}
@@ -327,6 +362,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;
}
@@ -397,6 +436,7 @@ public class PipeMemoryManager {
if (requestsOfPipeRegion == null ||
!requestsOfPipeRegion.remove(reservationKey)) {
return;
}
+ waitingTsFileParserCount--;
final PipeIdentity pipeIdentity = pipeRegionIdentity.pipeIdentity;
final ArrayDeque<PipeRegionIdentity> regionOrder =
@@ -430,6 +470,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) {
@@ -441,6 +489,10 @@ public class PipeMemoryManager {
private synchronized void signal() {
isSignaled = true;
notify();
+ final Runnable callback = wakeUpCallback;
+ if (callback != null) {
+ callback.run();
+ }
}
}
@@ -461,7 +513,8 @@ public class PipeMemoryManager {
return elapsedTimeInMs >= maxRetryTimeInMs;
}
- if (!isHardEnough4TabletParsingWithReservedParserMemory(0)) {
+ if (!isHardEnough4TabletParsingWithReservedParserMemory(
+ 0, getTotalNonFloatingMemorySizeInBytes())) {
return true;
}
@@ -470,35 +523,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;
}
@@ -1074,6 +1136,14 @@ public class PipeMemoryManager {
return usedMemorySizeInBytesOfTsFiles;
}
+ public long getReservedTsFileParserCount() {
+ return reservedTsFileParserCount;
+ }
+
+ public long getWaitingTsFileParserCount() {
+ return waitingTsFileParserCount;
+ }
+
public long getFreeMemorySizeInBytes() {
return Math.max(0, getTotalNonFloatingMemorySizeInBytes() -
memoryBlock.getUsedMemoryInBytes());
}
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 3438dc62c5d..612e9164fbd 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
@@ -89,7 +89,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();
@@ -107,10 +110,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);
@@ -119,6 +124,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 51a018d48d8..6bb09ea9c42 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;
@@ -27,12 +32,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;
@@ -41,6 +56,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 e6569beb7e6..b701df3848d 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
@@ -182,6 +182,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"),
