Caideyipi commented on code in PR #18051:
URL: https://github.com/apache/iotdb/pull/18051#discussion_r3517854500
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/parser/scan/TsFileInsertionEventScanParser.java:
##########
@@ -235,6 +231,7 @@ public TabletInsertionEvent next() {
final Tablet tablet = getNextTablet();
// Record tablet metrics
recordTabletMetrics(tablet);
+ releaseTabletMemoryBlock();
Review Comment:
Applied in c940475af1d: the previous parser-owned tablet memory is released
before fetching the next tablet, so the ownership/order is clearer.
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/batch/PipeTabletEventBatch.java:
##########
@@ -126,7 +129,28 @@ public synchronized boolean onEvent(final
TabletInsertionEvent event)
protected abstract boolean constructBatch(final TabletInsertionEvent event)
throws WALPipeException, IOException;
+ protected void increaseTotalBufferSizeAndUpdateMemoryBlock(final long
bufferSize) {
+ if (bufferSize <= 0) {
+ return;
+ }
+
+ final long newTotalBufferSize = totalBufferSize + bufferSize;
+ PipeDataNodeResourceManager.memory()
+ .forceResize(allocatedMemoryBlock, Math.min(newTotalBufferSize,
maxBatchSizeInBytes));
+ totalBufferSize = newTotalBufferSize;
Review Comment:
Applied in c940475af1d: totalBufferSize is clamped with
Math.min(totalBufferSize + bufferSize, maxBatchSizeInBytes) before resizing.
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/iotconsensusv2/payload/builder/IoTConsensusV2TransferBatchReqBuilder.java:
##########
@@ -137,27 +113,80 @@ public synchronized boolean onEvent(TabletInsertionEvent
event)
return false;
}
- final long requestCommitId = ((EnrichedEvent)
event).getReplicateIndexForIoTV2();
+ final EnrichedEvent enrichedEvent = (EnrichedEvent) event;
+ final long requestCommitId = enrichedEvent.getReplicateIndexForIoTV2();
// The deduplication logic here is to avoid the accumulation of the same
event in a batch when
// retrying.
if ((events.isEmpty() || !events.get(events.size() - 1).equals(event))) {
- events.add((EnrichedEvent) event);
- requestCommitIds.add(requestCommitId);
- final int bufferSize = buildTabletInsertionBuffer(event);
-
- ((EnrichedEvent) event)
-
.increaseReferenceCount(IoTConsensusV2TransferBatchReqBuilder.class.getName());
+ if (!enrichedEvent.increaseReferenceCount(
+ IoTConsensusV2TransferBatchReqBuilder.class.getName())) {
+
LOGGER.warn(DataNodePipeMessages.CANNOT_INCREASE_REFERENCE_COUNT_FOR_EVENT_IGNORE,
event);
+ return shouldEmit();
+ }
- if (firstEventProcessingTime == Long.MIN_VALUE) {
- firstEventProcessingTime = System.currentTimeMillis();
+ final int previousEventsSize = events.size();
+ final int previousRequestCommitIdsSize = requestCommitIds.size();
+ final int previousBatchReqsSize = batchReqs.size();
+ try {
+ events.add(enrichedEvent);
+ requestCommitIds.add(requestCommitId);
+ final int bufferSize = buildTabletInsertionBuffer(event);
+ increaseTotalBufferSizeAndUpdateMemoryBlock(bufferSize);
+
+ if (firstEventProcessingTime == Long.MIN_VALUE) {
+ firstEventProcessingTime = System.currentTimeMillis();
+ }
+ } catch (final Exception e) {
+ rollbackTo(previousEventsSize, previousRequestCommitIdsSize,
previousBatchReqsSize);
+ if (events.isEmpty()) {
+ resetMemoryUsage();
+ }
+ enrichedEvent.decreaseReferenceCount(
+ IoTConsensusV2TransferBatchReqBuilder.class.getName(), false);
+ throw e;
}
+ }
+
+ return shouldEmit();
+ }
- totalBufferSize += bufferSize;
+ private boolean shouldEmit() {
+ return !events.isEmpty()
+ && (totalBufferSize >= getMaxBatchSizeInBytes()
+ || System.currentTimeMillis() - firstEventProcessingTime >=
maxDelayInMs);
+ }
+
+ private void increaseTotalBufferSizeAndUpdateMemoryBlock(final long
bufferSize) {
+ if (bufferSize <= 0) {
+ return;
}
- return totalBufferSize >= getMaxBatchSizeInBytes()
- || System.currentTimeMillis() - firstEventProcessingTime >=
maxDelayInMs;
+ final long newTotalBufferSize = totalBufferSize + bufferSize;
+ PipeDataNodeResourceManager.memory()
+ .forceResize(allocatedMemoryBlock, Math.min(newTotalBufferSize,
getMaxBatchSizeInBytes()));
+ totalBufferSize = newTotalBufferSize;
+ }
+
+ private void rollbackTo(
+ final int previousEventsSize,
+ final int previousRequestCommitIdsSize,
+ final int previousBatchReqsSize) {
+ while (events.size() > previousEventsSize) {
+ events.remove(events.size() - 1);
+ }
+ while (requestCommitIds.size() > previousRequestCommitIdsSize) {
+ requestCommitIds.remove(requestCommitIds.size() - 1);
+ }
+ while (batchReqs.size() > previousBatchReqsSize) {
+ batchReqs.remove(batchReqs.size() - 1);
+ }
+ }
Review Comment:
Applied in c940475af1d: rollback now uses subList(...).clear() for events,
commit ids, and batch requests.
--
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]
