From 6a20e60a7d57cc1d5e741d21f20961ca8475ad7c Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Tue, 4 Aug 2026 19:35:54 +0800 Subject: [PATCH 1/4] Fix processor worker starvation on pipe stop --- .../task/connection/PipeEventCollector.java | 17 +- .../processor/PipeProcessorSubtask.java | 47 +++- .../PipeProcessorSubtaskExecutionGuard.java | 110 ++++++++ .../processor/PipeProcessorSubtaskWorker.java | 2 + .../PipeProcessorSubtaskYieldException.java | 53 ++++ .../tsfile/PipeTsFileInsertionEvent.java | 103 +++++++- ...ipeProcessorSubtaskExecutionGuardTest.java | 247 ++++++++++++++++++ .../pipe/agent/task/subtask/PipeSubtask.java | 15 +- 8 files changed, 577 insertions(+), 17 deletions(-) create mode 100644 iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtaskExecutionGuard.java create mode 100644 iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtaskYieldException.java create mode 100644 iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtaskExecutionGuardTest.java 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 cc6b126384f63..ac0f6c7bf0b74 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.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 PipeEventCollector( 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 void collect(final Event event) { } 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 @@ private void parseAndCollectEvent(final PipeRawTabletInsertionEvent sourceEvent) } 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 @@ private void parseAndCollectEvent(final PipeTsFileInsertionEvent sourceEvent) th } 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 a95c99ebd52e9..33465df453a24 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 @@ -56,6 +56,7 @@ 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 { @@ -73,6 +74,9 @@ 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); // This variable is used to distinguish between old and new subtasks before and after stuck // restart. @@ -93,6 +97,7 @@ public PipeProcessorSubtask( this.inputEventSupplier = inputEventSupplier; this.pipeProcessor = pipeProcessor; this.outputEventCollector = outputEventCollector; + this.outputEventCollector.setProcessorExecutionGuard(executionGuard); this.subtaskCreationTime = System.currentTimeMillis(); // Only register dataRegions @@ -123,12 +128,23 @@ public void bindExecutors( 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 @@ -140,7 +156,10 @@ protected boolean executeOnce() throws Exception { return false; } - outputEventCollector.resetFlags(); + executionGuard.check(); + if (!isResumingFromYield.getAndSet(false)) { + outputEventCollector.resetFlags(); + } try { if (event instanceof EnrichedEvent) { ((EnrichedEvent) event).throwIfNoPrivilege(); @@ -177,13 +196,16 @@ protected boolean executeOnce() throws Exception { 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( @@ -241,6 +263,9 @@ protected boolean executeOnce() throws Exception { .enrichWithCommitterKeyAndCommitId((EnrichedEvent) event, creationTime, regionId); } decreaseReferenceCountAndReleaseLastEvent(event, shouldReport); + } catch (final PipeProcessorSubtaskYieldException e) { + isResumingFromYield.set(true); + throw e; } catch (final PipeRuntimeOutOfMemoryCriticalException e) { PipeLogger.log( LOGGER::info, @@ -248,6 +273,10 @@ protected boolean executeOnce() throws Exception { e.getMessage()); return false; } catch (final Exception e) { + if (ExceptionUtils.getRootCause(e) instanceof PipeProcessorSubtaskYieldException) { + isResumingFromYield.set(true); + throw (PipeProcessorSubtaskYieldException) ExceptionUtils.getRootCause(e); + } if (ExceptionUtils.getRootCause(e) instanceof PipeRuntimeOutOfMemoryCriticalException) { PipeLogger.log( LOGGER::info, @@ -284,6 +313,20 @@ public void submitSelf() { // 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; } 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 0000000000000..a4eeab3482d0d --- /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. + * + *

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 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 2bb4acc123245..e011ca5b12b41 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 @@ -72,6 +72,8 @@ private boolean runSubtasks() { 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); 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 0000000000000..3fe5242fa078c --- /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 ba2a9d275c39d..b969ac4e61896 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.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 @@ private static String getDataRegionId(final TsFileResource resource) { * 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 boolean waitForTsFileClose() throws InterruptedException { synchronized (isClosed) { while (!isClosed.get()) { + processorExecutionGuard.check(); isClosed.wait(100); + processorExecutionGuard.check(); final boolean isClosedNow = resource.isClosed(); if (isClosedNow) { @@ -770,19 +781,41 @@ public interface TabletInsertionEventConsumer { 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 void consumeTabletInsertionEventsWithRetry( } } - 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 @@ private PipeRawTabletInsertionEvent getNextTabletInsertionEventFromSavedProgress Iterator 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 @@ private void consumeParsedTabletInsertionEventWithRetry( 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 @@ private void consumeParsedTabletInsertionEventWithRetry( } 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 @@ private void consumeParsedTabletInsertionEventWithRetry( } } + 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 Iterable toTabletInsertionEvents(final long timeout } } + 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 @@ private void releaseTsFileParserMemoryIfReserved() { } } - 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 0000000000000..3438dc62c5dc1 --- /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 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/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 d87fe3a5f3807..108f202e00293 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 synchronized void onSuccess(final Boolean hasAtLeastOneEventProcessed) { 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 void allowSubmittingSelf() { * {@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() { From f5cccc6c71b929b3df89e50f6742233a14c7bb76 Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Wed, 5 Aug 2026 09:56:37 +0800 Subject: [PATCH 2/4] Add multi-pipe processor worker test --- .../processor/PipeProcessorSubtaskWorker.java | 16 +++- .../PipeProcessorSubtaskWorkerTest.java | 73 +++++++++++++++++++ 2 files changed, 86 insertions(+), 3 deletions(-) create mode 100644 iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtaskWorkerTest.java 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 e011ca5b12b41..c4c5562d07e29 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 @@ -22,6 +22,7 @@ import org.apache.iotdb.commons.concurrent.WrappedRunnable; import org.apache.iotdb.db.i18n.DataNodePipeMessages; +import com.google.common.annotations.VisibleForTesting; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -38,8 +39,16 @@ public class PipeProcessorSubtaskWorker extends WrappedRunnable { private int workingRoundInAdjustmentInterval = 0; private long sleepingTimeInMilliSecond = 50; - private final Set subtasks = - Collections.newSetFromMap(new ConcurrentHashMap<>()); + private final Set subtasks; + + public PipeProcessorSubtaskWorker() { + this(Collections.newSetFromMap(new ConcurrentHashMap<>())); + } + + @VisibleForTesting + PipeProcessorSubtaskWorker(final Set subtasks) { + this.subtasks = subtasks; + } @Override @SuppressWarnings("squid:S2189") @@ -56,7 +65,8 @@ private void cleanupClosedSubtasksIfNecessary() { subtasks.removeIf(PipeProcessorSubtask::isClosed); } - private boolean runSubtasks() { + @VisibleForTesting + boolean runSubtasks() { ++totalRoundInAdjustmentInterval; boolean canSleepBeforeNextRound = true; 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 0000000000000..79cd4863398e7 --- /dev/null +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtaskWorkerTest.java @@ -0,0 +1,73 @@ +/* + * 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.junit.Assert; +import org.junit.Test; +import org.mockito.InOrder; + +import java.util.LinkedHashSet; + +import static org.mockito.ArgumentMatchers.any; +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()); + } + + 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; + } +} From e0f9652e267d4bb5e10d4a1981bebd1a8ee08bd9 Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Wed, 5 Aug 2026 11:49:15 +0800 Subject: [PATCH 3/4] Log long-running pipe processor event stacks --- .../iotdb/db/i18n/DataNodePipeMessages.java | 2 + .../iotdb/db/i18n/DataNodePipeMessages.java | 2 + .../processor/PipeProcessorSubtask.java | 35 +++++- .../processor/PipeProcessorSubtaskWorker.java | 111 ++++++++++++++++++ .../PipeProcessorSubtaskWorkerManager.java | 15 ++- .../PipeProcessorSubtaskWorkerTest.java | 78 ++++++++++++ 6 files changed, 240 insertions(+), 3 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 757529f3aa0c4..b9a773ee91c82 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 @@ -2592,4 +2592,6 @@ private 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 6fd1922c6a6fb..c7f564a05ea98 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 @@ -2420,4 +2420,6 @@ private 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/subtask/processor/PipeProcessorSubtask.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtask.java index 33465df453a24..4596e22110809 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 @@ -77,6 +77,8 @@ public class PipeProcessorSubtask extends PipeReportableSubtask { private final PipeProcessorSubtaskExecutionGuard executionGuard = new PipeProcessorSubtaskExecutionGuard(); private final AtomicBoolean isResumingFromYield = new AtomicBoolean(false); + private final AtomicReference eventProcessingContext = + new AtomicReference<>(); // This variable is used to distinguish between old and new subtasks before and after stuck // restart. @@ -110,7 +112,7 @@ public PipeProcessorSubtask( @Override public void bindExecutors( final ListeningExecutorService subtaskWorkerThreadPoolExecutor, - final ListeningScheduledExecutorService ignoredScheduledExecutor, + final ListeningScheduledExecutorService subtaskWorkerScheduledExecutor, final ExecutorService ignored, final PipeSubtaskScheduler subtaskScheduler) { this.subtaskWorkerThreadPoolExecutor = subtaskWorkerThreadPoolExecutor; @@ -121,7 +123,8 @@ public void bindExecutors( synchronized (PipeProcessorSubtaskWorkerManager.class) { if (subtaskWorkerManager.get() == null) { subtaskWorkerManager.set( - new PipeProcessorSubtaskWorkerManager(subtaskWorkerThreadPoolExecutor)); + new PipeProcessorSubtaskWorkerManager( + subtaskWorkerThreadPoolExecutor, subtaskWorkerScheduledExecutor)); } } } @@ -160,6 +163,9 @@ protected boolean executeOnce() throws Exception { 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(); @@ -301,6 +307,8 @@ protected boolean executeOnce() throws Exception { e.getMessage() != null ? " Message: " + e.getMessage() : ""); clearReferenceCountAndReleaseLastEvent(event); } + } finally { + eventProcessingContext.compareAndSet(currentEventProcessingContext, null); } return true; @@ -356,6 +364,29 @@ boolean isClosed() { 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/PipeProcessorSubtaskWorker.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtaskWorker.java index c4c5562d07e29..bc967cbc2e9ea 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,7 +20,9 @@ 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; @@ -29,18 +31,32 @@ 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 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<>())); } @@ -76,6 +92,8 @@ boolean runSubtasks() { continue; } + workerThread = Thread.currentThread(); + currentSubtask = subtask; try { final boolean hasAtLeastOneEventProcessed = subtask.call(); if (hasAtLeastOneEventProcessed) { @@ -90,6 +108,8 @@ boolean runSubtasks() { } else { subtask.onFailure(e); } + } finally { + currentSubtask = null; } } @@ -130,4 +150,95 @@ private void adjustSleepingTimeIfNecessary() { 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 33d58c4b5d491..ac2dd2cd7b575 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 PipeProcessorSubtaskWorkerManager(ListeningExecutorService workerThreadPo } 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/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 index 79cd4863398e7..51a018d48d8f6 100644 --- 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 @@ -19,13 +19,20 @@ 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; @@ -63,6 +70,77 @@ public void testYieldingPipesDoNotBlockAnotherPipeOnSameWorker() throws Exceptio 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); From 29513e2595af112e3d80a9555efc707ff51d98f0 Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Fri, 7 Aug 2026 17:15:43 +0800 Subject: [PATCH 4/4] refactor(pipe): reuse processor exception root cause --- .../agent/task/subtask/processor/PipeProcessorSubtask.java | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) 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 4596e22110809..1d1bfac94af2d 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 @@ -279,11 +279,12 @@ protected boolean executeOnce() throws Exception { e.getMessage()); return false; } catch (final Exception e) { - if (ExceptionUtils.getRootCause(e) instanceof PipeProcessorSubtaskYieldException) { + final Throwable rootCause = ExceptionUtils.getRootCause(e); + if (rootCause instanceof PipeProcessorSubtaskYieldException) { isResumingFromYield.set(true); - throw (PipeProcessorSubtaskYieldException) ExceptionUtils.getRootCause(e); + throw (PipeProcessorSubtaskYieldException) rootCause; } - if (ExceptionUtils.getRootCause(e) instanceof PipeRuntimeOutOfMemoryCriticalException) { + if (rootCause instanceof PipeRuntimeOutOfMemoryCriticalException) { PipeLogger.log( LOGGER::info, DataNodePipeMessages.TEMPORARILY_OUT_OF_MEMORY_IN_PIPE_EVENT_PROCESSING,