[To dev/1.3] Improve fair TsFile parser admission

iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java

@@ -23,6 +23,9 @@ public final class DataNodePipeMessages {
 
   private DataNodePipeMessages() {}
 
+  public static final String LOG_DECREASE_REFERENCE_COUNT_FOR_TSFILE_ARG_ERROR_8EF0D114 =
+      "Decrease reference count for TsFile {} error.";
+
   public static final String EXCEPTION_INCREASE_REFERENCE_COUNT_ERROR_HOLDER_MESSAGE_ARG_7E1D032D =
       "Increase reference count error. Holder Message: %s";
 

iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java

@@ -23,6 +23,8 @@ public final class DataNodePipeMessages {
 
   private DataNodePipeMessages() {}
 
+  public static final String LOG_DECREASE_REFERENCE_COUNT_FOR_TSFILE_ARG_ERROR_8EF0D114 =
+      "减少 TsFile {} 的引用计数时发生错误。";
   public static final String EXCEPTION_INCREASE_REFERENCE_COUNT_ERROR_HOLDER_MESSAGE_ARG_7E1D032D =
       "增加引用计数失败。持有者信息：%s";
 

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

@@ -293,6 +293,12 @@
     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

@@ -88,9 +88,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

@@ -28,6 +28,7 @@
 import org.apache.iotdb.commons.pipe.event.EnrichedEvent;
 import org.apache.iotdb.commons.pipe.resource.log.PipeLogger;
 import org.apache.iotdb.commons.pipe.resource.ref.PipePhantomReferenceManager.PipeEventResource;
+import org.apache.iotdb.db.i18n.DataNodePipeMessages;
 import org.apache.iotdb.db.pipe.agent.task.subtask.processor.PipeProcessorSubtaskExecutionGuard;
 import org.apache.iotdb.db.pipe.agent.task.subtask.processor.PipeProcessorSubtaskYieldException;
 import org.apache.iotdb.db.pipe.event.ReferenceTrackableEvent;
@@ -1064,6 +1065,10 @@ private void releaseTsFileParserMemoryIfReserved() {
     }
   }
 
+  public boolean isWaitingForTsFileParserMemory() {
+    return tsFileParserMemoryReservationKey.isWaitingForSignal();
+  }
+
   public void cancelTsFileParserMemoryReservationIfPending() {
     synchronized (isTsFileParserMemoryReserved) {
       if (!isTsFileParserMemoryReserved.get()) {
@@ -1142,12 +1147,8 @@ public long count(final boolean skipReportOnCommit) throws Exception {
   /** Release the resource of {@link TsFileInsertionDataContainer}. */
   @Override
   public void close() {
-    try {
-      cancelTsFileParserMemoryReservationIfPending();
-    } catch (final Exception e) {
-      LOGGER.warn("Decrease reference count for TsFile {} error.", tsFile, e);
-    }
-
+    // Every cleanup step is best effort and independent.  A parser/reader close failure must not
+    // strand its reservation or the pending tablet event.
     final PipeRawTabletInsertionEvent detachedPendingEvent;
     final boolean pendingEventIsBeingConsumed;
     final TsFileInsertionDataContainer containerToClose;
@@ -1168,6 +1169,18 @@ public void close() {
       }
     }
 
+    // Invalidate old invocations before cancelling: an invocation that passed its generation
+    // check can enqueue while close waits for dataContainer. 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.LOG_DECREASE_REFERENCE_COUNT_FOR_TSFILE_ARG_ERROR_8EF0D114,
+          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 @@
   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;
@@ -110,9 +128,22 @@
     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);
@@ -166,10 +197,19 @@
 
     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;
@@ -179,8 +219,11 @@
     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() {