This is an automated email from the ASF dual-hosted git repository.
jt2594838 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new 3c0c7fef006 [Pipe] Fix processor worker starvation on pipe stop
(#18396)
3c0c7fef006 is described below
commit 3c0c7fef0068162d7b3fc582df61a82050382ac8
Author: Caideyipi <[email protected]>
AuthorDate: Tue Aug 11 14:21:28 2026 +0800
[Pipe] Fix processor worker starvation on pipe stop (#18396)
* Fix processor worker starvation on pipe stop
* Add multi-pipe processor worker test
* Log long-running pipe processor event stacks
* refactor(pipe): reuse processor exception root cause
---
.../apache/iotdb/db/i18n/DataNodePipeMessages.java | 2 +
.../apache/iotdb/db/i18n/DataNodePipeMessages.java | 2 +
.../agent/task/connection/PipeEventCollector.java | 17 +-
.../subtask/processor/PipeProcessorSubtask.java | 85 ++++++-
.../PipeProcessorSubtaskExecutionGuard.java | 110 +++++++++
.../processor/PipeProcessorSubtaskWorker.java | 129 ++++++++++-
.../PipeProcessorSubtaskWorkerManager.java | 15 +-
.../PipeProcessorSubtaskYieldException.java | 53 +++++
.../common/tsfile/PipeTsFileInsertionEvent.java | 103 ++++++++-
.../PipeProcessorSubtaskExecutionGuardTest.java | 247 +++++++++++++++++++++
.../processor/PipeProcessorSubtaskWorkerTest.java | 151 +++++++++++++
.../pipe/agent/task/subtask/PipeSubtask.java | 15 +-
12 files changed, 905 insertions(+), 24 deletions(-)
diff --git
a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
index d7c54994a6d..7da8a596579 100644
---
a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
+++
b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
@@ -2588,4 +2588,6 @@ public final class DataNodePipeMessages {
"Topic config for %s is unavailable during consensus subscription setup";
public static final String
LOG_FAILED_TO_RELEASE_TSFILE_PARSER_MEMORY_FOR_PIPE_ARG_CREATION_TIME_ARG_IN_DATAREGION_ARG_BECAUSE_NO_RESERVATION_EXISTS_BB8321C0
=
"Failed to release TsFile parser memory for Pipe {} (creation time {})
in DataRegion {} because no reservation exists.";
+ public static final String
LOG_PIPE_PROCESSOR_WORKER_ARG_HAS_BEEN_PROCESSING_THE_SAME_EVENT_FOR_ARG_MS_PIPE_ARG_DATAREGION_ARG_SUBTASK_ARG_EVENT_ARG_THREAD_STATE_ARG_STACK_ARG_63B40775
=
+ "Pipe processor worker {} has been processing the same event for {} ms.
Pipe: {}, DataRegion: {}, subtask: {}, event: {}, thread state: {}. Stack:{}";
}
diff --git
a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
index beed68e1cde..35c88b9abfe 100644
---
a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
+++
b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
@@ -2416,4 +2416,6 @@ public final class DataNodePipeMessages {
"共识订阅设置期间 topic %s 的配置不可用";
public static final String
LOG_FAILED_TO_RELEASE_TSFILE_PARSER_MEMORY_FOR_PIPE_ARG_CREATION_TIME_ARG_IN_DATAREGION_ARG_BECAUSE_NO_RESERVATION_EXISTS_BB8321C0
=
"无法释放 Pipe {}（创建时间 {}）在 DataRegion {} 中的 TsFile 解析器内存，因为不存在对应的预留。";
+ public static final String
LOG_PIPE_PROCESSOR_WORKER_ARG_HAS_BEEN_PROCESSING_THE_SAME_EVENT_FOR_ARG_MS_PIPE_ARG_DATAREGION_ARG_SUBTASK_ARG_EVENT_ARG_THREAD_STATE_ARG_STACK_ARG_63B40775
=
+ "Pipe processor worker {} 已连续处理同一 event {}
ms。Pipe：{}，DataRegion：{}，subtask：{}，event：{}，线程状态：{}。栈：{}";
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/connection/PipeEventCollector.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/connection/PipeEventCollector.java
index cc6b126384f..ac0f6c7bf0b 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/connection/PipeEventCollector.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/connection/PipeEventCollector.java
@@ -28,6 +28,8 @@ import org.apache.iotdb.commons.pipe.event.EnrichedEvent;
import org.apache.iotdb.commons.pipe.event.ProgressReportEvent;
import org.apache.iotdb.db.i18n.DataNodePipeMessages;
import org.apache.iotdb.db.pipe.agent.PipeDataNodeAgent;
+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.common.deletion.PipeDeleteDataNodeEvent;
import org.apache.iotdb.db.pipe.event.common.heartbeat.PipeHeartbeatEvent;
import
org.apache.iotdb.db.pipe.event.common.tablet.PipeInsertNodeTabletInsertionEvent;
@@ -63,6 +65,8 @@ public class PipeEventCollector implements EventCollector {
private final boolean skipParsing;
private final boolean isUsedForConsensusPipe;
+ private PipeProcessorSubtaskExecutionGuard processorExecutionGuard =
+ PipeProcessorSubtaskExecutionGuard.disabled();
private final AtomicInteger collectInvocationCount = new AtomicInteger(0);
private boolean hasNoGeneratedEvent = true;
@@ -83,6 +87,11 @@ public class PipeEventCollector implements EventCollector {
this.isUsedForConsensusPipe = isUsedInConsensusPipe;
}
+ public void setProcessorExecutionGuard(
+ final PipeProcessorSubtaskExecutionGuard processorExecutionGuard) {
+ this.processorExecutionGuard = processorExecutionGuard;
+ }
+
@Override
public void collect(final Event event) {
try {
@@ -97,6 +106,8 @@ public class PipeEventCollector implements EventCollector {
} else if (!(event instanceof ProgressReportEvent)) {
collectEvent(event);
}
+ } catch (final PipeProcessorSubtaskYieldException e) {
+ throw e;
} catch (final PipeException e) {
throw e;
} catch (final Exception e) {
@@ -131,7 +142,7 @@ public class PipeEventCollector implements EventCollector {
}
private void parseAndCollectEvent(final PipeTsFileInsertionEvent
sourceEvent) throws Exception {
- if (!sourceEvent.waitForTsFileClose()) {
+ if (!sourceEvent.waitForTsFileClose(processorExecutionGuard)) {
LOGGER.warn(
DataNodePipeMessages.PIPE_SKIPPING_TEMPORARY_TSFILE_WHICH_SHOULDN_T,
sourceEvent.getTsFile());
@@ -148,7 +159,9 @@ public class PipeEventCollector implements EventCollector {
}
sourceEvent.consumeTabletInsertionEventsWithRetry(
- this::collectParsedRawTableEvent,
"PipeEventCollector::parseAndCollectEvent");
+ this::collectParsedRawTableEvent,
+ "PipeEventCollector::parseAndCollectEvent",
+ processorExecutionGuard);
sourceEvent.close();
if (sourceEvent.isGeneratedByHistoricalExtractor()) {
PipeTerminateEvent.markHistoricalTsFileSplit(
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtask.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtask.java
index 8c49e03cb06..0a95465b50e 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtask.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtask.java
@@ -57,6 +57,7 @@ import org.slf4j.LoggerFactory;
import java.util.Objects;
import java.util.concurrent.ExecutorService;
+import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;
public class PipeProcessorSubtask extends PipeReportableSubtask {
@@ -74,6 +75,11 @@ public class PipeProcessorSubtask extends
PipeReportableSubtask {
private final EventSupplier inputEventSupplier;
private final PipeProcessor pipeProcessor;
private final PipeEventCollector outputEventCollector;
+ private final PipeProcessorSubtaskExecutionGuard executionGuard =
+ new PipeProcessorSubtaskExecutionGuard();
+ private final AtomicBoolean isResumingFromYield = new AtomicBoolean(false);
+ private final AtomicReference<EventProcessingContext> eventProcessingContext
=
+ new AtomicReference<>();
// This variable is used to distinguish between old and new subtasks before
and after stuck
// restart.
@@ -94,6 +100,7 @@ public class PipeProcessorSubtask extends
PipeReportableSubtask {
this.inputEventSupplier = inputEventSupplier;
this.pipeProcessor = pipeProcessor;
this.outputEventCollector = outputEventCollector;
+ this.outputEventCollector.setProcessorExecutionGuard(executionGuard);
this.subtaskCreationTime = System.currentTimeMillis();
// Only register dataRegions
@@ -106,7 +113,7 @@ public class PipeProcessorSubtask extends
PipeReportableSubtask {
@Override
public void bindExecutors(
final ListeningExecutorService subtaskWorkerThreadPoolExecutor,
- final ListeningScheduledExecutorService ignoredScheduledExecutor,
+ final ListeningScheduledExecutorService subtaskWorkerScheduledExecutor,
final ExecutorService ignored,
final PipeSubtaskScheduler subtaskScheduler) {
this.subtaskWorkerThreadPoolExecutor = subtaskWorkerThreadPoolExecutor;
@@ -117,19 +124,31 @@ public class PipeProcessorSubtask extends
PipeReportableSubtask {
synchronized (PipeProcessorSubtaskWorkerManager.class) {
if (subtaskWorkerManager.get() == null) {
subtaskWorkerManager.set(
- new
PipeProcessorSubtaskWorkerManager(subtaskWorkerThreadPoolExecutor));
+ new PipeProcessorSubtaskWorkerManager(
+ subtaskWorkerThreadPoolExecutor,
subtaskWorkerScheduledExecutor));
}
}
}
subtaskWorkerManager.get().schedule(this);
}
+ @Override
+ public Boolean call() throws Exception {
+ executionGuard.enter();
+ try {
+ return super.call();
+ } finally {
+ executionGuard.exit();
+ }
+ }
+
@Override
protected boolean executeOnce() throws Exception {
if (isClosed.get()) {
return false;
}
+ executionGuard.check();
final Event event =
lastEvent != null
? lastEvent
@@ -141,7 +160,13 @@ public class PipeProcessorSubtask extends
PipeReportableSubtask {
return false;
}
- outputEventCollector.resetFlags();
+ executionGuard.check();
+ if (!isResumingFromYield.getAndSet(false)) {
+ outputEventCollector.resetFlags();
+ }
+ final EventProcessingContext currentEventProcessingContext =
+ new EventProcessingContext(event, System.nanoTime());
+ eventProcessingContext.set(currentEventProcessingContext);
try {
if (event instanceof EnrichedEvent) {
((EnrichedEvent) event).throwIfNoPrivilege();
@@ -178,13 +203,16 @@ public class PipeProcessorSubtask extends
PipeReportableSubtask {
event1 -> {
try {
pipeProcessor.process(event1, outputEventCollector);
+ } catch (PipeProcessorSubtaskYieldException e) {
+ throw e;
} catch (PipeRuntimeOutOfMemoryCriticalException e) {
throw e;
} catch (Exception e) {
throw new PipeException(e.getMessage(), e);
}
},
- "PipeProcessorSubtask::executeOnce");
+ "PipeProcessorSubtask::executeOnce",
+ executionGuard);
tsFileInsertionEvent.close();
if (tsFileInsertionEvent.isGeneratedByHistoricalExtractor()) {
PipeTerminateEvent.markHistoricalTsFileSplit(
@@ -242,6 +270,9 @@ public class PipeProcessorSubtask extends
PipeReportableSubtask {
.enrichWithCommitterKeyAndCommitId((EnrichedEvent) event,
creationTime, regionId);
}
decreaseReferenceCountAndReleaseLastEvent(event, shouldReport);
+ } catch (final PipeProcessorSubtaskYieldException e) {
+ isResumingFromYield.set(true);
+ throw e;
} catch (final PipeRuntimeOutOfMemoryCriticalException e) {
recordResourceFailure(event, PipeResourceFailureType.MEMORY_TIMEOUT);
PipeLogger.log(
@@ -250,7 +281,12 @@ public class PipeProcessorSubtask extends
PipeReportableSubtask {
e.getMessage());
return false;
} catch (final Exception e) {
- if (ExceptionUtils.getRootCause(e) instanceof
PipeRuntimeOutOfMemoryCriticalException) {
+ final Throwable rootCause = ExceptionUtils.getRootCause(e);
+ if (rootCause instanceof PipeProcessorSubtaskYieldException) {
+ isResumingFromYield.set(true);
+ throw (PipeProcessorSubtaskYieldException) rootCause;
+ }
+ if (rootCause instanceof PipeRuntimeOutOfMemoryCriticalException) {
recordResourceFailure(event, PipeResourceFailureType.MEMORY_TIMEOUT);
PipeLogger.log(
LOGGER::info,
@@ -275,6 +311,8 @@ public class PipeProcessorSubtask extends
PipeReportableSubtask {
e.getMessage() != null ? " Message: " + e.getMessage() : "");
clearReferenceCountAndReleaseLastEvent(event);
}
+ } finally {
+ eventProcessingContext.compareAndSet(currentEventProcessingContext,
null);
}
return true;
@@ -287,6 +325,20 @@ public class PipeProcessorSubtask extends
PipeReportableSubtask {
// and the worker will be submitted to the executor
}
+ @Override
+ protected void onAllowSubmittingSelf() {
+ executionGuard.start();
+ }
+
+ @Override
+ protected void onDisallowSubmittingSelf() {
+ executionGuard.stop();
+ final Event event = lastEvent;
+ if (event instanceof PipeTsFileInsertionEvent) {
+ ((PipeTsFileInsertionEvent)
event).cancelTsFileParserMemoryReservationIfPending();
+ }
+ }
+
public boolean isStoppedByException() {
return lastEvent instanceof EnrichedEvent && retryCount.get() >
MAX_RETRY_TIMES;
}
@@ -316,6 +368,29 @@ public class PipeProcessorSubtask extends
PipeReportableSubtask {
return isClosed.get();
}
+ EventProcessingContext getEventProcessingContext() {
+ return eventProcessingContext.get();
+ }
+
+ static final class EventProcessingContext {
+
+ private final Event event;
+ private final long startTimeInNanos;
+
+ EventProcessingContext(final Event event, final long startTimeInNanos) {
+ this.event = event;
+ this.startTimeInNanos = startTimeInNanos;
+ }
+
+ Event getEvent() {
+ return event;
+ }
+
+ long getStartTimeInNanos() {
+ return startTimeInNanos;
+ }
+ }
+
@Override
public boolean equals(final Object obj) {
if (this == obj) {
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtaskExecutionGuard.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtaskExecutionGuard.java
new file mode 100644
index 00000000000..a4eeab3482d
--- /dev/null
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtaskExecutionGuard.java
@@ -0,0 +1,110 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.db.pipe.agent.task.subtask.processor;
+
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
+
+/**
+ * Guards one processor subtask invocation against concurrent STOP/START
operations.
+ *
+ * <p>An invocation captures the current execution epoch. STOP invalidates
that epoch before START
+ * can enable a new one, so an invocation started before STOP must yield even
if the pipe is started
+ * again immediately.
+ */
+public class PipeProcessorSubtaskExecutionGuard {
+
+ private static final PipeProcessorSubtaskExecutionGuard DISABLED_GUARD =
+ new PipeProcessorSubtaskExecutionGuard(false);
+
+ private final boolean enabled;
+ private final AtomicBoolean isRunning = new AtomicBoolean(false);
+ private final AtomicLong executionEpoch = new AtomicLong(0);
+ private final ThreadLocal<Long> invocationEpoch = new ThreadLocal<>();
+
+ public PipeProcessorSubtaskExecutionGuard() {
+ this(true);
+ }
+
+ private PipeProcessorSubtaskExecutionGuard(final boolean enabled) {
+ this.enabled = enabled;
+ }
+
+ public static PipeProcessorSubtaskExecutionGuard disabled() {
+ return DISABLED_GUARD;
+ }
+
+ public boolean isEnabled() {
+ return enabled;
+ }
+
+ void start() {
+ if (enabled) {
+ isRunning.set(true);
+ }
+ }
+
+ void stop() {
+ if (enabled) {
+ isRunning.set(false);
+ executionEpoch.incrementAndGet();
+ }
+ }
+
+ void enter() {
+ if (!enabled) {
+ return;
+ }
+
+ final long currentEpoch = executionEpoch.get();
+ invocationEpoch.set(currentEpoch);
+ if (!isRunning.get() || currentEpoch != executionEpoch.get()) {
+ invocationEpoch.remove();
+ throw PipeProcessorSubtaskYieldException.pauseRequested();
+ }
+ }
+
+ void exit() {
+ if (enabled) {
+ invocationEpoch.remove();
+ }
+ }
+
+ public void check() {
+ if (!isCurrentInvocationValid()) {
+ throw PipeProcessorSubtaskYieldException.pauseRequested();
+ }
+ }
+
+ public boolean isCurrentInvocationValid() {
+ if (!enabled) {
+ return true;
+ }
+
+ final Long currentInvocationEpoch = invocationEpoch.get();
+ return currentInvocationEpoch != null
+ && isRunning.get()
+ && currentInvocationEpoch == executionEpoch.get();
+ }
+
+ public void yieldIfParserNotAdmitted() {
+ throw PipeProcessorSubtaskYieldException.parserNotAdmitted();
+ }
+}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtaskWorker.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtaskWorker.java
index 2bb4acc1232..bc967cbc2e9 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtaskWorker.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtaskWorker.java
@@ -20,26 +20,51 @@
package org.apache.iotdb.db.pipe.agent.task.subtask.processor;
import org.apache.iotdb.commons.concurrent.WrappedRunnable;
+import org.apache.iotdb.commons.pipe.event.EnrichedEvent;
import org.apache.iotdb.db.i18n.DataNodePipeMessages;
+import org.apache.iotdb.pipe.api.event.Event;
+import com.google.common.annotations.VisibleForTesting;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.Collections;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.TimeUnit;
public class PipeProcessorSubtaskWorker extends WrappedRunnable {
private static final Logger LOGGER =
LoggerFactory.getLogger(PipeProcessorSubtaskWorker.class);
private static final int SLEEP_INTERVAL_ADJUSTMENT_ROUND_INTERVAL = 100;
+ private static final long LONG_RUNNING_EVENT_INITIAL_REPORT_DELAY_IN_NANOS =
+ TimeUnit.MINUTES.toNanos(10);
+ private static final long LONG_RUNNING_EVENT_REPORT_INTERVAL_IN_NANOS =
+ TimeUnit.MINUTES.toNanos(30);
+ private static final int MAX_EVENT_REPORT_LENGTH = 1024;
+ private static final int MAX_STACK_TRACE_DEPTH = 64;
+
private int totalRoundInAdjustmentInterval = 0;
private int workingRoundInAdjustmentInterval = 0;
private long sleepingTimeInMilliSecond = 50;
- private final Set<PipeProcessorSubtask> subtasks =
- Collections.newSetFromMap(new ConcurrentHashMap<>());
+ private final Set<PipeProcessorSubtask> subtasks;
+
+ private volatile Thread workerThread;
+ private volatile PipeProcessorSubtask currentSubtask;
+
+ private PipeProcessorSubtask.EventProcessingContext
lastReportedEventProcessingContext;
+ private long lastEventReportTimeInNanos = Long.MIN_VALUE;
+
+ public PipeProcessorSubtaskWorker() {
+ this(Collections.newSetFromMap(new ConcurrentHashMap<>()));
+ }
+
+ @VisibleForTesting
+ PipeProcessorSubtaskWorker(final Set<PipeProcessorSubtask> subtasks) {
+ this.subtasks = subtasks;
+ }
@Override
@SuppressWarnings("squid:S2189")
@@ -56,7 +81,8 @@ public class PipeProcessorSubtaskWorker extends
WrappedRunnable {
subtasks.removeIf(PipeProcessorSubtask::isClosed);
}
- private boolean runSubtasks() {
+ @VisibleForTesting
+ boolean runSubtasks() {
++totalRoundInAdjustmentInterval;
boolean canSleepBeforeNextRound = true;
@@ -66,18 +92,24 @@ public class PipeProcessorSubtaskWorker extends
WrappedRunnable {
continue;
}
+ workerThread = Thread.currentThread();
+ currentSubtask = subtask;
try {
final boolean hasAtLeastOneEventProcessed = subtask.call();
if (hasAtLeastOneEventProcessed) {
canSleepBeforeNextRound = false;
}
subtask.onSuccess(hasAtLeastOneEventProcessed);
+ } catch (final PipeProcessorSubtaskYieldException ignored) {
+ // The subtask voluntarily yields this worker without succeeding,
failing, or retrying.
} catch (final Exception e) {
if (subtask.isClosed()) {
LOGGER.warn(DataNodePipeMessages.SUBTASK_IS_CLOSED_IGNORE_EXCEPTION,
subtask, e);
} else {
subtask.onFailure(e);
}
+ } finally {
+ currentSubtask = null;
}
}
@@ -118,4 +150,95 @@ public class PipeProcessorSubtaskWorker extends
WrappedRunnable {
public void schedule(final PipeProcessorSubtask pipeProcessorSubtask) {
subtasks.add(pipeProcessorSubtask);
}
+
+ void watchLongRunningEvent() {
+ final PipeProcessorSubtask subtask = currentSubtask;
+ final Thread thread = workerThread;
+ if (subtask == null || thread == null) {
+ return;
+ }
+
+ final PipeProcessorSubtask.EventProcessingContext context =
subtask.getEventProcessingContext();
+ final long currentTimeInNanos = System.nanoTime();
+ if (!isLongRunningEventReportDue(context, currentTimeInNanos)) {
+ return;
+ }
+
+ final StackTraceElement[] stackTrace = thread.getStackTrace();
+ // The event may finish while its stack is being captured. Do not
attribute a later event's
+ // stack to this event.
+ if (currentSubtask != subtask || subtask.getEventProcessingContext() !=
context) {
+ return;
+ }
+
+ markLongRunningEventReported(context, currentTimeInNanos);
+ LOGGER.warn(
+ DataNodePipeMessages
+
.LOG_PIPE_PROCESSOR_WORKER_ARG_HAS_BEEN_PROCESSING_THE_SAME_EVENT_FOR_ARG_MS_PIPE_ARG_DATAREGION_ARG_SUBTASK_ARG_EVENT_ARG_THREAD_STATE_ARG_STACK_ARG_63B40775,
+ thread.getName(),
+ TimeUnit.NANOSECONDS.toMillis(currentTimeInNanos -
context.getStartTimeInNanos()),
+ subtask.getPipeName(),
+ subtask.getRegionId(),
+ subtask.getDisplayTaskID(),
+ getEventReport(context.getEvent()),
+ thread.getState(),
+ formatStackTrace(stackTrace));
+ }
+
+ @VisibleForTesting
+ boolean isLongRunningEventReportDue(
+ final PipeProcessorSubtask.EventProcessingContext context, final long
currentTimeInNanos) {
+ if (context == null
+ || currentTimeInNanos - context.getStartTimeInNanos()
+ < LONG_RUNNING_EVENT_INITIAL_REPORT_DELAY_IN_NANOS) {
+ return false;
+ }
+
+ return lastReportedEventProcessingContext != context
+ || currentTimeInNanos - lastEventReportTimeInNanos
+ >= LONG_RUNNING_EVENT_REPORT_INTERVAL_IN_NANOS;
+ }
+
+ @VisibleForTesting
+ void markLongRunningEventReported(
+ final PipeProcessorSubtask.EventProcessingContext context, final long
currentTimeInNanos) {
+ lastReportedEventProcessingContext = context;
+ lastEventReportTimeInNanos = currentTimeInNanos;
+ }
+
+ @VisibleForTesting
+ static String getEventReport(final Event event) {
+ String report = event.getClass().getName();
+ if (event instanceof EnrichedEvent) {
+ try {
+ report =
+ event.getClass().getSimpleName() + ": " + ((EnrichedEvent)
event).coreReportMessage();
+ } catch (final RuntimeException ignored) {
+ // Keep the event class name if its diagnostic method fails.
+ }
+ }
+
+ report = report.replace('\n', ' ').replace('\r', ' ');
+ return report.length() <= MAX_EVENT_REPORT_LENGTH
+ ? report
+ : report.substring(0, MAX_EVENT_REPORT_LENGTH) + "...";
+ }
+
+ @VisibleForTesting
+ static String formatStackTrace(final StackTraceElement[] stackTrace) {
+ final StringBuilder builder = new StringBuilder();
+ final int frameCount = Math.min(stackTrace.length, MAX_STACK_TRACE_DEPTH);
+ for (int i = 0; i < frameCount; ++i) {
+ builder.append('\n').append('\t').append(stackTrace[i]);
+ }
+ if (stackTrace.length > frameCount) {
+ builder
+ .append('\n')
+ .append('\t')
+ .append("... (")
+ .append(stackTrace.length - frameCount)
+ .append(')');
+ }
+ return builder.toString();
+ }
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtaskWorkerManager.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtaskWorkerManager.java
index 33d58c4b5d4..ac2dd2cd7b5 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtaskWorkerManager.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtaskWorkerManager.java
@@ -19,10 +19,13 @@
package org.apache.iotdb.db.pipe.agent.task.subtask.processor;
+import org.apache.iotdb.commons.concurrent.threadpool.ScheduledExecutorUtil;
import org.apache.iotdb.commons.pipe.config.PipeConfig;
import com.google.common.util.concurrent.ListeningExecutorService;
+import com.google.common.util.concurrent.ListeningScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
public class PipeProcessorSubtaskWorkerManager {
@@ -34,7 +37,9 @@ public class PipeProcessorSubtaskWorkerManager {
private final AtomicLong scheduledTaskNumber;
- public PipeProcessorSubtaskWorkerManager(ListeningExecutorService
workerThreadPoolExecutor) {
+ public PipeProcessorSubtaskWorkerManager(
+ final ListeningExecutorService workerThreadPoolExecutor,
+ final ListeningScheduledExecutorService watcherScheduledExecutor) {
workers = new PipeProcessorSubtaskWorker[MAX_THREAD_NUM];
for (int i = 0; i < MAX_THREAD_NUM; i++) {
workers[i] = new PipeProcessorSubtaskWorker();
@@ -42,10 +47,18 @@ public class PipeProcessorSubtaskWorkerManager {
}
scheduledTaskNumber = new AtomicLong(0);
+ ScheduledExecutorUtil.safelyScheduleWithFixedDelay(
+ watcherScheduledExecutor, this::watchLongRunningEvents, 1, 1,
TimeUnit.MINUTES);
}
public void schedule(PipeProcessorSubtask pipeProcessorSubtask) {
workers[(int) (scheduledTaskNumber.getAndIncrement() %
MAX_THREAD_NUM)].schedule(
pipeProcessorSubtask);
}
+
+ private void watchLongRunningEvents() {
+ for (final PipeProcessorSubtaskWorker worker : workers) {
+ worker.watchLongRunningEvent();
+ }
+ }
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtaskYieldException.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtaskYieldException.java
new file mode 100644
index 00000000000..3fe5242fa07
--- /dev/null
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtaskYieldException.java
@@ -0,0 +1,53 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.db.pipe.agent.task.subtask.processor;
+
+/** Internal control-flow exception that immediately yields the current
processor worker. */
+public final class PipeProcessorSubtaskYieldException extends RuntimeException
{
+
+ private static final PipeProcessorSubtaskYieldException
PAUSE_REQUESTED_INSTANCE =
+ new PipeProcessorSubtaskYieldException(Reason.PAUSE_REQUESTED);
+ private static final PipeProcessorSubtaskYieldException
PARSER_NOT_ADMITTED_INSTANCE =
+ new PipeProcessorSubtaskYieldException(Reason.PARSER_NOT_ADMITTED);
+
+ private final Reason reason;
+
+ private PipeProcessorSubtaskYieldException(final Reason reason) {
+ super(null, null, false, false);
+ this.reason = reason;
+ }
+
+ public static PipeProcessorSubtaskYieldException pauseRequested() {
+ return PAUSE_REQUESTED_INSTANCE;
+ }
+
+ public static PipeProcessorSubtaskYieldException parserNotAdmitted() {
+ return PARSER_NOT_ADMITTED_INSTANCE;
+ }
+
+ public Reason getReason() {
+ return reason;
+ }
+
+ public enum Reason {
+ PAUSE_REQUESTED,
+ PARSER_NOT_ADMITTED
+ }
+}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeTsFileInsertionEvent.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeTsFileInsertionEvent.java
index ba2a9d275c3..b969ac4e618 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeTsFileInsertionEvent.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeTsFileInsertionEvent.java
@@ -38,6 +38,8 @@ import
org.apache.iotdb.commons.pipe.resource.ref.PipePhantomReferenceManager.Pi
import
org.apache.iotdb.commons.queryengine.plan.relational.metadata.QualifiedObjectName;
import org.apache.iotdb.db.auth.AuthorityChecker;
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;
import org.apache.iotdb.db.pipe.event.common.PipeInsertionEvent;
import
org.apache.iotdb.db.pipe.event.common.tablet.PipeRawTabletInsertionEvent;
@@ -314,6 +316,13 @@ public class PipeTsFileInsertionEvent extends
PipeInsertionEvent
* otherwise.
*/
public boolean waitForTsFileClose() throws InterruptedException {
+ return waitForTsFileClose(PipeProcessorSubtaskExecutionGuard.disabled());
+ }
+
+ public boolean waitForTsFileClose(
+ final PipeProcessorSubtaskExecutionGuard processorExecutionGuard)
+ throws InterruptedException {
+ processorExecutionGuard.check();
if (Objects.isNull(resource)) {
return true;
}
@@ -327,7 +336,9 @@ public class PipeTsFileInsertionEvent extends
PipeInsertionEvent
synchronized (isClosed) {
while (!isClosed.get()) {
+ processorExecutionGuard.check();
isClosed.wait(100);
+ processorExecutionGuard.check();
final boolean isClosedNow = resource.isClosed();
if (isClosedNow) {
@@ -770,19 +781,41 @@ public class PipeTsFileInsertionEvent extends
PipeInsertionEvent
public void consumeTabletInsertionEventsWithRetry(
final TabletInsertionEventConsumer consumer, final String callerName)
throws Exception {
+ consumeTabletInsertionEventsWithRetry(
+ consumer, callerName, PipeProcessorSubtaskExecutionGuard.disabled());
+ }
+
+ public void consumeTabletInsertionEventsWithRetry(
+ final TabletInsertionEventConsumer consumer,
+ final String callerName,
+ final PipeProcessorSubtaskExecutionGuard processorExecutionGuard)
+ throws Exception {
try {
while (true) {
+ processorExecutionGuard.check();
final PipeRawTabletInsertionEvent parsedEvent =
- getNextTabletInsertionEventFromSavedProgress();
+
getNextTabletInsertionEventFromSavedProgress(processorExecutionGuard);
if (parsedEvent == null) {
isTsFileParsingCompleted.set(true);
releaseTsFileParserMemoryIfReserved();
return;
}
+ processorExecutionGuard.check();
consumeParsedTabletInsertionEventWithRetry(
- consumer, callerName, parsedTabletInsertionEventCount.get(),
parsedEvent);
+ consumer,
+ callerName,
+ parsedTabletInsertionEventCount.get(),
+ parsedEvent,
+ processorExecutionGuard);
pendingTabletInsertionEvent.compareAndSet(parsedEvent, null);
+ processorExecutionGuard.check();
}
+ } catch (final PipeProcessorSubtaskYieldException e) {
+ releaseTsFileParserMemoryIfReserved();
+ if (!processorExecutionGuard.isCurrentInvocationValid()) {
+ cancelTsFileParserMemoryReservationIfPending();
+ }
+ throw e;
} catch (final PipeRuntimeOutOfMemoryCriticalException e) {
// Yield the active parser slot to the next pipe while retaining the
iterator and current
// tablet. The next retry resumes from this exact tablet instead of
reparsing the TsFile.
@@ -800,16 +833,15 @@ public class PipeTsFileInsertionEvent extends
PipeInsertionEvent
}
}
- private PipeRawTabletInsertionEvent
getNextTabletInsertionEventFromSavedProgress()
- throws Exception {
+ private PipeRawTabletInsertionEvent
getNextTabletInsertionEventFromSavedProgress(
+ final PipeProcessorSubtaskExecutionGuard processorExecutionGuard) throws
Exception {
if (isTsFileParsingCompleted.get()) {
return null;
}
- // Reacquire parser memory after a previous failure yielded the active
parser slot. This wait
- // is already bounded to 20-40 seconds, while the exponential backoff
below is only for retrying
- // the current tablet without yielding its parser slot.
- waitForResourceEnough4Parsing((long) ((1 + Math.random()) * 20 * 1000));
+ // Reacquire parser memory after a previous failure yielded the active
parser slot. Processor
+ // subtasks use non-blocking admission here, while other callers retain
the bounded wait.
+ reserveResource4Parsing(processorExecutionGuard);
final PipeRawTabletInsertionEvent pendingEvent =
pendingTabletInsertionEvent.get();
if (pendingEvent != null) {
@@ -818,7 +850,7 @@ public class PipeTsFileInsertionEvent extends
PipeInsertionEvent
Iterator<TabletInsertionEvent> iterator =
tabletInsertionEventIterator.get();
if (iterator == null) {
- if (!waitForTsFileClose()) {
+ if (!waitForTsFileClose(processorExecutionGuard)) {
LOGGER.warn(DataNodePipeMessages.PIPE_SKIPPING_TEMPORARY_TSFILE_S_PARSING_WHICH,
tsFile);
return null;
}
@@ -840,12 +872,14 @@ public class PipeTsFileInsertionEvent extends
PipeInsertionEvent
final TabletInsertionEventConsumer consumer,
final String callerName,
final int tabletEventCount,
- final TabletInsertionEvent parsedEvent)
+ final TabletInsertionEvent parsedEvent,
+ final PipeProcessorSubtaskExecutionGuard processorExecutionGuard)
throws Exception {
final PipeMemoryManager memoryManager =
PipeDataNodeResourceManager.memory();
long firstOutOfMemoryTimeInMs = Long.MIN_VALUE;
int retryCount = 0;
while (true) {
+ processorExecutionGuard.check();
try {
consumer.consume((PipeRawTabletInsertionEvent) parsedEvent);
return;
@@ -859,7 +893,7 @@ public class PipeTsFileInsertionEvent extends
PipeInsertionEvent
}
logParserRetryOnOutOfMemory(callerName, tabletEventCount, retryCount,
e);
try {
- Thread.sleep(getParserRetryBackoffInMs(retryCount));
+ sleepForParserRetry(getParserRetryBackoffInMs(retryCount),
processorExecutionGuard);
} catch (final InterruptedException interruptedException) {
Thread.currentThread().interrupt();
throw e;
@@ -868,6 +902,24 @@ public class PipeTsFileInsertionEvent extends
PipeInsertionEvent
}
}
+ private void sleepForParserRetry(
+ final long sleepTimeInMs, final PipeProcessorSubtaskExecutionGuard
processorExecutionGuard)
+ throws InterruptedException {
+ if (!processorExecutionGuard.isEnabled()) {
+ Thread.sleep(sleepTimeInMs);
+ return;
+ }
+
+ final long deadlineInMs = System.currentTimeMillis() + sleepTimeInMs;
+ long remainingTimeInMs = sleepTimeInMs;
+ while (remainingTimeInMs > 0) {
+ processorExecutionGuard.check();
+ Thread.sleep(Math.min(remainingTimeInMs, 100));
+ processorExecutionGuard.check();
+ remainingTimeInMs = deadlineInMs - System.currentTimeMillis();
+ }
+ }
+
private long getParserRetryBackoffInMs(final int retryCount) {
final long initialBackoffInMs =
Math.max(1,
PipeConfig.getInstance().getPipeMemoryAllocateRetryIntervalInMs());
@@ -950,6 +1002,33 @@ public class PipeTsFileInsertionEvent extends
PipeInsertionEvent
}
}
+ private void reserveResource4Parsing(
+ final PipeProcessorSubtaskExecutionGuard processorExecutionGuard)
+ throws InterruptedException {
+ if (!processorExecutionGuard.isEnabled()) {
+ waitForResourceEnough4Parsing((long) ((1 + Math.random()) * 20 * 1000));
+ return;
+ }
+
+ processorExecutionGuard.check();
+ final PipeMemoryManager memoryManager =
PipeDataNodeResourceManager.memory();
+ if (tryReserveTsFileParserMemory(memoryManager)) {
+ try {
+ processorExecutionGuard.check();
+ return;
+ } catch (final PipeProcessorSubtaskYieldException e) {
+ releaseTsFileParserMemoryIfReserved();
+ throw e;
+ }
+ }
+
+ if (!processorExecutionGuard.isCurrentInvocationValid()) {
+ cancelTsFileParserMemoryReservationIfPending();
+ processorExecutionGuard.check();
+ }
+ processorExecutionGuard.yieldIfParserNotAdmitted();
+ }
+
private void waitForResourceEnough4Parsing(final long timeoutMs) throws
InterruptedException {
final PipeMemoryManager memoryManager =
PipeDataNodeResourceManager.memory();
if (tryReserveTsFileParserMemory(memoryManager)) {
@@ -1026,7 +1105,7 @@ public class PipeTsFileInsertionEvent extends
PipeInsertionEvent
}
}
- private void cancelTsFileParserMemoryReservationIfPending() {
+ public void cancelTsFileParserMemoryReservationIfPending() {
if (!isTsFileParserMemoryReserved.get()) {
PipeDataNodeResourceManager.memory()
.cancelTsFileParserMemoryReservation(
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtaskExecutionGuardTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtaskExecutionGuardTest.java
new file mode 100644
index 00000000000..3438dc62c5d
--- /dev/null
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtaskExecutionGuardTest.java
@@ -0,0 +1,247 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.db.pipe.agent.task.subtask.processor;
+
+import org.apache.iotdb.commons.conf.CommonConfig;
+import org.apache.iotdb.commons.conf.CommonDescriptor;
+import org.apache.iotdb.commons.pipe.datastructure.pattern.PrefixTreePattern;
+import org.apache.iotdb.commons.utils.FileUtils;
+import org.apache.iotdb.db.pipe.event.common.tsfile.PipeTsFileInsertionEvent;
+import org.apache.iotdb.db.pipe.resource.PipeDataNodeResourceManager;
+import org.apache.iotdb.db.pipe.resource.memory.PipeMemoryManager;
+import
org.apache.iotdb.db.pipe.resource.memory.PipeMemoryManager.TsFileParserMemoryReservation;
+import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource;
+import
org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResourceStatus;
+
+import org.apache.tsfile.file.metadata.IDeviceID;
+import org.apache.tsfile.utils.TsFileGeneratorUtils;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.io.File;
+import java.nio.file.Files;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+
+public class PipeProcessorSubtaskExecutionGuardTest {
+
+ @Test
+ public void testStopAndImmediateRestartInvalidateCurrentInvocation() {
+ final PipeProcessorSubtaskExecutionGuard executionGuard =
+ new PipeProcessorSubtaskExecutionGuard();
+
+ executionGuard.start();
+ executionGuard.enter();
+ executionGuard.check();
+
+ executionGuard.stop();
+ executionGuard.start();
+ Assert.assertThrows(PipeProcessorSubtaskYieldException.class,
executionGuard::check);
+
+ executionGuard.exit();
+ executionGuard.enter();
+ executionGuard.check();
+ executionGuard.exit();
+ }
+
+ @Test(timeout = 60000)
+ public void testParserAdmissionYieldsWithoutBlockingAndResumes() throws
Exception {
+ final CommonConfig commonConfig =
CommonDescriptor.getInstance().getConfig();
+ final PipeMemoryManager memoryManager =
PipeDataNodeResourceManager.memory();
+ final long originalParserMemoryInBytes =
commonConfig.getPipeTsFileParserMemory();
+ final int originalGlobalLimit =
commonConfig.getPipeTsFileParserInFlightMaxNum();
+ final int originalPerPipeRegionLimit =
+ commonConfig.getPipeTsFileParserInFlightMaxNumPerPipeRegion();
+ final TsFileParserMemoryReservation blockerReservation = new
TsFileParserMemoryReservation();
+ final TsFileParserMemoryReservation competitorReservation = new
TsFileParserMemoryReservation();
+
+ final File tempDir =
Files.createTempDirectory("pipeProcessorAdmissionYield").toFile();
+ PipeTsFileInsertionEvent event = null;
+ boolean isBlockerReserved = false;
+ boolean isCompetitorReserved = false;
+ try {
+ commonConfig.setPipeTsFileParserMemory(1);
+ commonConfig.setPipeTsFileParserInFlightMaxNum(1);
+ commonConfig.setPipeTsFileParserInFlightMaxNumPerPipeRegion(1);
+ isBlockerReserved =
+ memoryManager.tryReserveTsFileParserMemory("blocker", 0, "0",
blockerReservation);
+ Assert.assertTrue(isBlockerReserved);
+
+ event = createEvent(tempDir, "admission.tsfile", "admissionPipe");
+ final PipeTsFileInsertionEvent eventToConsume = event;
+ final PipeProcessorSubtaskExecutionGuard executionGuard =
+ new PipeProcessorSubtaskExecutionGuard();
+ executionGuard.start();
+ executionGuard.enter();
+
+ final long startTimeInNanos = System.nanoTime();
+ final PipeProcessorSubtaskYieldException admissionYield =
+ Assert.assertThrows(
+ PipeProcessorSubtaskYieldException.class,
+ () ->
+ eventToConsume.consumeTabletInsertionEventsWithRetry(
+ parsedEvent ->
parsedEvent.clearReferenceCount(getClass().getName()),
+ "test",
+ executionGuard));
+ Assert.assertEquals(
+ PipeProcessorSubtaskYieldException.Reason.PARSER_NOT_ADMITTED,
+ admissionYield.getReason());
+ Assert.assertTrue(TimeUnit.NANOSECONDS.toMillis(System.nanoTime() -
startTimeInNanos) < 1000);
+ executionGuard.exit();
+
+ executionGuard.stop();
+ event.cancelTsFileParserMemoryReservationIfPending();
+ memoryManager.releaseTsFileParserMemory("blocker", 0, "0");
+ isBlockerReserved = false;
+ isCompetitorReserved =
+ memoryManager.tryReserveTsFileParserMemory("competitor", 0, "0",
competitorReservation);
+ Assert.assertTrue(isCompetitorReserved);
+ memoryManager.releaseTsFileParserMemory("competitor", 0, "0");
+ isCompetitorReserved = false;
+
+ final AtomicInteger consumedTabletCount = new AtomicInteger(0);
+ executionGuard.start();
+ executionGuard.enter();
+ event.consumeTabletInsertionEventsWithRetry(
+ parsedEvent -> {
+ consumedTabletCount.incrementAndGet();
+ parsedEvent.clearReferenceCount(getClass().getName());
+ },
+ "test",
+ executionGuard);
+ executionGuard.exit();
+ Assert.assertTrue(consumedTabletCount.get() > 0);
+ } finally {
+ if (event != null) {
+ event.close();
+ }
+ memoryManager.cancelTsFileParserMemoryReservation("blocker", 0, "0",
blockerReservation);
+ memoryManager.cancelTsFileParserMemoryReservation(
+ "competitor", 0, "0", competitorReservation);
+ if (isBlockerReserved) {
+ memoryManager.releaseTsFileParserMemory("blocker", 0, "0");
+ }
+ if (isCompetitorReserved) {
+ memoryManager.releaseTsFileParserMemory("competitor", 0, "0");
+ }
+ commonConfig.setPipeTsFileParserMemory(originalParserMemoryInBytes);
+ commonConfig.setPipeTsFileParserInFlightMaxNum(originalGlobalLimit);
+
commonConfig.setPipeTsFileParserInFlightMaxNumPerPipeRegion(originalPerPipeRegionLimit);
+ FileUtils.deleteFileOrDirectory(tempDir);
+ }
+ }
+
+ @Test(timeout = 60000)
+ public void testPauseAfterTabletResumesWithoutDuplicateConsumption() throws
Exception {
+ final CommonConfig commonConfig =
CommonDescriptor.getInstance().getConfig();
+ final long originalParserMemoryInBytes =
commonConfig.getPipeTsFileParserMemory();
+ final int originalGlobalLimit =
commonConfig.getPipeTsFileParserInFlightMaxNum();
+ final int originalPerPipeRegionLimit =
+ commonConfig.getPipeTsFileParserInFlightMaxNumPerPipeRegion();
+ final File tempDir =
Files.createTempDirectory("pipeProcessorPauseResume").toFile();
+ final PipeTsFileInsertionEvent event = createEvent(tempDir,
"resume.tsfile", "resumePipe");
+ final PipeProcessorSubtaskExecutionGuard executionGuard =
+ new PipeProcessorSubtaskExecutionGuard();
+ final AtomicInteger consumedTabletCount = new AtomicInteger(0);
+ final AtomicReference<Object> firstTablet = new AtomicReference<>();
+
+ try {
+ commonConfig.setPipeTsFileParserMemory(1);
+ commonConfig.setPipeTsFileParserInFlightMaxNum(1);
+ commonConfig.setPipeTsFileParserInFlightMaxNumPerPipeRegion(1);
+ executionGuard.start();
+ executionGuard.enter();
+ final PipeProcessorSubtaskYieldException pauseYield =
+ Assert.assertThrows(
+ PipeProcessorSubtaskYieldException.class,
+ () ->
+ event.consumeTabletInsertionEventsWithRetry(
+ parsedEvent -> {
+ firstTablet.set(parsedEvent);
+ consumedTabletCount.incrementAndGet();
+ parsedEvent.clearReferenceCount(getClass().getName());
+ executionGuard.stop();
+ },
+ "test",
+ executionGuard));
+ Assert.assertEquals(
+ PipeProcessorSubtaskYieldException.Reason.PAUSE_REQUESTED,
pauseYield.getReason());
+ executionGuard.exit();
+
+ executionGuard.start();
+ executionGuard.enter();
+ try {
+ event.consumeTabletInsertionEventsWithRetry(
+ parsedEvent -> {
+ Assert.assertNotSame(firstTablet.get(), parsedEvent);
+ consumedTabletCount.incrementAndGet();
+ parsedEvent.clearReferenceCount(getClass().getName());
+ },
+ "test",
+ executionGuard);
+ } catch (final PipeProcessorSubtaskYieldException e) {
+ Assert.fail("Unexpected yield reason: " + e.getReason());
+ }
+ executionGuard.exit();
+
+ Assert.assertTrue(consumedTabletCount.get() > 0);
+ } finally {
+ event.close();
+ commonConfig.setPipeTsFileParserMemory(originalParserMemoryInBytes);
+ commonConfig.setPipeTsFileParserInFlightMaxNum(originalGlobalLimit);
+
commonConfig.setPipeTsFileParserInFlightMaxNumPerPipeRegion(originalPerPipeRegionLimit);
+ FileUtils.deleteFileOrDirectory(tempDir);
+ }
+ }
+
+ private PipeTsFileInsertionEvent createEvent(
+ final File tempDir, final String fileName, final String pipeName) throws
Exception {
+ final File tsFile =
+ TsFileGeneratorUtils.generateNonAlignedTsFile(
+ new File(tempDir, fileName).getAbsolutePath(), 1, 1, 10, 0, 100,
10, 10);
+ final TsFileResource resource = new TsFileResource(tsFile);
+ resource.setStatusForTest(TsFileResourceStatus.NORMAL);
+ final IDeviceID deviceID =
IDeviceID.Factory.DEFAULT_FACTORY.create("root.testsg.d0");
+ resource.updateStartTime(deviceID, 0);
+ resource.updateEndTime(deviceID, 9);
+
+ return new PipeTsFileInsertionEvent(
+ false,
+ "root",
+ resource,
+ null,
+ false,
+ false,
+ false,
+ null,
+ pipeName,
+ 0,
+ null,
+ new PrefixTreePattern("root"),
+ null,
+ null,
+ null,
+ null,
+ true,
+ Long.MIN_VALUE,
+ Long.MAX_VALUE);
+ }
+}
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtaskWorkerTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtaskWorkerTest.java
new file mode 100644
index 00000000000..51a018d48d8
--- /dev/null
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtaskWorkerTest.java
@@ -0,0 +1,151 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.db.pipe.agent.task.subtask.processor;
+
+import org.apache.iotdb.commons.pipe.event.EnrichedEvent;
+
+import com.google.common.util.concurrent.ListeningExecutorService;
+import com.google.common.util.concurrent.ListeningScheduledExecutorService;
+import org.junit.Assert;
+import org.junit.Test;
+import org.mockito.InOrder;
+
+import java.util.LinkedHashSet;
+import java.util.concurrent.TimeUnit;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.atLeastOnce;
+import static org.mockito.Mockito.inOrder;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+public class PipeProcessorSubtaskWorkerTest {
+
+ @Test
+ public void testYieldingPipesDoNotBlockAnotherPipeOnSameWorker() throws
Exception {
+ final PipeProcessorSubtaskWorker worker = new
PipeProcessorSubtaskWorker(new LinkedHashSet<>());
+ final PipeProcessorSubtask stoppedPipe =
createRunnableSubtask("stoppedPipe");
+ final PipeProcessorSubtask parserWaitingPipe =
createRunnableSubtask("parserWaitingPipe");
+ final PipeProcessorSubtask runningPipe =
createRunnableSubtask("runningPipe");
+
+
when(stoppedPipe.call()).thenThrow(PipeProcessorSubtaskYieldException.pauseRequested());
+ when(parserWaitingPipe.call())
+ .thenThrow(PipeProcessorSubtaskYieldException.parserNotAdmitted());
+ when(runningPipe.call()).thenReturn(true);
+
+ worker.schedule(stoppedPipe);
+ worker.schedule(parserWaitingPipe);
+ worker.schedule(runningPipe);
+
+ Assert.assertFalse(worker.runSubtasks());
+
+ final InOrder inOrder = inOrder(stoppedPipe, parserWaitingPipe,
runningPipe);
+ inOrder.verify(stoppedPipe).call();
+ inOrder.verify(parserWaitingPipe).call();
+ inOrder.verify(runningPipe).call();
+ verify(runningPipe).onSuccess(true);
+ verify(stoppedPipe, never()).onSuccess(any());
+ verify(stoppedPipe, never()).onFailure(any());
+ verify(parserWaitingPipe, never()).onSuccess(any());
+ verify(parserWaitingPipe, never()).onFailure(any());
+ }
+
+ @Test
+ public void testLongRunningEventReportIsRateLimited() {
+ final PipeProcessorSubtaskWorker worker = new
PipeProcessorSubtaskWorker(new LinkedHashSet<>());
+ final long startTimeInNanos = 100;
+ final PipeProcessorSubtask.EventProcessingContext context =
+ new PipeProcessorSubtask.EventProcessingContext(
+ mock(EnrichedEvent.class), startTimeInNanos);
+ final long initialReportDelayInNanos = TimeUnit.MINUTES.toNanos(10);
+ final long reportIntervalInNanos = TimeUnit.MINUTES.toNanos(30);
+
+ Assert.assertFalse(
+ worker.isLongRunningEventReportDue(
+ context, startTimeInNanos + initialReportDelayInNanos - 1));
+ Assert.assertTrue(
+ worker.isLongRunningEventReportDue(context, startTimeInNanos +
initialReportDelayInNanos));
+
+ final long firstReportTimeInNanos = startTimeInNanos +
initialReportDelayInNanos;
+ worker.markLongRunningEventReported(context, firstReportTimeInNanos);
+ Assert.assertFalse(
+ worker.isLongRunningEventReportDue(
+ context, firstReportTimeInNanos + reportIntervalInNanos - 1));
+ Assert.assertTrue(
+ worker.isLongRunningEventReportDue(
+ context, firstReportTimeInNanos + reportIntervalInNanos));
+
+ final long nextEventStartTimeInNanos = firstReportTimeInNanos + 1;
+ final PipeProcessorSubtask.EventProcessingContext nextContext =
+ new PipeProcessorSubtask.EventProcessingContext(
+ mock(EnrichedEvent.class), nextEventStartTimeInNanos);
+ Assert.assertFalse(
+ worker.isLongRunningEventReportDue(
+ nextContext, nextEventStartTimeInNanos + initialReportDelayInNanos
- 1));
+ Assert.assertTrue(
+ worker.isLongRunningEventReportDue(
+ nextContext, nextEventStartTimeInNanos +
initialReportDelayInNanos));
+ }
+
+ @Test
+ public void testLongRunningEventLogPayloadIsBounded() {
+ final EnrichedEvent event = mock(EnrichedEvent.class);
+ when(event.coreReportMessage()).thenReturn("x".repeat(2048) + "\nmore");
+
+ final String eventReport =
PipeProcessorSubtaskWorker.getEventReport(event);
+ Assert.assertEquals(1027, eventReport.length());
+ Assert.assertFalse(eventReport.contains("\n"));
+ Assert.assertTrue(eventReport.endsWith("..."));
+
+ final StackTraceElement[] stackTrace = new StackTraceElement[100];
+ for (int i = 0; i < stackTrace.length; ++i) {
+ stackTrace[i] = new StackTraceElement("Class", "method" + i,
"File.java", i);
+ }
+ final String formattedStackTrace =
PipeProcessorSubtaskWorker.formatStackTrace(stackTrace);
+ Assert.assertTrue(formattedStackTrace.contains("method63"));
+ Assert.assertFalse(formattedStackTrace.contains("method64"));
+ Assert.assertTrue(formattedStackTrace.contains("... (36)"));
+ }
+
+ @Test
+ @SuppressWarnings("unsafeThreadSchedule")
+ public void testWorkerManagerSchedulesWatcher() {
+ final ListeningExecutorService workerThreadPoolExecutor =
mock(ListeningExecutorService.class);
+ final ListeningScheduledExecutorService watcherScheduledExecutor =
+ mock(ListeningScheduledExecutorService.class);
+
+ new PipeProcessorSubtaskWorkerManager(workerThreadPoolExecutor,
watcherScheduledExecutor);
+
+ verify(workerThreadPoolExecutor,
atLeastOnce()).submit(any(Runnable.class));
+ verify(watcherScheduledExecutor)
+ .scheduleWithFixedDelay(any(Runnable.class), eq(1L), eq(1L),
eq(TimeUnit.MINUTES));
+ }
+
+ private PipeProcessorSubtask createRunnableSubtask(final String mockName) {
+ final PipeProcessorSubtask subtask = mock(PipeProcessorSubtask.class,
mockName);
+ when(subtask.isClosed()).thenReturn(false);
+ when(subtask.isSubmittingSelf()).thenReturn(true);
+ when(subtask.isStoppedByException()).thenReturn(false);
+ return subtask;
+ }
+}
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/subtask/PipeSubtask.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/subtask/PipeSubtask.java
index d87fe3a5f38..108f202e002 100644
---
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/subtask/PipeSubtask.java
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/subtask/PipeSubtask.java
@@ -140,9 +140,14 @@ public abstract class PipeSubtask
public void allowSubmittingSelf() {
retryCount.set(0);
+ onAllowSubmittingSelf();
shouldStopSubmittingSelf.set(false);
}
+ protected void onAllowSubmittingSelf() {
+ // Do nothing by default.
+ }
+
/**
* Set the {@link PipeSubtask#shouldStopSubmittingSelf} state from {@code
false} to {@code true},
* in order to stop submitting the {@link PipeSubtask}.
@@ -151,7 +156,15 @@ public abstract class PipeSubtask
* {@code false} to {@code true}, {@code false} otherwise
*/
public boolean disallowSubmittingSelf() {
- return !shouldStopSubmittingSelf.getAndSet(true);
+ final boolean isChanged = !shouldStopSubmittingSelf.getAndSet(true);
+ if (isChanged) {
+ onDisallowSubmittingSelf();
+ }
+ return isChanged;
+ }
+
+ protected void onDisallowSubmittingSelf() {
+ // Do nothing by default.
}
public boolean isSubmittingSelf() {
