[Pipe] Improve fair TsFile parser admission

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtask.java

@@ -355,6 +355,12 @@ public boolean isStoppedByException() {
     return lastEvent instanceof EnrichedEvent && retryCount.get() > MAX_RETRY_TIMES;
   }
 
+  boolean isWaitingForTsFileParserMemory() {
+    final Event event = lastEvent;
+    return event instanceof PipeTsFileInsertionEvent
+        && ((PipeTsFileInsertionEvent) event).isWaitingForTsFileParserMemory();
+  }
+
   @Override
   public void close() {
     // Always deregister the metrics to avoid the deletion of the data region

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtaskWorker.java

@@ -89,9 +89,11 @@ boolean runSubtasks() {
     ++totalRoundInAdjustmentInterval;
 
     boolean canSleepBeforeNextRound = true;
-
     for (final PipeProcessorSubtask subtask : subtasks) {
-      if (subtask.isClosed() || !subtask.isSubmittingSelf() || subtask.isStoppedByException()) {
+      if (subtask.isClosed()
+          || !subtask.isSubmittingSelf()
+          || subtask.isStoppedByException()
+          || subtask.isWaitingForTsFileParserMemory()) {
         continue;
       }
 

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeTsFileInsertionEvent.java

@@ -1548,6 +1548,10 @@ private void releaseTsFileParserMemoryIfReserved() {
     }
   }
 
+  public boolean isWaitingForTsFileParserMemory() {
+    return tsFileParserMemoryReservationKey.isWaitingForSignal();
+  }
+
   public void cancelTsFileParserMemoryReservationIfPending() {
     synchronized (isTsFileParserMemoryReserved) {
       if (!isTsFileParserMemoryReserved.get()) {
@@ -1636,12 +1640,6 @@ public long count(final boolean skipReportOnCommit) throws Exception {
   public void close() {
     // Every cleanup step is best effort and independent.  A parser/reader close failure must not
     // strand its reservation or the pending tablet event.
-    try {
-      cancelTsFileParserMemoryReservationIfPending();
-    } catch (final Exception e) {
-      LOGGER.warn(DataNodePipeMessages.DECREASE_REFERENCE_COUNT_FOR_TSFILE_ERROR, tsFile, e);
-    }
-
     final PipeRawTabletInsertionEvent detachedPendingEvent;
     final boolean pendingEventIsBeingConsumed;
     final TsFileInsertionEventParser parserToClose;
@@ -1664,6 +1662,15 @@ public void close() {
       }
     }
 
+    // Invalidate old invocations before cancelling: an invocation that passed its generation
+    // check can enqueue while close waits for eventParser. It must either be removed here or
+    // observe the new generation and cancel its own request.
+    try {
+      cancelTsFileParserMemoryReservationIfPending();
+    } catch (final Exception e) {
+      LOGGER.warn(DataNodePipeMessages.DECREASE_REFERENCE_COUNT_FOR_TSFILE_ERROR, tsFile, e);
+    }
+
     if (detachedPendingEvent != null && !pendingEventIsBeingConsumed) {
       try {
         releaseParsedTabletEvent(detachedPendingEvent);

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/metric/overview/PipeResourceMetrics.java

@@ -33,6 +33,9 @@
 import org.apache.iotdb.metrics.utils.MetricLevel;
 import org.apache.iotdb.metrics.utils.MetricType;
 
+import java.util.Collections;
+import java.util.EnumMap;
+import java.util.Map;
 import java.util.concurrent.TimeUnit;
 
 public class PipeResourceMetrics implements IMetricSet {
@@ -49,6 +52,21 @@ public class PipeResourceMetrics implements IMetricSet {
   private Counter diskIOCounter = DoNothingMetricManager.DO_NOTHING_COUNTER;
   private volatile Counter tsFileParserAdmissionRejectCounter =
       DoNothingMetricManager.DO_NOTHING_COUNTER;
+
+  public enum TsFileParserAdmissionRejectReason {
+    CAP("cap"),
+    MEMORY("memory"),
+    FAIRNESS("fairness");
+
+    private final String tag;
+
+    TsFileParserAdmissionRejectReason(final String tag) {
+      this.tag = tag;
+    }
+  }
+
+  private volatile Map<TsFileParserAdmissionRejectReason, Counter>
+      tsFileParserAdmissionRejectCounters = Collections.emptyMap();
   private volatile Counter tsFileParserAdmissionYieldCounter =
       DoNothingMetricManager.DO_NOTHING_COUNTER;
   private volatile Timer tsFileParserAdmissionWaitTimer = DoNothingMetricManager.DO_NOTHING_TIMER;
@@ -111,9 +129,22 @@ public void bindTo(final AbstractMetricService metricService) {
     diskIOCounter =
         metricService.getOrCreateCounter(
             Metric.PIPE_TSFILE_SEND_DISK_IO.toString(), MetricLevel.IMPORTANT);
+    final Map<TsFileParserAdmissionRejectReason, Counter> rejectCounters =
+        new EnumMap<>(TsFileParserAdmissionRejectReason.class);
     tsFileParserAdmissionRejectCounter =
         metricService.getOrCreateCounter(
             Metric.PIPE_TSFILE_PARSER_ADMISSION_REJECT.toString(), MetricLevel.IMPORTANT);
+    for (final TsFileParserAdmissionRejectReason reason :
+        TsFileParserAdmissionRejectReason.values()) {
+      rejectCounters.put(
+          reason,
+          metricService.getOrCreateCounter(
+              Metric.PIPE_TSFILE_PARSER_ADMISSION_REJECT_REASON.toString(),
+              MetricLevel.IMPORTANT,
+              Tag.REASON.toString(),
+              reason.tag));
+    }
+    tsFileParserAdmissionRejectCounters = rejectCounters;
     tsFileParserAdmissionYieldCounter =
         metricService.getOrCreateCounter(
             Metric.PIPE_TSFILE_PARSER_ADMISSION_YIELD.toString(), MetricLevel.IMPORTANT);
@@ -167,10 +198,19 @@ public void unbindFrom(final AbstractMetricService metricService) {
 
     metricService.remove(MetricType.COUNTER, Metric.PIPE_TSFILE_SEND_DISK_IO.toString());
     metricService.remove(MetricType.COUNTER, Metric.PIPE_TSFILE_PARSER_ADMISSION_REJECT.toString());
+    for (final TsFileParserAdmissionRejectReason reason :
+        TsFileParserAdmissionRejectReason.values()) {
+      metricService.remove(
+          MetricType.COUNTER,
+          Metric.PIPE_TSFILE_PARSER_ADMISSION_REJECT_REASON.toString(),
+          Tag.REASON.toString(),
+          reason.tag);
+    }
     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());
+    tsFileParserAdmissionRejectCounters = Collections.emptyMap();
     tsFileParserAdmissionRejectCounter = DoNothingMetricManager.DO_NOTHING_COUNTER;
     tsFileParserAdmissionYieldCounter = DoNothingMetricManager.DO_NOTHING_COUNTER;
     tsFileParserAdmissionWaitTimer = DoNothingMetricManager.DO_NOTHING_TIMER;
@@ -180,8 +220,11 @@ public void recordDiskIO(final long bytes) {
     diskIOCounter.inc(bytes);
   }
 
-  public void recordTsFileParserAdmissionReject() {
+  public void recordTsFileParserAdmissionReject(final TsFileParserAdmissionRejectReason reason) {
     tsFileParserAdmissionRejectCounter.inc();
+    tsFileParserAdmissionRejectCounters
+        .getOrDefault(reason, DoNothingMetricManager.DO_NOTHING_COUNTER)
+        .inc();
   }
 
   public void recordTsFileParserAdmissionYield() {