This is an automated email from the ASF dual-hosted git repository.
rong pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new 1e37accd4e8 Pipe: Move parsing event logic into PipeProcessorSubtask
to separate CPU intensive and IO intensive tasks (#11965)
1e37accd4e8 is described below
commit 1e37accd4e80df866dd860c334bcfc01ad889cf1
Author: Zikun Ma <[email protected]>
AuthorDate: Wed Feb 7 13:02:58 2024 +0800
Pipe: Move parsing event logic into PipeProcessorSubtask to separate CPU
intensive and IO intensive tasks (#11965)
Co-authored-by: Caideyipi <[email protected]>
Co-authored-by: Steve Yurong Su <[email protected]>
---
.../protocol/airgap/IoTDBAirGapConnector.java | 31 --------
.../async/IoTDBThriftAsyncClientManager.java | 2 +-
.../thrift/async/IoTDBThriftAsyncConnector.java | 37 ----------
.../thrift/sync/IoTDBThriftSyncConnector.java | 34 ---------
.../protocol/writeback/WriteBackConnector.java | 20 -----
.../tablet/PipeInsertNodeTabletInsertionEvent.java | 2 +-
.../common/tablet/PipeRawTabletInsertionEvent.java | 4 +-
.../db/pipe/resource/memory/PipeMemoryManager.java | 77 +++++++++++++++++---
.../resource/memory/PipeTabletMemoryBlock.java | 64 ++++++++++++++++
.../pipe/task/connection/PipeEventCollector.java | 85 ++++++++++++++++++++++
.../subtask/processor/PipeProcessorSubtask.java | 5 +-
.../apache/iotdb/commons/conf/CommonConfig.java | 11 +++
.../iotdb/commons/conf/CommonDescriptor.java | 6 ++
.../iotdb/commons/pipe/config/PipeConfig.java | 7 ++
14 files changed, 245 insertions(+), 140 deletions(-)
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/airgap/IoTDBAirGapConnector.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/airgap/IoTDBAirGapConnector.java
index 854f527222c..b018ff0bb80 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/airgap/IoTDBAirGapConnector.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/airgap/IoTDBAirGapConnector.java
@@ -37,7 +37,6 @@ import
org.apache.iotdb.db.pipe.connector.payload.evolvable.request.PipeTransfer
import
org.apache.iotdb.db.pipe.connector.payload.evolvable.request.PipeTransferTabletBinaryReq;
import
org.apache.iotdb.db.pipe.connector.payload.evolvable.request.PipeTransferTabletInsertNodeReq;
import
org.apache.iotdb.db.pipe.connector.payload.evolvable.request.PipeTransferTabletRawReq;
-import org.apache.iotdb.db.pipe.event.EnrichedEvent;
import org.apache.iotdb.db.pipe.event.common.heartbeat.PipeHeartbeatEvent;
import
org.apache.iotdb.db.pipe.event.common.tablet.PipeInsertNodeTabletInsertionEvent;
import
org.apache.iotdb.db.pipe.event.common.tablet.PipeRawTabletInsertionEvent;
@@ -256,25 +255,6 @@ public class IoTDBAirGapConnector extends IoTDBConnector {
return;
}
- if (((EnrichedEvent) tabletInsertionEvent).shouldParsePatternOrTime()) {
- if (tabletInsertionEvent instanceof PipeInsertNodeTabletInsertionEvent) {
- transfer(
- ((PipeInsertNodeTabletInsertionEvent) tabletInsertionEvent)
- .parseEventWithPatternOrTime());
- } else { // tabletInsertionEvent instanceof PipeRawTabletInsertionEvent
- transfer(
- ((PipeRawTabletInsertionEvent)
tabletInsertionEvent).parseEventWithPatternOrTime());
- }
- return;
- } else {
- // ignore raw tablet event with zero rows
- if (tabletInsertionEvent instanceof PipeRawTabletInsertionEvent) {
- if (((PipeRawTabletInsertionEvent)
tabletInsertionEvent).hasNoNeedParsingAndIsEmpty()) {
- return;
- }
- }
- }
-
final int socketIndex = nextSocketIndex();
final Socket socket = sockets.get(socketIndex);
@@ -312,17 +292,6 @@ public class IoTDBAirGapConnector extends IoTDBConnector {
return;
}
- if (((EnrichedEvent) tsFileInsertionEvent).shouldParsePatternOrTime()) {
- try {
- for (final TabletInsertionEvent event :
tsFileInsertionEvent.toTabletInsertionEvents()) {
- transfer(event);
- }
- } finally {
- tsFileInsertionEvent.close();
- }
- return;
- }
-
final int socketIndex = nextSocketIndex();
final Socket socket = sockets.get(socketIndex);
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/thrift/async/IoTDBThriftAsyncClientManager.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/thrift/async/IoTDBThriftAsyncClientManager.java
index b70794667dd..59cedeb9d3c 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/thrift/async/IoTDBThriftAsyncClientManager.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/thrift/async/IoTDBThriftAsyncClientManager.java
@@ -186,7 +186,7 @@ public class IoTDBThriftAsyncClientManager extends
IoTDBThriftClientManager {
// Retry to handshake by PipeTransferHandshakeV1Req.
if (resp.get() != null
- && resp.get().getStatus().getCode() ==
TSStatusCode.PIPE_VERSION_ERROR.getStatusCode()) {
+ && resp.get().getStatus().getCode() ==
TSStatusCode.PIPE_TYPE_ERROR.getStatusCode()) {
LOGGER.info(
"Handshake error by PipeTransferHandshakeV2Req with receiver {}:{} "
+ "retry to handshake by PipeTransferHandshakeV1Req.",
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/thrift/async/IoTDBThriftAsyncConnector.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/thrift/async/IoTDBThriftAsyncConnector.java
index 185699cf3d9..b182e35adcd 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/thrift/async/IoTDBThriftAsyncConnector.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/thrift/async/IoTDBThriftAsyncConnector.java
@@ -152,25 +152,6 @@ public class IoTDBThriftAsyncConnector extends
IoTDBConnector {
return;
}
- if (((EnrichedEvent) tabletInsertionEvent).shouldParsePatternOrTime()) {
- if (tabletInsertionEvent instanceof PipeInsertNodeTabletInsertionEvent) {
- transfer(
- ((PipeInsertNodeTabletInsertionEvent) tabletInsertionEvent)
- .parseEventWithPatternOrTime());
- } else { // tabletInsertionEvent instanceof PipeRawTabletInsertionEvent
- transfer(
- ((PipeRawTabletInsertionEvent)
tabletInsertionEvent).parseEventWithPatternOrTime());
- }
- return;
- } else {
- // ignore raw tablet event with zero rows
- if (tabletInsertionEvent instanceof PipeRawTabletInsertionEvent) {
- if (((PipeRawTabletInsertionEvent)
tabletInsertionEvent).hasNoNeedParsingAndIsEmpty()) {
- return;
- }
- }
- }
-
if (isTabletBatchModeEnabled) {
if (tabletBatchBuilder.onEvent(tabletInsertionEvent)) {
final PipeTransferTabletBatchEventHandler
pipeTransferTabletBatchEventHandler =
@@ -259,24 +240,6 @@ public class IoTDBThriftAsyncConnector extends
IoTDBConnector {
final PipeTsFileInsertionEvent pipeTsFileInsertionEvent =
(PipeTsFileInsertionEvent) tsFileInsertionEvent;
- if (!pipeTsFileInsertionEvent.waitForTsFileClose()) {
- LOGGER.warn(
- "Pipe skipping temporary TsFile which shouldn't be transferred: {}",
- pipeTsFileInsertionEvent.getTsFile());
- return;
- }
-
- if ((pipeTsFileInsertionEvent).shouldParsePatternOrTime()) {
- try {
- for (final TabletInsertionEvent event :
- pipeTsFileInsertionEvent.toTabletInsertionEvents()) {
- transfer(event);
- }
- } finally {
- pipeTsFileInsertionEvent.close();
- }
- return;
- }
// Just in case. To avoid the case that exception occurred when
constructing the handler.
if (!pipeTsFileInsertionEvent.getTsFile().exists()) {
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/thrift/sync/IoTDBThriftSyncConnector.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/thrift/sync/IoTDBThriftSyncConnector.java
index 91eba79f4c4..0cdb4c6882f 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/thrift/sync/IoTDBThriftSyncConnector.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/thrift/sync/IoTDBThriftSyncConnector.java
@@ -34,7 +34,6 @@ import
org.apache.iotdb.db.pipe.connector.payload.evolvable.request.PipeTransfer
import
org.apache.iotdb.db.pipe.connector.payload.evolvable.request.PipeTransferTabletBinaryReq;
import
org.apache.iotdb.db.pipe.connector.payload.evolvable.request.PipeTransferTabletInsertNodeReq;
import
org.apache.iotdb.db.pipe.connector.payload.evolvable.request.PipeTransferTabletRawReq;
-import org.apache.iotdb.db.pipe.event.EnrichedEvent;
import org.apache.iotdb.db.pipe.event.common.heartbeat.PipeHeartbeatEvent;
import
org.apache.iotdb.db.pipe.event.common.tablet.PipeInsertNodeTabletInsertionEvent;
import
org.apache.iotdb.db.pipe.event.common.tablet.PipeRawTabletInsertionEvent;
@@ -196,25 +195,6 @@ public class IoTDBThriftSyncConnector extends
IoTDBConnector {
return;
}
- if (((EnrichedEvent) tabletInsertionEvent).shouldParsePatternOrTime()) {
- if (tabletInsertionEvent instanceof PipeInsertNodeTabletInsertionEvent) {
- transfer(
- ((PipeInsertNodeTabletInsertionEvent) tabletInsertionEvent)
- .parseEventWithPatternOrTime());
- } else { // tabletInsertionEvent instanceof PipeRawTabletInsertionEvent
- transfer(
- ((PipeRawTabletInsertionEvent)
tabletInsertionEvent).parseEventWithPatternOrTime());
- }
- return;
- } else {
- // ignore raw tablet event with zero rows
- if (tabletInsertionEvent instanceof PipeRawTabletInsertionEvent) {
- if (((PipeRawTabletInsertionEvent)
tabletInsertionEvent).hasNoNeedParsingAndIsEmpty()) {
- return;
- }
- }
- }
-
try {
if (isTabletBatchModeEnabled) {
if (tabletBatchBuilder.onEvent(tabletInsertionEvent)) {
@@ -246,20 +226,6 @@ public class IoTDBThriftSyncConnector extends
IoTDBConnector {
return;
}
- if (!((PipeTsFileInsertionEvent)
tsFileInsertionEvent).waitForTsFileClose()) {
- LOGGER.warn(
- "Pipe skipping temporary TsFile which shouldn't be transferred: {}",
- ((PipeTsFileInsertionEvent) tsFileInsertionEvent).getTsFile());
- return;
- }
-
- if (((EnrichedEvent) tsFileInsertionEvent).shouldParsePatternOrTime()) {
- for (final TabletInsertionEvent event :
tsFileInsertionEvent.toTabletInsertionEvents()) {
- transfer(event);
- }
- return;
- }
-
try {
// in order to commit in order
if (isTabletBatchModeEnabled && !tabletBatchBuilder.isEmpty()) {
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/writeback/WriteBackConnector.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/writeback/WriteBackConnector.java
index 40ff1fb6ea3..dface1e770e 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/writeback/WriteBackConnector.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/writeback/WriteBackConnector.java
@@ -26,7 +26,6 @@ import org.apache.iotdb.db.pipe.agent.PipeAgent;
import
org.apache.iotdb.db.pipe.connector.payload.evolvable.request.PipeTransferTabletBinaryReq;
import
org.apache.iotdb.db.pipe.connector.payload.evolvable.request.PipeTransferTabletInsertNodeReq;
import
org.apache.iotdb.db.pipe.connector.payload.evolvable.request.PipeTransferTabletRawReq;
-import org.apache.iotdb.db.pipe.event.EnrichedEvent;
import org.apache.iotdb.db.pipe.event.common.heartbeat.PipeHeartbeatEvent;
import
org.apache.iotdb.db.pipe.event.common.tablet.PipeInsertNodeTabletInsertionEvent;
import
org.apache.iotdb.db.pipe.event.common.tablet.PipeRawTabletInsertionEvent;
@@ -93,25 +92,6 @@ public class WriteBackConnector implements PipeConnector {
return;
}
- if (((EnrichedEvent) tabletInsertionEvent).shouldParsePatternOrTime()) {
- if (tabletInsertionEvent instanceof PipeInsertNodeTabletInsertionEvent) {
- transfer(
- ((PipeInsertNodeTabletInsertionEvent) tabletInsertionEvent)
- .parseEventWithPatternOrTime());
- } else { // tabletInsertionEvent instanceof PipeRawTabletInsertionEvent
- transfer(
- ((PipeRawTabletInsertionEvent)
tabletInsertionEvent).parseEventWithPatternOrTime());
- }
- return;
- } else {
- // ignore raw tablet event with zero rows
- if (tabletInsertionEvent instanceof PipeRawTabletInsertionEvent) {
- if (((PipeRawTabletInsertionEvent)
tabletInsertionEvent).hasNoNeedParsingAndIsEmpty()) {
- return;
- }
- }
- }
-
if (tabletInsertionEvent instanceof PipeInsertNodeTabletInsertionEvent) {
doTransfer((PipeInsertNodeTabletInsertionEvent) tabletInsertionEvent);
} else {
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tablet/PipeInsertNodeTabletInsertionEvent.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tablet/PipeInsertNodeTabletInsertionEvent.java
index 98ab1c7b2ac..4ae1b9ee64d 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tablet/PipeInsertNodeTabletInsertionEvent.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tablet/PipeInsertNodeTabletInsertionEvent.java
@@ -240,7 +240,7 @@ public class PipeInsertNodeTabletInsertionEvent extends
EnrichedEvent
/////////////////////////// parsePatternOrTime ///////////////////////////
- public TabletInsertionEvent parseEventWithPatternOrTime() {
+ public PipeRawTabletInsertionEvent parseEventWithPatternOrTime() {
return new PipeRawTabletInsertionEvent(
convertToTablet(), isAligned, pipeName, pipeTaskMeta, this, true);
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tablet/PipeRawTabletInsertionEvent.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tablet/PipeRawTabletInsertionEvent.java
index 03df841181c..93fe966409c 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tablet/PipeRawTabletInsertionEvent.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tablet/PipeRawTabletInsertionEvent.java
@@ -99,7 +99,7 @@ public class PipeRawTabletInsertionEvent extends
EnrichedEvent implements Tablet
@Override
public boolean internallyIncreaseResourceReferenceCount(String
holderMessage) {
- allocatedMemoryBlock = PipeResourceManager.memory().forceAllocate(tablet);
+ allocatedMemoryBlock =
PipeResourceManager.memory().forceAllocateWithRetry(tablet);
return true;
}
@@ -200,7 +200,7 @@ public class PipeRawTabletInsertionEvent extends
EnrichedEvent implements Tablet
/////////////////////////// parsePatternOrTime ///////////////////////////
- public TabletInsertionEvent parseEventWithPatternOrTime() {
+ public PipeRawTabletInsertionEvent parseEventWithPatternOrTime() {
return new PipeRawTabletInsertionEvent(
convertToTablet(), isAligned, pipeName, pipeTaskMeta, this,
needToReport);
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryManager.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryManager.java
index ad4e1951a1e..3f7a3eb2849 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryManager.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryManager.java
@@ -57,6 +57,12 @@ public class PipeMemoryManager {
private long usedMemorySizeInBytes;
+ // To avoid too much parsed events causing OOM. If total tablet memory size
exceeds this
+ // threshold, allocations of memory block for tablets will be rejected.
+ private static final double TABLET_MEMORY_REJECT_THRESHOLD =
+
PipeConfig.getInstance().getPipeDataStructureTabletMemoryBlockAllocationRejectThreshold();
+ private long usedMemorySizeInBytesOfTablets;
+
private final Set<PipeMemoryBlock> allocatedBlocks = new HashSet<>();
public PipeMemoryManager() {
@@ -69,13 +75,51 @@ public class PipeMemoryManager {
public synchronized PipeMemoryBlock forceAllocate(long sizeInBytes)
throws PipeRuntimeOutOfMemoryCriticalException {
+ return forceAllocate(sizeInBytes, false);
+ }
+
+ public PipeMemoryBlock forceAllocateWithRetry(Tablet tablet)
+ throws PipeRuntimeOutOfMemoryCriticalException {
+ for (int i = 1; i <= MEMORY_ALLOCATE_MAX_RETRIES; i++) {
+ if ((double) usedMemorySizeInBytesOfTablets / TOTAL_MEMORY_SIZE_IN_BYTES
+ < TABLET_MEMORY_REJECT_THRESHOLD) {
+ break;
+ }
+
+ try {
+ Thread.sleep(MEMORY_ALLOCATE_RETRY_INTERVAL_IN_MS);
+ } catch (InterruptedException ex) {
+ Thread.currentThread().interrupt();
+ LOGGER.warn("forceAllocateWithRetry: interrupted while waiting for
available memory", ex);
+ }
+ }
+
+ if ((double) usedMemorySizeInBytesOfTablets / TOTAL_MEMORY_SIZE_IN_BYTES
+ >= TABLET_MEMORY_REJECT_THRESHOLD) {
+ throw new PipeRuntimeOutOfMemoryCriticalException(
+ String.format(
+ "forceAllocateForTablet: failed to allocate because there's too
much memory for tablets, "
+ + "total memory size %d bytes, used memory for tablet size
%d bytes,",
+ TOTAL_MEMORY_SIZE_IN_BYTES, usedMemorySizeInBytesOfTablets));
+ }
+
+ synchronized (this) {
+ final PipeMemoryBlock block =
forceAllocate(calculateTabletSizeInBytes(tablet), true);
+ usedMemorySizeInBytesOfTablets += block.getMemoryUsageInBytes();
+ return block;
+ }
+ }
+
+ private PipeMemoryBlock forceAllocate(long sizeInBytes, boolean isForTablet)
+ throws PipeRuntimeOutOfMemoryCriticalException {
if (!PIPE_MEMORY_MANAGEMENT_ENABLED) {
+ // No need to consider isForTablet, for memory control is disabled
return new PipeMemoryBlock(sizeInBytes);
}
for (int i = 1; i <= MEMORY_ALLOCATE_MAX_RETRIES; i++) {
if (TOTAL_MEMORY_SIZE_IN_BYTES - usedMemorySizeInBytes >= sizeInBytes) {
- return registeredMemoryBlock(sizeInBytes);
+ return registerMemoryBlock(sizeInBytes, isForTablet);
}
try {
@@ -98,11 +142,6 @@ public class PipeMemoryManager {
sizeInBytes));
}
- public synchronized PipeMemoryBlock forceAllocate(Tablet tablet)
- throws PipeRuntimeOutOfMemoryCriticalException {
- return forceAllocate(calculateTabletSizeInBytes(tablet));
- }
-
/**
* Allocate a memory block for pipe only if memory already used is less than
specified threshold.
*
@@ -199,7 +238,7 @@ public class PipeMemoryManager {
}
if (TOTAL_MEMORY_SIZE_IN_BYTES - usedMemorySizeInBytes >= sizeInBytes) {
- return registeredMemoryBlock(sizeInBytes);
+ return registerMemoryBlock(sizeInBytes);
}
long sizeToAllocateInBytes = sizeInBytes;
@@ -214,7 +253,7 @@ public class PipeMemoryManager {
usedMemorySizeInBytes,
sizeInBytes,
sizeToAllocateInBytes);
- return registeredMemoryBlock(sizeToAllocateInBytes);
+ return registerMemoryBlock(sizeToAllocateInBytes);
}
sizeToAllocateInBytes =
@@ -233,7 +272,7 @@ public class PipeMemoryManager {
usedMemorySizeInBytes,
sizeInBytes,
sizeToAllocateInBytes);
- return registeredMemoryBlock(sizeToAllocateInBytes);
+ return registerMemoryBlock(sizeToAllocateInBytes);
} else {
LOGGER.warn(
"tryAllocate: failed to allocate memory, "
@@ -242,7 +281,7 @@ public class PipeMemoryManager {
TOTAL_MEMORY_SIZE_IN_BYTES,
usedMemorySizeInBytes,
sizeInBytes);
- return registeredMemoryBlock(0);
+ return registerMemoryBlock(0);
}
}
@@ -254,6 +293,9 @@ public class PipeMemoryManager {
if (TOTAL_MEMORY_SIZE_IN_BYTES - usedMemorySizeInBytes >=
memoryInBytesNeededToBeAllocated) {
usedMemorySizeInBytes += memoryInBytesNeededToBeAllocated;
+ if (block instanceof PipeTabletMemoryBlock) {
+ usedMemorySizeInBytesOfTablets += memoryInBytesNeededToBeAllocated;
+ }
block.setMemoryUsageInBytes(block.getMemoryUsageInBytes() +
memoryInBytesNeededToBeAllocated);
return true;
}
@@ -261,10 +303,15 @@ public class PipeMemoryManager {
return false;
}
- private PipeMemoryBlock registeredMemoryBlock(long sizeInBytes) {
+ private PipeMemoryBlock registerMemoryBlock(long sizeInBytes) {
+ return registerMemoryBlock(sizeInBytes, false);
+ }
+
+ private PipeMemoryBlock registerMemoryBlock(long sizeInBytes, boolean
isForTablet) {
usedMemorySizeInBytes += sizeInBytes;
- final PipeMemoryBlock returnedMemoryBlock = new
PipeMemoryBlock(sizeInBytes);
+ final PipeMemoryBlock returnedMemoryBlock =
+ isForTablet ? new PipeTabletMemoryBlock(sizeInBytes) : new
PipeMemoryBlock(sizeInBytes);
allocatedBlocks.add(returnedMemoryBlock);
return returnedMemoryBlock;
}
@@ -300,6 +347,9 @@ public class PipeMemoryManager {
allocatedBlocks.remove(block);
usedMemorySizeInBytes -= block.getMemoryUsageInBytes();
+ if (block instanceof PipeTabletMemoryBlock) {
+ usedMemorySizeInBytesOfTablets -= block.getMemoryUsageInBytes();
+ }
block.markAsReleased();
this.notifyAll();
@@ -311,6 +361,9 @@ public class PipeMemoryManager {
}
usedMemorySizeInBytes -= sizeInBytes;
+ if (block instanceof PipeTabletMemoryBlock) {
+ usedMemorySizeInBytesOfTablets -= sizeInBytes;
+ }
block.setMemoryUsageInBytes(block.getMemoryUsageInBytes() - sizeInBytes);
this.notifyAll();
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/PipeTabletMemoryBlock.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/PipeTabletMemoryBlock.java
new file mode 100644
index 00000000000..ed23215874a
--- /dev/null
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/PipeTabletMemoryBlock.java
@@ -0,0 +1,64 @@
+/*
+ * 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.resource.memory;
+
+import java.util.function.BiConsumer;
+import java.util.function.LongUnaryOperator;
+
+public class PipeTabletMemoryBlock extends PipeMemoryBlock {
+
+ public PipeTabletMemoryBlock(long memoryUsageInBytes) {
+ super(memoryUsageInBytes);
+ }
+
+ @Override
+ boolean shrink() {
+ return false;
+ }
+
+ @Override
+ boolean expand() {
+ return false;
+ }
+
+ @Override
+ public PipeMemoryBlock setShrinkMethod(LongUnaryOperator shrinkMethod) {
+ throw new UnsupportedOperationException(
+ "Shrink method is not supported in PipeTabletMemoryBlock");
+ }
+
+ @Override
+ public PipeMemoryBlock setShrinkCallback(BiConsumer<Long, Long>
shrinkCallback) {
+ throw new UnsupportedOperationException(
+ "Shrink callback is not supported in PipeTabletMemoryBlock");
+ }
+
+ @Override
+ public PipeMemoryBlock setExpandMethod(LongUnaryOperator extendMethod) {
+ throw new UnsupportedOperationException(
+ "Expand method is not supported in PipeTabletMemoryBlock");
+ }
+
+ @Override
+ public PipeMemoryBlock setExpandCallback(BiConsumer<Long, Long>
expandCallback) {
+ throw new UnsupportedOperationException(
+ "Expand callback is not supported in PipeTabletMemoryBlock");
+ }
+}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/task/connection/PipeEventCollector.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/task/connection/PipeEventCollector.java
index 727b976b098..50a91c82648 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/task/connection/PipeEventCollector.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/task/connection/PipeEventCollector.java
@@ -22,15 +22,26 @@ package org.apache.iotdb.db.pipe.task.connection;
import
org.apache.iotdb.commons.pipe.task.connection.BoundedBlockingPendingQueue;
import org.apache.iotdb.db.pipe.event.EnrichedEvent;
import org.apache.iotdb.db.pipe.event.common.heartbeat.PipeHeartbeatEvent;
+import
org.apache.iotdb.db.pipe.event.common.tablet.PipeInsertNodeTabletInsertionEvent;
+import
org.apache.iotdb.db.pipe.event.common.tablet.PipeRawTabletInsertionEvent;
+import org.apache.iotdb.db.pipe.event.common.tsfile.PipeTsFileInsertionEvent;
import org.apache.iotdb.db.pipe.progress.committer.PipeEventCommitManager;
import org.apache.iotdb.pipe.api.collector.EventCollector;
import org.apache.iotdb.pipe.api.event.Event;
+import org.apache.iotdb.pipe.api.event.dml.insertion.TabletInsertionEvent;
+import org.apache.iotdb.pipe.api.exception.PipeException;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import java.util.LinkedList;
import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
public class PipeEventCollector implements EventCollector, AutoCloseable {
+ private static final Logger LOGGER =
LoggerFactory.getLogger(PipeEventCollector.class);
+
private final BoundedBlockingPendingQueue<Event> pendingQueue;
private final EnrichedDeque<Event> bufferQueue;
@@ -39,6 +50,8 @@ public class PipeEventCollector implements EventCollector,
AutoCloseable {
private final AtomicBoolean isClosed = new AtomicBoolean(false);
+ private final AtomicInteger collectInvocationCount = new AtomicInteger(0);
+
public PipeEventCollector(BoundedBlockingPendingQueue<Event> pendingQueue,
int dataRegionId) {
this.pendingQueue = pendingQueue;
this.dataRegionId = dataRegionId;
@@ -47,6 +60,68 @@ public class PipeEventCollector implements EventCollector,
AutoCloseable {
@Override
public synchronized void collect(Event event) {
+ try {
+ if (event instanceof PipeInsertNodeTabletInsertionEvent) {
+ parseAndCollectEvent((PipeInsertNodeTabletInsertionEvent) event);
+ } else if (event instanceof PipeRawTabletInsertionEvent) {
+ parseAndCollectEvent((PipeRawTabletInsertionEvent) event);
+ } else if (event instanceof PipeTsFileInsertionEvent) {
+ parseAndCollectEvent((PipeTsFileInsertionEvent) event);
+ } else {
+ collectEvent(event);
+ }
+ } catch (Exception e) {
+ throw new PipeException("Error occurred when collecting events from
processor.", e);
+ }
+ }
+
+ private void parseAndCollectEvent(PipeInsertNodeTabletInsertionEvent
sourceEvent) {
+ if (sourceEvent.shouldParsePatternOrTime()) {
+ final PipeRawTabletInsertionEvent parsedEvent =
sourceEvent.parseEventWithPatternOrTime();
+ if (!parsedEvent.hasNoNeedParsingAndIsEmpty()) {
+ collectEvent(parsedEvent);
+ }
+ } else {
+ collectEvent(sourceEvent);
+ }
+ }
+
+ private void parseAndCollectEvent(PipeRawTabletInsertionEvent sourceEvent) {
+ if (sourceEvent.shouldParsePatternOrTime()) {
+ final PipeRawTabletInsertionEvent parsedEvent =
sourceEvent.parseEventWithPatternOrTime();
+ if (!parsedEvent.hasNoNeedParsingAndIsEmpty()) {
+ collectEvent(parsedEvent);
+ }
+ } else {
+ collectEvent(sourceEvent);
+ }
+ }
+
+ private void parseAndCollectEvent(PipeTsFileInsertionEvent sourceEvent)
throws Exception {
+ if (!sourceEvent.waitForTsFileClose()) {
+ LOGGER.warn(
+ "Pipe skipping temporary TsFile which shouldn't be transferred: {}",
+ sourceEvent.getTsFile());
+ return;
+ }
+
+ if (!sourceEvent.shouldParsePatternOrTime()) {
+ collectEvent(sourceEvent);
+ return;
+ }
+
+ try {
+ for (final TabletInsertionEvent parsedEvent :
sourceEvent.toTabletInsertionEvents()) {
+ collectEvent(parsedEvent);
+ }
+ } finally {
+ sourceEvent.close();
+ }
+ }
+
+ private void collectEvent(Event event) {
+ collectInvocationCount.incrementAndGet();
+
if (event instanceof EnrichedEvent) {
((EnrichedEvent)
event).increaseReferenceCount(PipeEventCollector.class.getName());
@@ -82,6 +157,14 @@ public class PipeEventCollector implements EventCollector,
AutoCloseable {
}
}
+ public void resetCollectInvocationCount() {
+ collectInvocationCount.set(0);
+ }
+
+ public boolean hasNoCollectInvocationAfterReset() {
+ return collectInvocationCount.get() == 0;
+ }
+
public boolean isBufferQueueEmpty() {
return bufferQueue.isEmpty();
}
@@ -118,6 +201,8 @@ public class PipeEventCollector implements EventCollector,
AutoCloseable {
bufferQueue.clear();
}
+ //////////////////////////// APIs provided for metric framework
////////////////////////////
+
public int getTabletInsertionEventCount() {
return bufferQueue.getTabletInsertionEventCount();
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/task/subtask/processor/PipeProcessorSubtask.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/task/subtask/processor/PipeProcessorSubtask.java
index 6ab50749760..2d3006b4166 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/task/subtask/processor/PipeProcessorSubtask.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/task/subtask/processor/PipeProcessorSubtask.java
@@ -104,6 +104,7 @@ public class PipeProcessorSubtask extends
PipeDataNodeSubtask {
: UserDefinedEnrichedEvent.maybeOf(inputEventSupplier.supply());
// Record the last event for retry when exception occurs
setLastEvent(event);
+
if (
// Though there is no event to process, there may still be some buffered
events
// in the outputEventCollector. Return true if there are still buffered
events,
@@ -115,6 +116,7 @@ public class PipeProcessorSubtask extends
PipeDataNodeSubtask {
return outputEventCollector.tryCollectBufferedEvents();
}
+ outputEventCollector.resetCollectInvocationCount();
try {
// event can be supplied after the subtask is closed, so we need to
check isClosed here
if (!isClosed.get()) {
@@ -136,8 +138,7 @@ public class PipeProcessorSubtask extends
PipeDataNodeSubtask {
outputEventCollector);
}
}
-
- releaseLastEvent(true);
+ releaseLastEvent(!isClosed.get() &&
outputEventCollector.hasNoCollectInvocationAfterReset());
} catch (Exception e) {
if (!isClosed.get()) {
throw new PipeException(
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonConfig.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonConfig.java
index c54a8325685..114f759d759 100644
---
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonConfig.java
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonConfig.java
@@ -151,6 +151,7 @@ public class CommonConfig {
Math.min(5, Math.max(1, Runtime.getRuntime().availableProcessors() / 2));
private int pipeDataStructureTabletRowSize = 2048;
+ private double pipeDataStructureTabletMemoryBlockAllocationRejectThreshold =
0.4;
private int pipeSubtaskExecutorBasicCheckPointIntervalByConsumedEventCount =
10_000;
private long pipeSubtaskExecutorBasicCheckPointIntervalByTimeDuration = 10 *
1000L;
@@ -531,6 +532,16 @@ public class CommonConfig {
this.pipeDataStructureTabletRowSize = pipeDataStructureTabletRowSize;
}
+ public double
getPipeDataStructureTabletMemoryBlockAllocationRejectThreshold() {
+ return pipeDataStructureTabletMemoryBlockAllocationRejectThreshold;
+ }
+
+ public void setPipeDataStructureTabletMemoryBlockAllocationRejectThreshold(
+ double pipeDataStructureTabletMemoryBlockAllocationRejectThreshold) {
+ this.pipeDataStructureTabletMemoryBlockAllocationRejectThreshold =
+ pipeDataStructureTabletMemoryBlockAllocationRejectThreshold;
+ }
+
public int getPipeExtractorAssignerDisruptorRingBufferSize() {
return pipeExtractorAssignerDisruptorRingBufferSize;
}
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonDescriptor.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonDescriptor.java
index 43ec8f911b4..f6dfe120db0 100644
---
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonDescriptor.java
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonDescriptor.java
@@ -258,6 +258,12 @@ public class CommonDescriptor {
properties.getProperty(
"pipe_data_structure_tablet_row_size",
String.valueOf(config.getPipeDataStructureTabletRowSize()))));
+ config.setPipeDataStructureTabletMemoryBlockAllocationRejectThreshold(
+ Double.parseDouble(
+ properties.getProperty(
+
"pipe_data_structure_tablet_memory_block_allocation_reject_threshold",
+ String.valueOf(
+
config.getPipeDataStructureTabletMemoryBlockAllocationRejectThreshold()))));
config.setPipeSubtaskExecutorMaxThreadNum(
Integer.parseInt(
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/PipeConfig.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/PipeConfig.java
index 74a6711a96b..219701f74ee 100644
---
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/PipeConfig.java
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/PipeConfig.java
@@ -53,6 +53,10 @@ public class PipeConfig {
return COMMON_CONFIG.getPipeDataStructureTabletRowSize();
}
+ public double
getPipeDataStructureTabletMemoryBlockAllocationRejectThreshold() {
+ return
COMMON_CONFIG.getPipeDataStructureTabletMemoryBlockAllocationRejectThreshold();
+ }
+
/////////////////////////////// Subtask Executor
///////////////////////////////
public int getPipeSubtaskExecutorMaxThreadNum() {
@@ -218,6 +222,9 @@ public class PipeConfig {
LOGGER.info("PipeHardLinkWALEnabled: {}", getPipeHardLinkWALEnabled());
LOGGER.info("PipeDataStructureTabletRowSize: {}",
getPipeDataStructureTabletRowSize());
+ LOGGER.info(
+ "PipeDataStructureTabletMemoryBlockAllocationRejectThreshold: {}",
+ getPipeDataStructureTabletMemoryBlockAllocationRejectThreshold());
LOGGER.info("PipeSubtaskExecutorMaxThreadNum: {}",
getPipeSubtaskExecutorMaxThreadNum());
LOGGER.info(
