You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by sp...@apache.org on 2025/04/08 04:35:03 UTC
(iotdb) branch feature/memory_heap_adapt created (now 96c5bad0345)
This is an automated email from the ASF dual-hosted git repository.
spricoder pushed a change to branch feature/memory_heap_adapt
in repository https://gitbox.apache.org/repos/asf/iotdb.git
at 96c5bad0345 Add heap memory adaption
This branch includes the following new commits:
new 96c5bad0345 Add heap memory adaption
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(iotdb) 01/01: Add heap memory adaption
Posted by sp...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
spricoder pushed a commit to branch feature/memory_heap_adapt
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 96c5bad0345d9e9e182c7814f3a7b2f2bc08011d
Author: spricoder <sp...@qq.com>
AuthorDate: Tue Apr 8 12:34:27 2025 +0800
Add heap memory adaption
---
.../iotdb/consensus/config/IoTConsensusConfig.java | 2 +-
.../logdispatcher/IoTConsensusMemoryManager.java | 2 +-
.../apache/iotdb/db/conf/DataNodeMemoryConfig.java | 23 +++--
.../org/apache/iotdb/db/conf/IoTDBDescriptor.java | 12 +--
.../db/consensus/DataRegionConsensusImpl.java | 2 +-
.../db/pipe/resource/memory/PipeMemoryManager.java | 4 +-
.../analyze/cache/partition/PartitionCache.java | 13 ++-
.../cache/schema/DataNodeDevicePathCache.java | 18 +++-
.../fetcher/cache/TableDeviceSchemaCache.java | 23 ++++-
.../java/org/apache/iotdb/db/service/DataNode.java | 4 +
.../db/storageengine/buffer/BloomFilterCache.java | 29 ++++--
.../iotdb/db/storageengine/buffer/ChunkCache.java | 30 ++++--
.../buffer/TimeSeriesMetadataCache.java | 29 ++++--
.../storageengine/dataregion/DataRegionInfo.java | 1 +
.../apache/iotdb/commons/conf/CommonConfig.java | 20 ++++
.../iotdb/commons/conf/CommonDescriptor.java | 8 ++
.../commons/memory/AtomicLongMemoryBlock.java | 30 +++---
.../apache/iotdb/commons/memory/IMemoryBlock.java | 48 +++++++--
.../apache/iotdb/commons/memory/MemoryManager.java | 76 ++++++++++----
.../memory/MemoryPeriodicalJobExecutor.java | 111 +++++++++++++++++++++
.../commons/memory/MemoryRuntimeController.java | 106 ++++++++++++++++++++
.../apache/iotdb/commons/service/ServiceType.java | 2 +-
22 files changed, 494 insertions(+), 99 deletions(-)
diff --git a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/config/IoTConsensusConfig.java b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/config/IoTConsensusConfig.java
index 0621aee23ef..c01553c1391 100644
--- a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/config/IoTConsensusConfig.java
+++ b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/config/IoTConsensusConfig.java
@@ -359,7 +359,7 @@ public class IoTConsensusConfig {
private long checkpointGap = 500;
private IMemoryBlock consensusMemoryBlock =
new AtomicLongMemoryBlock(
- "Consensus-Default", null, Runtime.getRuntime().maxMemory() / 10);
+ "Consensus-Default", null, Runtime.getRuntime().totalMemory() / 10);
private double maxMemoryRatioForQueue = 0.6;
private long regionMigrationSpeedLimitBytesPerSecond = 32 * 1024 * 1024L;
diff --git a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/logdispatcher/IoTConsensusMemoryManager.java b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/logdispatcher/IoTConsensusMemoryManager.java
index 7bee00588cd..4cb141f192f 100644
--- a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/logdispatcher/IoTConsensusMemoryManager.java
+++ b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/logdispatcher/IoTConsensusMemoryManager.java
@@ -33,7 +33,7 @@ public class IoTConsensusMemoryManager {
private final AtomicLong queueMemorySizeInByte = new AtomicLong(0);
private final AtomicLong syncMemorySizeInByte = new AtomicLong(0);
private IMemoryBlock memoryBlock =
- new AtomicLongMemoryBlock("Consensus-Default", null, Runtime.getRuntime().maxMemory() / 10);
+ new AtomicLongMemoryBlock("Consensus-Default", null, Runtime.getRuntime().totalMemory() / 10);
private Double maxMemoryRatioForQueue = 0.6;
private IoTConsensusMemoryManager() {
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/DataNodeMemoryConfig.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/DataNodeMemoryConfig.java
index 8de17422cc6..8fb4f86d7e6 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/DataNodeMemoryConfig.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/DataNodeMemoryConfig.java
@@ -23,6 +23,7 @@ import org.apache.iotdb.commons.conf.ConfigurationFileUtils;
import org.apache.iotdb.commons.conf.TrimProperties;
import org.apache.iotdb.commons.memory.MemoryConfig;
import org.apache.iotdb.commons.memory.MemoryManager;
+import org.apache.iotdb.db.storageengine.rescon.memory.SystemInfo;
import org.apache.iotdb.db.utils.MemUtils;
import org.slf4j.Logger;
@@ -59,8 +60,9 @@ public class DataNodeMemoryConfig {
private int queryThreadCount = Runtime.getRuntime().availableProcessors();
/** Max bytes of each FragmentInstance for DataExchange */
+ // TODO @spricoder : influence dynamic change of memory size
private long maxBytesPerFragmentInstance =
- Runtime.getRuntime().maxMemory() * 3 / 10 * 200 / 1001 / queryThreadCount;
+ Runtime.getRuntime().totalMemory() * 3 / 10 * 200 / 1001 / queryThreadCount;
/** The memory manager of on heap */
private MemoryManager onHeapMemoryManager;
@@ -151,18 +153,18 @@ public class DataNodeMemoryConfig {
}
}
- long storageEngineMemorySize = Runtime.getRuntime().maxMemory() * 3 / 10;
- long queryEngineMemorySize = Runtime.getRuntime().maxMemory() * 3 / 10;
- long schemaEngineMemorySize = Runtime.getRuntime().maxMemory() / 10;
- long consensusMemorySize = Runtime.getRuntime().maxMemory() / 10;
- long pipeMemorySize = Runtime.getRuntime().maxMemory() / 10;
+ long storageEngineMemorySize = Runtime.getRuntime().totalMemory() * 3 / 10;
+ long queryEngineMemorySize = Runtime.getRuntime().totalMemory() * 3 / 10;
+ long schemaEngineMemorySize = Runtime.getRuntime().totalMemory() / 10;
+ long consensusMemorySize = Runtime.getRuntime().totalMemory() / 10;
+ long pipeMemorySize = Runtime.getRuntime().totalMemory() / 10;
if (memoryAllocateProportion != null) {
String[] proportions = memoryAllocateProportion.split(":");
int proportionSum = 0;
for (String proportion : proportions) {
proportionSum += Integer.parseInt(proportion.trim());
}
- long maxMemoryAvailable = Runtime.getRuntime().maxMemory();
+ long maxMemoryAvailable = Runtime.getRuntime().totalMemory();
if (proportionSum != 0) {
storageEngineMemorySize =
@@ -189,9 +191,12 @@ public class DataNodeMemoryConfig {
}
}
onHeapMemoryManager =
- MemoryConfig.global().getOrCreateMemoryManager("OnHeap", Runtime.getRuntime().maxMemory());
+ MemoryConfig.global()
+ .getOrCreateMemoryManager("OnHeap", Runtime.getRuntime().totalMemory());
storageEngineMemoryManager =
- onHeapMemoryManager.getOrCreateMemoryManager("StorageEngine", storageEngineMemorySize);
+ onHeapMemoryManager
+ .getOrCreateMemoryManager("StorageEngine", storageEngineMemorySize)
+ .setMemoryUpdateCallback((before, after) -> SystemInfo.getInstance().loadWriteMemory());
queryEngineMemoryManager =
onHeapMemoryManager.getOrCreateMemoryManager("QueryEngine", queryEngineMemorySize);
schemaEngineMemoryManager =
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 c2d6f90e5ec..cf95e2582c4 100755
--- 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
@@ -50,7 +50,6 @@ import org.apache.iotdb.db.storageengine.dataregion.wal.WALManager;
import org.apache.iotdb.db.storageengine.dataregion.wal.utils.WALMode;
import org.apache.iotdb.db.storageengine.load.disk.ILoadDiskSelector;
import org.apache.iotdb.db.storageengine.rescon.disk.TierManager;
-import org.apache.iotdb.db.storageengine.rescon.memory.SystemInfo;
import org.apache.iotdb.db.utils.DateTimeUtils;
import org.apache.iotdb.db.utils.datastructure.TVListSortAlgorithm;
import org.apache.iotdb.external.api.IPropertiesLoader;
@@ -2333,6 +2332,7 @@ public class IoTDBDescriptor {
for (String proportion : proportions) {
proportionSum += Integer.parseInt(proportion.trim());
}
+ // TODO @spricoder: consider whether this part need change when total memory is dynamic
float maxMemoryAvailable = conf.getUdfMemoryBudgetInMB();
try {
conf.setUdfReaderMemoryBudgetInMB(
@@ -2630,12 +2630,12 @@ public class IoTDBDescriptor {
// first we need to release the memory allocated for consensus
MemoryManager storageEngineMemoryManager = memoryConfig.getStorageEngineMemoryManager();
MemoryManager consensusMemoryManager = memoryConfig.getConsensusMemoryManager();
+ long originSize = storageEngineMemoryManager.getInitialAllocatedMemorySizeInBytes();
long newSize =
- storageEngineMemoryManager.getTotalMemorySizeInBytes()
- + consensusMemoryManager.getTotalMemorySizeInBytes();
- consensusMemoryManager.setTotalMemorySizeInBytes(0);
- storageEngineMemoryManager.setTotalMemorySizeInBytesWithReload(newSize);
- SystemInfo.getInstance().loadWriteMemory();
+ storageEngineMemoryManager.getInitialAllocatedMemorySizeInBytes()
+ + consensusMemoryManager.getInitialAllocatedMemorySizeInBytes();
+ storageEngineMemoryManager.resizeByRatio((double) newSize / originSize);
+ consensusMemoryManager.resizeByRatio(0);
}
private static class IoTDBDescriptorHolder {
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/consensus/DataRegionConsensusImpl.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/consensus/DataRegionConsensusImpl.java
index 38ce6a0f158..22eed36d46c 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/consensus/DataRegionConsensusImpl.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/consensus/DataRegionConsensusImpl.java
@@ -118,7 +118,7 @@ public class DataRegionConsensusImpl {
IMemoryBlock memoryBlock =
MEMORY_CONFIG
.getConsensusMemoryManager()
- .exactAllocate("Consensus", MemoryBlockType.DYNAMIC);
+ .exactAllocate("Consensus", MemoryBlockType.STATIC);
return ConsensusConfig.newBuilder()
.setThisNodeId(CONF.getDataNodeId())
.setThisNode(new TEndPoint(CONF.getInternalAddress(), CONF.getDataRegionConsensusPort()))
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 0c7a473950d..8a2e370218b 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
@@ -52,11 +52,11 @@ public class PipeMemoryManager {
PipeConfig.getInstance().getPipeMemoryAllocateMinSizeInBytes();
// TODO @spricoder: consider combine memory block and used MemorySizeInBytes
- private IMemoryBlock memoryBlock =
+ private final IMemoryBlock memoryBlock =
IoTDBDescriptor.getInstance()
.getMemoryConfig()
.getPipeMemoryManager()
- .exactAllocate("Stream", MemoryBlockType.DYNAMIC);
+ .exactAllocate("Stream", MemoryBlockType.STATIC);
private static final double EXCEED_PROTECT_THRESHOLD = 0.95;
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/cache/partition/PartitionCache.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/cache/partition/PartitionCache.java
index 15828538c7f..e45ab7a5b3b 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/cache/partition/PartitionCache.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/cache/partition/PartitionCache.java
@@ -62,6 +62,7 @@ import org.apache.iotdb.rpc.TSStatusCode;
import com.github.benmanes.caffeine.cache.Cache;
import com.github.benmanes.caffeine.cache.Caffeine;
+import com.google.common.util.concurrent.AtomicDouble;
import org.apache.thrift.TException;
import org.apache.tsfile.file.metadata.IDeviceID;
import org.slf4j.Logger;
@@ -120,12 +121,22 @@ public class PartitionCache {
private final CacheMetrics cacheMetrics;
private final IMemoryBlock memoryBlock;
+ private final AtomicDouble memoryUsageCheatFactor = new AtomicDouble(1);
public PartitionCache() {
this.memoryBlock =
memoryConfig
.getPartitionCacheMemoryManager()
- .exactAllocate("PartitionCache", MemoryBlockType.STATIC);
+ .exactAllocate("PartitionCache", MemoryBlockType.STATIC)
+ .setMemoryUpdateCallback(
+ (oldMemory, newMemory) -> {
+ memoryUsageCheatFactor.updateAndGet(
+ factor -> factor / ((double) newMemory / oldMemory));
+ logger.debug(
+ "[MemoryUsageCheatFactor] PartitionCache has updated from {} to {}.",
+ oldMemory,
+ newMemory);
+ });
this.memoryBlock.allocate(this.memoryBlock.getTotalMemorySizeInBytes());
// TODO @spricoder: PartitionCache need to be controlled according to memory
this.schemaPartitionCache =
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/cache/schema/DataNodeDevicePathCache.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/cache/schema/DataNodeDevicePathCache.java
index 07293bc4ca3..480aee2b179 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/cache/schema/DataNodeDevicePathCache.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/cache/schema/DataNodeDevicePathCache.java
@@ -29,6 +29,7 @@ import org.apache.iotdb.db.conf.IoTDBDescriptor;
import com.github.benmanes.caffeine.cache.Cache;
import com.github.benmanes.caffeine.cache.Caffeine;
import com.github.benmanes.caffeine.cache.Weigher;
+import com.google.common.util.concurrent.AtomicDouble;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -39,6 +40,7 @@ public class DataNodeDevicePathCache {
private static final DataNodeMemoryConfig memoryConfig =
IoTDBDescriptor.getInstance().getMemoryConfig();
private final IMemoryBlock devicePathCacheMemoryBlock;
+ private final AtomicDouble memoryUsageCheatFactor = new AtomicDouble(1);
private final Cache<String, PartialPath> devicePathCache;
@@ -46,14 +48,26 @@ public class DataNodeDevicePathCache {
devicePathCacheMemoryBlock =
memoryConfig
.getDevicePathCacheMemoryManager()
- .exactAllocate("DevicePathCache", MemoryBlockType.STATIC);
+ .exactAllocate("DevicePathCache", MemoryBlockType.STATIC)
+ .setMemoryUpdateCallback(
+ (oldMemory, newMemory) -> {
+ memoryUsageCheatFactor.updateAndGet(
+ factor -> factor / ((double) newMemory / oldMemory));
+ LOGGER.debug(
+ "[MemoryUsageCheatFactor]DataNodeDevicePathCache has updated from {} to {}.",
+ oldMemory,
+ newMemory);
+ });
+ ;
// TODO @spricoder: later we can find a way to get the byte size of cache
devicePathCacheMemoryBlock.allocate(devicePathCacheMemoryBlock.getTotalMemorySizeInBytes());
devicePathCache =
Caffeine.newBuilder()
.maximumWeight(devicePathCacheMemoryBlock.getTotalMemorySizeInBytes())
.weigher(
- (Weigher<String, PartialPath>) (key, val) -> (PartialPath.estimateSize(val) + 32))
+ (Weigher<String, PartialPath>)
+ (key, val) ->
+ (int) ((PartialPath.estimateSize(val) + 32) * memoryUsageCheatFactor.get()))
.build();
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/cache/TableDeviceSchemaCache.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/cache/TableDeviceSchemaCache.java
index a724bc427c8..7ef79e8272f 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/cache/TableDeviceSchemaCache.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/cache/TableDeviceSchemaCache.java
@@ -38,6 +38,7 @@ import org.apache.iotdb.db.queryengine.plan.relational.metadata.QualifiedObjectN
import org.apache.iotdb.db.schemaengine.schemaregion.SchemaRegion;
import org.apache.iotdb.db.schemaengine.table.DataNodeTableCache;
+import com.google.common.util.concurrent.AtomicDouble;
import org.apache.tsfile.file.metadata.IDeviceID;
import org.apache.tsfile.file.metadata.StringArrayDeviceID;
import org.apache.tsfile.read.TimeValuePair;
@@ -106,20 +107,34 @@ public class TableDeviceSchemaCache {
private final ReentrantReadWriteLock readWriteLock = new ReentrantReadWriteLock(false);
private final IMemoryBlock memoryBlock;
+ private final AtomicDouble memoryUsageCheatFactor = new AtomicDouble(1);
private TableDeviceSchemaCache() {
memoryBlock =
memoryConfig
.getSchemaCacheMemoryManager()
- .exactAllocate("TableDeviceSchemaCache", MemoryBlockType.STATIC);
+ .exactAllocate("TableDeviceSchemaCache", MemoryBlockType.STATIC)
+ .setMemoryUpdateCallback(
+ (oldMemory, newMemory) -> {
+ memoryUsageCheatFactor.updateAndGet(
+ factor -> factor / ((double) newMemory / oldMemory));
+ logger.debug(
+ "[MemoryUsageCheatFactor]TableDeviceSchemaCache has updated from {} to {}.",
+ oldMemory,
+ newMemory);
+ });
dualKeyCache =
new DualKeyCacheBuilder<TableId, IDeviceID, TableDeviceCacheEntry>()
.cacheEvictionPolicy(
DualKeyCachePolicy.valueOf(config.getDataNodeSchemaCacheEvictionPolicy()))
.memoryCapacity(memoryBlock.getTotalMemorySizeInBytes())
- .firstKeySizeComputer(TableId::estimateSize)
- .secondKeySizeComputer(deviceID -> (int) deviceID.ramBytesUsed())
- .valueSizeComputer(TableDeviceCacheEntry::estimateSize)
+ .firstKeySizeComputer(
+ tableId -> (int) (tableId.estimateSize() * memoryUsageCheatFactor.get()))
+ .secondKeySizeComputer(
+ deviceID -> (int) (deviceID.ramBytesUsed() * memoryUsageCheatFactor.get()))
+ .valueSizeComputer(
+ tableDeviceCacheEntry ->
+ (int) (tableDeviceCacheEntry.estimateSize() * memoryUsageCheatFactor.get()))
.build();
memoryBlock.allocate(memoryBlock.getTotalMemorySizeInBytes());
MetricService.getInstance().addMetricSet(new TableDeviceSchemaCacheMetrics(this));
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/DataNode.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/DataNode.java
index b554eb92419..24c857eba8d 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/DataNode.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/DataNode.java
@@ -38,6 +38,7 @@ import org.apache.iotdb.commons.consensus.SchemaRegionId;
import org.apache.iotdb.commons.exception.IllegalPathException;
import org.apache.iotdb.commons.exception.IoTDBException;
import org.apache.iotdb.commons.exception.StartupException;
+import org.apache.iotdb.commons.memory.MemoryRuntimeController;
import org.apache.iotdb.commons.pipe.agent.plugin.meta.PipePluginMeta;
import org.apache.iotdb.commons.pipe.config.PipeConfig;
import org.apache.iotdb.commons.service.JMXService;
@@ -245,6 +246,9 @@ public class DataNode extends ServerCommandLine implements DataNodeMBean {
// Serialize mutable system properties
IoTDBStartCheck.getInstance().serializeMutableSystemPropertiesIfNecessary();
+ // Setup memory controller
+ registerManager.register(MemoryRuntimeController.getInstance());
+
logger.info("IoTDB configuration: {}", config.getConfigMessage());
logger.info("Congratulations, IoTDB DataNode is set up successfully. Now, enjoy yourself!");
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/buffer/BloomFilterCache.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/buffer/BloomFilterCache.java
index 7a41352098b..f3f58a24414 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/buffer/BloomFilterCache.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/buffer/BloomFilterCache.java
@@ -31,6 +31,7 @@ import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileID;
import com.github.benmanes.caffeine.cache.Cache;
import com.github.benmanes.caffeine.cache.Caffeine;
import com.github.benmanes.caffeine.cache.Weigher;
+import com.google.common.util.concurrent.AtomicDouble;
import org.apache.tsfile.read.TsFileSequenceReader;
import org.apache.tsfile.utils.BloomFilter;
import org.apache.tsfile.utils.RamUsageEstimator;
@@ -51,32 +52,44 @@ public class BloomFilterCache {
private static final Logger DEBUG_LOGGER = LoggerFactory.getLogger("QUERY_DEBUG");
private static final DataNodeMemoryConfig MEMORY_CONFIG =
IoTDBDescriptor.getInstance().getMemoryConfig();
- private static final IMemoryBlock CACHE_MEMORY_BLOCK;
+ private static final IMemoryBlock cacheMemoryBlock;
+ private static final AtomicDouble memoryUsageCheatFactor = new AtomicDouble(1);
private static final boolean CACHE_ENABLE = MEMORY_CONFIG.isMetaDataCacheEnable();
private final AtomicLong entryAverageSize = new AtomicLong(0);
private final Cache<BloomFilterCacheKey, BloomFilter> lruCache;
static {
- CACHE_MEMORY_BLOCK =
+ cacheMemoryBlock =
MEMORY_CONFIG
.getBloomFilterCacheMemoryManager()
- .exactAllocate("BloomFilterCache", MemoryBlockType.STATIC);
+ .exactAllocate("BloomFilterCache", MemoryBlockType.STATIC)
+ .setMemoryUpdateCallback(
+ (oldMemory, newMemory) -> {
+ memoryUsageCheatFactor.updateAndGet(
+ factor -> factor / ((double) newMemory / oldMemory));
+ LOGGER.debug(
+ "[MemoryUsageCheatFactor]BloomFilterCache has updated from {} to {}.",
+ oldMemory,
+ newMemory);
+ });
// TODO @spricoder: find a way to get the size of the BloomFilterCache
- CACHE_MEMORY_BLOCK.allocate(CACHE_MEMORY_BLOCK.getTotalMemorySizeInBytes());
+ cacheMemoryBlock.allocate(cacheMemoryBlock.getTotalMemorySizeInBytes());
}
private BloomFilterCache() {
if (CACHE_ENABLE) {
- LOGGER.info("BloomFilterCache size = {}", CACHE_MEMORY_BLOCK.getTotalMemorySizeInBytes());
+ LOGGER.info("BloomFilterCache size = {}", cacheMemoryBlock.getTotalMemorySizeInBytes());
}
lruCache =
Caffeine.newBuilder()
- .maximumWeight(CACHE_MEMORY_BLOCK.getTotalMemorySizeInBytes())
+ .maximumWeight(cacheMemoryBlock.getTotalMemorySizeInBytes())
.weigher(
(Weigher<BloomFilterCacheKey, BloomFilter>)
(key, bloomFilter) ->
- (int) (key.getRetainedSizeInBytes() + bloomFilter.getRetainedSizeInBytes()))
+ (int)
+ ((key.getRetainedSizeInBytes() + bloomFilter.getRetainedSizeInBytes())
+ * memoryUsageCheatFactor.get()))
.recordStats()
.build();
}
@@ -131,7 +144,7 @@ public class BloomFilterCache {
}
public long getMaxMemory() {
- return CACHE_MEMORY_BLOCK.getTotalMemorySizeInBytes();
+ return cacheMemoryBlock.getTotalMemorySizeInBytes();
}
public double getAverageLoadPenalty() {
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/buffer/ChunkCache.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/buffer/ChunkCache.java
index 1c39b993a3c..f84a1da7fe7 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/buffer/ChunkCache.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/buffer/ChunkCache.java
@@ -35,6 +35,7 @@ import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileID;
import com.github.benmanes.caffeine.cache.Cache;
import com.github.benmanes.caffeine.cache.Caffeine;
import com.github.benmanes.caffeine.cache.Weigher;
+import com.google.common.util.concurrent.AtomicDouble;
import org.apache.tsfile.file.metadata.statistics.Statistics;
import org.apache.tsfile.read.TsFileSequenceReader;
import org.apache.tsfile.read.common.Chunk;
@@ -63,7 +64,8 @@ public class ChunkCache {
private static final Logger DEBUG_LOGGER = LoggerFactory.getLogger("QUERY_DEBUG");
private static final DataNodeMemoryConfig MEMORY_CONFIG =
IoTDBDescriptor.getInstance().getMemoryConfig();
- private static final IMemoryBlock CACHE_MEMORY_BLOCK;
+ private static final IMemoryBlock cacheMemoryBlock;
+ private static final AtomicDouble memoryUsageCheatFactor = new AtomicDouble(1);
private static final boolean CACHE_ENABLE = MEMORY_CONFIG.isMetaDataCacheEnable();
private static final SeriesScanCostMetricSet SERIES_SCAN_COST_METRIC_SET =
@@ -73,25 +75,37 @@ public class ChunkCache {
private final Cache<ChunkCacheKey, Chunk> lruCache;
static {
- CACHE_MEMORY_BLOCK =
+ cacheMemoryBlock =
MEMORY_CONFIG
.getChunkCacheMemoryManager()
- .exactAllocate("ChunkCache", MemoryBlockType.STATIC);
+ .exactAllocate("ChunkCache", MemoryBlockType.STATIC)
+ .setMemoryUpdateCallback(
+ (oldMemory, newMemory) -> {
+ memoryUsageCheatFactor.updateAndGet(
+ factor -> factor / ((double) newMemory / oldMemory));
+ LOGGER.info(
+ "[MemoryUsageCheatFactor]ChunkCache has updated from {} to {}.",
+ oldMemory,
+ newMemory);
+ });
+ ;
// TODO @spricoder: find a way to get the size of the ChunkCache
- CACHE_MEMORY_BLOCK.allocate(CACHE_MEMORY_BLOCK.getTotalMemorySizeInBytes());
+ cacheMemoryBlock.allocate(cacheMemoryBlock.getTotalMemorySizeInBytes());
}
private ChunkCache() {
if (CACHE_ENABLE) {
- LOGGER.info("ChunkCache size = {}", CACHE_MEMORY_BLOCK.getTotalMemorySizeInBytes());
+ LOGGER.info("ChunkCache size = {}", cacheMemoryBlock.getTotalMemorySizeInBytes());
}
lruCache =
Caffeine.newBuilder()
- .maximumWeight(CACHE_MEMORY_BLOCK.getTotalMemorySizeInBytes())
+ .maximumWeight(cacheMemoryBlock.getTotalMemorySizeInBytes())
.weigher(
(Weigher<ChunkCacheKey, Chunk>)
(key, chunk) ->
- (int) (key.getRetainedSizeInBytes() + chunk.getRetainedSizeInBytes()))
+ (int)
+ ((key.getRetainedSizeInBytes() + chunk.getRetainedSizeInBytes())
+ * memoryUsageCheatFactor.get()))
.recordStats()
.build();
@@ -202,7 +216,7 @@ public class ChunkCache {
}
public long getMaxMemory() {
- return CACHE_MEMORY_BLOCK.getTotalMemorySizeInBytes();
+ return cacheMemoryBlock.getTotalMemorySizeInBytes();
}
public double getAverageLoadPenalty() {
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/buffer/TimeSeriesMetadataCache.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/buffer/TimeSeriesMetadataCache.java
index f2cd55c5e7d..822f18eaec1 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/buffer/TimeSeriesMetadataCache.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/buffer/TimeSeriesMetadataCache.java
@@ -34,6 +34,7 @@ import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileID;
import com.github.benmanes.caffeine.cache.Cache;
import com.github.benmanes.caffeine.cache.Caffeine;
import com.github.benmanes.caffeine.cache.Weigher;
+import com.google.common.util.concurrent.AtomicDouble;
import org.apache.tsfile.common.constant.TsFileConstant;
import org.apache.tsfile.file.metadata.IDeviceID;
import org.apache.tsfile.file.metadata.TimeseriesMetadata;
@@ -67,7 +68,8 @@ public class TimeSeriesMetadataCache {
private static final Logger DEBUG_LOGGER = LoggerFactory.getLogger("QUERY_DEBUG");
private static final DataNodeMemoryConfig memoryConfig =
IoTDBDescriptor.getInstance().getMemoryConfig();
- private static final IMemoryBlock CACHE_MEMORY_BLOCK;
+ private static final IMemoryBlock cacheMemoryBlock;
+ private static final AtomicDouble memoryUsageCheatFactor = new AtomicDouble(1);
private static final boolean CACHE_ENABLE = memoryConfig.isMetaDataCacheEnable();
private final Cache<TimeSeriesMetadataCacheKey, TimeseriesMetadata> lruCache;
@@ -79,26 +81,37 @@ public class TimeSeriesMetadataCache {
private static final String SEPARATOR = "$";
static {
- CACHE_MEMORY_BLOCK =
+ cacheMemoryBlock =
memoryConfig
.getTimeSeriesMetaDataCacheMemoryManager()
- .exactAllocate("TimeSeriesMetadataCache", MemoryBlockType.STATIC);
+ .exactAllocate("TimeSeriesMetadataCache", MemoryBlockType.STATIC)
+ .setMemoryUpdateCallback(
+ (oldMemory, newMemory) -> {
+ memoryUsageCheatFactor.updateAndGet(
+ factor -> factor / ((double) newMemory / oldMemory));
+ logger.info(
+ "[MemoryUsageCheatFactor]TimeSeriesMetadataCache has updated from {} to {}.",
+ oldMemory,
+ newMemory);
+ });
// TODO @spricoder find a better way to get the size of cache
- CACHE_MEMORY_BLOCK.allocate(CACHE_MEMORY_BLOCK.getTotalMemorySizeInBytes());
+ cacheMemoryBlock.allocate(cacheMemoryBlock.getTotalMemorySizeInBytes());
}
private TimeSeriesMetadataCache() {
if (CACHE_ENABLE) {
logger.info(
- "TimeSeriesMetadataCache size = {}", CACHE_MEMORY_BLOCK.getTotalMemorySizeInBytes());
+ "TimeSeriesMetadataCache size = {}", cacheMemoryBlock.getTotalMemorySizeInBytes());
}
lruCache =
Caffeine.newBuilder()
- .maximumWeight(CACHE_MEMORY_BLOCK.getTotalMemorySizeInBytes())
+ .maximumWeight(cacheMemoryBlock.getTotalMemorySizeInBytes())
.weigher(
(Weigher<TimeSeriesMetadataCacheKey, TimeseriesMetadata>)
(key, value) ->
- (int) (key.getRetainedSizeInBytes() + value.getRetainedSizeInBytes()))
+ (int)
+ ((key.getRetainedSizeInBytes() + value.getRetainedSizeInBytes())
+ * memoryUsageCheatFactor.get()))
.recordStats()
.build();
// add metrics
@@ -265,7 +278,7 @@ public class TimeSeriesMetadataCache {
}
public long getMaxMemory() {
- return CACHE_MEMORY_BLOCK.getTotalMemorySizeInBytes();
+ return cacheMemoryBlock.getTotalMemorySizeInBytes();
}
public double getAverageLoadPenalty() {
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegionInfo.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegionInfo.java
index 0e8e4902c8b..54e7dfda8b6 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegionInfo.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegionInfo.java
@@ -38,6 +38,7 @@ public class DataRegionInfo {
*/
private final AtomicLong memoryCost;
+ // TODO @spricoder dynamic threshold
/** The threshold of reporting it's size to SystemInfo */
private final long storageGroupSizeReportThreshold =
(long)
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 ef289d75c19..913f6fb780e 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
@@ -385,6 +385,10 @@ public class CommonConfig {
private volatile Pattern trustedUriPattern = Pattern.compile("file:.*");
+ // Whtether to enable memory adaption
+ private boolean enableMemoryAdapt = false;
+ private long memoryAdaptIntervalInS = 20;
+
CommonConfig() {
// Empty constructor
}
@@ -1821,4 +1825,20 @@ public class CommonConfig {
public void setTrustedUriPattern(Pattern trustedUriPattern) {
this.trustedUriPattern = trustedUriPattern;
}
+
+ public long getMemoryAdaptIntervalInS() {
+ return memoryAdaptIntervalInS;
+ }
+
+ public void setMemoryAdaptIntervalInS(long memoryAdaptIntervalInS) {
+ this.memoryAdaptIntervalInS = memoryAdaptIntervalInS;
+ }
+
+ public boolean isEnableMemoryAdapt() {
+ return enableMemoryAdapt;
+ }
+
+ public void setEnableMemoryAdapt(boolean enableMemoryAdapt) {
+ this.enableMemoryAdapt = enableMemoryAdapt;
+ }
}
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 d57c129d0a6..b21e9fc5ac4 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
@@ -251,6 +251,14 @@ public class CommonDescriptor {
properties.getProperty(
"cluster_device_limit_threshold",
String.valueOf(config.getDeviceLimitThreshold()))));
+ config.setEnableMemoryAdapt(
+ Boolean.parseBoolean(
+ properties.getProperty(
+ "enable_memory_adapt", Boolean.toString(config.isEnableMemoryAdapt()))));
+ config.setMemoryAdaptIntervalInS(
+ Integer.parseInt(
+ properties.getProperty(
+ "memory_check_interval", String.valueOf(config.getMemoryAdaptIntervalInS()))));
loadRetryProperties(properties);
loadBinaryAllocatorProps(properties);
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/memory/AtomicLongMemoryBlock.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/memory/AtomicLongMemoryBlock.java
index 1376d8ddb6d..e654213ea75 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/memory/AtomicLongMemoryBlock.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/memory/AtomicLongMemoryBlock.java
@@ -52,45 +52,45 @@ public class AtomicLongMemoryBlock extends IMemoryBlock {
}
@Override
- public long forceAllocateWithoutLimitation(long sizeInByte) {
- return usedMemoryInBytes.addAndGet(sizeInByte);
+ public long forceAllocateWithoutLimitation(long sizeInBytes) {
+ return usedMemoryInBytes.addAndGet(sizeInBytes);
}
@Override
- public boolean allocate(long sizeInByte) {
+ public boolean allocate(long sizeInBytes) {
AtomicBoolean result = new AtomicBoolean(false);
usedMemoryInBytes.updateAndGet(
memCost -> {
- if (memCost + sizeInByte > totalMemorySizeInBytes) {
+ if (memCost + sizeInBytes > totalMemorySizeInBytes) {
return memCost;
}
result.set(true);
- return memCost + sizeInByte;
+ return memCost + sizeInBytes;
});
return result.get();
}
@Override
- public boolean allocateIfSufficient(final long sizeInByte, final double maxRatio) {
+ public boolean allocateIfSufficient(final long sizeInBytes, final double maxRatio) {
AtomicBoolean result = new AtomicBoolean(false);
usedMemoryInBytes.updateAndGet(
memCost -> {
- if (memCost + sizeInByte > totalMemorySizeInBytes * maxRatio) {
+ if (memCost + sizeInBytes > totalMemorySizeInBytes * maxRatio) {
return memCost;
}
result.set(true);
- return memCost + sizeInByte;
+ return memCost + sizeInBytes;
});
return result.get();
}
@Override
- public boolean allocateUntilAvailable(long sizeInByte, long retryIntervalInMillis)
+ public boolean allocateUntilAvailable(long sizeInBytes, long retryIntervalInMillis)
throws InterruptedException {
long originSize = usedMemoryInBytes.get();
while (true) {
- boolean canUpdate = originSize + sizeInByte <= totalMemorySizeInBytes;
- if (canUpdate && usedMemoryInBytes.compareAndSet(originSize, originSize + sizeInByte)) {
+ boolean canUpdate = originSize + sizeInBytes <= totalMemorySizeInBytes;
+ if (canUpdate && usedMemoryInBytes.compareAndSet(originSize, originSize + sizeInBytes)) {
break;
}
Thread.sleep(TimeUnit.MILLISECONDS.toMillis(retryIntervalInMillis));
@@ -100,16 +100,16 @@ public class AtomicLongMemoryBlock extends IMemoryBlock {
}
@Override
- public long release(long sizeInByte) {
+ public long release(long sizeInBytes) {
return usedMemoryInBytes.updateAndGet(
memCost -> {
- if (sizeInByte > memCost) {
+ if (sizeInBytes > memCost) {
LOGGER.warn(
"The memory cost to be released is larger than the memory cost of memory block {}",
this);
return 0;
}
- return memCost - sizeInByte;
+ return memCost - sizeInBytes;
});
}
@@ -118,10 +118,12 @@ public class AtomicLongMemoryBlock extends IMemoryBlock {
this.usedMemoryInBytes.set(usedMemoryInBytes);
}
+ @Override
public long getUsedMemoryInBytes() {
return usedMemoryInBytes.get();
}
+ @Override
/** Get the free memory in byte of this memory block */
public long getFreeMemoryInBytes() {
return totalMemorySizeInBytes - usedMemoryInBytes.get();
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/memory/IMemoryBlock.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/memory/IMemoryBlock.java
index 416516bf2c7..5d3bf164dfc 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/memory/IMemoryBlock.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/memory/IMemoryBlock.java
@@ -22,6 +22,9 @@ package org.apache.iotdb.commons.memory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.BiConsumer;
+
public abstract class IMemoryBlock implements AutoCloseable {
private static final Logger LOGGER = LoggerFactory.getLogger(IMemoryBlock.class);
@@ -40,46 +43,50 @@ public abstract class IMemoryBlock implements AutoCloseable {
/** The total memory size in byte of this memory block */
protected long totalMemorySizeInBytes;
+ /** Update related parameters when totalMemorySizeInBytes update */
+ protected final AtomicReference<BiConsumer<Long, Long>> memoryUpdateCallback =
+ new AtomicReference<>();
+
/**
* Forcibly allocate memory without the limit of totalMemorySizeInBytes
*
- * @param sizeInByte the size of memory to be allocated, should be positive
+ * @param sizeInBytes the size of memory to be allocated, should be positive
* @return the number of bytes actually allocated
*/
- public abstract long forceAllocateWithoutLimitation(final long sizeInByte);
+ public abstract long forceAllocateWithoutLimitation(final long sizeInBytes);
/**
* Allocate memory managed by this memory block
*
- * @param sizeInByte the size of memory to be allocated, should be positive
+ * @param sizeInBytes the size of memory to be allocated, should be positive
*/
- public abstract boolean allocate(final long sizeInByte);
+ public abstract boolean allocate(final long sizeInBytes);
/**
* Allocate memory managed by this memory block. if the currently used ratio is already above
- * maxRatio, the allocation will fail".
+ * maxRatio, the allocation will fail.
*
- * @param sizeInByte the size of memory to be allocated, should be positive
+ * @param sizeInBytes the size of memory to be allocated, should be positive
* @param maxRatio the maximum ratio of memory can be allocated
*/
- public abstract boolean allocateIfSufficient(final long sizeInByte, final double maxRatio);
+ public abstract boolean allocateIfSufficient(final long sizeInBytes, final double maxRatio);
/**
* Allocate memory managed by this memory block until the required memory is available
*
- * @param sizeInByte the size of memory to be allocated, should be positive
+ * @param sizeInBytes the size of memory to be allocated, should be positive
* @param retryIntervalInMillis the time interval to wait after each allocation failure
*/
- public abstract boolean allocateUntilAvailable(final long sizeInByte, long retryIntervalInMillis)
+ public abstract boolean allocateUntilAvailable(final long sizeInBytes, long retryIntervalInMillis)
throws InterruptedException;
/**
* Try to release memory managed by this memory block
*
- * @param sizeInByte the size of memory to be released, should be positive
+ * @param sizeInBytes the size of memory to be released, should be positive
* @return the used size after release, zero if the release fails
*/
- public abstract long release(final long sizeInByte);
+ public abstract long release(final long sizeInBytes);
/**
* Try to set memory usage in byte of this memory block (for test only)
@@ -99,6 +106,25 @@ public abstract class IMemoryBlock implements AutoCloseable {
return name;
}
+ /** Update total memory size by ratio */
+ public long resizeByRatio(double ratio) {
+ long before = this.totalMemorySizeInBytes;
+ this.totalMemorySizeInBytes = (long) (before * ratio);
+ if (memoryUpdateCallback.get() != null) {
+ try {
+ memoryUpdateCallback.get().accept(before, totalMemorySizeInBytes);
+ } catch (Exception e) {
+ LOGGER.warn("Failed to execute the update callback.", e);
+ }
+ }
+ return this.totalMemorySizeInBytes - before;
+ }
+
+ public IMemoryBlock setMemoryUpdateCallback(BiConsumer<Long, Long> memoryUpdateCallback) {
+ this.memoryUpdateCallback.set(memoryUpdateCallback);
+ return this;
+ }
+
/** Update maximum memory size in byte of this memory block */
public void setTotalMemorySizeInBytes(final long totalMemorySizeInBytes) {
this.totalMemorySizeInBytes = totalMemorySizeInBytes;
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/memory/MemoryManager.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/memory/MemoryManager.java
index 03fb10d393a..072a45daac6 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/memory/MemoryManager.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/memory/MemoryManager.java
@@ -26,6 +26,8 @@ import org.slf4j.LoggerFactory;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.BiConsumer;
import java.util.function.LongUnaryOperator;
public class MemoryManager {
@@ -46,10 +48,13 @@ public class MemoryManager {
/** Whether memory management is enabled */
private final boolean enabled;
- /** The total memory size in byte of memory manager */
+ /** The memory size of this memory manager allocated by parent memory manager */
+ private volatile long initialAllocatedMemorySizeInBytes = 0L;
+
+ /** The max memory size of this memory manager */
private volatile long totalMemorySizeInBytes;
- /** The allocated memory size */
+ /** The allocated memory size of this memory manager */
private volatile long allocatedMemorySizeInBytes = 0L;
/** The parent memory manager */
@@ -61,10 +66,15 @@ public class MemoryManager {
/** The allocated memory blocks of this memory manager */
private final Map<String, IMemoryBlock> allocatedMemoryBlocks = new ConcurrentHashMap<>();
+ /** Update related parameters when totalMemorySizeInBytes update */
+ private final AtomicReference<BiConsumer<Long, Long>> memoryUpdateCallback =
+ new AtomicReference<>();
+
@TestOnly
public MemoryManager(long totalMemorySizeInBytes) {
this.name = "Test";
this.parentMemoryManager = null;
+ this.initialAllocatedMemorySizeInBytes = totalMemorySizeInBytes;
this.totalMemorySizeInBytes = totalMemorySizeInBytes;
this.enabled = false;
}
@@ -72,6 +82,7 @@ public class MemoryManager {
MemoryManager(String name, MemoryManager parentMemoryManager, long totalMemorySizeInBytes) {
this.name = name;
this.parentMemoryManager = parentMemoryManager;
+ this.initialAllocatedMemorySizeInBytes = totalMemorySizeInBytes;
this.totalMemorySizeInBytes = totalMemorySizeInBytes;
this.enabled = false;
}
@@ -83,6 +94,7 @@ public class MemoryManager {
boolean enabled) {
this.name = name;
this.parentMemoryManager = parentMemoryManager;
+ this.initialAllocatedMemorySizeInBytes = totalMemorySizeInBytes;
this.totalMemorySizeInBytes = totalMemorySizeInBytes;
this.enabled = enabled;
}
@@ -356,24 +368,6 @@ public class MemoryManager {
return getOrCreateMemoryManager(name, totalMemorySizeInBytes, false);
}
- /**
- * Re-allocate memory according to ratio
- *
- * @param ratio the ratio of new total memory size to old total memory size
- */
- private void reAllocateMemoryAccordingToRatio(double ratio) {
- // first increase the total memory size of this memory manager
- this.totalMemorySizeInBytes *= ratio;
- // then re-allocate memory for all memory blocks
- for (IMemoryBlock block : allocatedMemoryBlocks.values()) {
- block.setTotalMemorySizeInBytes((long) (block.getTotalMemorySizeInBytes() * ratio));
- }
- // finally re-allocate memory for all child memory managers
- for (Map.Entry<String, MemoryManager> entry : children.entrySet()) {
- entry.getValue().reAllocateMemoryAccordingToRatio(ratio);
- }
- }
-
/**
* Get the memory manager with specified names in levels
*
@@ -458,12 +452,13 @@ public class MemoryManager {
return totalMemorySizeInBytes;
}
+ @TestOnly
public void setTotalMemorySizeInBytes(long totalMemorySizeInBytes) {
this.totalMemorySizeInBytes = totalMemorySizeInBytes;
}
- public void setTotalMemorySizeInBytesWithReload(long totalMemorySizeInBytes) {
- reAllocateMemoryAccordingToRatio((double) totalMemorySizeInBytes / this.totalMemorySizeInBytes);
+ public long getInitialAllocatedMemorySizeInBytes() {
+ return initialAllocatedMemorySizeInBytes;
}
/** Get available memory size in bytes of memory manager */
@@ -486,8 +481,45 @@ public class MemoryManager {
return memorySize;
}
+ public MemoryManager setMemoryUpdateCallback(BiConsumer<Long, Long> memoryUpdateCallback) {
+ this.memoryUpdateCallback.set(memoryUpdateCallback);
+ return this;
+ }
+
// endregion
+ /**
+ * Resize memory by ratio
+ *
+ * @param ratio the ratio to resize memory, values [0.0, 1.0]
+ */
+ public synchronized void resizeByRatio(double ratio) {
+ // Update initial allocated memory size by ratio
+ long beforeInitialAllocatedMemorySizeInBytes = this.initialAllocatedMemorySizeInBytes;
+ this.initialAllocatedMemorySizeInBytes *= ratio;
+ // Update total memory size by actual size
+ long beforeTotalMemorySizeInBytes = this.totalMemorySizeInBytes;
+ this.totalMemorySizeInBytes +=
+ (this.initialAllocatedMemorySizeInBytes - beforeInitialAllocatedMemorySizeInBytes);
+ // Get actual ratio of re-allocate memory size
+ double actualRatio = (double) this.totalMemorySizeInBytes / beforeTotalMemorySizeInBytes;
+ // Re-allocate memory for all memory blocks
+ for (IMemoryBlock block : allocatedMemoryBlocks.values()) {
+ this.allocatedMemorySizeInBytes += block.resizeByRatio(actualRatio);
+ }
+ // Re-allocate memory for all child memory managers
+ for (Map.Entry<String, MemoryManager> entry : children.entrySet()) {
+ entry.getValue().resizeByRatio(ratio);
+ }
+ if (memoryUpdateCallback.get() != null) {
+ try {
+ memoryUpdateCallback.get().accept(beforeTotalMemorySizeInBytes, totalMemorySizeInBytes);
+ } catch (Exception e) {
+ LOGGER.warn("Failed to execute the update callback.", e);
+ }
+ }
+ }
+
@Override
public String toString() {
return "MemoryManager{"
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/memory/MemoryPeriodicalJobExecutor.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/memory/MemoryPeriodicalJobExecutor.java
new file mode 100644
index 00000000000..59662dc5a29
--- /dev/null
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/memory/MemoryPeriodicalJobExecutor.java
@@ -0,0 +1,111 @@
+/*
+ * 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.memory;
+
+import org.apache.iotdb.commons.concurrent.WrappedRunnable;
+import org.apache.iotdb.commons.concurrent.threadpool.ScheduledExecutorUtil;
+import org.apache.iotdb.commons.utils.TestOnly;
+
+import org.apache.tsfile.utils.Pair;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.Future;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+
+public class MemoryPeriodicalJobExecutor {
+ private static final Logger LOGGER = LoggerFactory.getLogger(MemoryPeriodicalJobExecutor.class);
+
+ private final ScheduledExecutorService executorService;
+ private final long minIntervalSeconds;
+
+ private long rounds;
+ private Future<?> executorFuture;
+
+ private final List<Pair<WrappedRunnable, Long>> periodicalJobs = new CopyOnWriteArrayList<>();
+
+ public MemoryPeriodicalJobExecutor(
+ final ScheduledExecutorService executorService, final long minIntervalSeconds) {
+ this.executorService = executorService;
+ this.minIntervalSeconds = minIntervalSeconds;
+ }
+
+ public void register(String id, Runnable periodicalJob, long intervalInSeconds) {
+ periodicalJobs.add(
+ new Pair<>(
+ new WrappedRunnable() {
+ @Override
+ public void runMayThrow() {
+ try {
+ periodicalJob.run();
+ } catch (Exception e) {
+ LOGGER.warn("Periodical job {} failed.", id, e);
+ }
+ }
+ },
+ Math.max(intervalInSeconds / minIntervalSeconds, 1)));
+ LOGGER.info(
+ "Memory periodical job {} is registered successfully. Interval: {} seconds.",
+ id,
+ Math.max(intervalInSeconds / minIntervalSeconds, 1) * minIntervalSeconds);
+ }
+
+ public synchronized void start() {
+ if (executorFuture == null) {
+ rounds = 0;
+
+ executorFuture =
+ ScheduledExecutorUtil.safelyScheduleWithFixedDelay(
+ executorService,
+ this::execute,
+ minIntervalSeconds,
+ minIntervalSeconds,
+ TimeUnit.SECONDS);
+ LOGGER.info("Memory periodical job executor is started successfully.");
+ }
+ }
+
+ protected void execute() {
+ ++rounds;
+
+ for (final Pair<WrappedRunnable, Long> periodicalJob : periodicalJobs) {
+ if (rounds % periodicalJob.right == 0) {
+ periodicalJob.left.run();
+ }
+ }
+ }
+
+ public synchronized void stop() {
+ if (executorFuture != null) {
+ executorFuture.cancel(false);
+ executorFuture = null;
+ LOGGER.info("Memory periodical job executor is stopped successfully.");
+ }
+ }
+
+ @TestOnly
+ public void clear() {
+ periodicalJobs.clear();
+ LOGGER.info("All memory periodical jobs are cleared successfully.");
+ }
+}
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/memory/MemoryRuntimeController.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/memory/MemoryRuntimeController.java
new file mode 100644
index 00000000000..89b1976b00c
--- /dev/null
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/memory/MemoryRuntimeController.java
@@ -0,0 +1,106 @@
+/*
+ * 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.memory;
+
+import org.apache.iotdb.commons.concurrent.IoTDBThreadPoolFactory;
+import org.apache.iotdb.commons.concurrent.ThreadName;
+import org.apache.iotdb.commons.conf.CommonConfig;
+import org.apache.iotdb.commons.conf.CommonDescriptor;
+import org.apache.iotdb.commons.exception.StartupException;
+import org.apache.iotdb.commons.service.IService;
+import org.apache.iotdb.commons.service.ServiceType;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.concurrent.atomic.AtomicBoolean;
+
+public class MemoryRuntimeController implements IService {
+ private static final Logger LOGGER = LoggerFactory.getLogger(MemoryRuntimeController.class);
+ private static final CommonConfig CONFIG = CommonDescriptor.getInstance().getConfig();
+ private static final MemoryManager ON_HEAP_MEMORY_MANAGER =
+ MemoryConfig.global().getMemoryManager("OnHeap");
+ private static final boolean ENABLE_MEMORY_ADAPT = CONFIG.isEnableMemoryAdapt();
+ private static final long MEMORY_ADAPT_INTERVAL_IN_S = CONFIG.getMemoryAdaptIntervalInS();
+
+ private static final AtomicBoolean isShutdown = new AtomicBoolean(false);
+
+ private static final MemoryPeriodicalJobExecutor memoryPeriodicalJobExecutor =
+ new MemoryPeriodicalJobExecutor(
+ IoTDBThreadPoolFactory.newSingleThreadScheduledExecutor(
+ ThreadName.MEMORY_PERIODICAL_JOB_EXECUTOR.getName()),
+ MEMORY_ADAPT_INTERVAL_IN_S);
+
+ @Override
+ public void start() throws StartupException {
+ memoryPeriodicalJobExecutor.start();
+
+ // Try to adapt total memory size according to the JVM total memory size
+ if (ENABLE_MEMORY_ADAPT) {
+ LOGGER.info(
+ "Enable automatic memory adapt with an interval of {} s", MEMORY_ADAPT_INTERVAL_IN_S);
+ MemoryRuntimeController.getInstance()
+ .registerPeriodicalJob(
+ "MemoryRuntimeAgent#adaptTotalMemory()",
+ this::adaptTotalMemory,
+ MEMORY_ADAPT_INTERVAL_IN_S);
+ }
+
+ isShutdown.set(false);
+ }
+
+ private void adaptTotalMemory() {
+ long totalMemory = Runtime.getRuntime().totalMemory();
+ if (ON_HEAP_MEMORY_MANAGER != null) {
+ long originMemorySize = ON_HEAP_MEMORY_MANAGER.getInitialAllocatedMemorySizeInBytes();
+ if (originMemorySize != totalMemory) {
+ LOGGER.info("Total memory size changed from {} to {}", originMemorySize, totalMemory);
+ ON_HEAP_MEMORY_MANAGER.resizeByRatio((double) totalMemory / originMemorySize);
+ }
+ }
+ }
+
+ @Override
+ public void stop() {
+ if (isShutdown.get()) {
+ return;
+ }
+ isShutdown.set(true);
+
+ memoryPeriodicalJobExecutor.stop();
+ }
+
+ public void registerPeriodicalJob(String id, Runnable periodicalJob, long intervalInSeconds) {
+ memoryPeriodicalJobExecutor.register(id, periodicalJob, intervalInSeconds);
+ }
+
+ @Override
+ public ServiceType getID() {
+ return ServiceType.MEMORY_RUNTIME_CONTROLLER;
+ }
+
+ private static class MemoryRuntimeAgentHolder {
+ private static final MemoryRuntimeController HANDLE = new MemoryRuntimeController();
+ }
+
+ public static MemoryRuntimeController getInstance() {
+ return MemoryRuntimeAgentHolder.HANDLE;
+ }
+}
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/service/ServiceType.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/service/ServiceType.java
index 7267c79a665..098891c2937 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/service/ServiceType.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/service/ServiceType.java
@@ -80,7 +80,7 @@ public enum ServiceType {
PIPE_PLUGIN_CLASSLOADER_MANAGER_SERVICE(
"Pipe Plugin Classloader Manager Service", "PipePluginClassLoader"),
AINode_RPC_SERVICE("Rpc Service for AINode", "AINodeRPCService"),
- MEMORY_RUNTIME_AGENT("Memory Runtime Agent", "MemoryRuntimeAgent"),
+ MEMORY_RUNTIME_CONTROLLER("Memory Runtime Controller", "MemoryRuntimeController"),
PIPE_RUNTIME_DATA_NODE_AGENT("Pipe Runtime Data Node Agent", "PipeRuntimeDataNodeAgent"),
PIPE_RUNTIME_CONFIG_NODE_AGENT("Pipe Runtime Config Node Agent", "PipeRuntimeConfigNodeAgent"),
SUBSCRIPTION_RUNTIME_AGENT("Subscription Runtime Agent", "SubscriptionRuntimeAgent"),