Caideyipi commented on code in PR #18261:
URL: https://github.com/apache/iotdb/pull/18261#discussion_r3628007495
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeTsFileInsertionEvent.java:
##########
@@ -747,26 +754,68 @@ public interface TabletInsertionEventConsumer {
public void consumeTabletInsertionEventsWithRetry(
final TabletInsertionEventConsumer consumer, final String callerName)
throws Exception {
- int tabletEventCount = 0;
try {
- final Iterable<TabletInsertionEvent> iterable =
toTabletInsertionEvents();
- final Iterator<TabletInsertionEvent> iterator = iterable.iterator();
- while (iterator.hasNext()) {
- final TabletInsertionEvent parsedEvent = iterator.next();
- tabletEventCount++;
+ while (true) {
+ final PipeRawTabletInsertionEvent parsedEvent =
+ getNextTabletInsertionEventFromSavedProgress();
+ if (parsedEvent == null) {
+ isTsFileParsingCompleted.set(true);
+ releaseTsFileParserMemoryIfReserved();
+ return;
+ }
consumeParsedTabletInsertionEventWithRetry(
- consumer, callerName, tabletEventCount, parsedEvent);
+ consumer, callerName, parsedTabletInsertionEventCount.get(),
parsedEvent);
+ pendingTabletInsertionEvent.compareAndSet(parsedEvent, null);
}
} catch (final PipeRuntimeOutOfMemoryCriticalException e) {
- close();
+ // 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.
+ releaseTsFileParserMemoryIfReserved();
LOGGER.warn(
DataNodePipeMessages.FAILED_TO_ALLOCATE_MEMORY_FOR_PARSING_TSFILE,
callerName,
getTsFile(),
- tabletEventCount,
+ parsedTabletInsertionEventCount.get(),
e);
throw e;
+ } catch (final Exception e) {
+ releaseTsFileParserMemoryIfReserved();
+ throw e;
+ }
+ }
+
+ private PipeRawTabletInsertionEvent
getNextTabletInsertionEventFromSavedProgress()
+ throws Exception {
+ if (isTsFileParsingCompleted.get()) {
+ return null;
+ }
+
+ // Reacquire parser memory after a previous failure yielded the active
parser slot.
+ waitForResourceEnough4Parsing((long) ((1 + Math.random()) * 20 * 1000));
Review Comment:
The parser-slot reacquisition has a separate bounded wait:
waitForResourceEnough4Parsing polls for availability for 20-40 seconds before
it can return OOM. The exponential backoff is only for retrying the current
tablet while retaining the parser slot. I clarified that distinction in
245c3358a7f.
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeTsFileInsertionEvent.java:
##########
@@ -788,21 +837,34 @@ private void consumeParsedTabletInsertionEventWithRetry(
}
if (memoryManager.shouldReleaseTsFileParserOnOutOfMemory(
firstOutOfMemoryTimeInMs, ++retryCount)) {
- releaseParsedTabletEvent(parsedEvent);
throw e;
}
logParserRetryOnOutOfMemory(callerName, tabletEventCount, retryCount,
e);
try {
-
Thread.sleep(PipeConfig.getInstance().getPipeMemoryAllocateRetryIntervalInMs());
+ Thread.sleep(getParserRetryBackoffInMs(retryCount));
} catch (final InterruptedException interruptedException) {
Thread.currentThread().interrupt();
- releaseParsedTabletEvent(parsedEvent);
throw e;
}
}
}
}
+ private long getParserRetryBackoffInMs(final int retryCount) {
+ final long initialBackoffInMs =
+ Math.max(1,
PipeConfig.getInstance().getPipeMemoryAllocateRetryIntervalInMs());
+ final int maxRetries = Math.max(1,
PipeConfig.getInstance().getPipeMemoryAllocateMaxRetries());
+ final long maxBackoffInMs =
+ initialBackoffInMs > Long.MAX_VALUE / maxRetries
+ ? Long.MAX_VALUE
+ : initialBackoffInMs * maxRetries;
+ final int shift = Math.min(30, Math.max(0, retryCount - 1));
+ if (initialBackoffInMs > (maxBackoffInMs >> shift)) {
+ return maxBackoffInMs;
+ }
+ return Math.min(maxBackoffInMs, initialBackoffInMs << shift);
Review Comment:
The 30 was an arbitrary shift cap intended to avoid overflow. I removed the
magic number in 245c3358a7f; the code now doubles iteratively and saturates at
maxBackoffInMs without overflow.
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtask.java:
##########
@@ -172,29 +172,24 @@ protected boolean executeOnce() throws Exception {
// We have to parse the privilege first, to avoid passing
no-privilege data to processor
if (event instanceof PipeTsFileInsertionEvent
&& ((PipeTsFileInsertionEvent) event).shouldParse4Privilege()) {
- try (final PipeTsFileInsertionEvent tsFileInsertionEvent =
- (PipeTsFileInsertionEvent) event) {
- final AtomicReference<Exception> ex = new AtomicReference<>();
- tsFileInsertionEvent.consumeTabletInsertionEventsWithRetry(
- event1 -> {
- try {
- pipeProcessor.process(event1, outputEventCollector);
- } catch (PipeRuntimeOutOfMemoryCriticalException e) {
- throw e;
- } catch (Exception e) {
- ex.set(e);
- }
- },
- "PipeProcessorSubtask::executeOnce");
- if (tsFileInsertionEvent.isGeneratedByHistoricalExtractor()) {
- PipeTerminateEvent.markHistoricalTsFileSplit(
- tsFileInsertionEvent.getPipeName(),
- tsFileInsertionEvent.getCreationTime(),
- regionId);
- }
- if (ex.get() != null) {
- throw ex.get();
- }
+ final PipeTsFileInsertionEvent tsFileInsertionEvent =
(PipeTsFileInsertionEvent) event;
+ tsFileInsertionEvent.consumeTabletInsertionEventsWithRetry(
+ event1 -> {
+ try {
+ pipeProcessor.process(event1, outputEventCollector);
+ } catch (PipeRuntimeOutOfMemoryCriticalException e) {
+ throw e;
+ } catch (Exception e) {
+ throw new PipeException(e.getMessage(), e);
+ }
+ },
+ "PipeProcessorSubtask::executeOnce");
+ tsFileInsertionEvent.close();
+ if (tsFileInsertionEvent.isGeneratedByHistoricalExtractor()) {
+ PipeTerminateEvent.markHistoricalTsFileSplit(
+ tsFileInsertionEvent.getPipeName(),
+ tsFileInsertionEvent.getCreationTime(),
+ regionId);
}
Review Comment:
Double-checked. During a retry the TsFile event is intentionally retained in
lastEvent. On success it is closed explicitly; on task drop or release,
PipeSubtask.clearReferenceCountAndReleaseLastEvent calls
PipeTsFileInsertionEvent.internallyDecreaseResourceReferenceCount, which calls
close() and releases the pending tablet, parser, and parser-memory reservation.
I also added
testConsumeTabletInsertionEventsWithRetryReleasesProgressWhenClosed in
245c3358a7f; the three targeted tests pass.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
