[Pipe] Fix processor worker starvation on pipe stop - #18396
Conversation
| if (ExceptionUtils.getRootCause(e) instanceof PipeProcessorSubtaskYieldException) { | ||
| isResumingFromYield.set(true); | ||
| throw (PipeProcessorSubtaskYieldException) ExceptionUtils.getRootCause(e); | ||
| } | ||
| if (ExceptionUtils.getRootCause(e) instanceof PipeRuntimeOutOfMemoryCriticalException) { |
There was a problem hiding this comment.
No need to call getRootCause multiple times.
There was a problem hiding this comment.
Addressed in 29513e2: the exception root cause is now computed once and reused for both control-flow checks, avoiding repeated traversal.
| 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); |
There was a problem hiding this comment.
Creating a static exception instance may miss the actual stack traces.
Is it expected?
There was a problem hiding this comment.
Yes, the static instances are intentional control-flow sentinels. They disable stack-trace creation because these exceptions are expected pause/yield signals rather than failures; the Reason enum identifies the requested action. A per-throw stack would add allocation cost without diagnostic value.
| while (remainingTimeInMs > 0) { | ||
| processorExecutionGuard.check(); | ||
| Thread.sleep(Math.min(remainingTimeInMs, 100)); | ||
| processorExecutionGuard.check(); | ||
| remainingTimeInMs = deadlineInMs - System.currentTimeMillis(); | ||
| } |
There was a problem hiding this comment.
Is it possible to register this thread with the guard and let it interrupt registered threads when the epoch changes?
Frequent sleep-and-wake may reduce performance.
There was a problem hiding this comment.
The guard currently has no registered-thread facility; its 100 ms polling interval bounds stop latency while keeping the guard usable across the existing worker and parser call paths. Registering and interrupting sleeping threads would require a broader guard lifecycle change, so I am leaving that as a separate optimization.
Description
Tests
mvn -o -nsu test -pl iotdb-core/datanode -Dtest=PipeProcessorSubtaskWorkerTest,PipeProcessorSubtaskExecutionGuardTest,PipeTsFileInsertionEventAdmissionTest,PipeMemoryManagerTest,PipeProcessorSubtaskExecutorTestmvn -o -nsu test -pl iotdb-core/datanode -Dtest=TsFileInsertionEventParserTestmvn -o test -pl iotdb-core/datanode -Dtest=PipeProcessorSubtaskWorkerTest,PipeProcessorSubtaskExecutionGuardTest -DfailIfNoTests=falsemvn -o compile -pl iotdb-core/datanode -P with-zh-locale