From 1112f70270248b4f4fea759d1d2a073443142164 Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Tue, 11 Aug 2026 14:21:28 +0800 Subject: [PATCH] [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 --- .../task/connection/PipeEventCollector.java | 18 +- .../processor/PipeProcessorSubtask.java | 81 +++++- .../PipeProcessorSubtaskExecutionGuard.java | 110 ++++++++ .../processor/PipeProcessorSubtaskWorker.java | 128 +++++++++- .../PipeProcessorSubtaskWorkerManager.java | 15 +- .../PipeProcessorSubtaskYieldException.java | 53 ++++ .../tsfile/PipeTsFileInsertionEvent.java | 103 +++++++- .../PipeProcessorSubtaskExecutorTest.java | 6 +- ...ipeProcessorSubtaskExecutionGuardTest.java | 240 ++++++++++++++++++ .../PipeProcessorSubtaskWorkerTest.java | 152 +++++++++++ .../task/execution/PipeSubtaskExecutor.java | 12 +- .../task/subtask/PipeAbstractSinkSubtask.java | 2 + .../pipe/agent/task/subtask/PipeSubtask.java | 17 +- 13 files changed, 913 insertions(+), 24 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 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/connection/PipeEventCollector.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/connection/PipeEventCollector.java index df72ccb830dbc..d6710c83bd72b 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 @@ -25,6 +25,8 @@ import org.apache.iotdb.commons.pipe.event.EnrichedEvent; import org.apache.iotdb.commons.pipe.event.ProgressReportEvent; 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.heartbeat.PipeHeartbeatEvent; import org.apache.iotdb.db.pipe.event.common.schema.PipeSchemaRegionWritePlanEvent; import org.apache.iotdb.db.pipe.event.common.tablet.PipeInsertNodeTabletInsertionEvent; @@ -56,6 +58,9 @@ public class PipeEventCollector implements EventCollector { private final boolean skipParsing; + private PipeProcessorSubtaskExecutionGuard processorExecutionGuard = + PipeProcessorSubtaskExecutionGuard.disabled(); + private final AtomicInteger collectInvocationCount = new AtomicInteger(0); private boolean hasNoGeneratedEvent = true; private boolean isFailedToIncreaseReferenceCount = false; @@ -73,6 +78,11 @@ public PipeEventCollector( this.skipParsing = skipParsing; } + public void setProcessorExecutionGuard( + final PipeProcessorSubtaskExecutionGuard processorExecutionGuard) { + this.processorExecutionGuard = processorExecutionGuard; + } + @Override public void collect(final Event event) { try { @@ -91,6 +101,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) { @@ -123,7 +135,7 @@ private void parseAndCollectEvent(final PipeRawTabletInsertionEvent sourceEvent) } private void parseAndCollectEvent(final PipeTsFileInsertionEvent sourceEvent) throws Exception { - if (!sourceEvent.waitForTsFileClose()) { + if (!sourceEvent.waitForTsFileClose(processorExecutionGuard)) { LOGGER.warn( "Pipe skipping temporary TsFile which shouldn't be transferred: {}", sourceEvent.getTsFile()); @@ -140,7 +152,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 193693c5a95bf..0f29a238e4e76 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 @@ -33,6 +33,7 @@ import org.apache.iotdb.db.pipe.agent.task.connection.PipeEventCollector; import org.apache.iotdb.db.pipe.event.UserDefinedEnrichedEvent; import org.apache.iotdb.db.pipe.event.common.heartbeat.PipeHeartbeatEvent; +import org.apache.iotdb.db.pipe.event.common.tsfile.PipeTsFileInsertionEvent; import org.apache.iotdb.db.pipe.metric.overview.PipeDataNodeSinglePipeMetrics; import org.apache.iotdb.db.pipe.metric.processor.PipeProcessorMetrics; import org.apache.iotdb.db.pipe.processor.pipeconsensus.PipeConsensusProcessor; @@ -44,12 +45,14 @@ import org.apache.iotdb.pipe.api.exception.PipeException; import com.google.common.util.concurrent.ListeningExecutorService; +import com.google.common.util.concurrent.ListeningScheduledExecutorService; import org.apache.commons.lang3.exception.ExceptionUtils; import org.slf4j.Logger; 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 { @@ -67,6 +70,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 = + new AtomicReference<>(); // This variable is used to distinguish between old and new subtasks before and after stuck // restart. @@ -87,6 +95,7 @@ public PipeProcessorSubtask( this.inputEventSupplier = inputEventSupplier; this.pipeProcessor = pipeProcessor; this.outputEventCollector = outputEventCollector; + this.outputEventCollector.setProcessorExecutionGuard(executionGuard); this.subtaskCreationTime = System.currentTimeMillis(); // Only register dataRegions @@ -98,6 +107,7 @@ public PipeProcessorSubtask( @Override public void bindExecutors( final ListeningExecutorService subtaskWorkerThreadPoolExecutor, + final ListeningScheduledExecutorService subtaskWorkerScheduledExecutor, final ExecutorService ignored, final PipeSubtaskScheduler subtaskScheduler) { this.subtaskWorkerThreadPoolExecutor = subtaskWorkerThreadPoolExecutor; @@ -108,19 +118,31 @@ public void bindExecutors( 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 @@ -132,7 +154,13 @@ protected boolean executeOnce() throws Exception { return false; } - outputEventCollector.resetFlags(); + executionGuard.check(); + if (!isResumingFromYield.getAndSet(false)) { + outputEventCollector.resetFlags(); + } + final EventProcessingContext currentEventProcessingContext = + new EventProcessingContext(event, System.nanoTime()); + eventProcessingContext.set(currentEventProcessingContext); try { // event can be supplied after the subtask is closed, so we need to check isClosed here if (!isClosed.get()) { @@ -188,6 +216,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, @@ -195,7 +226,12 @@ protected boolean executeOnce() throws Exception { 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) { PipeLogger.log( LOGGER::info, "Temporarily out of memory in pipe event processing, will wait for the memory to release. Message: %s", @@ -218,6 +254,8 @@ protected boolean executeOnce() throws Exception { e.getMessage() != null ? " Message: " + e.getMessage() : ""); clearReferenceCountAndReleaseLastEvent(event); } + } finally { + eventProcessingContext.compareAndSet(currentEventProcessingContext, null); } return true; @@ -230,6 +268,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; } @@ -259,6 +311,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/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 b9584d2c586b3..ad1548d9432be 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,25 +20,50 @@ 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.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 subtasks = - Collections.newSetFromMap(new ConcurrentHashMap<>()); + 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<>())); + } + + @VisibleForTesting + PipeProcessorSubtaskWorker(final Set subtasks) { + this.subtasks = subtasks; + } @Override @SuppressWarnings("squid:S2189") @@ -55,7 +80,8 @@ private void cleanupClosedSubtasksIfNecessary() { subtasks.removeIf(PipeProcessorSubtask::isClosed); } - private boolean runSubtasks() { + @VisibleForTesting + boolean runSubtasks() { ++totalRoundInAdjustmentInterval; boolean canSleepBeforeNextRound = true; @@ -65,18 +91,24 @@ private boolean runSubtasks() { 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("subtask {} is closed, ignore exception", subtask, e); } else { subtask.onFailure(e); } + } finally { + currentSubtask = null; } } @@ -117,4 +149,94 @@ 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( + "Pipe processor worker {} has been processing the same event for {} ms. Pipe: {}, DataRegion: {}, subtask: {}, event: {}, thread state: {}. Stack:{}", + thread.getName(), + TimeUnit.NANOSECONDS.toMillis(currentTimeInNanos - context.getStartTimeInNanos()), + subtask.getPipeName(), + subtask.getRegionId(), + subtask.getTaskID(), + 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/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 a4859c33579de..35b87436dee09 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 @@ -28,6 +28,8 @@ import org.apache.iotdb.commons.pipe.event.EnrichedEvent; import org.apache.iotdb.commons.pipe.resource.log.PipeLogger; import org.apache.iotdb.commons.pipe.resource.ref.PipePhantomReferenceManager.PipeEventResource; +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.tablet.PipeRawTabletInsertionEvent; import org.apache.iotdb.db.pipe.event.common.tsfile.container.TsFileInsertionDataContainer; @@ -246,6 +248,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; } @@ -259,7 +268,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) { @@ -504,19 +515,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. @@ -534,16 +567,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) { @@ -552,7 +584,7 @@ private PipeRawTabletInsertionEvent getNextTabletInsertionEventFromSavedProgress Iterator iterator = tabletInsertionEventIterator.get(); if (iterator == null) { - if (!waitForTsFileClose()) { + if (!waitForTsFileClose(processorExecutionGuard)) { LOGGER.warn( "Pipe skipping temporary TsFile's parsing which shouldn't be transferred: {}", tsFile); return null; @@ -575,12 +607,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; @@ -594,7 +628,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; @@ -603,6 +637,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()); @@ -683,6 +735,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)) { @@ -757,7 +836,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/PipeProcessorSubtaskExecutorTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/PipeProcessorSubtaskExecutorTest.java index a403e83329ef1..72daec87488d9 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/PipeProcessorSubtaskExecutorTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/PipeProcessorSubtaskExecutorTest.java @@ -21,6 +21,7 @@ import org.apache.iotdb.commons.exception.pipe.PipeRuntimeOutOfMemoryCriticalException; import org.apache.iotdb.commons.pipe.agent.task.connection.EventSupplier; +import org.apache.iotdb.commons.pipe.agent.task.execution.PipeSubtaskScheduler; import org.apache.iotdb.db.pipe.agent.task.connection.PipeEventCollector; import org.apache.iotdb.db.pipe.agent.task.execution.PipeProcessorSubtaskExecutor; import org.apache.iotdb.db.pipe.agent.task.subtask.processor.PipeProcessorSubtask; @@ -143,7 +144,10 @@ private TestablePipeProcessorSubtask( } private boolean executeOnceForTest() throws Exception { - return executeOnce(); + subtaskScheduler = mock(PipeSubtaskScheduler.class); + when(subtaskScheduler.schedule()).thenReturn(true, false); + allowSubmittingSelf(); + return call(); } } } 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..b7c5fdc554203 --- /dev/null +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtaskExecutionGuardTest.java @@ -0,0 +1,240 @@ +/* + * 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.PrefixPipePattern; +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.file.metadata.PlainDeviceID; +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 = new PlainDeviceID("root.testsg.d0"); + resource.updateStartTime(deviceID, 0); + resource.updateEndTime(deviceID, 9); + + return new PipeTsFileInsertionEvent( + resource, + null, + false, + false, + false, + pipeName, + 0, + null, + new PrefixPipePattern("root"), + 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 0000000000000..d1c34ccf47160 --- /dev/null +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtaskWorkerTest.java @@ -0,0 +1,152 @@ +/* + * 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.apache.commons.lang3.StringUtils; +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(StringUtils.repeat('x', 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/execution/PipeSubtaskExecutor.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/execution/PipeSubtaskExecutor.java index f16c6387cf570..61b821bba7f0a 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/execution/PipeSubtaskExecutor.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/execution/PipeSubtaskExecutor.java @@ -26,6 +26,7 @@ import org.apache.iotdb.commons.utils.TestOnly; import com.google.common.util.concurrent.ListeningExecutorService; +import com.google.common.util.concurrent.ListeningScheduledExecutorService; import com.google.common.util.concurrent.MoreExecutors; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -36,6 +37,7 @@ import java.util.Objects; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ExecutorService; +import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.ThreadPoolExecutor; public abstract class PipeSubtaskExecutor { @@ -50,6 +52,7 @@ public abstract class PipeSubtaskExecutor { protected final WrappedThreadPoolExecutor underlyingThreadPool; protected final ListeningExecutorService subtaskWorkerThreadPoolExecutor; + protected final ListeningScheduledExecutorService subtaskWorkerScheduledExecutor; private final Map registeredIdSubtaskMapper; @@ -82,6 +85,9 @@ protected PipeSubtaskExecutor( underlyingThreadPool.disableErrorLog(); } subtaskWorkerThreadPoolExecutor = MoreExecutors.listeningDecorator(underlyingThreadPool); + final ScheduledExecutorService underlyingScheduledExecutor = + IoTDBThreadPoolFactory.newSingleThreadScheduledExecutor(workingThreadName + "-Scheduler"); + subtaskWorkerScheduledExecutor = MoreExecutors.listeningDecorator(underlyingScheduledExecutor); subtaskCallbackListeningExecutor = Objects.nonNull(callbackThreadName) ? IoTDBThreadPoolFactory.newSingleThreadExecutor( @@ -104,7 +110,10 @@ public final synchronized void register(final PipeSubtask subtask) { registeredIdSubtaskMapper.put(subtask.getTaskID(), subtask); subtask.bindExecutors( - subtaskWorkerThreadPoolExecutor, subtaskCallbackListeningExecutor, schedulerSupplier(this)); + subtaskWorkerThreadPoolExecutor, + subtaskWorkerScheduledExecutor, + subtaskCallbackListeningExecutor, + schedulerSupplier(this)); } protected PipeSubtaskScheduler schedulerSupplier(final PipeSubtaskExecutor executor) { @@ -179,6 +188,7 @@ public final synchronized void shutdown() { } subtaskWorkerThreadPoolExecutor.shutdown(); + subtaskWorkerScheduledExecutor.shutdown(); if (subtaskCallbackListeningExecutor != globalSubtaskCallbackListeningExecutor) { subtaskCallbackListeningExecutor.shutdown(); } diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/subtask/PipeAbstractSinkSubtask.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/subtask/PipeAbstractSinkSubtask.java index f3c1d72605c64..f9d7c2be472b9 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/subtask/PipeAbstractSinkSubtask.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/subtask/PipeAbstractSinkSubtask.java @@ -35,6 +35,7 @@ import com.google.common.util.concurrent.Futures; import com.google.common.util.concurrent.ListenableFuture; import com.google.common.util.concurrent.ListeningExecutorService; +import com.google.common.util.concurrent.ListeningScheduledExecutorService; import org.apache.commons.lang3.exception.ExceptionUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -71,6 +72,7 @@ protected PipeAbstractSinkSubtask( @Override public void bindExecutors( final ListeningExecutorService subtaskWorkerThreadPoolExecutor, + final ListeningScheduledExecutorService ignoredScheduledExecutor, final ExecutorService subtaskCallbackListeningExecutor, final PipeSubtaskScheduler subtaskScheduler) { this.subtaskWorkerThreadPoolExecutor = subtaskWorkerThreadPoolExecutor; 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 1b58d1d61784e..b583276fa90f5 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 @@ -25,6 +25,7 @@ import com.google.common.util.concurrent.FutureCallback; import com.google.common.util.concurrent.ListeningExecutorService; +import com.google.common.util.concurrent.ListeningScheduledExecutorService; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -65,6 +66,7 @@ protected PipeSubtask(final String taskID, final long creationTime) { public abstract void bindExecutors( ListeningExecutorService subtaskWorkerThreadPoolExecutor, + ListeningScheduledExecutorService subtaskWorkerScheduledExecutor, ExecutorService subtaskCallbackListeningExecutor, PipeSubtaskScheduler subtaskScheduler); @@ -128,9 +130,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}. @@ -139,7 +146,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() {