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 ea3537bf7ed [IOTDB-6223] Pipe: Pipe Memory Management integrates into
the Pipe Module (#11424)
ea3537bf7ed is described below
commit ea3537bf7ed68ba6e99f7930018b0f78bd2e3b4b
Author: Itami Sho <[email protected]>
AuthorDate: Tue Oct 31 18:34:13 2023 +0800
[IOTDB-6223] Pipe: Pipe Memory Management integrates into the Pipe Module
(#11424)
Co-authored-by: Steve Yurong Su <[email protected]>
---
.../org/apache/iotdb/db/conf/IoTDBDescriptor.java | 8 +-
.../builder/PipeTransferBatchReqBuilder.java | 26 +++-
.../protocol/legacy/IoTDBLegacyPipeConnector.java | 1 -
.../connector/protocol/opcua/OpcUaConnector.java | 2 +-
.../thrift/async/IoTDBThriftAsyncConnector.java | 6 +
.../thrift/sync/IoTDBThriftSyncConnector.java | 4 +
.../common/tablet/PipeRawTabletInsertionEvent.java | 6 +
.../tsfile/TsFileInsertionDataContainer.java | 12 ++
.../realtime/assigner/DisruptorQueue.java | 17 ++-
.../db/pipe/resource/memory/PipeMemoryBlock.java | 4 +-
.../db/pipe/resource/memory/PipeMemoryManager.java | 141 +++++++++++++++++++--
.../dataregion/wal/utils/WALInsertNodeCache.java | 16 ++-
.../resources/conf/iotdb-common.properties | 4 +-
.../apache/iotdb/commons/conf/CommonConfig.java | 40 ++++++
.../iotdb/commons/conf/CommonDescriptor.java | 22 +++-
.../exception/pipe/PipeRuntimeExceptionType.java | 5 +
.../PipeRuntimeOutOfMemoryCriticalException.java | 100 +++++++++++++++
.../iotdb/commons/pipe/config/PipeConfig.java | 24 ++++
18 files changed, 414 insertions(+), 24 deletions(-)
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java
index e6de5b42b0d..19c00c4c941 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java
@@ -1736,7 +1736,13 @@ public class IoTDBDescriptor {
conf.setAllocateMemoryForPipe(
maxMemoryAvailable * Integer.parseInt(proportions[4].trim()) /
proportionSum);
} else {
- conf.setAllocateMemoryForPipe(0);
+ conf.setAllocateMemoryForPipe(
+ (maxMemoryAvailable
+ - (conf.getAllocateMemoryForStorageEngine()
+ + conf.getAllocateMemoryForRead()
+ + conf.getAllocateMemoryForSchema()
+ + conf.getAllocateMemoryForConsensus()))
+ / 2);
}
}
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/payload/evolvable/builder/PipeTransferBatchReqBuilder.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/payload/evolvable/builder/PipeTransferBatchReqBuilder.java
index 9521e6a97aa..e301645e865 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/payload/evolvable/builder/PipeTransferBatchReqBuilder.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/payload/evolvable/builder/PipeTransferBatchReqBuilder.java
@@ -24,12 +24,17 @@ import
org.apache.iotdb.db.pipe.connector.payload.evolvable.request.PipeTransfer
import
org.apache.iotdb.db.pipe.connector.payload.evolvable.request.PipeTransferTabletRawReq;
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.resource.PipeResourceManager;
+import org.apache.iotdb.db.pipe.resource.memory.PipeMemoryBlock;
import
org.apache.iotdb.db.storageengine.dataregion.wal.exception.WALPipeException;
import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters;
import org.apache.iotdb.pipe.api.event.Event;
import org.apache.iotdb.pipe.api.event.dml.insertion.TabletInsertionEvent;
import org.apache.iotdb.service.rpc.thrift.TPipeTransferReq;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
@@ -42,7 +47,9 @@ import static
org.apache.iotdb.db.pipe.config.constant.PipeConnectorConstant.CON
import static
org.apache.iotdb.db.pipe.config.constant.PipeConnectorConstant.SINK_IOTDB_BATCH_DELAY_KEY;
import static
org.apache.iotdb.db.pipe.config.constant.PipeConnectorConstant.SINK_IOTDB_BATCH_SIZE_KEY;
-public abstract class PipeTransferBatchReqBuilder {
+public abstract class PipeTransferBatchReqBuilder implements AutoCloseable {
+
+ private static final Logger LOGGER =
LoggerFactory.getLogger(PipeTransferBatchReqBuilder.class);
protected final List<TPipeTransferReq> reqs = new ArrayList<>();
protected final List<Event> events = new ArrayList<>();
@@ -52,6 +59,7 @@ public abstract class PipeTransferBatchReqBuilder {
protected long firstEventProcessingTime = Long.MIN_VALUE;
// limit in buffer size
+ protected final PipeMemoryBlock allocatedMemoryBlock;
protected final long maxBatchSizeInBytes;
protected long bufferSize = 0;
@@ -61,10 +69,19 @@ public abstract class PipeTransferBatchReqBuilder {
Arrays.asList(CONNECTOR_IOTDB_BATCH_DELAY_KEY,
SINK_IOTDB_BATCH_DELAY_KEY),
CONNECTOR_IOTDB_BATCH_DELAY_DEFAULT_VALUE)
* 1000;
- maxBatchSizeInBytes =
+
+ final long requestMaxBatchSizeInBytes =
parameters.getLongOrDefault(
Arrays.asList(CONNECTOR_IOTDB_BATCH_SIZE_KEY,
SINK_IOTDB_BATCH_SIZE_KEY),
CONNECTOR_IOTDB_BATCH_SIZE_DEFAULT_VALUE);
+ allocatedMemoryBlock =
PipeResourceManager.memory().tryAllocate(requestMaxBatchSizeInBytes);
+ maxBatchSizeInBytes = allocatedMemoryBlock.getMemoryUsageInBytes();
+ if (maxBatchSizeInBytes != requestMaxBatchSizeInBytes) {
+ LOGGER.info(
+ "PipeTransferBatchReqBuilder: the max batch size is adjusted from {}
to {}.",
+ requestMaxBatchSizeInBytes,
+ maxBatchSizeInBytes);
+ }
}
public List<TPipeTransferReq> getTPipeTransferReqs() {
@@ -99,4 +116,9 @@ public abstract class PipeTransferBatchReqBuilder {
}
return req;
}
+
+ @Override
+ public void close() {
+ allocatedMemoryBlock.close();
+ }
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/legacy/IoTDBLegacyPipeConnector.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/legacy/IoTDBLegacyPipeConnector.java
index f202fca7f23..fabf87f4bdf 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/legacy/IoTDBLegacyPipeConnector.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/legacy/IoTDBLegacyPipeConnector.java
@@ -246,7 +246,6 @@ public class IoTDBLegacyPipeConnector implements
PipeConnector {
throw new NotImplementedException(
"IoTDBLegacyPipeConnector only support PipeTsFileInsertionEvent.");
}
-
if (!((PipeTsFileInsertionEvent)
tsFileInsertionEvent).waitForTsFileClose()) {
LOGGER.warn(
"Pipe skipping temporary TsFile which shouldn't be transferred: {}",
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/opcua/OpcUaConnector.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/opcua/OpcUaConnector.java
index 30f2dbea963..9c2a695006d 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/opcua/OpcUaConnector.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/opcua/OpcUaConnector.java
@@ -161,7 +161,7 @@ public class OpcUaConnector implements PipeConnector {
if (!(tabletInsertionEvent instanceof PipeInsertNodeTabletInsertionEvent)
&& !(tabletInsertionEvent instanceof PipeRawTabletInsertionEvent)) {
LOGGER.warn(
- "IoTDBThriftSyncConnector only support "
+ "OpcUaConnector only support "
+ "PipeInsertNodeTabletInsertionEvent and
PipeRawTabletInsertionEvent. "
+ "Ignore {}.",
tabletInsertionEvent);
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 84a13d33a74..82966915c7d 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
@@ -71,6 +71,7 @@ import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
+import static
org.apache.iotdb.db.pipe.config.constant.PipeConnectorConstant.CONNECTOR_IOTDB_BATCH_MODE_ENABLE_KEY;
import static
org.apache.iotdb.db.pipe.config.constant.PipeConnectorConstant.SINK_IOTDB_BATCH_MODE_ENABLE_KEY;
public class IoTDBThriftAsyncConnector extends IoTDBConnector {
@@ -125,6 +126,7 @@ public class IoTDBThriftAsyncConnector extends
IoTDBConnector {
// Disable batch mode for retry connector, in case retry events are never
sent again
PipeParameters retryParameters = new PipeParameters(new
HashMap<>(parameters.getAttribute()));
retryParameters.getAttribute().put(SINK_IOTDB_BATCH_MODE_ENABLE_KEY,
"false");
+ retryParameters.getAttribute().put(CONNECTOR_IOTDB_BATCH_MODE_ENABLE_KEY,
"false");
retryConnector.customize(retryParameters, configuration);
if (isTabletBatchModeEnabled) {
@@ -561,5 +563,9 @@ public class IoTDBThriftAsyncConnector extends
IoTDBConnector {
// synchronized to avoid close connector when transfer event
public synchronized void close() throws Exception {
retryConnector.close();
+
+ if (tabletBatchBuilder != null) {
+ tabletBatchBuilder.close();
+ }
}
}
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 bfa2297086d..ef84812be7a 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
@@ -436,5 +436,9 @@ public class IoTDBThriftSyncConnector extends
IoTDBConnector {
isClientAlive.set(i, false);
}
}
+
+ if (tabletBatchBuilder != null) {
+ tabletBatchBuilder.close();
+ }
}
}
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 a3f1925c7d5..2ad67057995 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
@@ -24,6 +24,8 @@ import
org.apache.iotdb.commons.consensus.index.impl.MinimumProgressIndex;
import org.apache.iotdb.commons.pipe.task.meta.PipeTaskMeta;
import org.apache.iotdb.commons.utils.TestOnly;
import org.apache.iotdb.db.pipe.event.EnrichedEvent;
+import org.apache.iotdb.db.pipe.resource.PipeResourceManager;
+import org.apache.iotdb.db.pipe.resource.memory.PipeMemoryBlock;
import org.apache.iotdb.pipe.api.access.Row;
import org.apache.iotdb.pipe.api.collector.RowCollector;
import org.apache.iotdb.pipe.api.event.dml.insertion.TabletInsertionEvent;
@@ -40,6 +42,8 @@ public class PipeRawTabletInsertionEvent extends
EnrichedEvent implements Tablet
private final EnrichedEvent sourceEvent;
private boolean needToReport;
+ private PipeMemoryBlock allocatedMemoryBlock;
+
private TabletInsertionDataContainer dataContainer;
private PipeRawTabletInsertionEvent(
@@ -77,11 +81,13 @@ public class PipeRawTabletInsertionEvent extends
EnrichedEvent implements Tablet
@Override
public boolean internallyIncreaseResourceReferenceCount(String
holderMessage) {
+ allocatedMemoryBlock = PipeResourceManager.memory().forceAllocate(tablet);
return true;
}
@Override
public boolean internallyDecreaseResourceReferenceCount(String
holderMessage) {
+ allocatedMemoryBlock.close();
return true;
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/TsFileInsertionDataContainer.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/TsFileInsertionDataContainer.java
index 8769018bd2f..f83bca354ee 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/TsFileInsertionDataContainer.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/TsFileInsertionDataContainer.java
@@ -19,9 +19,12 @@
package org.apache.iotdb.db.pipe.event.common.tsfile;
+import org.apache.iotdb.commons.pipe.config.PipeConfig;
import org.apache.iotdb.commons.pipe.task.meta.PipeTaskMeta;
import org.apache.iotdb.db.pipe.event.EnrichedEvent;
import
org.apache.iotdb.db.pipe.event.common.tablet.PipeRawTabletInsertionEvent;
+import org.apache.iotdb.db.pipe.resource.PipeResourceManager;
+import org.apache.iotdb.db.pipe.resource.memory.PipeMemoryBlock;
import org.apache.iotdb.pipe.api.event.dml.insertion.TabletInsertionEvent;
import org.apache.iotdb.pipe.api.exception.PipeException;
import org.apache.iotdb.tsfile.common.constant.TsFileConstant;
@@ -58,6 +61,8 @@ public class TsFileInsertionDataContainer implements
AutoCloseable {
private final PipeTaskMeta pipeTaskMeta; // used to report progress
private final EnrichedEvent sourceEvent; // used to report progress
+ private final PipeMemoryBlock allocatedMemoryBlock;
+
private final TsFileSequenceReader tsFileSequenceReader;
private final TsFileReader tsFileReader;
@@ -90,6 +95,11 @@ public class TsFileInsertionDataContainer implements
AutoCloseable {
this.sourceEvent = sourceEvent;
try {
+ allocatedMemoryBlock =
+ PipeResourceManager.memory()
+ .forceAllocate(
+
PipeConfig.getInstance().getPipeMemoryAllocateForTsFileSequenceReaderInBytes());
+
tsFileSequenceReader = new
TsFileSequenceReader(tsFile.getAbsolutePath(), true, true);
tsFileReader = new TsFileReader(tsFileSequenceReader);
@@ -229,5 +239,7 @@ public class TsFileInsertionDataContainer implements
AutoCloseable {
} catch (IOException e) {
LOGGER.warn("Failed to close TsFileSequenceReader", e);
}
+
+ allocatedMemoryBlock.close();
}
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/realtime/assigner/DisruptorQueue.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/realtime/assigner/DisruptorQueue.java
index 5a0a0142e29..67acb16540f 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/realtime/assigner/DisruptorQueue.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/realtime/assigner/DisruptorQueue.java
@@ -25,6 +25,8 @@ 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.realtime.PipeRealtimeEvent;
import org.apache.iotdb.db.pipe.metric.PipeEventCounter;
+import org.apache.iotdb.db.pipe.resource.PipeResourceManager;
+import org.apache.iotdb.db.pipe.resource.memory.PipeMemoryBlock;
import com.lmax.disruptor.BlockingWaitStrategy;
import com.lmax.disruptor.EventHandler;
@@ -39,16 +41,28 @@ public class DisruptorQueue {
private static final IoTDBDaemonThreadFactory THREAD_FACTORY =
new IoTDBDaemonThreadFactory(PIPE_EXTRACTOR_DISRUPTOR.getName());
+ private final PipeMemoryBlock allocatedMemoryBlock;
private final Disruptor<EventContainer> disruptor;
private final RingBuffer<EventContainer> ringBuffer;
private final PipeEventCounter eventCounter = new PipeEventCounter();
public DisruptorQueue(EventHandler<PipeRealtimeEvent> eventHandler) {
+ final PipeConfig config = PipeConfig.getInstance();
+ final int ringBufferSize =
config.getPipeExtractorAssignerDisruptorRingBufferSize();
+ final long ringBufferEntrySizeInBytes =
+ config.getPipeExtractorAssignerDisruptorRingBufferEntrySizeInBytes();
+
+ allocatedMemoryBlock =
+ PipeResourceManager.memory().tryAllocate(ringBufferSize *
ringBufferEntrySizeInBytes);
+
disruptor =
new Disruptor<>(
EventContainer::new,
-
PipeConfig.getInstance().getPipeExtractorAssignerDisruptorRingBufferSize(),
+ Math.max(
+ 32,
+ Math.toIntExact(
+ allocatedMemoryBlock.getMemoryUsageInBytes() /
ringBufferEntrySizeInBytes)),
THREAD_FACTORY,
ProducerType.MULTI,
new BlockingWaitStrategy());
@@ -74,6 +88,7 @@ public class DisruptorQueue {
public void clear() {
disruptor.halt();
+ allocatedMemoryBlock.close();
}
private static class EventContainer {
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryBlock.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryBlock.java
index 166f7924b82..331eb32bd3c 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryBlock.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryBlock.java
@@ -31,7 +31,7 @@ public class PipeMemoryBlock implements AutoCloseable {
this.memoryUsageInBytes = memoryUsageInBytes;
}
- long getMemoryUsageInBytes() {
+ public long getMemoryUsageInBytes() {
return memoryUsageInBytes;
}
@@ -44,7 +44,7 @@ public class PipeMemoryBlock implements AutoCloseable {
}
@Override
- public void close() throws Exception {
+ public void close() {
PipeResourceManager.memory().release(this);
}
}
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 86a8020b7f3..895d5890a03 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
@@ -19,18 +19,26 @@
package org.apache.iotdb.db.pipe.resource.memory;
-import org.apache.iotdb.commons.exception.pipe.PipeRuntimeCriticalException;
-import org.apache.iotdb.commons.exception.pipe.PipeRuntimeException;
+import
org.apache.iotdb.commons.exception.pipe.PipeRuntimeOutOfMemoryCriticalException;
import org.apache.iotdb.commons.pipe.config.PipeConfig;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.tsfile.enums.TSDataType;
+import org.apache.iotdb.tsfile.utils.Binary;
+import org.apache.iotdb.tsfile.write.record.Tablet;
+import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.util.List;
+
public class PipeMemoryManager {
private static final Logger LOGGER =
LoggerFactory.getLogger(PipeMemoryManager.class);
+ private static final boolean PIPE_MEMORY_MANAGEMENT_ENABLED =
+ PipeConfig.getInstance().getPipeMemoryManagementEnabled();
+
private static final int MEMORY_ALLOCATE_MAX_RETRIES =
PipeConfig.getInstance().getPipeMemoryAllocateMaxRetries();
private static final long MEMORY_ALLOCATE_RETRY_INTERVAL_IN_MS =
@@ -38,10 +46,17 @@ public class PipeMemoryManager {
private static final long TOTAL_MEMORY_SIZE_IN_BYTES =
IoTDBDescriptor.getInstance().getConfig().getAllocateMemoryForPipe();
+ private static final long MEMORY_ALLOCATE_MIN_SIZE_IN_BYTES =
+ PipeConfig.getInstance().getPipeMemoryAllocateMinSizeInBytes();
+
private long usedMemorySizeInBytes = 0;
- public synchronized PipeMemoryBlock allocate(long sizeInBytes)
- throws PipeRuntimeException, InterruptedException {
+ public synchronized PipeMemoryBlock forceAllocate(long sizeInBytes)
+ throws PipeRuntimeOutOfMemoryCriticalException {
+ if (!PIPE_MEMORY_MANAGEMENT_ENABLED) {
+ return new PipeMemoryBlock(sizeInBytes);
+ }
+
for (int i = 1; i <= MEMORY_ALLOCATE_MAX_RETRIES; i++) {
if (TOTAL_MEMORY_SIZE_IN_BYTES - usedMemorySizeInBytes >= sizeInBytes) {
usedMemorySizeInBytes += sizeInBytes;
@@ -52,13 +67,13 @@ public class PipeMemoryManager {
this.wait(MEMORY_ALLOCATE_RETRY_INTERVAL_IN_MS);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
- LOGGER.warn("allocate: interrupted while waiting for available
memory", e);
+ LOGGER.warn("forceAllocate: interrupted while waiting for available
memory", e);
}
}
- throw new PipeRuntimeCriticalException(
+ throw new PipeRuntimeOutOfMemoryCriticalException(
String.format(
- "failed to allocate memory after %d retries, "
+ "forceAllocate: failed to allocate memory after %d retries, "
+ "total memory size %d bytes, used memory size %d bytes, "
+ "requested memory size %d bytes",
MEMORY_ALLOCATE_MAX_RETRIES,
@@ -67,8 +82,118 @@ public class PipeMemoryManager {
sizeInBytes));
}
+ public synchronized PipeMemoryBlock forceAllocate(Tablet tablet)
+ throws PipeRuntimeOutOfMemoryCriticalException {
+ return forceAllocate(calculateTabletSizeInBytes(tablet));
+ }
+
+ private long calculateTabletSizeInBytes(Tablet tablet) {
+ long totalSizeInBytes = 0;
+
+ if (tablet == null) {
+ return totalSizeInBytes;
+ }
+
+ // timestamps
+ if (tablet.timestamps != null) {
+ totalSizeInBytes += tablet.timestamps.length * 8L;
+ }
+
+ // values
+ final List<MeasurementSchema> timeseries = tablet.getSchemas();
+ if (timeseries != null) {
+ for (int column = 0; column < timeseries.size(); column++) {
+ final MeasurementSchema measurementSchema = timeseries.get(column);
+ if (measurementSchema == null) {
+ continue;
+ }
+
+ final TSDataType tsDataType = measurementSchema.getType();
+ if (tsDataType == null) {
+ continue;
+ }
+
+ if (tsDataType == TSDataType.TEXT) {
+ if (tablet.values == null || tablet.values.length <= column) {
+ continue;
+ }
+ final Binary[] values = ((Binary[]) tablet.values[column]);
+ if (values == null) {
+ continue;
+ }
+ for (Binary value : values) {
+ totalSizeInBytes +=
+ value == null ? 0 : (value.getLength() == -1 ? 0 :
value.getLength());
+ }
+ } else {
+ totalSizeInBytes += (long) tablet.timestamps.length *
tsDataType.getDataTypeSize();
+ }
+ }
+ }
+
+ // bitMaps
+ if (tablet.bitMaps != null) {
+ for (int i = 0; i < tablet.bitMaps.length; i++) {
+ totalSizeInBytes += tablet.bitMaps[i] == null ? 0 :
tablet.bitMaps[i].getSize();
+ }
+ }
+
+ // estimate other dataStructures size
+ totalSizeInBytes += 100;
+
+ return totalSizeInBytes;
+ }
+
+ public synchronized PipeMemoryBlock tryAllocate(long sizeInBytes) {
+ if (!PIPE_MEMORY_MANAGEMENT_ENABLED) {
+ return new PipeMemoryBlock(sizeInBytes);
+ }
+
+ if (TOTAL_MEMORY_SIZE_IN_BYTES - usedMemorySizeInBytes >= sizeInBytes) {
+ usedMemorySizeInBytes += sizeInBytes;
+ return new PipeMemoryBlock(sizeInBytes);
+ }
+
+ long sizeToAllocateInBytes = sizeInBytes;
+ while (sizeToAllocateInBytes > MEMORY_ALLOCATE_MIN_SIZE_IN_BYTES) {
+ if (TOTAL_MEMORY_SIZE_IN_BYTES - usedMemorySizeInBytes >=
sizeToAllocateInBytes) {
+ LOGGER.info(
+ "tryAllocate: allocated memory, "
+ + "total memory size {} bytes, used memory size {} bytes, "
+ + "original requested memory size {} bytes,"
+ + "actual requested memory size {} bytes",
+ TOTAL_MEMORY_SIZE_IN_BYTES,
+ usedMemorySizeInBytes,
+ sizeInBytes,
+ sizeToAllocateInBytes);
+ usedMemorySizeInBytes += sizeToAllocateInBytes;
+ return new PipeMemoryBlock(sizeToAllocateInBytes);
+ }
+
+ try {
+ this.wait(MEMORY_ALLOCATE_RETRY_INTERVAL_IN_MS);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ LOGGER.warn("tryAllocate: interrupted while waiting for available
memory", e);
+ }
+
+ sizeToAllocateInBytes =
+ Math.max(sizeToAllocateInBytes * 2 / 3,
MEMORY_ALLOCATE_MIN_SIZE_IN_BYTES);
+ }
+
+ LOGGER.warn(
+ "tryAllocate: failed to allocate memory, "
+ + "total memory size {} bytes, used memory size {} bytes, "
+ + "requested memory size {} bytes",
+ TOTAL_MEMORY_SIZE_IN_BYTES,
+ usedMemorySizeInBytes,
+ sizeInBytes);
+
+ return new PipeMemoryBlock(0);
+ }
+
public synchronized void release(PipeMemoryBlock block) {
- if (block == null || block.isReleased()) {
+ if (!PIPE_MEMORY_MANAGEMENT_ENABLED || block == null ||
block.isReleased()) {
return;
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/utils/WALInsertNodeCache.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/utils/WALInsertNodeCache.java
index d60b964c9ac..44e2c287622 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/utils/WALInsertNodeCache.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/utils/WALInsertNodeCache.java
@@ -23,6 +23,8 @@ import org.apache.iotdb.commons.utils.TestOnly;
import org.apache.iotdb.db.conf.IoTDBConfig;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.pipe.metric.PipeWALInsertNodeCacheMetrics;
+import org.apache.iotdb.db.pipe.resource.PipeResourceManager;
+import org.apache.iotdb.db.pipe.resource.memory.PipeMemoryBlock;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertNode;
import org.apache.iotdb.db.storageengine.dataregion.wal.buffer.WALEntry;
@@ -51,12 +53,14 @@ import java.util.concurrent.ConcurrentHashMap;
/** This cache is used by {@link WALEntryPosition}. */
public class WALInsertNodeCache {
+
private static final Logger LOGGER =
LoggerFactory.getLogger(WALInsertNodeCache.class);
private static final IoTDBConfig CONFIG =
IoTDBDescriptor.getInstance().getConfig();
// LRU cache, find Pair<ByteBuffer, InsertNode> by WALEntryPosition
- private final LoadingCache<WALEntryPosition, Pair<ByteBuffer, InsertNode>>
lruCache;
+ private final PipeMemoryBlock allocatedMemoryBlock;
private boolean isBatchLoadEnabled;
+ private final LoadingCache<WALEntryPosition, Pair<ByteBuffer, InsertNode>>
lruCache;
// ids of all pinned memTables
private final Set<Long> memTablesNeedSearch = ConcurrentHashMap.newKeySet();
@@ -64,12 +68,13 @@ public class WALInsertNodeCache {
private volatile boolean hasPipeRunning = false;
private WALInsertNodeCache(Integer dataRegionId) {
- // TODO: try allocate memory 2 * config.getWalFileSizeThresholdInByte()
for the cache
- // If allocate memory failed, disable batch load
- isBatchLoadEnabled = true;
+ allocatedMemoryBlock =
+ PipeResourceManager.memory().tryAllocate(2 *
CONFIG.getWalFileSizeThresholdInByte());
+ isBatchLoadEnabled =
+ allocatedMemoryBlock.getMemoryUsageInBytes() >=
CONFIG.getWalFileSizeThresholdInByte();
lruCache =
Caffeine.newBuilder()
- .maximumWeight(2 * CONFIG.getWalFileSizeThresholdInByte())
+ .maximumWeight(allocatedMemoryBlock.getMemoryUsageInBytes())
.weigher(
(Weigher<WALEntryPosition, Pair<ByteBuffer, InsertNode>>)
(position, pair) -> position.getSize())
@@ -287,6 +292,7 @@ public class WALInsertNodeCache {
@TestOnly
public void clear() {
lruCache.invalidateAll();
+ allocatedMemoryBlock.close();
memTablesNeedSearch.clear();
}
}
diff --git
a/iotdb-core/node-commons/src/assembly/resources/conf/iotdb-common.properties
b/iotdb-core/node-commons/src/assembly/resources/conf/iotdb-common.properties
index 918d1cc3bb5..a67d3ada788 100644
---
a/iotdb-core/node-commons/src/assembly/resources/conf/iotdb-common.properties
+++
b/iotdb-core/node-commons/src/assembly/resources/conf/iotdb-common.properties
@@ -172,9 +172,9 @@ cluster_name=defaultCluster
# Datatype: boolean
# enable_mem_control=true
-# Memory Allocation Ratio: StorageEngine, QueryEngine, SchemaEngine,
StreamingEngine, Consensus and Free Memory.
+# Memory Allocation Ratio: StorageEngine, QueryEngine, SchemaEngine,
Consensus, StreamingEngine and Free Memory.
# The parameter form is a:b:c:d:e:f, where a, b, c, d, e and f are integers.
for example: 1:1:1:1:1:1 , 6:2:1:1:1:1
-# If you have high level of writing pressure and low level of reading
pressure, please adjust it to for example 6:1:1:1:1:2
+# If you have high level of writing pressure and low level of reading
pressure, please adjust it to for example 6:1:1:1:1:1
# datanode_memory_proportion=3:3:1:1:1:1
# Schema Memory Allocation Ratio: SchemaRegion, SchemaCache, and
PartitionCache.
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 66c3e93c4df..df94ca21013 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
@@ -158,6 +158,7 @@ public class CommonConfig {
private long pipeSubtaskExecutorPendingQueueMaxBlockingTimeMs = 1000;
private int pipeExtractorAssignerDisruptorRingBufferSize = 65536;
+ private long pipeExtractorAssignerDisruptorRingBufferEntrySizeInBytes = 50;
// 50B
private int pipeExtractorMatcherCacheSize = 1024;
private long pipeConnectorTimeoutMs = 15 * 60 * 1000L; // 15 minutes
@@ -184,8 +185,11 @@ public class CommonConfig {
private int pipeMaxAllowedPendingTsFileEpochPerDataRegion = 2;
+ private boolean pipeMemoryManagementEnabled = true;
private long pipeMemoryAllocateRetryIntervalMs = 1000;
private int pipeMemoryAllocateMaxRetries = 10;
+ private long pipeMemoryAllocateMinSizeInBytes = 32;
+ private long pipeMemoryAllocateForTsFileSequenceReaderInBytes = 2 * 1024 *
1024; // 2MB
/** Whether to use persistent schema mode. */
private String schemaEngineMode = "Memory";
@@ -529,6 +533,16 @@ public class CommonConfig {
pipeExtractorAssignerDisruptorRingBufferSize;
}
+ public long getPipeExtractorAssignerDisruptorRingBufferEntrySizeInBytes() {
+ return pipeExtractorAssignerDisruptorRingBufferEntrySizeInBytes;
+ }
+
+ public void setPipeExtractorAssignerDisruptorRingBufferEntrySizeInBytes(
+ long pipeExtractorAssignerDisruptorRingBufferEntrySize) {
+ this.pipeExtractorAssignerDisruptorRingBufferEntrySizeInBytes =
+ pipeExtractorAssignerDisruptorRingBufferEntrySize;
+ }
+
public int getPipeExtractorMatcherCacheSize() {
return pipeExtractorMatcherCacheSize;
}
@@ -720,6 +734,24 @@ public class CommonConfig {
this.pipeMaxAllowedPendingTsFileEpochPerDataRegion =
pipeExtractorPendingQueueTsfileLimit;
}
+ public boolean getPipeMemoryManagementEnabled() {
+ return pipeMemoryManagementEnabled;
+ }
+
+ public void setPipeMemoryManagementEnabled(boolean
pipeMemoryManagementEnabled) {
+ this.pipeMemoryManagementEnabled = pipeMemoryManagementEnabled;
+ }
+
+ public long getPipeMemoryAllocateForTsFileSequenceReaderInBytes() {
+ return pipeMemoryAllocateForTsFileSequenceReaderInBytes;
+ }
+
+ public void setPipeMemoryAllocateForTsFileSequenceReaderInBytes(
+ long pipeMemoryAllocateForTsFileSequenceReaderInBytes) {
+ this.pipeMemoryAllocateForTsFileSequenceReaderInBytes =
+ pipeMemoryAllocateForTsFileSequenceReaderInBytes;
+ }
+
public int getPipeMemoryAllocateMaxRetries() {
return pipeMemoryAllocateMaxRetries;
}
@@ -736,6 +768,14 @@ public class CommonConfig {
this.pipeMemoryAllocateRetryIntervalMs = pipeMemoryAllocateRetryIntervalMs;
}
+ public long getPipeMemoryAllocateMinSizeInBytes() {
+ return pipeMemoryAllocateMinSizeInBytes;
+ }
+
+ public void setPipeMemoryAllocateMinSizeInBytes(long
pipeMemoryAllocateMinSizeInBytes) {
+ this.pipeMemoryAllocateMinSizeInBytes = pipeMemoryAllocateMinSizeInBytes;
+ }
+
public String getSchemaEngineMode() {
return schemaEngineMode;
}
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 2a569327072..bc254f832b1 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
@@ -283,6 +283,12 @@ public class CommonDescriptor {
properties.getProperty(
"pipe_extractor_assigner_disruptor_ring_buffer_size",
String.valueOf(config.getPipeExtractorAssignerDisruptorRingBufferSize()))));
+ config.setPipeExtractorAssignerDisruptorRingBufferEntrySizeInBytes( // 1MB
+ Integer.parseInt(
+ properties.getProperty(
+
"pipe_extractor_assigner_disruptor_ring_buffer_entry_size_in_bytes",
+ String.valueOf(
+
config.getPipeExtractorAssignerDisruptorRingBufferEntrySizeInBytes()))));
config.setPipeExtractorMatcherCacheSize(
Integer.parseInt(
properties.getProperty(
@@ -377,17 +383,31 @@ public class CommonDescriptor {
"pipe_max_allowed_pending_tsfile_epoch_per_data_region",
String.valueOf(config.getPipeMaxAllowedPendingTsFileEpochPerDataRegion()))));
+ config.setPipeMemoryManagementEnabled(
+ Boolean.parseBoolean(
+ properties.getProperty(
+ "pipe_memory_management_enabled",
+ String.valueOf(config.getPipeMemoryManagementEnabled()))));
config.setPipeMemoryAllocateMaxRetries(
Integer.parseInt(
properties.getProperty(
"pipe_memory_allocate_max_retries",
String.valueOf(config.getPipeMemoryAllocateMaxRetries()))));
-
config.setPipeMemoryAllocateRetryIntervalInMs(
Long.parseLong(
properties.getProperty(
"pipe_memory_allocate_retry_interval_in_ms",
String.valueOf(config.getPipeMemoryAllocateRetryIntervalInMs()))));
+ config.setPipeMemoryAllocateMinSizeInBytes(
+ Long.parseLong(
+ properties.getProperty(
+ "pipe_memory_allocate_min_size_in_bytes",
+
String.valueOf(config.getPipeMemoryAllocateMinSizeInBytes()))));
+ config.setPipeMemoryAllocateForTsFileSequenceReaderInBytes(
+ Long.parseLong(
+ properties.getProperty(
+ "pipe_memory_allocate_for_tsfile_sequence_reader_in_bytes",
+
String.valueOf(config.getPipeMemoryAllocateForTsFileSequenceReaderInBytes()))));
}
public void loadGlobalConfig(TGlobalConfig globalConfig) {
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/exception/pipe/PipeRuntimeExceptionType.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/exception/pipe/PipeRuntimeExceptionType.java
index 4feb7a9c48f..287d0d11b06 100644
---
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/exception/pipe/PipeRuntimeExceptionType.java
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/exception/pipe/PipeRuntimeExceptionType.java
@@ -31,6 +31,7 @@ public enum PipeRuntimeExceptionType {
NON_CRITICAL_EXCEPTION((short) 1),
CRITICAL_EXCEPTION((short) 2),
CONNECTOR_CRITICAL_EXCEPTION((short) 3),
+ OUT_OF_MEMORY_CRITICAL_EXCEPTION((short) 4),
;
private final short type;
@@ -61,6 +62,8 @@ public enum PipeRuntimeExceptionType {
return PipeRuntimeCriticalException.deserializeFrom(version,
byteBuffer);
case 3:
return PipeRuntimeConnectorCriticalException.deserializeFrom(version,
byteBuffer);
+ case 4:
+ return
PipeRuntimeOutOfMemoryCriticalException.deserializeFrom(version, byteBuffer);
default:
throw new UnsupportedOperationException(
String.format("Unsupported PipeRuntimeException type %s.", type));
@@ -77,6 +80,8 @@ public enum PipeRuntimeExceptionType {
return PipeRuntimeCriticalException.deserializeFrom(version, stream);
case 3:
return PipeRuntimeConnectorCriticalException.deserializeFrom(version,
stream);
+ case 4:
+ return
PipeRuntimeOutOfMemoryCriticalException.deserializeFrom(version, stream);
default:
throw new UnsupportedOperationException(
String.format("Unsupported PipeRuntimeException type %s.", type));
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/exception/pipe/PipeRuntimeOutOfMemoryCriticalException.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/exception/pipe/PipeRuntimeOutOfMemoryCriticalException.java
new file mode 100644
index 00000000000..b073152a3e5
--- /dev/null
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/exception/pipe/PipeRuntimeOutOfMemoryCriticalException.java
@@ -0,0 +1,100 @@
+/*
+ * 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.commons.exception.pipe;
+
+import org.apache.iotdb.commons.pipe.task.meta.PipeRuntimeMetaVersion;
+import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.nio.ByteBuffer;
+import java.util.Objects;
+
+public class PipeRuntimeOutOfMemoryCriticalException extends
PipeRuntimeCriticalException {
+
+ public PipeRuntimeOutOfMemoryCriticalException(String message) {
+ super(message);
+ }
+
+ public PipeRuntimeOutOfMemoryCriticalException(String message, long
timeStamp) {
+ super(message, timeStamp);
+ }
+
+ @Override
+ public boolean equals(Object obj) {
+ return obj instanceof PipeRuntimeOutOfMemoryCriticalException
+ && Objects.equals(
+ getMessage(), ((PipeRuntimeOutOfMemoryCriticalException)
obj).getMessage())
+ && Objects.equals(getTimeStamp(), ((PipeRuntimeException)
obj).getTimeStamp());
+ }
+
+ @Override
+ public void serialize(ByteBuffer byteBuffer) {
+
PipeRuntimeExceptionType.OUT_OF_MEMORY_CRITICAL_EXCEPTION.serialize(byteBuffer);
+ ReadWriteIOUtils.write(getMessage(), byteBuffer);
+ ReadWriteIOUtils.write(getTimeStamp(), byteBuffer);
+ }
+
+ @Override
+ public void serialize(OutputStream stream) throws IOException {
+
PipeRuntimeExceptionType.OUT_OF_MEMORY_CRITICAL_EXCEPTION.serialize(stream);
+ ReadWriteIOUtils.write(getMessage(), stream);
+ ReadWriteIOUtils.write(getTimeStamp(), stream);
+ }
+
+ public static PipeRuntimeOutOfMemoryCriticalException deserializeFrom(
+ PipeRuntimeMetaVersion version, ByteBuffer byteBuffer) {
+ final String message = ReadWriteIOUtils.readString(byteBuffer);
+ switch (version) {
+ case VERSION_1:
+ return new PipeRuntimeOutOfMemoryCriticalException(message);
+ case VERSION_2:
+ return new PipeRuntimeOutOfMemoryCriticalException(
+ message, ReadWriteIOUtils.readLong(byteBuffer));
+ default:
+ throw new UnsupportedOperationException(String.format("Unsupported
version %s", version));
+ }
+ }
+
+ public static PipeRuntimeOutOfMemoryCriticalException deserializeFrom(
+ PipeRuntimeMetaVersion version, InputStream stream) throws IOException {
+ final String message = ReadWriteIOUtils.readString(stream);
+ switch (version) {
+ case VERSION_1:
+ return new PipeRuntimeOutOfMemoryCriticalException(message);
+ case VERSION_2:
+ return new PipeRuntimeOutOfMemoryCriticalException(
+ message, ReadWriteIOUtils.readLong(stream));
+ default:
+ throw new UnsupportedOperationException(String.format("Unsupported
version %s", version));
+ }
+ }
+
+ @Override
+ public String toString() {
+ return "PipeRuntimeOutOfMemoryException{"
+ + "message='"
+ + getMessage()
+ + "', timeStamp="
+ + getTimeStamp()
+ + "}";
+ }
+}
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 05517648845..47e3f1d5003 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
@@ -77,6 +77,10 @@ public class PipeConfig {
return COMMON_CONFIG.getPipeExtractorAssignerDisruptorRingBufferSize();
}
+ public long getPipeExtractorAssignerDisruptorRingBufferEntrySizeInBytes() {
+ return
COMMON_CONFIG.getPipeExtractorAssignerDisruptorRingBufferEntrySizeInBytes();
+ }
+
public int getPipeExtractorMatcherCacheSize() {
return COMMON_CONFIG.getPipeExtractorMatcherCacheSize();
}
@@ -159,6 +163,10 @@ public class PipeConfig {
/////////////////////////////// Memory ///////////////////////////////
+ public boolean getPipeMemoryManagementEnabled() {
+ return COMMON_CONFIG.getPipeMemoryManagementEnabled();
+ }
+
public int getPipeMemoryAllocateMaxRetries() {
return COMMON_CONFIG.getPipeMemoryAllocateMaxRetries();
}
@@ -167,6 +175,14 @@ public class PipeConfig {
return COMMON_CONFIG.getPipeMemoryAllocateRetryIntervalInMs();
}
+ public long getPipeMemoryAllocateMinSizeInBytes() {
+ return COMMON_CONFIG.getPipeMemoryAllocateMinSizeInBytes();
+ }
+
+ public long getPipeMemoryAllocateForTsFileSequenceReaderInBytes() {
+ return COMMON_CONFIG.getPipeMemoryAllocateForTsFileSequenceReaderInBytes();
+ }
+
/////////////////////////////// Utils ///////////////////////////////
private static final Logger LOGGER =
LoggerFactory.getLogger(PipeConfig.class);
@@ -193,6 +209,9 @@ public class PipeConfig {
LOGGER.info(
"PipeExtractorAssignerDisruptorRingBufferSize: {}",
getPipeExtractorAssignerDisruptorRingBufferSize());
+ LOGGER.info(
+ "PipeExtractorAssignerDisruptorRingBufferEntrySizeInBytes: {}",
+ getPipeExtractorAssignerDisruptorRingBufferEntrySizeInBytes());
LOGGER.info("PipeExtractorMatcherCacheSize: {}",
getPipeExtractorMatcherCacheSize());
LOGGER.info("PipeConnectorTimeoutMs: {}", getPipeConnectorTimeoutMs());
@@ -226,9 +245,14 @@ public class PipeConfig {
"PipeMaxAllowedPendingTsFileEpochPerDataRegion: {}",
getPipeMaxAllowedPendingTsFileEpochPerDataRegion());
+ LOGGER.info("PipeMemoryManagementEnabled: {}",
getPipeMemoryManagementEnabled());
LOGGER.info("PipeMemoryAllocateMaxRetries: {}",
getPipeMemoryAllocateMaxRetries());
LOGGER.info(
"PipeMemoryAllocateRetryIntervalInMs: {}",
getPipeMemoryAllocateRetryIntervalInMs());
+ LOGGER.info("PipeMemoryAllocateMinSizeInBytes: {}",
getPipeMemoryAllocateMinSizeInBytes());
+ LOGGER.info(
+ "PipeMemoryAllocateForTsFileSequenceReaderInBytes: {}",
+ getPipeMemoryAllocateForTsFileSequenceReaderInBytes());
}
/////////////////////////////// Singleton ///////////////////////////////
