Skip to content

Commit 6527fb0

Browse files
CaideyipiCaideyipi
authored
[Pipe] Wake parser workers when admission is available (#18843) (#18865)
* Fix pipe parser admission wake-ups * Avoid reflective test dependency (cherry picked from commit aec1ad6)
1 parent dfcd489 commit 6527fb0

10 files changed

Lines changed: 403 additions & 50 deletions

File tree

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

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -271,6 +271,10 @@ public void submitSelf() {
271271
// and the worker will be submitted to the executor
272272
}
273273

274+
void setWorkerWakeUpCallback(final Runnable wakeUpCallback) {
275+
executionGuard.setWorkerWakeUpCallback(wakeUpCallback);
276+
}
277+
274278
@Override
275279
protected void onAllowSubmittingSelf() {
276280
executionGuard.start();

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

Lines changed: 16 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,8 @@
1919

2020
package org.apache.iotdb.db.pipe.agent.task.subtask.processor;
2121

22+
import org.apache.iotdb.db.pipe.metric.overview.PipeResourceMetrics;
23+
2224
import java.util.concurrent.atomic.AtomicBoolean;
2325
import java.util.concurrent.atomic.AtomicLong;
2426

@@ -38,6 +40,7 @@ public class PipeProcessorSubtaskExecutionGuard {
3840
private final AtomicBoolean isRunning = new AtomicBoolean(false);
3941
private final AtomicLong executionEpoch = new AtomicLong(0);
4042
private final ThreadLocal<Long> invocationEpoch = new ThreadLocal<>();
43+
private volatile Runnable workerWakeUpCallback;
4144

4245
public PipeProcessorSubtaskExecutionGuard() {
4346
this(true);
@@ -55,9 +58,21 @@ public boolean isEnabled() {
5558
return enabled;
5659
}
5760

61+
void setWorkerWakeUpCallback(final Runnable workerWakeUpCallback) {
62+
this.workerWakeUpCallback = workerWakeUpCallback;
63+
}
64+
65+
public Runnable getWorkerWakeUpCallback() {
66+
return workerWakeUpCallback;
67+
}
68+
5869
void start() {
5970
if (enabled) {
6071
isRunning.set(true);
72+
final Runnable wakeUpCallback = workerWakeUpCallback;
73+
if (wakeUpCallback != null) {
74+
wakeUpCallback.run();
75+
}
6176
}
6277
}
6378

@@ -105,6 +120,7 @@ public boolean isCurrentInvocationValid() {
105120
}
106121

107122
public void yieldIfParserNotAdmitted() {
123+
PipeResourceMetrics.getInstance().recordTsFileParserAdmissionYield();
108124
throw PipeProcessorSubtaskYieldException.parserNotAdmitted();
109125
}
110126
}

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

Lines changed: 27 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -49,6 +49,9 @@ public class PipeProcessorSubtaskWorker extends WrappedRunnable {
4949
private long sleepingTimeInMilliSecond = 50;
5050

5151
private final Set<PipeProcessorSubtask> subtasks;
52+
private final Object wakeUpMonitor = new Object();
53+
private final Runnable wakeUpCallback = this::wakeUp;
54+
private boolean isWakeUpRequested;
5255

5356
private volatile Thread workerThread;
5457
private volatile PipeProcessorSubtask currentSubtask;
@@ -115,10 +118,18 @@ boolean runSubtasks() {
115118
return canSleepBeforeNextRound;
116119
}
117120

118-
private void sleepIfNecessary(final boolean canSleepBeforeNextRound) {
121+
@VisibleForTesting
122+
void sleepIfNecessary(final boolean canSleepBeforeNextRound) {
119123
if (canSleepBeforeNextRound) {
120124
try {
121-
Thread.sleep(sleepingTimeInMilliSecond);
125+
synchronized (wakeUpMonitor) {
126+
// A parser can become eligible before this worker starts waiting. Keep that notification
127+
// until the idle round consumes it, instead of sleeping with the fair queue head ready.
128+
if (!isWakeUpRequested) {
129+
wakeUpMonitor.wait(sleepingTimeInMilliSecond);
130+
}
131+
isWakeUpRequested = false;
132+
}
122133
} catch (final InterruptedException e) {
123134
LOGGER.warn("subtask worker is interrupted", e);
124135
Thread.currentThread().interrupt();
@@ -128,6 +139,11 @@ private void sleepIfNecessary(final boolean canSleepBeforeNextRound) {
128139
}
129140
}
130141

142+
@VisibleForTesting
143+
void setSleepingTimeInMilliSecond(final long sleepingTimeInMilliSecond) {
144+
this.sleepingTimeInMilliSecond = sleepingTimeInMilliSecond;
145+
}
146+
131147
private void adjustSleepingTimeIfNecessary() {
132148
if (totalRoundInAdjustmentInterval % SLEEP_INTERVAL_ADJUSTMENT_ROUND_INTERVAL == 0) {
133149
final double workingRatioInAdjustmentInterval =
@@ -147,7 +163,16 @@ private void adjustSleepingTimeIfNecessary() {
147163
}
148164

149165
public void schedule(final PipeProcessorSubtask pipeProcessorSubtask) {
166+
pipeProcessorSubtask.setWorkerWakeUpCallback(wakeUpCallback);
150167
subtasks.add(pipeProcessorSubtask);
168+
wakeUp();
169+
}
170+
171+
private void wakeUp() {
172+
synchronized (wakeUpMonitor) {
173+
isWakeUpRequested = true;
174+
wakeUpMonitor.notifyAll();
175+
}
151176
}
152177

153178
void watchLongRunningEvent() {

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

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -923,6 +923,10 @@ private boolean reserveResource4Parsing(
923923
final PipeProcessorSubtaskExecutionGuard processorExecutionGuard,
924924
final long expectedParserStateGeneration)
925925
throws InterruptedException {
926+
// Register before enqueueing: memory may be released as soon as admission fails, before this
927+
// invocation yields or its worker enters the idle wait.
928+
tsFileParserMemoryReservationKey.setWakeUpCallback(
929+
processorExecutionGuard.getWorkerWakeUpCallback());
926930
if (!processorExecutionGuard.isEnabled()) {
927931
return waitForResourceEnough4Parsing(
928932
(long) ((1 + Math.random()) * 20 * 1000), expectedParserStateGeneration);

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

Lines changed: 47 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -29,9 +29,12 @@
2929
import org.apache.iotdb.metrics.impl.DoNothingMetricManager;
3030
import org.apache.iotdb.metrics.metricsets.IMetricSet;
3131
import org.apache.iotdb.metrics.type.Counter;
32+
import org.apache.iotdb.metrics.type.Timer;
3233
import org.apache.iotdb.metrics.utils.MetricLevel;
3334
import org.apache.iotdb.metrics.utils.MetricType;
3435

36+
import java.util.concurrent.TimeUnit;
37+
3538
public class PipeResourceMetrics implements IMetricSet {
3639

3740
private static final String PIPE_USED_MEMORY = "PipeUsedMemory";
@@ -44,6 +47,11 @@ public class PipeResourceMetrics implements IMetricSet {
4447
private static final String PIPE_TOTAL_MEMORY = "PipeTotalMemory";
4548
private static final String PIPE_FLOATING_MEMORY = "PipeFloatingMemory";
4649
private Counter diskIOCounter = DoNothingMetricManager.DO_NOTHING_COUNTER;
50+
private volatile Counter tsFileParserAdmissionRejectCounter =
51+
DoNothingMetricManager.DO_NOTHING_COUNTER;
52+
private volatile Counter tsFileParserAdmissionYieldCounter =
53+
DoNothingMetricManager.DO_NOTHING_COUNTER;
54+
private volatile Timer tsFileParserAdmissionWaitTimer = DoNothingMetricManager.DO_NOTHING_TIMER;
4755

4856
//////////////////////////// bindTo & unbindFrom (metric framework) ////////////////////////////
4957

@@ -102,6 +110,25 @@ public void bindTo(final AbstractMetricService metricService) {
102110
diskIOCounter =
103111
metricService.getOrCreateCounter(
104112
Metric.PIPE_TSFILE_SEND_DISK_IO.toString(), MetricLevel.IMPORTANT);
113+
tsFileParserAdmissionRejectCounter =
114+
metricService.getOrCreateCounter(
115+
Metric.PIPE_TSFILE_PARSER_ADMISSION_REJECT.toString(), MetricLevel.IMPORTANT);
116+
tsFileParserAdmissionYieldCounter =
117+
metricService.getOrCreateCounter(
118+
Metric.PIPE_TSFILE_PARSER_ADMISSION_YIELD.toString(), MetricLevel.IMPORTANT);
119+
tsFileParserAdmissionWaitTimer =
120+
metricService.getOrCreateTimer(
121+
Metric.PIPE_TSFILE_PARSER_ADMISSION_WAIT.toString(), MetricLevel.IMPORTANT);
122+
metricService.createAutoGauge(
123+
Metric.PIPE_TSFILE_PARSER_IN_FLIGHT.toString(),
124+
MetricLevel.IMPORTANT,
125+
PipeDataNodeResourceManager.memory(),
126+
PipeMemoryManager::getReservedTsFileParserCount);
127+
metricService.createAutoGauge(
128+
Metric.PIPE_TSFILE_PARSER_WAITING.toString(),
129+
MetricLevel.IMPORTANT,
130+
PipeDataNodeResourceManager.memory(),
131+
PipeMemoryManager::getWaitingTsFileParserCount);
105132
}
106133

107134
@Override
@@ -138,12 +165,32 @@ public void unbindFrom(final AbstractMetricService metricService) {
138165
metricService.remove(MetricType.AUTO_GAUGE, Metric.PIPE_PHANTOM_REFERENCE_COUNT.toString());
139166

140167
metricService.remove(MetricType.COUNTER, Metric.PIPE_TSFILE_SEND_DISK_IO.toString());
168+
metricService.remove(MetricType.COUNTER, Metric.PIPE_TSFILE_PARSER_ADMISSION_REJECT.toString());
169+
metricService.remove(MetricType.COUNTER, Metric.PIPE_TSFILE_PARSER_ADMISSION_YIELD.toString());
170+
metricService.remove(MetricType.TIMER, Metric.PIPE_TSFILE_PARSER_ADMISSION_WAIT.toString());
171+
metricService.remove(MetricType.AUTO_GAUGE, Metric.PIPE_TSFILE_PARSER_IN_FLIGHT.toString());
172+
metricService.remove(MetricType.AUTO_GAUGE, Metric.PIPE_TSFILE_PARSER_WAITING.toString());
173+
tsFileParserAdmissionRejectCounter = DoNothingMetricManager.DO_NOTHING_COUNTER;
174+
tsFileParserAdmissionYieldCounter = DoNothingMetricManager.DO_NOTHING_COUNTER;
175+
tsFileParserAdmissionWaitTimer = DoNothingMetricManager.DO_NOTHING_TIMER;
141176
}
142177

143178
public void recordDiskIO(final long bytes) {
144179
diskIOCounter.inc(bytes);
145180
}
146181

182+
public void recordTsFileParserAdmissionReject() {
183+
tsFileParserAdmissionRejectCounter.inc();
184+
}
185+
186+
public void recordTsFileParserAdmissionYield() {
187+
tsFileParserAdmissionYieldCounter.inc();
188+
}
189+
190+
public void recordTsFileParserAdmissionWait(final long waitTimeInNanos) {
191+
tsFileParserAdmissionWaitTimer.update(waitTimeInNanos, TimeUnit.NANOSECONDS);
192+
}
193+
147194
//////////////////////////// singleton ////////////////////////////
148195

149196
private static class PipeResourceMetricsHolder {

0 commit comments

Comments
 (0)