[To dev/1.3] [Pipe] Wake parser workers when admission is available

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

@@ -271,6 +271,10 @@ public void submitSelf() {
     // and the worker will be submitted to the executor
   }
 
+  void setWorkerWakeUpCallback(final Runnable wakeUpCallback) {
+    executionGuard.setWorkerWakeUpCallback(wakeUpCallback);
+  }
+
   @Override
   protected void onAllowSubmittingSelf() {
     executionGuard.start();

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 boolean isEnabled() {
     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 boolean isCurrentInvocationValid() {
   }
 
   public void yieldIfParserNotAdmitted() {
+    PipeResourceMetrics.getInstance().recordTsFileParserAdmissionYield();
     throw PipeProcessorSubtaskYieldException.parserNotAdmitted();
   }
 }

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 @@ boolean runSubtasks() {
     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 @@ private void sleepIfNecessary(final boolean canSleepBeforeNextRound) {
     }
   }
 
+  @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 @@ private void adjustSleepingTimeIfNecessary() {
   }
 
   public void schedule(final PipeProcessorSubtask pipeProcessorSubtask) {
+    pipeProcessorSubtask.setWorkerWakeUpCallback(wakeUpCallback);
     subtasks.add(pipeProcessorSubtask);
+    wakeUp();
+  }
+
+  private void wakeUp() {
+    synchronized (wakeUpMonitor) {
+      isWakeUpRequested = true;
+      wakeUpMonitor.notifyAll();
+    }
   }
 
   void watchLongRunningEvent() {

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

@@ -923,6 +923,10 @@ private boolean reserveResource4Parsing(
       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);

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

@@ -29,9 +29,12 @@
 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 void bindTo(final AbstractMetricService metricService) {
     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 void unbindFrom(final AbstractMetricService metricService) {
     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 {