Skip to content

Commit aec1ad6

Browse files
CaideyipiCaideyipi
authored
[Pipe] Wake parser workers when admission is available (#18843)
* Fix pipe parser admission wake-ups * Avoid reflective test dependency
1 parent 72b4d16 commit aec1ad6

10 files changed

Lines changed: 397 additions & 36 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
@@ -325,6 +325,10 @@ public void submitSelf() {
325325
// and the worker will be submitted to the executor
326326
}
327327

328+
void setWorkerWakeUpCallback(final Runnable wakeUpCallback) {
329+
executionGuard.setWorkerWakeUpCallback(wakeUpCallback);
330+
}
331+
328332
@Override
329333
protected void onAllowSubmittingSelf() {
330334
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
@@ -50,6 +50,9 @@ public class PipeProcessorSubtaskWorker extends WrappedRunnable {
5050
private long sleepingTimeInMilliSecond = 50;
5151

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

5457
private volatile Thread workerThread;
5558
private volatile PipeProcessorSubtask currentSubtask;
@@ -116,10 +119,18 @@ boolean runSubtasks() {
116119
return canSleepBeforeNextRound;
117120
}
118121

119-
private void sleepIfNecessary(final boolean canSleepBeforeNextRound) {
122+
@VisibleForTesting
123+
void sleepIfNecessary(final boolean canSleepBeforeNextRound) {
120124
if (canSleepBeforeNextRound) {
121125
try {
122-
Thread.sleep(sleepingTimeInMilliSecond);
126+
synchronized (wakeUpMonitor) {
127+
// A parser can become eligible before this worker starts waiting. Keep that notification
128+
// until the idle round consumes it, instead of sleeping with the fair queue head ready.
129+
if (!isWakeUpRequested) {
130+
wakeUpMonitor.wait(sleepingTimeInMilliSecond);
131+
}
132+
isWakeUpRequested = false;
133+
}
123134
} catch (final InterruptedException e) {
124135
LOGGER.warn(DataNodePipeMessages.SUBTASK_WORKER_IS_INTERRUPTED, e);
125136
Thread.currentThread().interrupt();
@@ -129,6 +140,11 @@ private void sleepIfNecessary(final boolean canSleepBeforeNextRound) {
129140
}
130141
}
131142

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

150166
public void schedule(final PipeProcessorSubtask pipeProcessorSubtask) {
167+
pipeProcessorSubtask.setWorkerWakeUpCallback(wakeUpCallback);
151168
subtasks.add(pipeProcessorSubtask);
169+
wakeUp();
170+
}
171+
172+
private void wakeUp() {
173+
synchronized (wakeUpMonitor) {
174+
isWakeUpRequested = true;
175+
wakeUpMonitor.notifyAll();
176+
}
152177
}
153178

154179
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
@@ -1376,6 +1376,10 @@ private boolean reserveResource4Parsing(
13761376
final PipeProcessorSubtaskExecutionGuard processorExecutionGuard,
13771377
final long expectedParserStateGeneration)
13781378
throws InterruptedException {
1379+
// Register before enqueueing: memory may be released as soon as admission fails, before this
1380+
// invocation yields or its worker enters the idle wait.
1381+
tsFileParserMemoryReservationKey.setWakeUpCallback(
1382+
processorExecutionGuard.getWorkerWakeUpCallback());
13791383
if (!processorExecutionGuard.isEnabled()) {
13801384
return waitForResourceEnough4Parsing(
13811385
(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

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
private static final String PIPE_FLOATING_MEMORY = "PipeFloatingMemory";
4856

4957
//////////////////////////// bindTo & unbindFrom (metric framework) ////////////////////////////
@@ -103,6 +111,25 @@ public void bindTo(final AbstractMetricService metricService) {
103111
diskIOCounter =
104112
metricService.getOrCreateCounter(
105113
Metric.PIPE_TSFILE_SEND_DISK_IO.toString(), MetricLevel.IMPORTANT);
114+
tsFileParserAdmissionRejectCounter =
115+
metricService.getOrCreateCounter(
116+
Metric.PIPE_TSFILE_PARSER_ADMISSION_REJECT.toString(), MetricLevel.IMPORTANT);
117+
tsFileParserAdmissionYieldCounter =
118+
metricService.getOrCreateCounter(
119+
Metric.PIPE_TSFILE_PARSER_ADMISSION_YIELD.toString(), MetricLevel.IMPORTANT);
120+
tsFileParserAdmissionWaitTimer =
121+
metricService.getOrCreateTimer(
122+
Metric.PIPE_TSFILE_PARSER_ADMISSION_WAIT.toString(), MetricLevel.IMPORTANT);
123+
metricService.createAutoGauge(
124+
Metric.PIPE_TSFILE_PARSER_IN_FLIGHT.toString(),
125+
MetricLevel.IMPORTANT,
126+
PipeDataNodeResourceManager.memory(),
127+
PipeMemoryManager::getReservedTsFileParserCount);
128+
metricService.createAutoGauge(
129+
Metric.PIPE_TSFILE_PARSER_WAITING.toString(),
130+
MetricLevel.IMPORTANT,
131+
PipeDataNodeResourceManager.memory(),
132+
PipeMemoryManager::getWaitingTsFileParserCount);
106133
}
107134

108135
@Override
@@ -139,12 +166,32 @@ public void unbindFrom(final AbstractMetricService metricService) {
139166
metricService.remove(MetricType.AUTO_GAUGE, Metric.PIPE_PHANTOM_REFERENCE_COUNT.toString());
140167

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

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

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

150197
private static class PipeResourceMetricsHolder {

0 commit comments

Comments
 (0)