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 4757f4d2960 [IOTDB-6198] Pipe: Memory Management Framework (#11350)
4757f4d2960 is described below
commit 4757f4d29602d1404443fa032dc6e18c9d967d17
Author: Itami Sho <[email protected]>
AuthorDate: Wed Oct 25 09:40:19 2023 +0800
[IOTDB-6198] Pipe: Memory Management Framework (#11350)
Co-authored-by: Steve Yurong Su <[email protected]>
---
.../java/org/apache/iotdb/db/conf/IoTDBConfig.java | 11 +++
.../org/apache/iotdb/db/conf/IoTDBDescriptor.java | 6 ++
.../db/pipe/resource/PipeResourceManager.java | 7 ++
.../db/pipe/resource/memory/PipeMemoryBlock.java | 50 ++++++++++++
.../db/pipe/resource/memory/PipeMemoryManager.java | 88 ++++++++++++++++++++++
.../apache/iotdb/commons/conf/CommonConfig.java | 19 +++++
.../iotdb/commons/conf/CommonDescriptor.java | 12 +++
.../iotdb/commons/pipe/config/PipeConfig.java | 14 ++++
8 files changed, 207 insertions(+)
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java
index e7685e25c0d..15aa68d6d9f 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java
@@ -145,6 +145,9 @@ public class IoTDBConfig {
/** Memory allocated for the consensus layer */
private long allocateMemoryForConsensus = Runtime.getRuntime().maxMemory() /
10;
+ /** Memory allocated for the pipe */
+ private long allocateMemoryForPipe = Runtime.getRuntime().maxMemory() / 10;
+
/** Ratio of memory allocated for buffered arrays */
private double bufferedArraysMemoryProportion = 0.6;
@@ -1899,6 +1902,14 @@ public class IoTDBConfig {
this.allocateMemoryForTimeIndex = allocateMemoryForRead * 200 / 1001;
}
+ public long getAllocateMemoryForPipe() {
+ return allocateMemoryForPipe;
+ }
+
+ public void setAllocateMemoryForPipe(long allocateMemoryForPipe) {
+ this.allocateMemoryForPipe = allocateMemoryForPipe;
+ }
+
public long getAllocateMemoryForFree() {
return Runtime.getRuntime().maxMemory()
- allocateMemoryForStorageEngine
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 7d0f230bc16..5bf62aa2eee 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
@@ -1712,6 +1712,11 @@ public class IoTDBDescriptor {
maxMemoryAvailable * Integer.parseInt(proportions[2].trim()) /
proportionSum);
conf.setAllocateMemoryForConsensus(
maxMemoryAvailable * Integer.parseInt(proportions[3].trim()) /
proportionSum);
+ // if pipe proportion is set, use it, otherwise use the default value
+ if (proportions.length >= 6) {
+ conf.setAllocateMemoryForPipe(
+ maxMemoryAvailable * Integer.parseInt(proportions[4].trim()) /
proportionSum);
+ }
}
}
@@ -1719,6 +1724,7 @@ public class IoTDBDescriptor {
logger.info("initial allocateMemoryForWrite = {}",
conf.getAllocateMemoryForStorageEngine());
logger.info("initial allocateMemoryForSchema = {}",
conf.getAllocateMemoryForSchema());
logger.info("initial allocateMemoryForConsensus = {}",
conf.getAllocateMemoryForConsensus());
+ logger.info("initial allocateMemoryForPipe = {}",
conf.getAllocateMemoryForPipe());
initSchemaMemoryAllocate(properties);
initStorageEngineAllocate(properties);
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/PipeResourceManager.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/PipeResourceManager.java
index 519aa0b56ab..56fe8cd6bd0 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/PipeResourceManager.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/PipeResourceManager.java
@@ -20,6 +20,7 @@
package org.apache.iotdb.db.pipe.resource;
import org.apache.iotdb.commons.pipe.config.PipeConfig;
+import org.apache.iotdb.db.pipe.resource.memory.PipeMemoryManager;
import org.apache.iotdb.db.pipe.resource.tsfile.PipeTsFileResourceManager;
import org.apache.iotdb.db.pipe.resource.wal.PipeWALResourceManager;
import
org.apache.iotdb.db.pipe.resource.wal.hardlink.PipeWALHardlinkResourceManager;
@@ -31,6 +32,7 @@ public class PipeResourceManager {
private final PipeTsFileResourceManager pipeTsFileResourceManager;
private final AtomicReference<PipeWALResourceManager> pipeWALResourceManager;
+ private final PipeMemoryManager pipeMemoryManager;
public static PipeTsFileResourceManager tsfile() {
return PipeResourceManagerHolder.INSTANCE.pipeTsFileResourceManager;
@@ -50,11 +52,16 @@ public class PipeResourceManager {
return PipeResourceManagerHolder.INSTANCE.pipeWALResourceManager.get();
}
+ public static PipeMemoryManager memory() {
+ return PipeResourceManagerHolder.INSTANCE.pipeMemoryManager;
+ }
+
///////////////////////////// SINGLETON /////////////////////////////
private PipeResourceManager() {
pipeTsFileResourceManager = new PipeTsFileResourceManager();
pipeWALResourceManager = new AtomicReference<>();
+ pipeMemoryManager = new PipeMemoryManager();
}
private static class PipeResourceManagerHolder {
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
new file mode 100644
index 00000000000..166f7924b82
--- /dev/null
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryBlock.java
@@ -0,0 +1,50 @@
+/*
+ * 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 org.apache.iotdb.db.pipe.resource.PipeResourceManager;
+
+public class PipeMemoryBlock implements AutoCloseable {
+
+ private final long memoryUsageInBytes;
+
+ private volatile boolean isReleased = false;
+
+ public PipeMemoryBlock(long memoryUsageInBytes) {
+ this.memoryUsageInBytes = memoryUsageInBytes;
+ }
+
+ long getMemoryUsageInBytes() {
+ return memoryUsageInBytes;
+ }
+
+ boolean isReleased() {
+ return isReleased;
+ }
+
+ void markAsReleased() {
+ isReleased = true;
+ }
+
+ @Override
+ public void close() throws Exception {
+ 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
new file mode 100644
index 00000000000..19a6f80246c
--- /dev/null
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryManager.java
@@ -0,0 +1,88 @@
+/*
+ * 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 org.apache.iotdb.commons.exception.pipe.PipeRuntimeCriticalException;
+import org.apache.iotdb.commons.exception.pipe.PipeRuntimeException;
+import org.apache.iotdb.commons.pipe.config.PipeConfig;
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class PipeMemoryManager {
+
+ private static final Logger LOGGER =
LoggerFactory.getLogger(PipeMemoryManager.class);
+
+ private static final int MEMORY_ALLOCATE_MAX_RETRIES =
+ PipeConfig.getInstance().getPipeMemoryAllocateMaxRetries();
+ private static final long MEMORY_ALLOCATE_RETRY_INTERVAL_IN_MS =
+ PipeConfig.getInstance().getPipeMemoryAllocateRetryIntervalInMs();
+
+ private static final long TOTAL_MEMORY_SIZE_IN_BYTES =
+ IoTDBDescriptor.getInstance().getConfig().getAllocateMemoryForPipe();
+ private long usedMemorySizeInBytes = 0;
+
+ public synchronized PipeMemoryBlock allocate(long sizeInBytes)
+ throws PipeRuntimeException, InterruptedException {
+ for (int i = 1; i <= MEMORY_ALLOCATE_MAX_RETRIES; i++) {
+ if (TOTAL_MEMORY_SIZE_IN_BYTES - usedMemorySizeInBytes >= sizeInBytes) {
+ usedMemorySizeInBytes += sizeInBytes;
+ return new PipeMemoryBlock(sizeInBytes);
+ }
+
+ try {
+ this.wait(MEMORY_ALLOCATE_RETRY_INTERVAL_IN_MS);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ LOGGER.warn("allocate: interrupted while waiting for available
memory", e);
+ }
+ }
+
+ throw new PipeRuntimeCriticalException(
+ String.format(
+ "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,
+ TOTAL_MEMORY_SIZE_IN_BYTES,
+ usedMemorySizeInBytes,
+ sizeInBytes));
+ }
+
+ public synchronized void release(PipeMemoryBlock block) {
+ if (block == null || block.isReleased()) {
+ return;
+ }
+
+ usedMemorySizeInBytes -= block.getMemoryUsageInBytes();
+ block.markAsReleased();
+
+ this.notifyAll();
+ }
+
+ public long getUsedMemorySizeInBytes() {
+ return usedMemorySizeInBytes;
+ }
+
+ public long getTotalMemorySizeInBytes() {
+ return TOTAL_MEMORY_SIZE_IN_BYTES;
+ }
+}
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 25c751e83ed..a94c7889a45 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
@@ -191,6 +191,9 @@ public class CommonConfig {
private int pipeMaxAllowedPendingTsFileEpochPerDataRegion = 1;
+ private long pipeMemoryAllocateRetryIntervalMs = 1000;
+ private int pipeMemoryAllocateMaxRetries = 10;
+
/** Whether to use persistent schema mode. */
private String schemaEngineMode = "Memory";
@@ -732,6 +735,22 @@ public class CommonConfig {
this.pipeMaxAllowedPendingTsFileEpochPerDataRegion =
pipeExtractorPendingQueueTsfileLimit;
}
+ public int getPipeMemoryAllocateMaxRetries() {
+ return pipeMemoryAllocateMaxRetries;
+ }
+
+ public void setPipeMemoryAllocateMaxRetries(int
pipeMemoryAllocateMaxRetries) {
+ this.pipeMemoryAllocateMaxRetries = pipeMemoryAllocateMaxRetries;
+ }
+
+ public long getPipeMemoryAllocateRetryIntervalInMs() {
+ return pipeMemoryAllocateRetryIntervalMs;
+ }
+
+ public void setPipeMemoryAllocateRetryIntervalInMs(long
pipeMemoryAllocateRetryIntervalMs) {
+ this.pipeMemoryAllocateRetryIntervalMs = pipeMemoryAllocateRetryIntervalMs;
+ }
+
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 047b5a67515..901c67a2da9 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
@@ -389,6 +389,18 @@ public class CommonDescriptor {
properties.getProperty(
"pipe_max_allowed_pending_tsfile_epoch_per_data_region",
String.valueOf(config.getPipeMaxAllowedPendingTsFileEpochPerDataRegion()))));
+
+ 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()))));
}
public void loadGlobalConfig(TGlobalConfig globalConfig) {
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 f44dfb60bac..05517648845 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
@@ -157,6 +157,16 @@ public class PipeConfig {
return COMMON_CONFIG.getPipeMaxAllowedPendingTsFileEpochPerDataRegion();
}
+ /////////////////////////////// Memory ///////////////////////////////
+
+ public int getPipeMemoryAllocateMaxRetries() {
+ return COMMON_CONFIG.getPipeMemoryAllocateMaxRetries();
+ }
+
+ public long getPipeMemoryAllocateRetryIntervalInMs() {
+ return COMMON_CONFIG.getPipeMemoryAllocateRetryIntervalInMs();
+ }
+
/////////////////////////////// Utils ///////////////////////////////
private static final Logger LOGGER =
LoggerFactory.getLogger(PipeConfig.class);
@@ -215,6 +225,10 @@ public class PipeConfig {
LOGGER.info(
"PipeMaxAllowedPendingTsFileEpochPerDataRegion: {}",
getPipeMaxAllowedPendingTsFileEpochPerDataRegion());
+
+ LOGGER.info("PipeMemoryAllocateMaxRetries: {}",
getPipeMemoryAllocateMaxRetries());
+ LOGGER.info(
+ "PipeMemoryAllocateRetryIntervalInMs: {}",
getPipeMemoryAllocateRetryIntervalInMs());
}
/////////////////////////////// Singleton ///////////////////////////////
