diff --git a/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBWriteMemoryLifecycleIT.java b/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBWriteMemoryLifecycleIT.java new file mode 100644 index 0000000000000..6f743ee307397 --- /dev/null +++ b/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBWriteMemoryLifecycleIT.java @@ -0,0 +1,382 @@ +/* + * 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.it; + +import org.apache.iotdb.db.it.utils.TestUtils; +import org.apache.iotdb.isession.ISession; +import org.apache.iotdb.it.env.EnvFactory; +import org.apache.iotdb.it.env.cluster.config.MppCommonConfig; +import org.apache.iotdb.it.framework.IoTDBTestRunner; +import org.apache.iotdb.itbase.category.LocalStandaloneIT; +import org.apache.iotdb.itbase.env.BaseEnv; +import org.apache.iotdb.rpc.StatementExecutionException; +import org.apache.iotdb.rpc.TSStatusCode; + +import org.apache.tsfile.enums.TSDataType; +import org.apache.tsfile.write.record.Tablet; +import org.apache.tsfile.write.schema.MeasurementSchema; +import org.junit.After; +import org.junit.Before; +import org.junit.Test; +import org.junit.experimental.categories.Category; +import org.junit.runner.RunWith; + +import java.sql.Connection; +import java.sql.ResultSet; +import java.sql.Statement; +import java.util.Arrays; +import java.util.Collections; +import java.util.Map; +import java.util.Properties; +import java.util.TreeMap; +import java.util.concurrent.TimeUnit; + +import static org.awaitility.Awaitility.await; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertThrows; +import static org.junit.Assert.assertTrue; + +@RunWith(IoTDBTestRunner.class) +@Category(LocalStandaloneIT.class) +public class IoTDBWriteMemoryLifecycleIT { + private static final String DATABASE = "root.write_memory"; + private static final String[] DEVICES = {DATABASE + ".plain", DATABASE + ".aligned"}; + private static final String OVERSIZED_TEXT = "x".repeat(32 * 1024); + private final Map expected = new TreeMap<>(); + + @Before + public void setUp() throws Exception { + EnvFactory.getEnv() + .getConfig() + .getCommonConfig() + .setDataRegionConsensusProtocolClass("org.apache.iotdb.consensus.simple.SimpleConsensus") + .setWalMode("SYNC") + .setEnableMemControl(true) + .setEnableLastCache(false) + .setPrimitiveArraySize(64) + .setTimePartitionInterval(1000) + .setMemtableSizeThreshold(64L * 1024 * 1024) + .setEnableSeqSpaceCompaction(false) + .setEnableUnseqSpaceCompaction(false) + .setEnableCrossSpaceCompaction(false); + // Keep flushes deterministic and reject WAL entries only after memory admission. + Properties properties = new Properties(); + properties.setProperty("enable_timed_flush_seq_memtable", "false"); + properties.setProperty("enable_timed_flush_unseq_memtable", "false"); + properties.setProperty("wal_entry_size_limit_in_byte", "16384"); + // Read live accounting, not the reporter's 15-second cached snapshot. + properties.setProperty("prometheus_reporter_async_update", "false"); + ((MppCommonConfig) EnvFactory.getEnv().getConfig().getDataNodeCommonConfig()) + .updateProperties(properties); + EnvFactory.getEnv() + .getConfig() + .getConfigNodeConfig() + .setMetricReporterType(Collections.singletonList("PROMETHEUS")); + EnvFactory.getEnv() + .getConfig() + .getDataNodeConfig() + .setMetricReporterType(Collections.singletonList("PROMETHEUS")); + EnvFactory.getEnv().initClusterEnvironment(); + try (ISession session = EnvFactory.getEnv().getSessionConnection()) { + session.executeNonQueryStatement("CREATE DATABASE " + DATABASE); + session.executeNonQueryStatement("CREATE TIMESERIES " + DEVICES[0] + ".n INT64"); + session.executeNonQueryStatement("CREATE TIMESERIES " + DEVICES[0] + ".t TEXT"); + session.executeNonQueryStatement( + "CREATE ALIGNED TIMESERIES " + DEVICES[1] + "(n INT64, t TEXT)"); + } + } + + @After + public void tearDown() throws Exception { + EnvFactory.getEnv().cleanClusterEnvironment(); + } + + @Test + public void testArrayBoundariesSparseColumnsAndRepeatedFlush() throws Exception { + try (ISession session = EnvFactory.getEnv().getSessionConnection()) { + // Exercise 63 -> 64 -> 65 and multiple array allocations, for both TVList types. + insert(session, 0, 63); + insert(session, 63, 1); + insert(session, 64, 1); + insert(session, 65, 64); + long beforeNewColumn = memoryUsage(); + session.insertAlignedRecord( + DEVICES[1], + 128, + Collections.singletonList("late"), + Collections.singletonList(TSDataType.TEXT), + Collections.singletonList("new-column")); + assertTrue("A late aligned column must be charged", memoryUsage() > beforeNewColumn); + assertData(); + flushAndAssertReleased(session); + for (int cycle = 1; cycle <= 3; cycle++) { + insert(session, cycle * 129, 129); + // This timestamp is already on disk: exercise unsequence processors and overwrite merge. + overwrite(session, 0, "replacement-" + cycle); + assertTrue("Live writes must be charged after reusing pooled arrays", memoryUsage() > 0); + assertData(); + flushAndAssertReleased(session); + assertData(); + } + } + try (Connection connection = EnvFactory.getEnv().getConnection(); + Statement statement = connection.createStatement(); + ResultSet result = statement.executeQuery("SELECT late FROM " + DEVICES[1])) { + assertTrue(result.next()); + assertEquals(128, result.getLong(1)); + assertEquals("new-column", result.getString(2)); + assertFalse(result.next()); + } + } + + @Test + public void testWalRejectionRollsBackAndAllowsFurtherWrites() throws Exception { + try (ISession session = EnvFactory.getEnv().getSessionConnection()) { + insert(session, 0, 63); + long baseline = memoryUsage(); + assertTrue(baseline > 0); + for (int attempt = 0; attempt < 3; attempt++) { + for (int device = 0; device < DEVICES.length; device++) { + final boolean aligned = device == 1; + final String path = DEVICES[device]; + // A new measurement exercises metadata rollback as well as array and Binary rollback. + StatementExecutionException failure = + assertThrows( + StatementExecutionException.class, + () -> { + if (aligned) { + session.insertAlignedRecord( + path, + 63, + Collections.singletonList("rejected"), + Collections.singletonList(TSDataType.TEXT), + Collections.singletonList(OVERSIZED_TEXT)); + } else { + session.insertRecord( + path, + 63, + Collections.singletonList("rejected"), + Collections.singletonList(TSDataType.TEXT), + Collections.singletonList(OVERSIZED_TEXT)); + } + }); + assertEquals(TSStatusCode.WAL_ENTRY_TOO_LARGE.getStatusCode(), failure.getStatusCode()); + assertEquals("Rejected row leaked memory", baseline, memoryUsage()); + Tablet tablet = tablet(path, 63, 2, OVERSIZED_TEXT); + StatementExecutionException batchFailure = + assertThrows( + StatementExecutionException.class, + () -> { + if (aligned) { + session.insertAlignedTablet(tablet); + } else { + session.insertTablet(tablet); + } + }); + assertEquals( + TSStatusCode.WAL_ENTRY_TOO_LARGE.getStatusCode(), batchFailure.getStatusCode()); + assertEquals("Rejected tablet leaked memory", baseline, memoryUsage()); + } + } + assertData(); + insert(session, 63, 66); + assertData(); + flushAndAssertReleased(session); + assertData(); + } + try (Connection connection = EnvFactory.getEnv().getConnection(); + Statement statement = connection.createStatement()) { + for (String device : DEVICES) { + try (ResultSet result = statement.executeQuery("SELECT rejected FROM " + device)) { + assertFalse("Rejected WAL entry must not insert data", result.next()); + } + } + } + } + + @Test + public void testCrashRecoveryThenWriteAndRelease() throws Exception { + try (ISession session = EnvFactory.getEnv().getSessionConnection()) { + insert(session, 0, 129); + assertTrue(memoryUsage() > 0); + } + // No FLUSH and no graceful shutdown: acknowledged data must come back through WAL recovery. + TestUtils.stopForciblyAndRestartDataNodes(); + assertData(); + // Recovery flushes the replayed memtable and seals its TsFile before accepting requests. + assertEquals("Recovery must leave no live write memory", 0, memoryUsage()); + try (ISession session = EnvFactory.getEnv().getSessionConnection()) { + insert(session, 129, 129); + overwrite(session, 0, "after-recovery"); + assertTrue("New writes after recovery must be charged", memoryUsage() > 0); + assertData(); + flushAndAssertReleased(session); + } + TestUtils.stopForciblyAndRestartDataNodes(); + assertData(); + assertEquals("Closed files must not retain write memory after restart", 0, memoryUsage()); + try (ISession session = EnvFactory.getEnv().getSessionConnection()) { + insert(session, 258, 65); + assertTrue(memoryUsage() > 0); + flushAndAssertReleased(session); + assertData(); + } + } + + @Test + public void testTableTabletSplitAcrossDevicesAndTimePartitions() throws Exception { + String database = "write_memory_table"; + try (Connection connection = EnvFactory.getEnv().getConnection(BaseEnv.TABLE_SQL_DIALECT); + Statement statement = connection.createStatement()) { + statement.execute("CREATE DATABASE " + database); + statement.execute("USE " + database); + statement.execute("CREATE TABLE readings(device STRING TAG, n INT64 FIELD, t STRING FIELD)"); + // Each request contains two devices and spans the 1000 ms partition boundary. + for (int cycle = 0; cycle < 2; cycle++) { + StringBuilder sql = new StringBuilder("INSERT INTO readings(time,device,n,t) VALUES "); + for (int row = 0; row < 130; row++) { + for (int device = 0; device < 2; device++) { + if (row != 0 || device != 0) { + sql.append(','); + } + sql.append('(') + .append(935 + row) + .append(", 'd") + .append(device) + .append("', ") + .append(row + cycle * 1000) + .append(", ") + .append(row % 3 == 0 ? "null" : "'value-" + cycle + "-" + row + "'") + .append(')'); + } + } + statement.execute(sql.toString()); + assertTrue("All split processors must account for live writes", memoryUsage(database) > 0); + // The second cycle overwrites flushed data, exercising unsequence ranges in both + // partitions. + for (int phase = 0; phase < 2; phase++) { + try (ResultSet result = + statement.executeQuery("SELECT time,device,n,t FROM readings ORDER BY time,device")) { + for (int row = 0; row < 130; row++) { + for (int device = 0; device < 2; device++) { + assertTrue(result.next()); + assertEquals(935 + row, result.getLong(1)); + assertEquals("d" + device, result.getString(2)); + assertEquals(row + cycle * 1000, result.getLong(3)); + assertEquals( + row % 3 == 0 ? null : "value-" + cycle + "-" + row, result.getString(4)); + } + } + assertFalse(result.next()); + } + if (phase == 0) { + statement.execute("FLUSH"); + await() + .atMost(30, TimeUnit.SECONDS) + .untilAsserted(() -> assertEquals(0, memoryUsage(database))); + } + } + } + } + } + + private void insert(ISession session, int start, int count) throws Exception { + session.insertTablet(tablet(DEVICES[0], start, count, null)); + session.insertAlignedTablet(tablet(DEVICES[1], start, count, null)); + for (int time = start; time < start + count; time++) { + expected.put((long) time, text(time)); + } + } + + private static Tablet tablet(String device, int start, int count, String override) { + Tablet tablet = + new Tablet( + device, + Arrays.asList( + new MeasurementSchema("n", TSDataType.INT64), + new MeasurementSchema("t", TSDataType.TEXT)), + count); + for (int row = 0; row < count; row++) { + int time = start + row; + tablet.addTimestamp(row, time); + tablet.addValue("n", row, (long) time); + tablet.addValue("t", row, override == null ? text(time) : override); + } + return tablet; + } + + private static String text(int time) { + return time % 3 == 0 ? null : "value-" + time; + } + + private void overwrite(ISession session, long time, String value) throws Exception { + session.insertRecord( + DEVICES[0], + time, + Collections.singletonList("t"), + Collections.singletonList(TSDataType.TEXT), + Collections.singletonList(value)); + session.insertAlignedRecord( + DEVICES[1], + time, + Collections.singletonList("t"), + Collections.singletonList(TSDataType.TEXT), + Collections.singletonList(value)); + expected.put(time, value); + } + + private void assertData() throws Exception { + try (Connection connection = EnvFactory.getEnv().getConnection(); + Statement statement = connection.createStatement()) { + for (String device : DEVICES) { + try (ResultSet result = statement.executeQuery("SELECT n,t FROM " + device)) { + for (Map.Entry row : expected.entrySet()) { + assertTrue("Missing timestamp " + row.getKey() + " in " + device, result.next()); + assertEquals((long) row.getKey(), result.getLong(1)); + assertEquals((long) row.getKey(), result.getLong(2)); + assertFalse(result.wasNull()); + assertEquals(row.getValue(), result.getString(3)); + } + assertFalse("Unexpected extra row in " + device, result.next()); + } + } + } + } + + private static void flushAndAssertReleased(ISession session) throws Exception { + session.executeNonQueryStatement("FLUSH"); + await().atMost(30, TimeUnit.SECONDS).untilAsserted(() -> assertEquals(0, memoryUsage())); + } + + private static long memoryUsage() { + return memoryUsage(DATABASE); + } + + private static long memoryUsage(String database) { + String label = "name=\"database_" + database + "\""; + return EnvFactory.getEnv().getMetricPrometheusReporterContents(null).stream() + .flatMap(content -> content.lines()) + .filter(line -> line.startsWith("mem{") && line.contains(label)) + .mapToLong(line -> (long) Double.parseDouble(line.substring(line.lastIndexOf(' ') + 1))) + .findFirst() + .orElseThrow(() -> new AssertionError("Missing database memory gauge: " + label)); + } +} diff --git a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/StorageEngineMessages.java b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/StorageEngineMessages.java index d385331319126..1adfd8ac3cc8e 100644 --- a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/StorageEngineMessages.java +++ b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/StorageEngineMessages.java @@ -1670,6 +1670,15 @@ private StorageEngineMessages() {} public static final String MESSAGE_THE_ASSOCIATED_RESOURCE_FILE_OF_ARG_IS_NOT_FOUND_IN_THE_SNAPSHOT_CB9152B5 = "The associated resource file of {} is not found in the snapshot"; public static final String MESSAGE_EVICTED_NON_EXISTING_EXISTING_SERIES_COUNT_ARG_ARG_ARG_TOTAL_REQUEST_ARG_3026ADBD = "Evicted non-existing/existing series count: {}/{}({}), total request: {}"; + public static final String EXCEPTION_WRITE_MEMORY_BLOCK_ARG_IS_CLOSED_9E2561DF = + "Write memory block %s is closed"; + public static final String EXCEPTION_WRITE_MEMORY_SIZE_MUST_BE_NON_NEGATIVE_ARG_2CBCF358 = + "Write memory size must be non-negative: %s"; + public static final String EXCEPTION_CANNOT_RELEASE_ARG_BYTES_FROM_WRITE_MEMORY_BLOCK_ARG_WITH_ARG_BYTES_3783A614 = + "Cannot release %s bytes from write memory block %s with %s bytes"; + public static final String EXCEPTION_WRITE_MEMORY_BLOCK_ARG_ALREADY_BELONGS_TO_ANOTHER_MANAGER_A05A8229 = + "Write memory block %s already belongs to another manager"; + public static final String EXCEPTION_CHUNK_LAYOUT_IS_MISSING_E87C71C0 = "Chunk layout is missing"; public static final String EXCEPTION_CHUNK_HEADER_LIST_IS_EMPTY_5DE68C78 = "Chunk header list is empty."; diff --git a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/StorageEngineMessages.java b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/StorageEngineMessages.java index 53b828387b558..0a0d3b266a4bd 100644 --- a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/StorageEngineMessages.java +++ b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/StorageEngineMessages.java @@ -1591,6 +1591,15 @@ private StorageEngineMessages() {} public static final String MESSAGE_THE_ASSOCIATED_RESOURCE_FILE_OF_ARG_IS_NOT_FOUND_IN_THE_SNAPSHOT_CB9152B5 = "在快照中未找到 {} 关联的资源文件"; public static final String MESSAGE_EVICTED_NON_EXISTING_EXISTING_SERIES_COUNT_ARG_ARG_ARG_TOTAL_REQUEST_ARG_3026ADBD = "淘汰的不存在/已存在时间序列数量:{}/{}({}),总请求数:{}"; + public static final String EXCEPTION_WRITE_MEMORY_BLOCK_ARG_IS_CLOSED_9E2561DF = + "写入内存块 %s 已关闭"; + public static final String EXCEPTION_WRITE_MEMORY_SIZE_MUST_BE_NON_NEGATIVE_ARG_2CBCF358 = + "写入内存大小不能为负数:%s"; + public static final String EXCEPTION_CANNOT_RELEASE_ARG_BYTES_FROM_WRITE_MEMORY_BLOCK_ARG_WITH_ARG_BYTES_3783A614 = + "无法释放 %s 字节:写入内存块 %s 当前占用 %s 字节"; + public static final String EXCEPTION_WRITE_MEMORY_BLOCK_ARG_ALREADY_BELONGS_TO_ANOTHER_MANAGER_A05A8229 = + "写入内存块 %s 已属于其他管理器"; + public static final String EXCEPTION_CHUNK_LAYOUT_IS_MISSING_E87C71C0 = "缺少 Chunk layout"; public static final String EXCEPTION_CHUNK_HEADER_LIST_IS_EMPTY_5DE68C78 = "Chunk header 列表为空。"; public static final String EXCEPTION_FAILED_TO_ENCODE_CHUNK_USING_ALIGNEDCHUNKWRITERIMPL_5AAF9AA8 = diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java index 3c94440301dda..4ee3a2a6827f1 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java @@ -372,7 +372,7 @@ public class DataRegion implements IDataRegionForQuery { .build(); /** database info for mem control. */ - private final DataRegionInfo dataRegionInfo = new DataRegionInfo(this); + private final DataRegionInfo dataRegionInfo; /** whether it's ready from recovery. */ private boolean isReady = false; @@ -431,6 +431,7 @@ public DataRegion( this.dataRegionIdString = dataRegionIdString; this.dataRegionId = new DataRegionId(Integer.parseInt(dataRegionIdString)); this.databaseName = databaseName; + this.dataRegionInfo = new DataRegionInfo(this); this.fileFlushPolicy = fileFlushPolicy; this.isTableModel = isTableModelDatabase(databaseName); this.delayAnalyzer = @@ -512,6 +513,7 @@ public DataRegion(String databaseName, String dataRegionIdString) { this.isTableModel = isTableModelDatabase(databaseName); this.dataRegionIdString = dataRegionIdString; this.dataRegionId = new DataRegionId(Integer.parseInt(this.dataRegionIdString)); + this.dataRegionInfo = new DataRegionInfo(this); this.tsFileManager = new TsFileManager(databaseName, dataRegionIdString, ""); this.partitionMaxFileVersions = new HashMap<>(); partitionMaxFileVersions.put(0L, 0L); @@ -1073,7 +1075,9 @@ private void recoverUnsealedTsFileCallBack(UnsealedTsFileRecoverPerformer recove tsFileResource.removeResourceFile(); tsFileProcessor.setTimeRangeId(timePartitionId); writer.makeMetadataVisible(); - TsFileProcessorInfo tsFileProcessorInfo = new TsFileProcessorInfo(dataRegionInfo); + TsFileProcessorInfo tsFileProcessorInfo = + new TsFileProcessorInfo( + dataRegionInfo, tsFileProcessor.getTsFileResource().getTsFilePath()); tsFileProcessor.setTsFileProcessorInfo(tsFileProcessorInfo); this.dataRegionInfo.initTsFileProcessorInfo(tsFileProcessor); // get chunkMetadata size @@ -2322,7 +2326,9 @@ private TsFileProcessor getTsFileProcessor( this::flushCallback, sequence); - TsFileProcessorInfo tsFileProcessorInfo = new TsFileProcessorInfo(dataRegionInfo); + TsFileProcessorInfo tsFileProcessorInfo = + new TsFileProcessorInfo( + dataRegionInfo, tsFileProcessor.getTsFileResource().getTsFilePath()); tsFileProcessor.setTsFileProcessorInfo(tsFileProcessorInfo); this.dataRegionInfo.initTsFileProcessorInfo(tsFileProcessor); 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 0e8e4902c8b10..ca5357ad7d7af 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 @@ -21,6 +21,7 @@ import org.apache.iotdb.db.conf.IoTDBDescriptor; import org.apache.iotdb.db.storageengine.dataregion.memtable.TsFileProcessor; import org.apache.iotdb.db.storageengine.rescon.memory.SystemInfo; +import org.apache.iotdb.db.storageengine.rescon.memory.WriteMemoryManager; import java.util.List; import java.util.concurrent.CopyOnWriteArrayList; @@ -32,11 +33,7 @@ public class DataRegionInfo { private final DataRegion dataRegion; - /** - * The total database memory cost, including unsealed TsFileResource, ChunkMetadata, WAL, - * primitive arrays and TEXT values - */ - private final AtomicLong memoryCost; + private final WriteMemoryManager writeMemoryManager; /** The threshold of reporting it's size to SystemInfo */ private final long storageGroupSizeReportThreshold = @@ -57,7 +54,13 @@ public class DataRegionInfo { public DataRegionInfo(DataRegion dataRegion) { this.dataRegion = dataRegion; - memoryCost = new AtomicLong(); + this.writeMemoryManager = + new WriteMemoryManager( + "DataRegion[" + + dataRegion.getDatabaseName() + + "/" + + dataRegion.getDataRegionIdString() + + "]"); } public DataRegion getDataRegion() { @@ -69,16 +72,12 @@ public void initTsFileProcessorInfo(TsFileProcessor tsFileProcessor) { reportedTsps.add(tsFileProcessor); } - public void addStorageGroupMemCost(long cost) { - memoryCost.getAndAdd(cost); - } - - public void releaseStorageGroupMemCost(long cost) { - memoryCost.getAndAdd(-cost); + public WriteMemoryManager getWriteMemoryManager() { + return writeMemoryManager; } public long getMemCost() { - return memoryCost.get(); + return writeMemoryManager.getMemoryUsageInBytes(); } public List getAllReportedTsp() { @@ -87,7 +86,7 @@ public List getAllReportedTsp() { public boolean needToReportToSystem() { boolean needToReport = - memoryCost.get() - lastReportedSize.get() > storageGroupSizeReportThreshold + getMemCost() - lastReportedSize.get() > storageGroupSizeReportThreshold || needToReportToSystem.get(); // report once and then reset flag to false if (needToReportToSystem.get()) { diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/flush/MemTableFlushTask.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/flush/MemTableFlushTask.java index 4a2dfd88cd604..57b1f73de7c1c 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/flush/MemTableFlushTask.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/flush/MemTableFlushTask.java @@ -45,14 +45,17 @@ import org.slf4j.LoggerFactory; import java.io.IOException; +import java.io.UncheckedIOException; import java.util.ArrayList; import java.util.Collections; import java.util.List; import java.util.Map; import java.util.concurrent.BlockingQueue; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CompletionException; import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutionException; -import java.util.concurrent.Future; import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.TimeUnit; @@ -72,13 +75,15 @@ public class MemTableFlushTask { /* storage group name -> last time */ private static final Map flushPointsCache = new ConcurrentHashMap<>(); - private final Future encodingTaskFuture; - private final Future ioTaskFuture; + private final CompletableFuture subTaskFailure = new CompletableFuture<>(); + private FlushSubTask encodingSubTask; + private FlushSubTask ioSubTask; + private final boolean encodingFasterThanIo = SystemInfo.getInstance().isEncodingFasterThanIo(); private RestorableTsFileIOWriter writer; private final BlockingQueue encodingTaskQueue = new LinkedBlockingQueue<>(); private final BlockingQueue ioTaskQueue = - (SystemInfo.getInstance().isEncodingFasterThanIo()) + encodingFasterThanIo ? new LinkedBlockingQueue<>(config.getIoTaskQueueSizeForFlushing()) : new LinkedBlockingQueue<>(); @@ -118,8 +123,6 @@ public MemTableFlushTask( this.storageGroup = storageGroup; this.dataRegionId = dataRegionId; this.ignoreAllNullRows = !isTableModel; - this.encodingTaskFuture = SUB_TASK_POOL_MANAGER.submit(encodingTask); - this.ioTaskFuture = SUB_TASK_POOL_MANAGER.submit(ioTask); long MAX_NUMBER_OF_POINTS_IN_CHUNK = config.getTargetChunkPointNum(); long TARGET_CHUNK_SIZE = config.getTargetChunkSize(); @@ -141,6 +144,44 @@ public MemTableFlushTask( /** the function for flushing memtable. */ @SuppressWarnings("squid:S3776") public void syncFlushMemTable() throws ExecutionException, InterruptedException { + long estimatedBytes = + encodingFasterThanIo && memTable.getSeriesNumber() != 0 + ? memTable.memSize() + / memTable.getSeriesNumber() + * config.getIoTaskQueueSizeForFlushing() + : 0; + try (SystemInfo.FlushMemoryReservation ignored = + SystemInfo.getInstance() + .reserveTemporaryMemoryForFlushing( + "Flush[" + storageGroup + "/" + dataRegionId + "/" + memTable.getMemTableId() + "]", + estimatedBytes)) { + encodingSubTask = new FlushSubTask(encodingTask); + ioSubTask = new FlushSubTask(ioTask); + try { + SUB_TASK_POOL_MANAGER.submit(encodingSubTask); + SUB_TASK_POOL_MANAGER.submit(ioSubTask); + doSyncFlushMemTable(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw e; + } finally { + // Future.cancel/get alone does not establish that a running task has exited. + encodingSubTask.cancel(); + ioSubTask.cancel(); + boolean interrupted = Thread.interrupted(); + interrupted |= encodingSubTask.awaitExit(); + interrupted |= ioSubTask.awaitExit(); + encodingTaskQueue.clear(); + ioTaskQueue.clear(); + times = null; + if (interrupted) { + Thread.currentThread().interrupt(); + } + } + } + } + + private void doSyncFlushMemTable() throws ExecutionException, InterruptedException { long avgSeriesPointsNum = memTable.getSeriesNumber() == 0 ? 0 @@ -152,16 +193,6 @@ public void syncFlushMemTable() throws ExecutionException, InterruptedException memTable.getTotalPointsNum(), avgSeriesPointsNum); - long estimatedTemporaryMemSize = 0L; - if (SystemInfo.getInstance().isEncodingFasterThanIo()) { - estimatedTemporaryMemSize = - memTable.getSeriesNumber() == 0 - ? 0 - : memTable.memSize() - / memTable.getSeriesNumber() - * config.getIoTaskQueueSizeForFlushing(); - SystemInfo.getInstance().applyTemporaryMemoryForFlushing(estimatedTemporaryMemSize); - } long start = System.currentTimeMillis(); long sortTime = 0; @@ -207,17 +238,9 @@ public void syncFlushMemTable() throws ExecutionException, InterruptedException sortTime); WRITING_METRICS.recordFlushCost(WritingMetrics.FLUSH_STAGE_SORT, sortTime); - try { - encodingTaskFuture.get(); - } catch (InterruptedException | ExecutionException e) { - ioTaskFuture.cancel(true); - if (e instanceof InterruptedException) { - Thread.currentThread().interrupt(); - } - throw e; - } - - ioTaskFuture.get(); + CompletableFuture.anyOf( + subTaskFailure, CompletableFuture.allOf(encodingSubTask.finished, ioSubTask.finished)) + .get(); try { long writePlanIndicesStartTime = System.currentTimeMillis(); @@ -230,9 +253,6 @@ public void syncFlushMemTable() throws ExecutionException, InterruptedException throw new ExecutionException(e); } - if (estimatedTemporaryMemSize != 0) { - SystemInfo.getInstance().releaseTemporaryMemoryForFlushing(estimatedTemporaryMemSize); - } SystemInfo.getInstance().setEncodingFasterThanIo(ioTime >= memSerializeTime); MetricService.getInstance() @@ -258,14 +278,16 @@ public void run() { storageGroup, writer.getFile().getName()); while (true) { - + if (encodingSubTask.isCancelled()) { + return; + } Object task; try { task = encodingTaskQueue.take(); } catch (InterruptedException e1) { LOGGER.error(StorageEngineMessages.TAKE_TASK_INTO_IO_QUEUE_INTERRUPTED); Thread.currentThread().interrupt(); - break; + throw new CompletionException(e1); } if (task instanceof StartFlushGroupIOTask || task instanceof EndChunkGroupIoTask) { try { @@ -279,8 +301,8 @@ public void run() { storageGroup, writer.getFile().getName(), e); - // generally it is because the thread pool is shutdown so the task should be aborted - break; + Thread.currentThread().interrupt(); + throw new CompletionException(e); } } else if (task instanceof TaskEnd) { break; @@ -301,6 +323,7 @@ public void run() { } catch (InterruptedException e) { LOGGER.error(StorageEngineMessages.PUT_TASK_INTO_IO_QUEUE_INTERRUPTED); Thread.currentThread().interrupt(); + throw new CompletionException(e); } DataRegion.getNonSystemDatabaseName(storageGroup) @@ -350,13 +373,16 @@ public static void recordFlushPointsMetricInternal( storageGroup, writer.getFile().getName()); while (true) { + if (ioSubTask.isCancelled()) { + return; + } Object ioMessage = null; try { ioMessage = ioTaskQueue.take(); } catch (InterruptedException e1) { LOGGER.error(StorageEngineMessages.TAKE_TASK_FROM_IO_QUEUE_INTERRUPTED); Thread.currentThread().interrupt(); - break; + throw new CompletionException(e1); } long starTime = System.currentTimeMillis(); try { @@ -378,7 +404,7 @@ public static void recordFlushPointsMetricInternal( memTable, e); DataNodeExceptionMetrics.getInstance().recordSuspiciousDiskException(e); - return; + throw new UncheckedIOException(e); } long subTaskTime = System.currentTimeMillis() - starTime; ioTime += subTaskTime; @@ -394,6 +420,72 @@ public static void recordFlushPointsMetricInternal( WRITING_METRICS.recordFlushCost(WritingMetrics.FLUSH_STAGE_IO, ioTime); }; + /** Tracks actual worker exit, including cancellation before the executor starts it. */ + private final class FlushSubTask implements Runnable { + private final Runnable task; + private final CompletableFuture finished = new CompletableFuture<>(); + private final CountDownLatch exited = new CountDownLatch(1); + private Thread runner; + private boolean cancelled; + + private FlushSubTask(Runnable task) { + this.task = task; + } + + @Override + public void run() { + synchronized (this) { + if (cancelled) { + return; + } + runner = Thread.currentThread(); + } + try { + task.run(); + } catch (Throwable failure) { + subTaskFailure.completeExceptionally( + failure instanceof CompletionException || failure instanceof UncheckedIOException + ? failure.getCause() + : failure); + } finally { + synchronized (this) { + runner = null; + finished.complete(null); + exited.countDown(); + } + } + } + + private synchronized boolean isCancelled() { + return cancelled; + } + + private synchronized void cancel() { + cancelled = true; + if (runner != null) { + runner.interrupt(); + } else { + finished.complete(null); + exited.countDown(); + } + } + + private boolean awaitExit() { + boolean interrupted = false; + while (true) { + try { + exited.await(); + return interrupted; + } catch (InterruptedException e) { + // Cancellation has already interrupted the worker. It may still own buffers or the + // writer, so cleanup must wait for its actual exit before releasing the reservation. + // Restore the caller's interrupt flag after both workers have exited. + interrupted = true; + } + } + } + } + static class TaskEnd { TaskEnd() {} diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AbstractMemTable.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AbstractMemTable.java index 382063359c186..f51f2c6810564 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AbstractMemTable.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AbstractMemTable.java @@ -44,6 +44,8 @@ import org.apache.iotdb.db.storageengine.dataregion.read.filescan.impl.MemChunkHandleImpl; import org.apache.iotdb.db.storageengine.dataregion.wal.buffer.IWALByteBufferView; import org.apache.iotdb.db.storageengine.dataregion.wal.utils.WALWriteUtils; +import org.apache.iotdb.db.storageengine.rescon.memory.WriteMemoryBlock; +import org.apache.iotdb.db.storageengine.rescon.memory.WriteMemoryManager; import org.apache.iotdb.db.utils.EncryptDBUtils; import org.apache.iotdb.db.utils.MemUtils; import org.apache.iotdb.db.utils.ModificationUtils; @@ -95,11 +97,20 @@ public abstract class AbstractMemTable implements IMemTable { /** Memory size of data points, including TEXT values. */ private long memSize = 0; - /** - * Memory usage of all TVLists memory usage regardless of whether these TVLists are full, - * including TEXT values. - */ - private long tvListRamCost = 0; + private final long memTableId = memTableIdCounter.incrementAndGet(); + + private final WriteMemoryBlock arrayMemoryBlock = + new WriteMemoryBlock( + "MemTable[" + memTableId + "].Arrays", WriteMemoryBlock.Category.MEMTABLE_ARRAY); + private final WriteMemoryBlock binaryMemoryBlock = + new WriteMemoryBlock( + "MemTable[" + memTableId + "].Binary", WriteMemoryBlock.Category.MEMTABLE_BINARY); + private final WriteMemoryBlock structureMemoryBlock = + new WriteMemoryBlock( + "MemTable[" + memTableId + "].Structure", WriteMemoryBlock.Category.MEMTABLE_STRUCTURE); + private final WriteMemoryBlock recoveredMemoryBlock = + new WriteMemoryBlock( + "MemTable[" + memTableId + "].Recovered", WriteMemoryBlock.Category.MEMTABLE_RECOVERED); private int seriesNumber = 0; @@ -115,8 +126,6 @@ public abstract class AbstractMemTable implements IMemTable { private long minPlanIndex = Long.MAX_VALUE; - private final long memTableId = memTableIdCounter.incrementAndGet(); - private final long createdTime = System.currentTimeMillis(); /** this time is updated by the timed flush, same as createdTime when the feature is disabled. */ @@ -584,7 +593,10 @@ public void clear() { totalValueCount = 0; nullValueCount = 0; totalPointsNumThreshold = 0; - tvListRamCost = 0; + arrayMemoryBlock.clear(); + binaryMemoryBlock.clear(); + structureMemoryBlock.clear(); + recoveredMemoryBlock.clear(); maxPlanIndex = 0; minPlanIndex = 0; } @@ -956,19 +968,45 @@ public long delete(ModEntry modEntry, boolean ignoreAllNullRows) { return pointDeleted; } + @Override + public void setWriteMemoryManager(WriteMemoryManager manager) { + arrayMemoryBlock.attachTo(manager); + binaryMemoryBlock.attachTo(manager); + structureMemoryBlock.attachTo(manager); + recoveredMemoryBlock.attachTo(manager); + } + + @Override + public WriteMemoryBlock getArrayMemoryBlock() { + return arrayMemoryBlock; + } + + @Override + public WriteMemoryBlock getBinaryMemoryBlock() { + return binaryMemoryBlock; + } + + @Override + public WriteMemoryBlock getStructureMemoryBlock() { + return structureMemoryBlock; + } + @Override public void addTVListRamCost(long cost) { - this.tvListRamCost += cost; + arrayMemoryBlock.add(cost); } @Override public void releaseTVListRamCost(long cost) { - this.tvListRamCost -= cost; + arrayMemoryBlock.release(cost); } @Override public long getTVListsRamCost() { - return tvListRamCost; + return arrayMemoryBlock.getUsedMemoryInBytes() + + binaryMemoryBlock.getUsedMemoryInBytes() + + structureMemoryBlock.getUsedMemoryInBytes() + + recoveredMemoryBlock.getUsedMemoryInBytes(); } @Override @@ -993,9 +1031,18 @@ public boolean shouldFlush() { @Override public void release() { + if (arrayMemoryBlock.isClosed()) { + return; + } for (Entry entry : memTableMap.entrySet()) { entry.getValue().release(); } + // Preserve the existing write-accounting release point. Query-held TVLists and the array + // pool retain their existing resource ownership; this is not a measurement of heap liveness. + arrayMemoryBlock.close(); + binaryMemoryBlock.close(); + structureMemoryBlock.close(); + recoveredMemoryBlock.close(); } @Override @@ -1064,7 +1111,7 @@ public void serializeToWAL(IWALByteBufferView buffer) { WALWriteUtils.write((byte) -1, buffer); buffer.putInt(seriesNumber); buffer.putLong(memSize); - buffer.putLong(tvListRamCost); + buffer.putLong(getTVListsRamCost()); buffer.putLong(totalPointsNum); buffer.putLong(totalPointsNumThreshold); buffer.putLong(maxPlanIndex); @@ -1087,7 +1134,7 @@ protected void deserialize(DataInputStream stream, boolean multiTvListInMemChunk throws IOException { seriesNumber = stream.readInt(); memSize = stream.readLong(); - tvListRamCost = stream.readLong(); + recoveredMemoryBlock.add(stream.readLong()); totalPointsNum = stream.readLong(); totalPointsNumThreshold = stream.readLong(); maxPlanIndex = stream.readLong(); @@ -1119,7 +1166,7 @@ public void deserializeFromOldMemTableSnapshot( DataInputStream stream, boolean multiTvListInMemChunk) throws IOException { seriesNumber = stream.readInt(); memSize = stream.readLong(); - tvListRamCost = stream.readLong(); + recoveredMemoryBlock.add(stream.readLong()); totalPointsNum = stream.readLong(); totalPointsNumThreshold = stream.readLong(); maxPlanIndex = stream.readLong(); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/IMemTable.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/IMemTable.java index 48d43644c8dc3..066b8931be47d 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/IMemTable.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/IMemTable.java @@ -32,6 +32,8 @@ import org.apache.iotdb.db.storageengine.dataregion.modification.ModEntry; import org.apache.iotdb.db.storageengine.dataregion.read.filescan.IChunkHandle; import org.apache.iotdb.db.storageengine.dataregion.wal.buffer.WALEntryValue; +import org.apache.iotdb.db.storageengine.rescon.memory.WriteMemoryBlock; +import org.apache.iotdb.db.storageengine.rescon.memory.WriteMemoryManager; import org.apache.tsfile.file.metadata.IChunkMetadata; import org.apache.tsfile.file.metadata.IDeviceID; @@ -79,7 +81,15 @@ void writeAlignedRow( */ long memSize(); - /** only used when mem control enabled */ + void setWriteMemoryManager(WriteMemoryManager manager); + + WriteMemoryBlock getArrayMemoryBlock(); + + WriteMemoryBlock getBinaryMemoryBlock(); + + WriteMemoryBlock getStructureMemoryBlock(); + + /** Adds TVList array memory; containers and Binary payload have separate blocks. */ void addTVListRamCost(long cost); /** only used when mem control enabled */ diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java index 52dab8856a4c4..76fca09afec47 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java @@ -76,6 +76,7 @@ import org.apache.iotdb.db.storageengine.rescon.memory.MemTableManager; import org.apache.iotdb.db.storageengine.rescon.memory.PrimitiveArrayManager; import org.apache.iotdb.db.storageengine.rescon.memory.SystemInfo; +import org.apache.iotdb.db.storageengine.rescon.memory.WriteMemoryBlock; import org.apache.iotdb.db.utils.CommonUtils; import org.apache.iotdb.db.utils.EncryptDBUtils; import org.apache.iotdb.db.utils.MemUtils; @@ -134,8 +135,6 @@ public class TsFileProcessor { /** Logger fot this class. */ private static final Logger logger = LoggerFactory.getLogger(TsFileProcessor.class); - private static final int NUM_MEM_TO_ESTIMATE = 3; - /** Data Region name of this tsfile, databaseName-dataRegionId. */ private final String dataRegionName; @@ -323,7 +322,7 @@ public void insert(InsertRowNode insertRowNode, long[] infoForMetrics) ? new AlignedTVListRamCostSnapshot(workMemTable, insertRowNode.getDeviceID()) : null; - long[] memIncrements; + WriteMemoryDelta memIncrements; long memControlStartTime = System.nanoTime(); if (insertRowNode.isAligned()) { @@ -396,7 +395,7 @@ public void insert(InsertRowNode insertRowNode, long[] infoForMetrics) pointInserted = workMemTable.insert(insertRowNode); } } finally { - reconcileAlignedTVListRamCost(alignedRamCostSnapshot, memIncrements[0]); + reconcileAlignedTVListRamCost(alignedRamCostSnapshot, memIncrements.arrayBytes); } // Update start time of this memtable @@ -420,7 +419,7 @@ public void insertRows(InsertRowsNode insertRowsNode, long[] infoForMetrics) ensureMemTable(infoForMetrics); workMemTable.checkDataType(insertRowsNode); - long[] memIncrements; + WriteMemoryDelta memIncrements; long alignedMemTableIncrement = 0; Set alignedDeviceIds = new HashSet<>(); for (InsertRowNode insertRowNode : insertRowsNode.getInsertRowNodeList()) { @@ -444,24 +443,24 @@ public void insertRows(InsertRowsNode insertRowsNode, long[] infoForMetrics) nonAlignedList.add(insertRowNode); } } - long[] alignedMemIncrements = checkAlignedMemCostAndAddToTspInfoForRows(alignedList); - alignedMemTableIncrement = alignedMemIncrements[0]; - final long[] nonAlignedMemIncrements; + WriteMemoryDelta alignedMemIncrements = + checkAlignedMemCostAndAddToTspInfoForRows(alignedList); + alignedMemTableIncrement = alignedMemIncrements.arrayBytes; + final WriteMemoryDelta nonAlignedMemIncrements; try { nonAlignedMemIncrements = checkMemCostAndAddToTspInfoForRows(nonAlignedList); } catch (final WriteProcessException e) { rollbackMemoryInfoIfNeeded(alignedMemIncrements); throw e; } - memIncrements = new long[3]; - for (int i = 0; i < 3; i++) { - memIncrements[i] = alignedMemIncrements[i] + nonAlignedMemIncrements[i]; - } + memIncrements = new WriteMemoryDelta(); + memIncrements.add(alignedMemIncrements); + memIncrements.add(nonAlignedMemIncrements); } else { if (insertRowsNode.isAligned()) { memIncrements = checkAlignedMemCostAndAddToTspInfoForRows(insertRowsNode.getInsertRowNodeList()); - alignedMemTableIncrement = memIncrements[0]; + alignedMemTableIncrement = memIncrements.arrayBytes; } else { memIncrements = checkMemCostAndAddToTspInfoForRows(insertRowsNode.getInsertRowNodeList()); } @@ -544,6 +543,7 @@ private void createNewWorkingMemTable() { .getAvailableMemTable( dataRegionInfo.getDataRegion().getDatabaseName(), dataRegionInfo.getDataRegion().getDataRegionIdString()); + workMemTable.setWriteMemoryManager(dataRegionInfo.getWriteMemoryManager()); walNode.onMemTableCreated(workMemTable, tsFileResource.getTsFilePath()); } @@ -557,7 +557,7 @@ private static List>> splitTabletByDevice( return deviceEndOffsetPairsList; } - private long[] scheduleMemoryBlock( + private WriteMemoryDelta scheduleMemoryBlock( InsertTabletNode insertTabletNode, List rangeList, List>> deviceEndOffsetPairsList, @@ -565,18 +565,16 @@ private long[] scheduleMemoryBlock( long[] infoForMetrics) throws WriteProcessException { long memControlStartTime = System.nanoTime(); - long[] totalMemIncrements = new long[NUM_MEM_TO_ESTIMATE]; + WriteMemoryDelta totalMemIncrements = new WriteMemoryDelta(); try { for (int rangeIndex = 0; rangeIndex < rangeList.size(); ++rangeIndex) { int[] range = rangeList.get(rangeIndex); int start = range[0]; int end = range[1]; - long[] memIncrements = + WriteMemoryDelta memIncrements = checkMemCost( insertTabletNode, start, end, deviceEndOffsetPairsList.get(rangeIndex), results); - for (int i = 0; i < memIncrements.length; i++) { - totalMemIncrements[i] += memIncrements[i]; - } + totalMemIncrements.add(memIncrements); } return totalMemIncrements; } catch (final WriteProcessException e) { @@ -594,14 +592,14 @@ private long[] scheduleMemoryBlock( } } - private long[] checkMemCost( + private WriteMemoryDelta checkMemCost( InsertTabletNode insertTabletNode, int start, int end, List> deviceEndOffsetPairs, TSStatus[] results) throws WriteProcessException { - long[] memIncrements; + WriteMemoryDelta memIncrements; if (insertTabletNode.isAligned()) { memIncrements = checkAlignedMemCost(insertTabletNode, start, deviceEndOffsetPairs, results); } else { @@ -618,18 +616,18 @@ private long[] checkMemCost( return memIncrements; } - private long[] checkAlignedMemCost( + private WriteMemoryDelta checkAlignedMemCost( InsertTabletNode insertTabletNode, int start, List> deviceEndPosList, TSStatus[] results) throws WriteProcessException { - long[] memIncrements = new long[NUM_MEM_TO_ESTIMATE]; + WriteMemoryDelta memIncrements = new WriteMemoryDelta(); int splitStart = start; for (Pair iDeviceIDIntegerPair : deviceEndPosList) { int splitEnd = iDeviceIDIntegerPair.getRight(); IDeviceID deviceID = iDeviceIDIntegerPair.getLeft(); - final long[] splitMemIncrements; + final WriteMemoryDelta splitMemIncrements; try { splitMemIncrements = checkAlignedMemCostAndAddToTspForTablet( @@ -646,9 +644,7 @@ private long[] checkAlignedMemCost( rollbackMemoryInfoIfNeeded(memIncrements); throw e; } - for (int i = 0; i < NUM_MEM_TO_ESTIMATE; i++) { - memIncrements[i] += splitMemIncrements[i]; - } + memIncrements.add(splitMemIncrements); splitStart = splitEnd; } return memIncrements; @@ -679,7 +675,7 @@ public void insertTablet( takeAlignedTVListRamCostSnapshotFromDeviceRanges( workMemTable, insertTabletNode, deviceEndOffsetPairsList); - long[] memIncrements = + WriteMemoryDelta memIncrements = scheduleMemoryBlock( insertTabletNode, rangeList, deviceEndOffsetPairsList, results, infoForMetrics); @@ -786,7 +782,7 @@ public void insertTablet( } } } finally { - reconcileAlignedTVListRamCost(alignedRamCostSnapshot, memIncrements[0]); + reconcileAlignedTVListRamCost(alignedRamCostSnapshot, memIncrements.arrayBytes); } tsFileResource.updateProgressIndex(insertTabletNode.getProgressIndex()); @@ -797,7 +793,7 @@ public void insertTablet( } @SuppressWarnings("squid:S3776") // High Cognitive Complexity - private long[] checkMemCostAndAddToTspInfoForRow( + private WriteMemoryDelta checkMemCostAndAddToTspInfoForRow( IDeviceID deviceId, String[] measurements, TSDataType[] dataTypes, @@ -834,11 +830,11 @@ private long[] checkMemCostAndAddToTspInfoForRow( } } updateMemoryInfo(memTableIncrement, chunkMetadataIncrement, textDataIncrement); - return new long[] {memTableIncrement, textDataIncrement, chunkMetadataIncrement}; + return new WriteMemoryDelta(memTableIncrement, textDataIncrement, chunkMetadataIncrement); } @SuppressWarnings("squid:S3776") // High Cognitive Complexity - private long[] checkMemCostAndAddToTspInfoForRows(List insertRowNodeList) + private WriteMemoryDelta checkMemCostAndAddToTspInfoForRows(List insertRowNodeList) throws WriteProcessException { // Memory of increased PrimitiveArray and TEXT values, e.g., add a long[128], add 128*8 long memTableIncrement = 0L; @@ -889,11 +885,11 @@ private long[] checkMemCostAndAddToTspInfoForRows(List insertRowN } } updateMemoryInfo(memTableIncrement, chunkMetadataIncrement, textDataIncrement); - return new long[] {memTableIncrement, textDataIncrement, chunkMetadataIncrement}; + return new WriteMemoryDelta(memTableIncrement, textDataIncrement, chunkMetadataIncrement); } @SuppressWarnings("squid:S3776") // high Cognitive Complexity - private long[] checkAlignedMemCostAndAddToTspInfoForRow( + private WriteMemoryDelta checkAlignedMemCostAndAddToTspInfoForRow( IDeviceID deviceId, String[] measurements, TSDataType[] dataTypes, @@ -952,12 +948,12 @@ private long[] checkAlignedMemCostAndAddToTspInfoForRow( } } updateMemoryInfo(memTableIncrement, chunkMetadataIncrement, textDataIncrement); - return new long[] {memTableIncrement, textDataIncrement, chunkMetadataIncrement}; + return new WriteMemoryDelta(memTableIncrement, textDataIncrement, chunkMetadataIncrement); } @SuppressWarnings("squid:S3776") // high Cognitive Complexity - private long[] checkAlignedMemCostAndAddToTspInfoForRows(List insertRowNodeList) - throws WriteProcessException { + private WriteMemoryDelta checkAlignedMemCostAndAddToTspInfoForRows( + List insertRowNodeList) throws WriteProcessException { // Fixed-size TVList structures and materialized value primitive arrays. long memTableIncrement = 0L; // Variable-length payload retained by Binary values. @@ -1069,10 +1065,10 @@ private long[] checkAlignedMemCostAndAddToTspInfoForRows(List ins } } updateMemoryInfo(memTableIncrement, chunkMetadataIncrement, textDataIncrement); - return new long[] {memTableIncrement, textDataIncrement, chunkMetadataIncrement}; + return new WriteMemoryDelta(memTableIncrement, textDataIncrement, chunkMetadataIncrement); } - private long[] checkMemCostAndAddToTspInfoForTablet( + private WriteMemoryDelta checkMemCostAndAddToTspInfoForTablet( IDeviceID deviceId, String[] measurements, TSDataType[] dataTypes, @@ -1082,9 +1078,9 @@ private long[] checkMemCostAndAddToTspInfoForTablet( int end) throws WriteProcessException { if (start >= end) { - return new long[] {0, 0, 0}; + return new WriteMemoryDelta(); } - long[] memIncrements = new long[3]; // memTable, text, chunk metadata + WriteMemoryDelta memIncrements = new WriteMemoryDelta(); for (int i = 0; dataTypes != null && i < dataTypes.length; i++) { // Skip failed Measurements @@ -1093,14 +1089,14 @@ private long[] checkMemCostAndAddToTspInfoForTablet( } updateMemCost(dataTypes[i], measurements[i], deviceId, start, end, memIncrements, columns[i]); } - long memTableIncrement = memIncrements[0]; - long textDataIncrement = memIncrements[1]; - long chunkMetadataIncrement = memIncrements[2]; + long memTableIncrement = memIncrements.arrayBytes; + long textDataIncrement = memIncrements.binaryBytes; + long chunkMetadataIncrement = memIncrements.metadataBytes; updateMemoryInfo(memTableIncrement, chunkMetadataIncrement, textDataIncrement); return memIncrements; } - private long[] checkAlignedMemCostAndAddToTspForTablet( + private WriteMemoryDelta checkAlignedMemCostAndAddToTspForTablet( IDeviceID deviceId, String[] measurements, TSDataType[] dataTypes, @@ -1112,9 +1108,9 @@ private long[] checkAlignedMemCostAndAddToTspForTablet( TSStatus[] results) throws WriteProcessException { if (start >= end) { - return new long[] {0, 0, 0}; + return new WriteMemoryDelta(); } - long[] memIncrements = new long[3]; // memTable, text, chunk metadata + WriteMemoryDelta memIncrements = new WriteMemoryDelta(); updateAlignedMemCost( dataTypes, @@ -1127,9 +1123,9 @@ private long[] checkAlignedMemCostAndAddToTspForTablet( bitMaps, columnCategories, results); - long memTableIncrement = memIncrements[0]; - long textDataIncrement = memIncrements[1]; - long chunkMetadataIncrement = memIncrements[2]; + long memTableIncrement = memIncrements.arrayBytes; + long textDataIncrement = memIncrements.binaryBytes; + long chunkMetadataIncrement = memIncrements.metadataBytes; updateMemoryInfo(memTableIncrement, chunkMetadataIncrement, textDataIncrement); return memIncrements; } @@ -1140,23 +1136,24 @@ private void updateMemCost( IDeviceID deviceId, int start, int end, - long[] memIncrements, + WriteMemoryDelta memIncrements, Object column) { - // memIncrements[0]: fixed-size TVList structures and materialized value primitive arrays. - // memIncrements[1]: variable-length payload retained by Binary values. - // memIncrements[2]: ChunkMetadata entries for newly introduced writable fields. + // memIncrements.arrayBytes: fixed-size TVList structures and materialized value primitive + // arrays. + // memIncrements.binaryBytes: variable-length payload retained by Binary values. + // memIncrements.metadataBytes: ChunkMetadata entries for newly introduced writable fields. IWritableMemChunk memChunk = workMemTable.getWritableMemChunk(deviceId, measurement); if (memChunk == null) { // ChunkMetadataIncrement - memIncrements[2] += ChunkMetadata.calculateRamSize(measurement, dataType); - memIncrements[0] += + memIncrements.metadataBytes += ChunkMetadata.calculateRamSize(measurement, dataType); + memIncrements.arrayBytes += ((end - start) / PrimitiveArrayManager.ARRAY_SIZE + 1) * TVList.tvListArrayMemCost(dataType); } else { long currentChunkPointNum = memChunk.rowCount(); if (currentChunkPointNum % PrimitiveArrayManager.ARRAY_SIZE == 0) { - memIncrements[0] += + memIncrements.arrayBytes += ((end - start) / PrimitiveArrayManager.ARRAY_SIZE + 1) * memChunk.getWorkingTVList().tvListArrayMemCost(); } else { @@ -1164,14 +1161,15 @@ private void updateMemCost( (end - start - 1 + (currentChunkPointNum % PrimitiveArrayManager.ARRAY_SIZE)) / PrimitiveArrayManager.ARRAY_SIZE; if (acquireArray != 0) { - memIncrements[0] += acquireArray * memChunk.getWorkingTVList().tvListArrayMemCost(); + memIncrements.arrayBytes += + acquireArray * memChunk.getWorkingTVList().tvListArrayMemCost(); } } } // TEXT data size if (dataType.isBinary()) { Binary[] binColumn = (Binary[]) column; - memIncrements[1] += MemUtils.getBinaryColumnSize(binColumn, start, end, null); + memIncrements.binaryBytes += MemUtils.getBinaryColumnSize(binColumn, start, end, null); } } @@ -1181,7 +1179,7 @@ private void updateAlignedMemCost( String[] measurementIds, int start, int end, - long[] memIncrements, + WriteMemoryDelta memIncrements, Object[] columns, BitMap[] bitMaps, TsTableColumnCategory[] columnCategories, @@ -1191,12 +1189,11 @@ private void updateAlignedMemCost( TSDataType[] writableFieldDataTypes = getWritableFieldDataTypes(measurementIds, dataTypes, columns, columnCategories); - // memIncrements = [memTable, text, chunk metadata] respectively IWritableMemChunk memChunk = workMemTable.getWritableMemChunk(deviceId, AlignedPath.VECTOR_PLACEHOLDER); if (memChunk == null) { // A new aligned device creates one value ChunkMetadata entry for each writable field. - memIncrements[2] += + memIncrements.metadataBytes += writableFieldDataTypes.length * ChunkMetadata.calculateRamSize(AlignedPath.VECTOR_PLACEHOLDER, TSDataType.VECTOR); @@ -1205,13 +1202,13 @@ private void updateAlignedMemCost( + (incomingPointNum % PrimitiveArrayManager.ARRAY_SIZE > 0 ? 1 : 0); // Each new block allocates its timestamp array. Value arrays and their references are // excluded here because all-null and failed-only column segments leave them unmaterialized. - memIncrements[0] += + memIncrements.arrayBytes += numArraysToAdd * AlignedTVList.alignedTvListArrayMemCostWithoutPrimitiveArrays( writableFieldDataTypes, null); // Add the payload and header of a value primitive array only for a column/block segment that // contains at least one successful non-null value. - memIncrements[0] += + memIncrements.arrayBytes += calculateTabletValueArrayMemCost( null, measurementIds, @@ -1239,7 +1236,7 @@ private void updateAlignedMemCost( if (acquireArray != 0) { // Each acquired block adds a timestamp array and optional sort-index array. Value arrays // and their references are charged separately below. - memIncrements[0] += + memIncrements.arrayBytes += acquireArray * alignedMemChunk .getWorkingTVList() @@ -1247,7 +1244,7 @@ private void updateAlignedMemCost( } // Charge the payload and header of each value primitive array that this tablet actually // materializes. Arrays already present in the working TVList are not charged again. - memIncrements[0] += + memIncrements.arrayBytes += calculateTabletValueArrayMemCost( alignedMemChunk, measurementIds, @@ -1271,7 +1268,7 @@ private void updateAlignedMemCost( Binary[] binColumn = (Binary[]) columns[i]; // Binary values keep their variable-length content outside the fixed Binary[] arrays. // Failed rows are excluded because their values are not inserted into the memtable. - memIncrements[1] += MemUtils.getBinaryColumnSize(binColumn, start, end, results); + memIncrements.binaryBytes += MemUtils.getBinaryColumnSize(binColumn, start, end, results); } } } @@ -1404,14 +1401,22 @@ private void reconcileAlignedTVListRamCost( return; } - long correction = snapshot.getMemoryCorrection(estimatedMemTableIncrement); + long totalCorrection = snapshot.getMemoryCorrection(estimatedMemTableIncrement); + long structureIncrement = snapshot.structureRamCost - snapshot.structureRamCostBeforeWrite; + WriteMemoryBlock structureBlock = snapshot.memTable.getStructureMemoryBlock(); + if (structureIncrement > 0) { + structureBlock.add(structureIncrement); + } else if (structureIncrement < 0) { + structureBlock.release(-structureIncrement); + } + long correction = totalCorrection - structureIncrement; if (correction > 0) { - dataRegionInfo.addStorageGroupMemCost(correction); snapshot.memTable.addTVListRamCost(correction); } else if (correction < 0) { long releasedMemory = -correction; - dataRegionInfo.releaseStorageGroupMemCost(releasedMemory); snapshot.memTable.releaseTVListRamCost(releasedMemory); + } + if (totalCorrection < 0) { SystemInfo.getInstance().resetStorageGroupStatus(dataRegionInfo); } } @@ -1478,12 +1483,15 @@ static final class AlignedTVListRamCostSnapshot { private final IDeviceID deviceId; private final Set deviceIds; private final long ramCostBeforeWrite; + private final long structureRamCostBeforeWrite; + private long structureRamCost; AlignedTVListRamCostSnapshot(IMemTable memTable, IDeviceID deviceId) { this.memTable = memTable; this.deviceId = deviceId; this.deviceIds = null; this.ramCostBeforeWrite = getRamCost(memTable, deviceId); + this.structureRamCostBeforeWrite = structureRamCost; } AlignedTVListRamCostSnapshot(IMemTable memTable, Set deviceIds) { @@ -1491,15 +1499,17 @@ static final class AlignedTVListRamCostSnapshot { this.deviceId = null; this.deviceIds = deviceIds; this.ramCostBeforeWrite = getRamCost(memTable, deviceIds); + this.structureRamCostBeforeWrite = structureRamCost; } long getMemoryCorrection(long estimatedMemTableIncrement) { + structureRamCost = 0; return (deviceId == null ? getRamCost(memTable, deviceIds) : getRamCost(memTable, deviceId)) - ramCostBeforeWrite - estimatedMemTableIncrement; } - private static long getRamCost(IMemTable memTable, Set deviceIds) { + private long getRamCost(IMemTable memTable, Set deviceIds) { long ramCost = 0; for (IDeviceID currentDeviceId : deviceIds) { ramCost += getRamCost(memTable, currentDeviceId); @@ -1507,7 +1517,7 @@ private static long getRamCost(IMemTable memTable, Set deviceIds) { return ramCost; } - private static long getRamCost(IMemTable memTable, IDeviceID deviceId) { + private long getRamCost(IMemTable memTable, IDeviceID deviceId) { IWritableMemChunk memChunk = memTable.getWritableMemChunk(deviceId, AlignedPath.VECTOR_PLACEHOLDER); if (!(memChunk instanceof AlignedWritableMemChunk)) { @@ -1515,72 +1525,116 @@ private static long getRamCost(IMemTable memTable, IDeviceID deviceId) { } AlignedWritableMemChunk alignedMemChunk = (AlignedWritableMemChunk) memChunk; - long ramCost = alignedMemChunk.getWorkingTVList().getRamSize(); + long ramCost = getRamCost(alignedMemChunk.getWorkingTVList()); for (AlignedTVList sortedTVList : alignedMemChunk.getSortedList()) { - ramCost += sortedTVList.getRamSize(); + ramCost += getRamCost(sortedTVList); } return ramCost; } + + private long getRamCost(AlignedTVList list) { + synchronized (list) { + long structureBytes = list.getStructureRamSize(); + structureRamCost += structureBytes; + return list.getArrayRamSize() + structureBytes; + } + } } private void updateMemoryInfo( long memTableIncrement, long chunkMetadataIncrement, long textDataIncrement) throws WriteProcessRejectException { - memTableIncrement += textDataIncrement; - dataRegionInfo.addStorageGroupMemCost(memTableIncrement); - tsFileProcessorInfo.addTSPMemCost(chunkMetadataIncrement); - if (dataRegionInfo.needToReportToSystem()) { - try { - if (!SystemInfo.getInstance().reportStorageGroupStatus(dataRegionInfo, this)) { - long startTime = System.currentTimeMillis(); - while (SystemInfo.getInstance().isRejected()) { - if (workMemTable.shouldFlush()) { - break; - } - try { - TimeUnit.MILLISECONDS.sleep(config.getCheckPeriodWhenInsertBlocked()); - if (System.currentTimeMillis() - startTime - > config.getMaxWaitingTimeWhenInsertBlocked()) { - throw new WriteProcessRejectException( - String.format( - StorageEngineMessages.STORAGE_EXCEPTION_SYSTEM_REJECTED_OVER_SMS_94CEF932, - (System.currentTimeMillis() - startTime))); - } - } catch (InterruptedException e) { - Thread.currentThread().interrupt(); - } + // Reserve before reporting, but expose the new MemTable bytes to flush selection only + // after admission, preserving the previous pre-report/post-report accounting order. + // Most rows reuse existing arrays and metadata. Avoid reservations and atomic updates + // for these zero increments, but still check whether the region needs to report. + try (WriteMemoryBlock.Reservation arrays = + memTableIncrement == 0 + ? null + : workMemTable.getArrayMemoryBlock().reserve(memTableIncrement); + WriteMemoryBlock.Reservation binary = + textDataIncrement == 0 + ? null + : workMemTable.getBinaryMemoryBlock().reserve(textDataIncrement); + WriteMemoryBlock.Reservation metadata = + chunkMetadataIncrement == 0 + ? null + : tsFileProcessorInfo.getMemoryBlock().reserve(chunkMetadataIncrement)) { + if (dataRegionInfo.needToReportToSystem()) { + checkMemoryPressure(); + } + if (arrays != null) { + arrays.commit(); + } + if (binary != null) { + binary.commit(); + } + if (metadata != null) { + metadata.commit(); + } + workMemTable.addTextDataSize(textDataIncrement); + } catch (WriteProcessRejectException e) { + // try-with-resources has already rolled back every reservation before re-reporting. + SystemInfo.getInstance().resetStorageGroupStatus(dataRegionInfo); + throw e; + } + } + + private void checkMemoryPressure() throws WriteProcessRejectException { + if (!SystemInfo.getInstance().reportStorageGroupStatus(dataRegionInfo, this)) { + long startTime = System.currentTimeMillis(); + while (SystemInfo.getInstance().isRejected()) { + if (workMemTable.shouldFlush()) { + break; + } + try { + TimeUnit.MILLISECONDS.sleep(config.getCheckPeriodWhenInsertBlocked()); + if (System.currentTimeMillis() - startTime + > config.getMaxWaitingTimeWhenInsertBlocked()) { + throw new WriteProcessRejectException( + String.format( + StorageEngineMessages.STORAGE_EXCEPTION_SYSTEM_REJECTED_OVER_SMS_94CEF932, + (System.currentTimeMillis() - startTime))); } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); } - } catch (WriteProcessRejectException e) { - dataRegionInfo.releaseStorageGroupMemCost(memTableIncrement); - tsFileProcessorInfo.releaseTSPMemCost(chunkMetadataIncrement); - SystemInfo.getInstance().resetStorageGroupStatus(dataRegionInfo); - throw e; } } - workMemTable.addTVListRamCost(memTableIncrement); - workMemTable.addTextDataSize(textDataIncrement); } - private void rollbackMemoryInfo(long[] memIncrements) { - long memTableIncrement = memIncrements[0]; - long textDataIncrement = memIncrements[1]; - long chunkMetadataIncrement = memIncrements[2]; - - memTableIncrement += textDataIncrement; - dataRegionInfo.releaseStorageGroupMemCost(memTableIncrement); - tsFileProcessorInfo.releaseTSPMemCost(chunkMetadataIncrement); + private void rollbackMemoryInfo(WriteMemoryDelta delta) { + workMemTable.getArrayMemoryBlock().release(delta.arrayBytes); + workMemTable.getBinaryMemoryBlock().release(delta.binaryBytes); + tsFileProcessorInfo.releaseTSPMemCost(delta.metadataBytes); + workMemTable.releaseTextDataSize(delta.binaryBytes); SystemInfo.getInstance().resetStorageGroupStatus(dataRegionInfo); - workMemTable.releaseTVListRamCost(memTableIncrement); - workMemTable.releaseTextDataSize(textDataIncrement); } - private void rollbackMemoryInfoIfNeeded(final long[] memIncrements) { - for (final long memIncrement : memIncrements) { - if (memIncrement != 0) { - rollbackMemoryInfo(memIncrements); - return; - } + private void rollbackMemoryInfoIfNeeded(final WriteMemoryDelta delta) { + if (delta.arrayBytes != 0 || delta.binaryBytes != 0 || delta.metadataBytes != 0) { + rollbackMemoryInfo(delta); + } + } + + /** Named estimates shared by row, mixed-row and split-tablet admission/rollback paths. */ + private static final class WriteMemoryDelta { + private long arrayBytes; + private long binaryBytes; + private long metadataBytes; + + private WriteMemoryDelta() {} + + private WriteMemoryDelta(long arrayBytes, long binaryBytes, long metadataBytes) { + this.arrayBytes = arrayBytes; + this.binaryBytes = binaryBytes; + this.metadataBytes = metadataBytes; + } + + private void add(WriteMemoryDelta delta) { + arrayBytes += delta.arrayBytes; + binaryBytes += delta.binaryBytes; + metadataBytes += delta.metadataBytes; } } @@ -1862,10 +1916,9 @@ private void releaseFlushedMemTable(IMemTable memTable) { memTable.isSignalMemTable(), flushingMemTables.size()); } + long releasedMemTableCost = memTable.getTVListsRamCost(); memTable.release(); MemTableManager.getInstance().decreaseMemtableNumber(); - // Reset the mem cost in StorageGroupProcessorInfo - dataRegionInfo.releaseStorageGroupMemCost(memTable.getTVListsRamCost()); if (logger.isDebugEnabled()) { logger.debug( StorageEngineMessages @@ -1876,7 +1929,7 @@ private void releaseFlushedMemTable(IMemTable memTable) { } // Report to System SystemInfo.getInstance().resetStorageGroupStatus(dataRegionInfo); - SystemInfo.getInstance().resetFlushingMemTableCost(memTable.getTVListsRamCost()); + SystemInfo.getInstance().resetFlushingMemTableCost(releasedMemTableCost); if (logger.isDebugEnabled()) { logger.debug( StorageEngineMessages @@ -2715,7 +2768,6 @@ public void setTimeRangeId(long timeRangeId) { public void putMemTableBackAndClose() throws TsFileProcessorException { if (workMemTable != null) { workMemTable.release(); - dataRegionInfo.releaseStorageGroupMemCost(workMemTable.getTVListsRamCost()); workMemTable = null; } try { diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessorInfo.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessorInfo.java index b06ed63cba025..b243605164990 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessorInfo.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessorInfo.java @@ -20,21 +20,25 @@ import org.apache.iotdb.commons.service.metric.MetricService; import org.apache.iotdb.db.storageengine.dataregion.DataRegionInfo; +import org.apache.iotdb.db.storageengine.rescon.memory.WriteMemoryBlock; /** The TsFileProcessorInfo records the memory cost of this TsFileProcessor. */ public class TsFileProcessorInfo { - /** Once tspInfo updated, report to storageGroupInfo that this TSP belongs to. */ - private final DataRegionInfo dataRegionInfo; - /** memory occupation of unsealed TsFileResource, ChunkMetadata, WAL */ - private long memCost; + private final WriteMemoryBlock memoryBlock; private final TsFileProcessorInfoMetrics metrics; public TsFileProcessorInfo(DataRegionInfo dataRegionInfo) { - this.dataRegionInfo = dataRegionInfo; - this.memCost = 0L; + this(dataRegionInfo, "TsFileMetadata"); + } + + public TsFileProcessorInfo(DataRegionInfo dataRegionInfo, String tsFilePath) { + this.memoryBlock = + dataRegionInfo + .getWriteMemoryManager() + .createBlock(tsFilePath, WriteMemoryBlock.Category.TSFILE_METADATA); this.metrics = new TsFileProcessorInfoMetrics(dataRegionInfo.getDataRegion().getDatabaseName(), this); MetricService.getInstance().addMetricSet(metrics); @@ -42,25 +46,26 @@ public TsFileProcessorInfo(DataRegionInfo dataRegionInfo) { /** called in each insert */ public void addTSPMemCost(long cost) { - memCost += cost; - dataRegionInfo.addStorageGroupMemCost(cost); + memoryBlock.add(cost); } /** called when meet exception */ public void releaseTSPMemCost(long cost) { - dataRegionInfo.releaseStorageGroupMemCost(cost); - memCost -= cost; + memoryBlock.release(cost); } /** called when closing TSP */ public void clear() { - dataRegionInfo.releaseStorageGroupMemCost(memCost); - memCost = 0L; + memoryBlock.close(); MetricService.getInstance().removeMetricSet(metrics); } + public WriteMemoryBlock getMemoryBlock() { + return memoryBlock; + } + /** get memCost */ public long getMemCost() { - return memCost; + return memoryBlock.getAccountedMemoryInBytes(); } } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/utils/MemoryControlledWALEntryQueue.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/utils/MemoryControlledWALEntryQueue.java index 0974aef999a3b..1bc5915aca70c 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/utils/MemoryControlledWALEntryQueue.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/utils/MemoryControlledWALEntryQueue.java @@ -31,7 +31,7 @@ import static org.apache.iotdb.rpc.TSStatusCode.WAL_ENTRY_TOO_LARGE; public class MemoryControlledWALEntryQueue { - private final BlockingQueue queue; + private final BlockingQueue queue; private static final Object nonFullCondition = new Object(); public MemoryControlledWALEntryQueue() { @@ -39,14 +39,14 @@ public MemoryControlledWALEntryQueue() { } public WALEntry poll(long timeout, TimeUnit unit) throws InterruptedException { - WALEntry e = queue.poll(timeout, unit); - if (e != null) { - SystemInfo.getInstance().getWalBufferQueueMemoryBlock().release(getElementSize(e)); + ChargedEntry entry = queue.poll(timeout, unit); + if (entry != null) { + SystemInfo.getInstance().getWalBufferQueueMemoryBlock().release(entry.bytes); synchronized (nonFullCondition) { nonFullCondition.notifyAll(); } } - return e; + return entry == null ? null : entry.entry; } public void put(WALEntry e) throws InterruptedException { @@ -68,16 +68,27 @@ public void put(WALEntry e) throws InterruptedException { nonFullCondition.wait(); } } - queue.put(e); + boolean enqueued = false; + try { + queue.put(new ChargedEntry(e, elementSize)); + enqueued = true; + } finally { + if (!enqueued) { + SystemInfo.getInstance().getWalBufferQueueMemoryBlock().release(elementSize); + synchronized (nonFullCondition) { + nonFullCondition.notifyAll(); + } + } + } } public WALEntry take() throws InterruptedException { - WALEntry e = queue.take(); - SystemInfo.getInstance().getWalBufferQueueMemoryBlock().release(getElementSize(e)); + ChargedEntry entry = queue.take(); + SystemInfo.getInstance().getWalBufferQueueMemoryBlock().release(entry.bytes); synchronized (nonFullCondition) { nonFullCondition.notifyAll(); } - return e; + return entry.entry; } public int size() { @@ -88,6 +99,16 @@ public boolean isEmpty() { return queue.isEmpty(); } + private static final class ChargedEntry { + private final WALEntry entry; + private final long bytes; + + private ChargedEntry(WALEntry entry, long bytes) { + this.entry = entry; + this.bytes = bytes; + } + } + private long getElementSize(WALEntry walEntry) { return walEntry.getMemorySize(); } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/rescon/memory/PrimitiveArrayManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/rescon/memory/PrimitiveArrayManager.java index f46fd6de554b5..90d7424ce19ef 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/rescon/memory/PrimitiveArrayManager.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/rescon/memory/PrimitiveArrayManager.java @@ -31,6 +31,7 @@ import org.apache.tsfile.enums.TSDataType; import org.apache.tsfile.read.common.type.Type; import org.apache.tsfile.utils.Binary; +import org.apache.tsfile.utils.RamUsageEstimator; import org.apache.tsfile.write.UnSupportedDataTypeException; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -40,6 +41,7 @@ import java.util.Arrays; import java.util.concurrent.atomic.AtomicLong; import java.util.function.IntFunction; +import java.util.function.ToLongFunction; /** Manage all primitive data lists in memory, including get and release operations. */ public class PrimitiveArrayManager { @@ -65,7 +67,8 @@ public class PrimitiveArrayManager { private static final IMemoryBlock POOLED_ARRAYS_MEMORY_BLOCK = MEMORY_CONFIG .getBufferedArraysMemoryManager() - .exactAllocate("BufferedArrays", MemoryBlockType.DYNAMIC); + .exactAllocateReadOnly( + "BufferedArrays", MemoryBlockType.DYNAMIC, () -> getPooledArrayMemoryInBytes(0)); /** threshold total size of arrays for all data types */ private static final double POOLED_ARRAYS_MEMORY_THRESHOLD = @@ -74,6 +77,31 @@ public class PrimitiveArrayManager { /** TSDataType#serialize() -> ArrayDeque, VECTOR and UNKNOWN are ignored */ private static final ArrayDeque[] POOLED_ARRAYS = new ArrayDeque[TSDataType.values().length]; + // Indexed by TSDataType#serialize(), like POOLED_ARRAYS. Resolve the array type once at startup. + private static final ToLongFunction[] ARRAY_SIZE_CALCULATORS = + createArraySizeCalculators(); + + @SuppressWarnings("unchecked") + private static ToLongFunction[] createArraySizeCalculators() { + ToLongFunction[] calculators = new ToLongFunction[TSDataType.values().length]; + calculators[TSDataType.BOOLEAN.serialize()] = a -> RamUsageEstimator.sizeOf((boolean[]) a); + calculators[TSDataType.INT32.serialize()] = a -> RamUsageEstimator.sizeOf((int[]) a); + calculators[TSDataType.DATE.serialize()] = calculators[TSDataType.INT32.serialize()]; + calculators[TSDataType.INT64.serialize()] = a -> RamUsageEstimator.sizeOf((long[]) a); + calculators[TSDataType.TIMESTAMP.serialize()] = calculators[TSDataType.INT64.serialize()]; + calculators[TSDataType.FLOAT.serialize()] = a -> RamUsageEstimator.sizeOf((float[]) a); + calculators[TSDataType.DOUBLE.serialize()] = a -> RamUsageEstimator.sizeOf((double[]) a); + // Binary[] entries are cleared on return; only the shallow reference array belongs to the pool. + calculators[TSDataType.TEXT.serialize()] = a -> RamUsageEstimator.shallowSizeOf((Binary[]) a); + calculators[TSDataType.STRING.serialize()] = calculators[TSDataType.TEXT.serialize()]; + calculators[TSDataType.BLOB.serialize()] = calculators[TSDataType.TEXT.serialize()]; + calculators[TSDataType.OBJECT.serialize()] = calculators[TSDataType.TEXT.serialize()]; + return calculators; + } + + // Each entry is guarded by the corresponding queue's monitor. + private static final long[] POOLED_ARRAY_BYTES = new long[TSDataType.values().length]; + /** TSDataType#serialize() -> max size of ArrayDeque, VECTOR and UNKNOWN are ignored */ private static final int[] LIMITS = new int[TSDataType.values().length]; @@ -119,7 +147,14 @@ private static void init() { limitUpdateThreshold = (long) ((TSDataType.values().length - 2) * limit); for (int i = 0; i < POOLED_ARRAYS.length; ++i) { - POOLED_ARRAYS[i] = new ArrayDeque<>((int) limit); + if (POOLED_ARRAYS[i] == null) { + POOLED_ARRAYS[i] = new ArrayDeque<>((int) limit); + } else { + synchronized (POOLED_ARRAYS[i]) { + POOLED_ARRAYS[i].clear(); + POOLED_ARRAY_BYTES[i] = 0; + } + } } for (AtomicLong allocationRequestCount : ALLOCATION_REQUEST_COUNTS) { @@ -159,6 +194,9 @@ public static Object allocate(TSDataType dataType) { Object array; synchronized (POOLED_ARRAYS[order]) { array = POOLED_ARRAYS[order].poll(); + if (array != null) { + POOLED_ARRAY_BYTES[order] -= ARRAY_SIZE_CALCULATORS[order].applyAsLong(array); + } } StorageEngineMemoryMetrics.getInstance().incPamAllocation(); if (array == null) { @@ -268,12 +306,28 @@ public static void release(Object array) { ArrayDeque arrays = POOLED_ARRAYS[order]; if (arrays.size() < LIMITS[order]) { arrays.add(array); + POOLED_ARRAY_BYTES[order] += ARRAY_SIZE_CALCULATORS[order].applyAsLong(array); } else { StorageEngineMemoryMetrics.getInstance().incPamReleaseFailure(); } } } + public static IMemoryBlock getMemoryBlock() { + return POOLED_ARRAYS_MEMORY_BLOCK; + } + + private static long getPooledArrayMemoryInBytes(int order) { + if (order == POOLED_ARRAYS.length) { + return 0; + } + // Hold every queue lock in ascending order for a linearizable total. Borrow/return paths + // hold only one queue lock and never call this method or the memory manager. + synchronized (POOLED_ARRAYS[order]) { + return POOLED_ARRAY_BYTES[order] + getPooledArrayMemoryInBytes(order + 1); + } + } + public static void close() { init(); } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/rescon/memory/SystemInfo.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/rescon/memory/SystemInfo.java index 5491f113ba7e4..8373c53b975d3 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/rescon/memory/SystemInfo.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/rescon/memory/SystemInfo.java @@ -58,6 +58,8 @@ public class SystemInfo { private volatile boolean rejected = false; private long memorySizeForMemtable; + // Flush reservations reduce the MemTable budget; they must not also enter DataRegion usage. + private final WriteMemoryManager flushMemoryManager = new WriteMemoryManager("FlushTasks"); private final Map reportedStorageGroupMemCostMap = new HashMap<>(); private long flushingMemTablesCost = 0L; @@ -369,8 +371,10 @@ public long getMemorySizeForCompaction() { return compactionMemoryBlock.getTotalMemorySizeInBytes(); } - public void loadWriteMemory() { - memorySizeForMemtable = memoryConfig.getMemtableMemoryManager().getTotalMemorySizeInBytes(); + public synchronized void loadWriteMemory() { + memorySizeForMemtable = + memoryConfig.getMemtableMemoryManager().getTotalMemorySizeInBytes() + - flushMemoryManager.getMemoryUsageInBytes(); FLUSH_THRESHOLD = memorySizeForMemtable * config.getFlushProportion(); REJECT_THRESHOLD = memorySizeForMemtable * memoryConfig.getRejectProportion(); WritingMetrics.getInstance().recordFlushThreshold(FLUSH_THRESHOLD); @@ -487,24 +491,39 @@ private InstanceHolder() {} private static final SystemInfo instance = new SystemInfo(); } - public synchronized void applyTemporaryMemoryForFlushing(long estimatedTemporaryMemSize) { - memorySizeForMemtable -= estimatedTemporaryMemSize; - FLUSH_THRESHOLD = memorySizeForMemtable * config.getFlushProportion(); - REJECT_THRESHOLD = memorySizeForMemtable * memoryConfig.getRejectProportion(); - WritingMetrics.getInstance().recordFlushThreshold(FLUSH_THRESHOLD); - WritingMetrics.getInstance().recordRejectThreshold(REJECT_THRESHOLD); - WritingMetrics.getInstance() - .recordWALQueueMaxMemorySize(walBufferQueueMemoryBlock.getTotalMemorySizeInBytes()); + public synchronized FlushMemoryReservation reserveTemporaryMemoryForFlushing( + String name, long estimatedBytes) { + WriteMemoryBlock block = + flushMemoryManager.createBlock(name, WriteMemoryBlock.Category.FLUSH_TEMPORARY); + try { + block.add(estimatedBytes); + loadWriteMemory(); + return new FlushMemoryReservation(block); + } catch (RuntimeException e) { + block.close(); + loadWriteMemory(); + throw e; + } } - public synchronized void releaseTemporaryMemoryForFlushing(long estimatedTemporaryMemSize) { - memorySizeForMemtable += estimatedTemporaryMemSize; - FLUSH_THRESHOLD = memorySizeForMemtable * config.getFlushProportion(); - REJECT_THRESHOLD = memorySizeForMemtable * memoryConfig.getRejectProportion(); - WritingMetrics.getInstance().recordFlushThreshold(FLUSH_THRESHOLD); - WritingMetrics.getInstance().recordRejectThreshold(REJECT_THRESHOLD); - WritingMetrics.getInstance() - .recordWALQueueMaxMemorySize(walBufferQueueMemoryBlock.getTotalMemorySizeInBytes()); + public WriteMemoryManager getFlushMemoryManager() { + return flushMemoryManager; + } + + public final class FlushMemoryReservation implements AutoCloseable { + private final WriteMemoryBlock block; + + private FlushMemoryReservation(WriteMemoryBlock block) { + this.block = block; + } + + @Override + public void close() { + synchronized (SystemInfo.this) { + block.close(); + loadWriteMemory(); + } + } } public long getTotalMemTableSize() { diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/rescon/memory/WriteMemoryBlock.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/rescon/memory/WriteMemoryBlock.java new file mode 100644 index 0000000000000..495da430cc047 --- /dev/null +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/rescon/memory/WriteMemoryBlock.java @@ -0,0 +1,218 @@ +/* + * 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.storageengine.rescon.memory; + +import org.apache.iotdb.db.i18n.StorageEngineMessages; + +import java.util.concurrent.atomic.AtomicLong; + +/** + * A named owner's estimated write-memory usage. All changes propagate to its accounting scope. The + * block does not allocate heap memory or impose a per-owner quota. + */ +public final class WriteMemoryBlock implements AutoCloseable { + public enum Category { + MEMTABLE_ARRAY, + MEMTABLE_STRUCTURE, + MEMTABLE_BINARY, + // WAL snapshots contain only a total; do not guess an array/Binary breakdown on recovery. + MEMTABLE_RECOVERED, + TSFILE_METADATA, + FLUSH_TEMPORARY + } + + private static final AtomicLong NEXT_ID = new AtomicLong(); + private final long id = NEXT_ID.incrementAndGet(); + private final Category category; + private final String name; + // Guarded by this block's monitor, just like reservations and ownership. + private long usedBytes; + private WriteMemoryManager manager; + private long reservedBytes; + private boolean closed; + + public WriteMemoryBlock(String name, Category category) { + this.category = category; + this.name = name; + } + + /** Standalone/recovery MemTables are attached only when owned by a live writer. */ + public synchronized void attachTo(WriteMemoryManager manager) { + checkOpen(); + if (this.manager == manager) { + return; + } + if (this.manager != null) { + throw new IllegalStateException( + String.format( + StorageEngineMessages + .EXCEPTION_WRITE_MEMORY_BLOCK_ARG_ALREADY_BELONGS_TO_ANOTHER_MANAGER_A05A8229, + getName())); + } + this.manager = manager; + manager.addMemory(getAccountedMemoryInBytes()); + manager.register(this); + } + + public synchronized void add(long bytes) { + checkOpen(); + checkNonNegative(bytes); + usedBytes += bytes; + + if (manager != null) { + manager.addMemory(bytes); + } + } + + public synchronized void release(long bytes) { + checkOpen(); + checkNonNegative(bytes); + if (bytes > usedBytes) { + throw new IllegalArgumentException( + String.format( + StorageEngineMessages + .EXCEPTION_CANNOT_RELEASE_ARG_BYTES_FROM_WRITE_MEMORY_BLOCK_ARG_WITH_ARG_BYTES_3783A614, + bytes, + getName(), + usedBytes)); + } + usedBytes -= bytes; + if (manager != null) { + manager.releaseMemory(bytes); + } + } + + /** A pending allocation counts towards rejection but is not yet flushable MemTable memory. */ + public synchronized Reservation reserve(long bytes) { + checkOpen(); + checkNonNegative(bytes); + reservedBytes += bytes; + if (manager != null) { + manager.addMemory(bytes); + } + return new Reservation(bytes); + } + + public synchronized long getUsedMemoryInBytes() { + return usedBytes; + } + + public synchronized long getReservedMemoryInBytes() { + return reservedBytes; + } + + public synchronized long getAccountedMemoryInBytes() { + return usedBytes + reservedBytes; + } + + /** Clears live accounting, for a MemTable that is explicitly cleared and reused. */ + public synchronized void clear() { + checkOpen(); + // Reservations must be settled by their caller before a live owner is reused. + release(usedBytes); + } + + @Override + public synchronized void close() { + if (closed) { + return; + } + if (manager != null) { + manager.releaseMemory(getAccountedMemoryInBytes()); + manager.unregister(this); + manager = null; + } + usedBytes = 0; + reservedBytes = 0; + closed = true; + } + + public long getId() { + return id; + } + + public String getName() { + return name; + } + + public Category getCategory() { + return category; + } + + public synchronized boolean isClosed() { + return closed; + } + + private void checkOpen() { + if (closed) { + throw new IllegalStateException( + String.format( + StorageEngineMessages.EXCEPTION_WRITE_MEMORY_BLOCK_ARG_IS_CLOSED_9E2561DF, + getName())); + } + } + + private static void checkNonNegative(long bytes) { + if (bytes < 0) { + throw new IllegalArgumentException( + String.format( + StorageEngineMessages.EXCEPTION_WRITE_MEMORY_SIZE_MUST_BE_NON_NEGATIVE_ARG_2CBCF358, + bytes)); + } + } + + /** Closing an uncommitted reservation rolls it back; committing changes no region total. */ + public final class Reservation implements AutoCloseable { + private final long bytes; + private boolean completed; + + private Reservation(long bytes) { + this.bytes = bytes; + } + + public void commit() { + synchronized (WriteMemoryBlock.this) { + if (completed) { + return; + } + checkOpen(); + reservedBytes -= bytes; + usedBytes += bytes; + + completed = true; + } + } + + @Override + public void close() { + synchronized (WriteMemoryBlock.this) { + if (completed) { + return; + } + if (!closed) { + reservedBytes -= bytes; + if (manager != null) { + manager.releaseMemory(bytes); + } + } + completed = true; + } + } + } +} diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/rescon/memory/WriteMemoryManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/rescon/memory/WriteMemoryManager.java new file mode 100644 index 0000000000000..7337dedf2d12c --- /dev/null +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/rescon/memory/WriteMemoryManager.java @@ -0,0 +1,76 @@ +/* + * 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.storageengine.rescon.memory; + +import org.apache.iotdb.commons.memory.AtomicLongMemoryBlock; +import org.apache.iotdb.commons.memory.IMemoryBlock; + +import java.util.ArrayList; +import java.util.List; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; + +/** + * Owns a write-memory accounting scope (a DataRegion or flush reservations). This is not a new + * budget: SystemInfo retains the existing reporting, flush and rejection policy. + */ +public final class WriteMemoryManager { + private final IMemoryBlock memoryBlock; + + public WriteMemoryManager() { + this("WriteMemory"); + } + + public WriteMemoryManager(String scope) { + memoryBlock = new AtomicLongMemoryBlock(scope, null, Long.MAX_VALUE); + } + + private final Set blocks = ConcurrentHashMap.newKeySet(); + + public WriteMemoryBlock createBlock(String name, WriteMemoryBlock.Category category) { + WriteMemoryBlock block = new WriteMemoryBlock(name, category); + block.attachTo(this); + return block; + } + + void register(WriteMemoryBlock block) { + blocks.add(block); + } + + void unregister(WriteMemoryBlock block) { + blocks.remove(block); + } + + void addMemory(long bytes) { + memoryBlock.forceAllocateWithoutLimitation(bytes); + } + + void releaseMemory(long bytes) { + memoryBlock.release(bytes); + } + + /** Includes pending reservations, just like the previous DataRegion memory counter. */ + public long getMemoryUsageInBytes() { + return memoryBlock.getUsedMemoryInBytes(); + } + + public List getMemoryBlocks() { + return new ArrayList<>(blocks); + } +} diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/AlignedTVList.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/AlignedTVList.java index 4f8694f9f1c22..749756f7adcfe 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/AlignedTVList.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/AlignedTVList.java @@ -1416,10 +1416,18 @@ public synchronized RamInfo calculateRamSize(Set columnsToClone) { } public synchronized long getRamSize() { + return getArrayRamSize() + getStructureRamSize(); + } + + /** Arrays and bitmaps, excluding the containers that own them. */ + public synchronized long getArrayRamSize() { return (long) timestamps.size() * alignedTvListArrayMemCostWithoutPrimitiveArrays() + materializedValueArrayMemCost - + materializedBitmapMemoryCost - + calculateContainerRamCost(null); + + materializedBitmapMemoryCost; + } + + public synchronized long getStructureRamSize() { + return calculateContainerRamCost(null); } public synchronized long getRamSize(Set columnsToClone) { diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/MemTableFlushTaskTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/MemTableFlushTaskTest.java index ef4c520b13554..82470bef0222b 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/MemTableFlushTaskTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/MemTableFlushTaskTest.java @@ -22,6 +22,7 @@ import org.apache.iotdb.commons.exception.IllegalPathException; import org.apache.iotdb.db.exception.WriteProcessException; import org.apache.iotdb.db.storageengine.dataregion.flush.MemTableFlushTask; +import org.apache.iotdb.db.storageengine.rescon.memory.SystemInfo; import org.apache.iotdb.db.utils.EnvironmentUtils; import org.apache.iotdb.db.utils.constant.TestConstant; @@ -53,13 +54,23 @@ import java.util.Arrays; import java.util.List; import java.util.concurrent.BlockingQueue; +import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutionException; import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; import static org.apache.iotdb.db.storageengine.rescon.memory.PrimitiveArrayManager.ARRAY_SIZE; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertSame; +import static org.junit.Assert.assertThrows; import static org.junit.Assert.assertTrue; +import static org.mockito.Mockito.any; +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.doThrow; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; public class MemTableFlushTaskTest { @@ -86,6 +97,95 @@ public void tearDown() throws Exception { EnvironmentUtils.cleanDir(TestConstant.OUTPUT_DATA_DIR); } + @Test(timeout = 20000) + public void testIoFailureReleasesTemporaryReservation() throws Exception { + MemTableTestUtils.produceVectorData(memTable); + SystemInfo system = SystemInfo.getInstance(); + boolean old = system.isEncodingFasterThanIo(); + long before = system.getFlushMemoryManager().getMemoryUsageInBytes(); + RestorableTsFileIOWriter failing = mock(RestorableTsFileIOWriter.class); + when(failing.getFile()).thenReturn(new File(filePath)); + IOException error = new IOException("injected IO failure"); + doThrow(error).when(failing).startChunkGroup(any(IDeviceID.class)); + try { + system.setEncodingFasterThanIo(true); + ExecutionException actual = + assertThrows( + ExecutionException.class, + () -> + new MemTableFlushTask(memTable, failing, storageGroup, dataRegionId, true) + .syncFlushMemTable()); + assertSame(error, actual.getCause()); + assertEquals(before, system.getFlushMemoryManager().getMemoryUsageInBytes()); + } finally { + system.setEncodingFasterThanIo(old); + } + } + + @Test(timeout = 20000) + public void testInterruptedFlushWaitsForIoExitBeforeRelease() throws Exception { + MemTableTestUtils.produceVectorData(memTable); + SystemInfo system = SystemInfo.getInstance(); + boolean old = system.isEncodingFasterThanIo(); + long before = system.getFlushMemoryManager().getMemoryUsageInBytes(); + CountDownLatch entered = new CountDownLatch(1); + CountDownLatch interrupted = new CountDownLatch(1); + CountDownLatch mayExit = new CountDownLatch(1); + RestorableTsFileIOWriter slow = mock(RestorableTsFileIOWriter.class); + when(slow.getFile()).thenReturn(new File(filePath)); + doAnswer( + invocation -> { + entered.countDown(); + while (true) { + try { + mayExit.await(); + break; + } catch (InterruptedException e) { + interrupted.countDown(); + } + } + return null; + }) + .when(slow) + .startChunkGroup(any(IDeviceID.class)); + AtomicReference failure = new AtomicReference<>(); + Thread caller = + new Thread( + () -> { + try { + new MemTableFlushTask(memTable, slow, storageGroup, dataRegionId, true) + .syncFlushMemTable(); + failure.set(new AssertionError("flush should be interrupted")); + } catch (InterruptedException expected) { + if (!Thread.currentThread().isInterrupted()) + failure.set(new AssertionError("interrupt flag lost")); + } catch (Throwable e) { + failure.set(e); + } + }); + try { + system.setEncodingFasterThanIo(true); + caller.start(); + assertTrue(entered.await(5, TimeUnit.SECONDS)); + caller.interrupt(); + assertTrue(interrupted.await(5, TimeUnit.SECONDS)); + // Another interruption during cancellation must not let cleanup release live worker memory. + caller.interrupt(); + assertTrue(caller.isAlive()); + assertTrue(system.getFlushMemoryManager().getMemoryUsageInBytes() > before); + mayExit.countDown(); + caller.join(5000); + assertFalse(caller.isAlive()); + if (failure.get() != null) throw new AssertionError(failure.get()); + assertEquals(before, system.getFlushMemoryManager().getMemoryUsageInBytes()); + } finally { + mayExit.countDown(); + caller.interrupt(); + caller.join(5000); + system.setEncodingFasterThanIo(old); + } + } + @Test public void testFlushMemTable() throws ExecutionException, InterruptedException, IllegalPathException { diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/MemTableMemoryBlockTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/MemTableMemoryBlockTest.java new file mode 100644 index 0000000000000..166cd94642e34 --- /dev/null +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/MemTableMemoryBlockTest.java @@ -0,0 +1,167 @@ +/* + * 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.storageengine.dataregion.memtable; + +import org.apache.iotdb.calc.plan.planner.memory.MemoryReservationManager; +import org.apache.iotdb.db.queryengine.execution.fragment.FragmentInstanceContext; +import org.apache.iotdb.db.storageengine.dataregion.flush.FlushStatus; +import org.apache.iotdb.db.storageengine.dataregion.wal.utils.WALByteBufferForTest; +import org.apache.iotdb.db.storageengine.rescon.memory.WriteMemoryBlock; +import org.apache.iotdb.db.storageengine.rescon.memory.WriteMemoryManager; +import org.apache.iotdb.db.utils.datastructure.TVList; + +import org.junit.Test; + +import java.io.ByteArrayInputStream; +import java.io.DataInputStream; +import java.nio.ByteBuffer; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertSame; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +public class MemTableMemoryBlockTest { + @Test + public void queryOwnedTvListSurvivesWriteBlockRelease() throws Exception { + IMemTable table = new PrimitiveMemTable("root.sg", "0"); + WriteMemoryManager manager = new WriteMemoryManager(); + table.setWriteMemoryManager(manager); + MemTableTestUtils.produceData( + table, + 1, + 10, + MemTableTestUtils.deviceId0, + MemTableTestUtils.measurementId0, + MemTableTestUtils.dataType0); + TVList list = + table + .getMemTableMap() + .values() + .iterator() + .next() + .getMemChunkMap(true) + .values() + .iterator() + .next() + .getWorkingTVList(); + long bytes = list.calculateRamSize().getRamSize(); + table.getArrayMemoryBlock().add(bytes); + MemoryReservationManager readMemory = mock(MemoryReservationManager.class); + FragmentInstanceContext query = mock(FragmentInstanceContext.class); + when(query.getMemoryReservationContext()).thenReturn(readMemory); + list.getQueryContextSet().add(query); + try { + int rows = list.rowCount(); + table.release(); + verify(readMemory).reserveMemoryCumulatively(bytes); + assertSame(query, list.getOwnerQuery()); + assertEquals(bytes, list.getReservedMemoryBytes()); + assertEquals(rows, list.rowCount()); + assertEquals(0, manager.getMemoryUsageInBytes()); + assertEquals(0, manager.getMemoryBlocks().size()); + } finally { + list.getQueryContextSet().clear(); + list.clear(); + } + } + + @Test + public void flushingDoesNotDoubleCountAndReleaseKeepsMetadata() { + WriteMemoryManager manager = new WriteMemoryManager(); + WriteMemoryBlock metadata = + manager.createBlock("metadata", WriteMemoryBlock.Category.TSFILE_METADATA); + metadata.add(11); + IMemTable memTable = new PrimitiveMemTable("root.sg", "0"); + memTable.setWriteMemoryManager(manager); + memTable.getArrayMemoryBlock().add(70); + memTable.getBinaryMemoryBlock().add(20); + memTable.getStructureMemoryBlock().add(10); + assertEquals(100, memTable.getTVListsRamCost()); + assertEquals(111, manager.getMemoryUsageInBytes()); + memTable.setFlushStatus(FlushStatus.FLUSHING); + assertEquals(111, manager.getMemoryUsageInBytes()); + memTable.release(); + memTable.setFlushStatus(FlushStatus.FLUSHED); + memTable.release(); + assertEquals(0, memTable.getTVListsRamCost()); + assertEquals(11, manager.getMemoryUsageInBytes()); + assertEquals(1, manager.getMemoryBlocks().size()); + metadata.close(); + assertEquals(0, manager.getMemoryUsageInBytes()); + } + + @Test + public void walSnapshotRetainsTotalAndFormatWithoutInventingCategories() throws Exception { + IMemTable original = new PrimitiveMemTable("root.sg", "0"); + original.getArrayMemoryBlock().add(70); + original.getBinaryMemoryBlock().add(20); + original.getStructureMemoryBlock().add(10); + ByteBuffer bytes = ByteBuffer.allocate(original.serializedSize()); + original.serializeToWAL(new WALByteBufferForTest(bytes)); + assertEquals(0, bytes.remaining()); + // The persisted total remains at the same offset after marker, series count and memSize. + assertEquals(100, bytes.getLong(Byte.BYTES + Integer.BYTES + Long.BYTES)); + for (boolean oldSnapshot : new boolean[] {false, true}) { + DataInputStream input = new DataInputStream(new ByteArrayInputStream(bytes.array())); + IMemTable restored = + oldSnapshot + ? AbstractMemTable.Factory.createFromOldMemTableSnapshot(input) + : AbstractMemTable.Factory.create(input); + assertEquals(0, input.available()); + assertEquals(100, restored.getTVListsRamCost()); + assertEquals(0, restored.getArrayMemoryBlock().getUsedMemoryInBytes()); + assertEquals(0, restored.getBinaryMemoryBlock().getUsedMemoryInBytes()); + assertEquals(0, restored.getStructureMemoryBlock().getUsedMemoryInBytes()); + WriteMemoryManager manager = new WriteMemoryManager(); + restored.setWriteMemoryManager(manager); + assertEquals(100, manager.getMemoryUsageInBytes()); + assertEquals( + 100, + manager.getMemoryBlocks().stream() + .filter(b -> b.getCategory() == WriteMemoryBlock.Category.MEMTABLE_RECOVERED) + .mapToLong(WriteMemoryBlock::getUsedMemoryInBytes) + .sum()); + restored.release(); + assertEquals(0, manager.getMemoryUsageInBytes()); + } + original.release(); + } + + @Test + public void clearResetsAccountingWithoutChangingLogicalSizeSemantics() { + IMemTable table = new PrimitiveMemTable("root.sg", "0"); + WriteMemoryManager manager = new WriteMemoryManager(); + table.setWriteMemoryManager(manager); + table.addTVListRamCost(70); + table.getStructureMemoryBlock().add(10); + table.getBinaryMemoryBlock().add(20); + table.addTextDataSize(20); + assertEquals(20, table.memSize()); + assertEquals(100, table.getTVListsRamCost()); + table.clear(); + assertEquals(0, manager.getMemoryUsageInBytes()); + assertEquals(0, table.memSize()); + assertEquals(0, table.getStructureMemoryBlock().getUsedMemoryInBytes()); + table.addTVListRamCost(5); + assertEquals(5, manager.getMemoryUsageInBytes()); + table.release(); + } +} diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessorTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessorTest.java index 61a672ebdcfe8..116782cdff73a 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessorTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessorTest.java @@ -34,6 +34,7 @@ import org.apache.iotdb.db.exception.DataRegionException; import org.apache.iotdb.db.exception.TsFileProcessorException; import org.apache.iotdb.db.exception.WriteProcessException; +import org.apache.iotdb.db.exception.WriteProcessRejectException; import org.apache.iotdb.db.queryengine.common.QueryId; import org.apache.iotdb.db.queryengine.execution.fragment.QueryContext; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.DeleteDataNode; @@ -53,12 +54,15 @@ import org.apache.iotdb.db.storageengine.dataregion.wal.WALManager; import org.apache.iotdb.db.storageengine.dataregion.wal.buffer.WALEntry; import org.apache.iotdb.db.storageengine.dataregion.wal.io.WALReader; +import org.apache.iotdb.db.storageengine.dataregion.wal.node.WALFakeNode; import org.apache.iotdb.db.storageengine.dataregion.wal.node.WALNode; import org.apache.iotdb.db.storageengine.dataregion.wal.utils.WALFileUtils; import org.apache.iotdb.db.storageengine.dataregion.wal.utils.WALMode; +import org.apache.iotdb.db.storageengine.dataregion.wal.utils.listener.AbstractResultListener; import org.apache.iotdb.db.storageengine.dataregion.wal.utils.listener.WALFlushListener; import org.apache.iotdb.db.storageengine.rescon.memory.PrimitiveArrayManager; import org.apache.iotdb.db.storageengine.rescon.memory.SystemInfo; +import org.apache.iotdb.db.storageengine.rescon.memory.WriteMemoryBlock; import org.apache.iotdb.db.utils.EnvironmentUtils; import org.apache.iotdb.db.utils.constant.TestConstant; import org.apache.iotdb.db.utils.datastructure.AlignedTVList; @@ -97,6 +101,7 @@ import java.io.IOException; import java.lang.management.ManagementFactory; import java.lang.management.ThreadInfo; +import java.lang.reflect.Field; import java.nio.charset.StandardCharsets; import java.util.ArrayList; import java.util.Arrays; @@ -1769,6 +1774,162 @@ public void testWriteAndClose() assertTrue(processor.getTsFileResource().isClosed()); } + @Test + public void testWriteMemoryCategoriesAndClose() throws Exception { + processor = newTestProcessor(filePath); + TSRecord record = new TSRecord(deviceId, 1); + record.addTuple(DataPoint.getDataPoint(TSDataType.TEXT, "text", "payload")); + processor.insert(buildInsertRowNodeByTSRecord(record), new long[5]); + IMemTable memTable = processor.getWorkMemTable(); + assertTrue(memTable.getArrayMemoryBlock().getUsedMemoryInBytes() > 0); + assertTrue(memTable.getBinaryMemoryBlock().getUsedMemoryInBytes() > 0); + long metadataBytes = + sgInfo.getWriteMemoryManager().getMemoryBlocks().stream() + .filter(b -> b.getCategory() == WriteMemoryBlock.Category.TSFILE_METADATA) + .mapToLong(WriteMemoryBlock::getUsedMemoryInBytes) + .sum(); + assertTrue(metadataBytes > 0); + assertEquals(memTable.getTVListsRamCost() + metadataBytes, sgInfo.getMemCost()); + processor.syncClose(); + assertEquals(0, sgInfo.getMemCost()); + assertTrue(sgInfo.getWriteMemoryManager().getMemoryBlocks().isEmpty()); + } + + @Test + public void testWalFailureRollsBackRowMixedRowsAndSplitTablet() throws Exception { + processor = newTestProcessor(filePath); + insertAlignedRow(processor, deviceId, 0, false); + IMemTable memTable = processor.getWorkMemTable(); + long regionBefore = sgInfo.getMemCost(); + long ramBefore = memTable.getTVListsRamCost(); + long logicalBefore = memTable.memSize(); + long structureBefore = memTable.getStructureMemoryBlock().getUsedMemoryInBytes(); + assertTrue(structureBefore > 0); + Field field = TsFileProcessor.class.getDeclaredField("walNode"); + field.setAccessible(true); + Object originalWal = field.get(processor); + field.set( + processor, + new WALFakeNode( + AbstractResultListener.Status.FAILURE, new IOException("injected WAL failure"))); + try { + TSRecord record = new TSRecord(deviceId, 1); + record.addTuple(DataPoint.getDataPoint(TSDataType.TEXT, "text", "payload")); + InsertRowNode row = buildInsertRowNodeByTSRecord(record); + row.setAligned(true); + Assert.assertThrows(WriteProcessException.class, () -> processor.insert(row, new long[5])); + assertMemoryUnchanged(regionBefore, ramBefore, logicalBefore); + assertEquals(structureBefore, memTable.getStructureMemoryBlock().getUsedMemoryInBytes()); + + InsertRowsNode rows = new InsertRowsNode(new PlanNodeId("mixed")); + rows.addOneInsertRowNode(row, 0); + TSRecord nonAligned = new TSRecord("root.vehicle.other", 1); + nonAligned.addTuple(DataPoint.getDataPoint(TSDataType.TEXT, "text", "payload")); + rows.addOneInsertRowNode(buildInsertRowNodeByTSRecord(nonAligned), 1); + Assert.assertThrows( + WriteProcessException.class, () -> processor.insertRows(rows, new long[5])); + assertMemoryUnchanged(regionBefore, ramBefore, logicalBefore); + assertEquals(structureBefore, memTable.getStructureMemoryBlock().getUsedMemoryInBytes()); + + InsertTabletNode tablet = genSingleMeasurementTablet(10, true); + TSStatus[] results = new TSStatus[10]; + Assert.assertThrows( + WriteProcessException.class, + () -> + processor.insertTablet( + tablet, + Arrays.asList(new int[] {0, 5}, new int[] {5, 10}), + results, + true, + new long[5])); + assertMemoryUnchanged(regionBefore, ramBefore, logicalBefore); + assertEquals(structureBefore, memTable.getStructureMemoryBlock().getUsedMemoryInBytes()); + } finally { + field.set(processor, originalWal); + processor.syncClose(); + } + } + + @Test + public void testRejectionRollsBackAllMemoryReservations() throws Exception { + processor = newTestProcessor(filePath); + int originalWait = config.getMaxWaitingTimeWhenInsertBlocked(); + int originalPeriod = config.getCheckPeriodWhenInsertBlocked(); + // Keep selection empty so rejection cannot launch an asynchronous flush in this test. + sgInfo.getAllReportedTsp().clear(); + try (WriteMemoryBlock pressure = + sgInfo + .getWriteMemoryManager() + .createBlock("pressure", WriteMemoryBlock.Category.TSFILE_METADATA)) { + pressure.add((long) SystemInfo.getInstance().getRejectThreshold() + 1); + long before = sgInfo.getMemCost(); + sgInfo.setNeedToReportToSystem(true); + config.setMaxWaitingTimeWhenInsertBlocked(0); + config.setCheckPeriodWhenInsertBlocked(1); + TSRecord record = new TSRecord(deviceId, 1); + record.addTuple(DataPoint.getDataPoint(TSDataType.TEXT, "text", "payload")); + Assert.assertThrows( + WriteProcessRejectException.class, + () -> processor.insert(buildInsertRowNodeByTSRecord(record), new long[5])); + assertMemoryUnchanged(before, 0, 0); + assertEquals(before, SystemInfo.getInstance().getTotalMemTableSize()); + } finally { + config.setMaxWaitingTimeWhenInsertBlocked(originalWait); + config.setCheckPeriodWhenInsertBlocked(originalPeriod); + SystemInfo.getInstance().resetStorageGroupStatus(sgInfo); + processor.putMemTableBackAndClose(); + } + assertEquals(0, sgInfo.getMemCost()); + assertTrue(sgInfo.getWriteMemoryManager().getMemoryBlocks().isEmpty()); + } + + @Test + public void testZeroMemoryIncrementStillChecksPressure() throws Exception { + processor = newTestProcessor(filePath); + TSRecord first = new TSRecord(deviceId, 0); + first.addTuple(DataPoint.getDataPoint(TSDataType.INT64, "s1", "1")); + processor.insert(buildInsertRowNodeByTSRecord(first), new long[5]); + IMemTable memTable = processor.getWorkMemTable(); + long ramBefore = memTable.getTVListsRamCost(); + long logicalBefore = memTable.memSize(); + int originalWait = config.getMaxWaitingTimeWhenInsertBlocked(); + int originalPeriod = config.getCheckPeriodWhenInsertBlocked(); + sgInfo.getAllReportedTsp().clear(); + try (WriteMemoryBlock pressure = + sgInfo + .getWriteMemoryManager() + .createBlock("pressure", WriteMemoryBlock.Category.TSFILE_METADATA)) { + pressure.add((long) SystemInfo.getInstance().getRejectThreshold() + 1); + long before = sgInfo.getMemCost(); + sgInfo.setNeedToReportToSystem(true); + config.setMaxWaitingTimeWhenInsertBlocked(0); + config.setCheckPeriodWhenInsertBlocked(1); + // The next numeric row reuses both its value/timestamp arrays and its metadata. + TSRecord next = new TSRecord(deviceId, 1); + next.addTuple(DataPoint.getDataPoint(TSDataType.INT64, "s1", "2")); + Assert.assertThrows( + WriteProcessRejectException.class, + () -> processor.insert(buildInsertRowNodeByTSRecord(next), new long[5])); + assertMemoryUnchanged(before, ramBefore, logicalBefore); + assertEquals(1, memTable.getTotalPointsNum()); + } finally { + config.setMaxWaitingTimeWhenInsertBlocked(originalWait); + config.setCheckPeriodWhenInsertBlocked(originalPeriod); + SystemInfo.getInstance().resetStorageGroupStatus(sgInfo); + processor.putMemTableBackAndClose(); + } + assertEquals(0, sgInfo.getMemCost()); + } + + private void assertMemoryUnchanged(long regionBytes, long ramBytes, long logicalBytes) { + assertEquals(regionBytes, sgInfo.getMemCost()); + assertEquals(ramBytes, processor.getWorkMemTable().getTVListsRamCost()); + assertEquals(logicalBytes, processor.getWorkMemTable().memSize()); + for (WriteMemoryBlock block : sgInfo.getWriteMemoryManager().getMemoryBlocks()) { + assertEquals(0, block.getReservedMemoryInBytes()); + } + } + private void closeTsFileProcessor(TsFileProcessor unsealedTsFileProcessor) throws TsFileProcessorException { TsFileResource resource = unsealedTsFileProcessor.getTsFileResource(); @@ -1828,9 +1989,13 @@ private AlignedWritableMemChunk getAlignedMemChunk(String targetDevice) { } private void assertAlignedTvListRamCostMatchesActual(String targetDevice) { + AlignedTVList list = getAlignedMemChunk(targetDevice).getWorkingTVList(); + IMemTable table = processor.getWorkMemTable(); + Assert.assertEquals(list.calculateRamSize().getRamSize(), table.getTVListsRamCost()); + Assert.assertEquals(list.getArrayRamSize(), table.getArrayMemoryBlock().getUsedMemoryInBytes()); Assert.assertEquals( - getAlignedMemChunk(targetDevice).getWorkingTVList().calculateRamSize().getRamSize(), - processor.getWorkMemTable().getTVListsRamCost()); + list.getStructureRamSize(), table.getStructureMemoryBlock().getUsedMemoryInBytes()); + assertTrue(table.getStructureMemoryBlock().getUsedMemoryInBytes() > 0); } private InsertTabletNode genSingleMeasurementTablet(int rowCount, boolean isAligned) diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/rescon/memory/WriteMemoryBlockTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/rescon/memory/WriteMemoryBlockTest.java new file mode 100644 index 0000000000000..0af93211c4d7d --- /dev/null +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/rescon/memory/WriteMemoryBlockTest.java @@ -0,0 +1,184 @@ +/* + * 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.storageengine.rescon.memory; + +import org.junit.Test; + +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotEquals; +import static org.junit.Assert.assertThrows; +import static org.junit.Assert.assertTrue; + +public class WriteMemoryBlockTest { + @Test + public void reservationCountsAgainstRegionButNotFlushableMemory() { + WriteMemoryManager manager = new WriteMemoryManager(); + WriteMemoryBlock block = manager.createBlock("array", WriteMemoryBlock.Category.MEMTABLE_ARRAY); + block.add(8); + try (WriteMemoryBlock.Reservation reservation = block.reserve(10)) { + assertEquals(18, manager.getMemoryUsageInBytes()); + assertEquals(8, block.getUsedMemoryInBytes()); + assertEquals(10, block.getReservedMemoryInBytes()); + reservation.commit(); + reservation.commit(); + assertEquals(18, manager.getMemoryUsageInBytes()); + assertEquals(18, block.getUsedMemoryInBytes()); + assertEquals(0, block.getReservedMemoryInBytes()); + } + block.close(); + block.close(); + assertEquals(0, manager.getMemoryUsageInBytes()); + assertTrue(manager.getMemoryBlocks().isEmpty()); + } + + @Test + public void rejectedReservationPreservesOtherOwners() { + WriteMemoryManager manager = new WriteMemoryManager(); + WriteMemoryBlock first = manager.createBlock("same", WriteMemoryBlock.Category.MEMTABLE_ARRAY); + WriteMemoryBlock second = manager.createBlock("same", WriteMemoryBlock.Category.MEMTABLE_ARRAY); + first.add(7); + second.add(11); + assertNotEquals(first.getId(), second.getId()); + assertEquals(2, manager.getMemoryBlocks().size()); + try (WriteMemoryBlock.Reservation ignored = first.reserve(13)) { + assertEquals(31, manager.getMemoryUsageInBytes()); + } + assertEquals(18, manager.getMemoryUsageInBytes()); + assertThrows(IllegalArgumentException.class, () -> first.release(8)); + assertEquals(18, manager.getMemoryUsageInBytes()); + first.close(); + assertEquals(11, manager.getMemoryUsageInBytes()); + second.close(); + } + + @Test + public void ownerCloseCancelsPendingReservationOnlyOnce() { + WriteMemoryManager manager = new WriteMemoryManager(); + WriteMemoryBlock block = manager.createBlock("array", WriteMemoryBlock.Category.MEMTABLE_ARRAY); + WriteMemoryBlock.Reservation reservation = block.reserve(10); + block.close(); + assertThrows(IllegalStateException.class, reservation::commit); + reservation.close(); + reservation.close(); + assertEquals(0, manager.getMemoryUsageInBytes()); + assertThrows(IllegalStateException.class, () -> block.add(1)); + assertThrows(IllegalStateException.class, () -> block.reserve(1)); + } + + @Test + public void attachingExistingUsageIsIdempotent() { + WriteMemoryBlock block = + new WriteMemoryBlock("recovery", WriteMemoryBlock.Category.MEMTABLE_RECOVERED); + block.add(30); + WriteMemoryManager manager = new WriteMemoryManager(); + block.attachTo(manager); + block.attachTo(manager); + assertEquals(30, manager.getMemoryUsageInBytes()); + assertThrows(IllegalStateException.class, () -> block.attachTo(new WriteMemoryManager())); + assertThrows(IllegalArgumentException.class, () -> block.add(-1)); + block.clear(); + assertEquals(0, manager.getMemoryUsageInBytes()); + block.add(5); + assertEquals(5, manager.getMemoryUsageInBytes()); + block.close(); + } + + @Test + public void concurrentReservationsOnOneBlockKeepExactUsage() throws Exception { + WriteMemoryManager manager = new WriteMemoryManager(); + WriteMemoryBlock block = + manager.createBlock("shared", WriteMemoryBlock.Category.MEMTABLE_ARRAY); + block.add(17); + ExecutorService executor = Executors.newFixedThreadPool(4); + CountDownLatch start = new CountDownLatch(1); + List> tasks = new ArrayList<>(); + try { + for (int i = 0; i < 4; i++) { + tasks.add( + executor.submit( + () -> { + start.await(); + for (int j = 0; j < 1000; j++) { + try (WriteMemoryBlock.Reservation reservation = block.reserve(5)) { + reservation.commit(); + } + try (WriteMemoryBlock.Reservation ignored = block.reserve(11)) { + block.release(3); + } + } + return null; + })); + } + start.countDown(); + for (Future task : tasks) { + task.get(30, TimeUnit.SECONDS); + } + assertEquals(8017, block.getUsedMemoryInBytes()); + assertEquals(0, block.getReservedMemoryInBytes()); + assertEquals(8017, manager.getMemoryUsageInBytes()); + } finally { + start.countDown(); + executor.shutdownNow(); + assertTrue(executor.awaitTermination(30, TimeUnit.SECONDS)); + block.close(); + } + assertEquals(0, manager.getMemoryUsageInBytes()); + assertTrue(manager.getMemoryBlocks().isEmpty()); + } + + @Test + public void concurrentWritersAndReleasesKeepRegionTotal() throws Exception { + WriteMemoryManager manager = new WriteMemoryManager(); + ExecutorService executor = Executors.newFixedThreadPool(4); + List> tasks = new ArrayList<>(); + try { + for (int i = 0; i < 4; i++) { + tasks.add( + executor.submit( + () -> { + try (WriteMemoryBlock block = + manager.createBlock("array", WriteMemoryBlock.Category.MEMTABLE_ARRAY)) { + for (int j = 0; j < 1000; j++) { + try (WriteMemoryBlock.Reservation reservation = block.reserve(5)) { + reservation.commit(); + } + block.release(3); + } + assertEquals(2000, block.getUsedMemoryInBytes()); + } + })); + } + for (Future task : tasks) { + task.get(); + } + assertEquals(0, manager.getMemoryUsageInBytes()); + assertTrue(manager.getMemoryBlocks().isEmpty()); + } finally { + executor.shutdownNow(); + } + } +} diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/rescon/memory/WriteMemoryLifecycleTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/rescon/memory/WriteMemoryLifecycleTest.java new file mode 100644 index 0000000000000..52c410bc2f6fe --- /dev/null +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/rescon/memory/WriteMemoryLifecycleTest.java @@ -0,0 +1,133 @@ +/* + * 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.storageengine.rescon.memory; + +import org.apache.iotdb.db.storageengine.dataregion.wal.buffer.WALEntry; +import org.apache.iotdb.db.storageengine.dataregion.wal.utils.MemoryControlledWALEntryQueue; + +import org.apache.tsfile.enums.TSDataType; +import org.apache.tsfile.utils.Binary; +import org.apache.tsfile.utils.RamUsageEstimator; +import org.junit.Test; + +import java.util.concurrent.TimeUnit; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertSame; +import static org.junit.Assert.assertThrows; +import static org.junit.Assert.assertTrue; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +public class WriteMemoryLifecycleTest { + @Test + public void testFlushReservationsSurviveReloadAndCloseOnce() { + SystemInfo system = SystemInfo.getInstance(); + double threshold = system.getFlushThreshold(); + long before = system.getFlushMemoryManager().getMemoryUsageInBytes(); + try (SystemInfo.FlushMemoryReservation first = + system.reserveTemporaryMemoryForFlushing("first", 1024); + SystemInfo.FlushMemoryReservation second = + system.reserveTemporaryMemoryForFlushing("second", 2048)) { + assertEquals(before + 3072, system.getFlushMemoryManager().getMemoryUsageInBytes()); + double reduced = system.getFlushThreshold(); + assertTrue(reduced < threshold); + system.loadWriteMemory(); + assertEquals(reduced, system.getFlushThreshold(), 0); + first.close(); + first.close(); + assertEquals(before + 2048, system.getFlushMemoryManager().getMemoryUsageInBytes()); + } + assertEquals(before, system.getFlushMemoryManager().getMemoryUsageInBytes()); + assertEquals(threshold, system.getFlushThreshold(), 0); + } + + @Test + public void testWalUsesOriginalChargeWhenEntryChanges() throws Exception { + MemoryControlledWALEntryQueue queue = new MemoryControlledWALEntryQueue(); + WALEntry entry = mock(WALEntry.class); + when(entry.getMemorySize()).thenReturn(128L, 4096L); + long before = SystemInfo.getInstance().getWalBufferQueueMemoryBlock().getUsedMemoryInBytes(); + queue.put(entry); + assertEquals( + before + 128, + SystemInfo.getInstance().getWalBufferQueueMemoryBlock().getUsedMemoryInBytes()); + assertSame(entry, queue.take()); + assertEquals( + before, SystemInfo.getInstance().getWalBufferQueueMemoryBlock().getUsedMemoryInBytes()); + verify(entry, times(1)).getMemorySize(); + assertNull(queue.poll(0, TimeUnit.MILLISECONDS)); + } + + @Test + public void testInterruptedWalPutReturnsCharge() { + MemoryControlledWALEntryQueue queue = new MemoryControlledWALEntryQueue(); + WALEntry entry = mock(WALEntry.class); + when(entry.getMemorySize()).thenReturn(128L); + long before = SystemInfo.getInstance().getWalBufferQueueMemoryBlock().getUsedMemoryInBytes(); + try { + Thread.currentThread().interrupt(); + assertThrows(InterruptedException.class, () -> queue.put(entry)); + } finally { + Thread.interrupted(); + } + assertTrue(queue.isEmpty()); + assertEquals( + before, SystemInfo.getInstance().getWalBufferQueueMemoryBlock().getUsedMemoryInBytes()); + } + + @Test + public void testArrayPoolCountsOnlyIdleArrays() { + PrimitiveArrayManager.close(); + try { + for (TSDataType type : + new TSDataType[] { + TSDataType.BOOLEAN, + TSDataType.INT32, + TSDataType.INT64, + TSDataType.FLOAT, + TSDataType.DOUBLE, + TSDataType.TEXT + }) { + Object array = PrimitiveArrayManager.allocate(type); + if (array instanceof Binary[]) { + ((Binary[]) array)[0] = new Binary(new byte[32]); + } + assertEquals(0, PrimitiveArrayManager.getMemoryBlock().getUsedMemoryInBytes()); + PrimitiveArrayManager.release(array); + assertEquals( + RamUsageEstimator.shallowSizeOf(array), + PrimitiveArrayManager.getMemoryBlock().getUsedMemoryInBytes()); + if (array instanceof Binary[]) { + assertNull(((Binary[]) array)[0]); + } + assertSame(array, PrimitiveArrayManager.allocate(type)); + assertEquals(0, PrimitiveArrayManager.getMemoryBlock().getUsedMemoryInBytes()); + PrimitiveArrayManager.release(array); + PrimitiveArrayManager.close(); + assertEquals(0, PrimitiveArrayManager.getMemoryBlock().getUsedMemoryInBytes()); + } + } finally { + PrimitiveArrayManager.close(); + } + } +} diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/utils/datastructure/PrimitiveArrayManagerTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/utils/datastructure/PrimitiveArrayManagerTest.java index d5c7b3e865f99..18a3d29a843a9 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/utils/datastructure/PrimitiveArrayManagerTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/utils/datastructure/PrimitiveArrayManagerTest.java @@ -26,9 +26,18 @@ import org.apache.tsfile.enums.TSDataType; import org.apache.tsfile.utils.Binary; +import org.apache.tsfile.utils.RamUsageEstimator; import org.junit.Assert; import org.junit.Test; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; + import static org.apache.iotdb.db.storageengine.rescon.memory.PrimitiveArrayManager.ARRAY_SIZE; public class PrimitiveArrayManagerTest { @@ -101,4 +110,117 @@ public void testUpdateLimits() { } } } + + @Test + public void countsActualLengthsAndClearsOwnership() { + PrimitiveArrayManager.close(); + try { + int[] ints = new int[3]; + long[] longs = new long[0]; + Binary[] binaries = new Binary[97]; + binaries[0] = new Binary(new byte[32]); + PrimitiveArrayManager.release(ints); + PrimitiveArrayManager.release(longs); + PrimitiveArrayManager.release(binaries); + Assert.assertNull(binaries[0]); + long bytes = + RamUsageEstimator.shallowSizeOf(ints) + + RamUsageEstimator.shallowSizeOf(longs) + + RamUsageEstimator.shallowSizeOf(binaries); + Assert.assertEquals(bytes, PrimitiveArrayManager.getMemoryBlock().getUsedMemoryInBytes()); + Assert.assertSame(ints, PrimitiveArrayManager.allocate(TSDataType.INT32)); + Assert.assertEquals( + bytes - RamUsageEstimator.shallowSizeOf(ints), + PrimitiveArrayManager.getMemoryBlock().getUsedMemoryInBytes()); + PrimitiveArrayManager.close(); + Assert.assertEquals(0, PrimitiveArrayManager.getMemoryBlock().getUsedMemoryInBytes()); + } finally { + PrimitiveArrayManager.close(); + } + } + + @Test + public void cachedCalculatorsCoverAllSupportedTypes() { + PrimitiveArrayManager.close(); + try { + for (TSDataType type : TSDataTypeTestUtils.getSupportedTypes()) { + Object array = PrimitiveArrayManager.allocate(type); + long bytes = RamUsageEstimator.shallowSizeOf(array); + PrimitiveArrayManager.release(array); + Assert.assertEquals(bytes, PrimitiveArrayManager.getMemoryBlock().getUsedMemoryInBytes()); + // Alias types return to their primitive pool (e.g. DATE -> INT32, STRING -> TEXT). + TSDataType poolType; + switch (type) { + case DATE: + poolType = TSDataType.INT32; + break; + case TIMESTAMP: + poolType = TSDataType.INT64; + break; + case STRING: + case BLOB: + case OBJECT: + poolType = TSDataType.TEXT; + break; + default: + poolType = type; + } + Assert.assertSame(array, PrimitiveArrayManager.allocate(poolType)); + Assert.assertEquals(0, PrimitiveArrayManager.getMemoryBlock().getUsedMemoryInBytes()); + } + } finally { + PrimitiveArrayManager.close(); + } + } + + @Test + public void concurrentBorrowReturnAndSnapshotsKeepExactTotal() throws Exception { + PrimitiveArrayManager.close(); + TSDataType[] types = {TSDataType.INT64, TSDataType.INT32, TSDataType.BOOLEAN, TSDataType.TEXT}; + long expected = 0; + for (TSDataType type : types) { + Object array = PrimitiveArrayManager.allocate(type); + expected += RamUsageEstimator.shallowSizeOf(array); + PrimitiveArrayManager.release(array); + } + final long upperBound = expected; + ExecutorService executor = Executors.newFixedThreadPool(5); + CountDownLatch start = new CountDownLatch(1); + List> futures = new ArrayList<>(); + try { + for (TSDataType type : types) { + futures.add( + executor.submit( + () -> { + start.await(); + for (int i = 0; i < 2000; i++) { + Object array = PrimitiveArrayManager.allocate(type); + PrimitiveArrayManager.release(array); + } + return null; + })); + } + futures.add( + executor.submit( + () -> { + start.await(); + for (int i = 0; i < 8000; i++) { + long actual = PrimitiveArrayManager.getMemoryBlock().getUsedMemoryInBytes(); + Assert.assertTrue(actual >= 0 && actual <= upperBound); + } + return null; + })); + start.countDown(); + for (Future future : futures) { + future.get(30, TimeUnit.SECONDS); + } + Assert.assertEquals(expected, PrimitiveArrayManager.getMemoryBlock().getUsedMemoryInBytes()); + } finally { + start.countDown(); + executor.shutdownNow(); + Assert.assertTrue(executor.awaitTermination(30, TimeUnit.SECONDS)); + PrimitiveArrayManager.close(); + } + Assert.assertEquals(0, PrimitiveArrayManager.getMemoryBlock().getUsedMemoryInBytes()); + } } diff --git a/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/CommonMessages.java b/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/CommonMessages.java index b8f18377293aa..a0bcb6078e220 100644 --- a/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/CommonMessages.java +++ b/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/CommonMessages.java @@ -341,4 +341,8 @@ private CommonMessages() {} public static final String EXCEPTION_THE_TABLE_ARG_ARG_IS_IN_THE_PRE_DELETE_STATE_PLEASE_WAIT_A_FEW_SECONDS_IF_THE_TABLE_IS_STILL_IN_THIS_STATE_PLEASE_DROP_IT_AGAIN_13B9D39F = "The table %s.%s is in the pre-delete state. Please wait a few seconds. If the table is still in this state, please drop it again."; + public static final String EXCEPTION_MEMORY_BLOCK_ARG_IS_READ_ONLY_D10B3340 = + "Memory block %s is read-only"; + public static final String EXCEPTION_MEMORY_BLOCK_ARG_ALREADY_USES_MUTABLE_ACCOUNTING_6F7DCB3F = + "Memory block %s already uses mutable accounting"; } diff --git a/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/CommonMessages.java b/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/CommonMessages.java index 0028dfcd7d3ee..75873a976719e 100644 --- a/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/CommonMessages.java +++ b/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/CommonMessages.java @@ -238,4 +238,7 @@ private CommonMessages() {} public static final String EXCEPTION_THE_TABLE_ARG_ARG_IS_IN_THE_PRE_DELETE_STATE_PLEASE_WAIT_A_FEW_SECONDS_IF_THE_TABLE_IS_STILL_IN_THIS_STATE_PLEASE_DROP_IT_AGAIN_13B9D39F = "表 %s.%s 处于预删除的状态,请稍等,如之后重试还是此状态,请输入sql再次删除"; + public static final String EXCEPTION_MEMORY_BLOCK_ARG_IS_READ_ONLY_D10B3340 = "内存块 %s 为只读"; + public static final String EXCEPTION_MEMORY_BLOCK_ARG_ALREADY_USES_MUTABLE_ACCOUNTING_6F7DCB3F = + "内存块 %s 已使用可变计量"; } 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 2aa480b9060cc..5b0b2f0ea491f 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,7 +26,9 @@ import org.slf4j.LoggerFactory; import java.util.Map; +import java.util.Objects; import java.util.concurrent.ConcurrentHashMap; +import java.util.function.LongSupplier; import java.util.function.LongUnaryOperator; public class MemoryManager { @@ -101,12 +103,27 @@ private MemoryManager( */ public synchronized IMemoryBlock exactAllocate( String name, long sizeInBytes, MemoryBlockType type) { + return exactAllocate(name, sizeInBytes, type, null); + } + + /** Reserves the remaining budget for an owner that supplies an exact, read-only usage view. */ + public synchronized IMemoryBlock exactAllocateReadOnly( + String name, MemoryBlockType type, LongSupplier usedMemory) { + return exactAllocate( + name, + totalMemorySizeInBytes - allocatedMemorySizeInBytes, + type, + Objects.requireNonNull(usedMemory)); + } + + private IMemoryBlock exactAllocate( + String name, long sizeInBytes, MemoryBlockType type, LongSupplier usedMemory) { if (!enabled) { - return getOrRegisterMemoryBlock(name, sizeInBytes, type); + return getOrRegisterMemoryBlock(name, sizeInBytes, type, usedMemory); } for (int i = 0; i < MEMORY_ALLOCATE_MAX_RETRIES; i++) { if (totalMemorySizeInBytes - allocatedMemorySizeInBytes >= sizeInBytes) { - return getOrRegisterMemoryBlock(name, sizeInBytes, type); + return getOrRegisterMemoryBlock(name, sizeInBytes, type, usedMemory); } try { @@ -241,6 +258,11 @@ public synchronized IMemoryBlock tryAllocate( */ private IMemoryBlock getOrRegisterMemoryBlock( String name, long sizeInBytes, MemoryBlockType type) { + return getOrRegisterMemoryBlock(name, sizeInBytes, type, null); + } + + private IMemoryBlock getOrRegisterMemoryBlock( + String name, long sizeInBytes, MemoryBlockType type, LongSupplier usedMemory) { if (sizeInBytes < 0) { throw new MemoryException( String.format( @@ -252,6 +274,13 @@ private IMemoryBlock getOrRegisterMemoryBlock( name, (blockName, block) -> { if (block != null) { + if (usedMemory != null && !(block instanceof ReadOnlyMemoryBlock)) { + throw new MemoryException( + String.format( + CommonMessages + .EXCEPTION_MEMORY_BLOCK_ARG_ALREADY_USES_MUTABLE_ACCOUNTING_6F7DCB3F, + name)); + } if (block.getTotalMemorySizeInBytes() != sizeInBytes) { LOGGER.warn( CommonMessages @@ -264,7 +293,9 @@ private IMemoryBlock getOrRegisterMemoryBlock( return block; } else { allocatedMemorySizeInBytes += sizeInBytes; - return new AtomicLongMemoryBlock(name, this, sizeInBytes, type); + return usedMemory == null + ? new AtomicLongMemoryBlock(name, this, sizeInBytes, type) + : new ReadOnlyMemoryBlock(name, this, sizeInBytes, type, usedMemory); } }); } diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/memory/ReadOnlyMemoryBlock.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/memory/ReadOnlyMemoryBlock.java new file mode 100644 index 0000000000000..d5bbb757d3efa --- /dev/null +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/memory/ReadOnlyMemoryBlock.java @@ -0,0 +1,110 @@ +/* + * 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.i18n.CommonMessages; + +import java.util.function.LongSupplier; + +/** + * A budgeted view of memory owned and counted by another component. The supplier must provide its + * own synchronization; allocation and release remain the owner's responsibility. + */ +public final class ReadOnlyMemoryBlock extends IMemoryBlock { + private final LongSupplier usedMemory; + + ReadOnlyMemoryBlock( + String name, + MemoryManager manager, + long totalMemorySizeInBytes, + MemoryBlockType type, + LongSupplier usedMemory) { + this.name = name; + this.memoryManager = manager; + this.totalMemorySizeInBytes = totalMemorySizeInBytes; + this.memoryBlockType = type; + this.usedMemory = usedMemory; + } + + @Override + public long getUsedMemoryInBytes() { + return usedMemory.getAsLong(); + } + + @Override + public long getFreeMemoryInBytes() { + return totalMemorySizeInBytes - getUsedMemoryInBytes(); + } + + @Override + public long forceAllocateWithoutLimitation(long sizeInByte) { + throw readOnly(); + } + + @Override + public boolean allocate(long sizeInByte) { + throw readOnly(); + } + + @Override + public boolean allocateIfSufficient(long sizeInByte, double maxRatio) { + throw readOnly(); + } + + @Override + public boolean allocateUntilAvailable(long sizeInByte, long retryIntervalInMillis) { + throw readOnly(); + } + + @Override + public long release(long sizeInByte) { + throw readOnly(); + } + + @Override + public void setUsedMemoryInBytes(long usedMemoryInBytes) { + throw readOnly(); + } + + private UnsupportedOperationException readOnly() { + return new UnsupportedOperationException( + String.format(CommonMessages.EXCEPTION_MEMORY_BLOCK_ARG_IS_READ_ONLY_D10B3340, name)); + } + + @Override + public void close() { + memoryManager.release(this); + } + + @Override + public String toString() { + return "ReadOnlyMemoryBlock{name=" + + name + + ", isReleased=" + + isReleased + + ", memoryBlockType=" + + memoryBlockType + + ", totalMemorySizeInBytes=" + + totalMemorySizeInBytes + + ", usedMemoryInBytes=" + + getUsedMemoryInBytes() + + '}'; + } +} diff --git a/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/memory/ReadOnlyMemoryBlockTest.java b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/memory/ReadOnlyMemoryBlockTest.java new file mode 100644 index 0000000000000..b04e314d404e6 --- /dev/null +++ b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/memory/ReadOnlyMemoryBlockTest.java @@ -0,0 +1,81 @@ +/* + * 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.junit.Test; + +import java.util.concurrent.atomic.AtomicLong; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertThrows; +import static org.junit.Assert.assertTrue; + +public class ReadOnlyMemoryBlockTest { + @Test + public void externalUsageAndBudgetHaveSeparateLifetimes() throws Exception { + MemoryManager manager = new MemoryManager(100); + IMemoryBlock other = manager.exactAllocate("other", 20, MemoryBlockType.DYNAMIC); + other.allocate(7); + AtomicLong owner = new AtomicLong(11); + IMemoryBlock view = manager.exactAllocateReadOnly("view", MemoryBlockType.DYNAMIC, owner::get); + assertEquals(100, manager.getAllocatedMemorySizeInBytes()); + assertEquals(80, view.getTotalMemorySizeInBytes()); + assertEquals(69, view.getFreeMemoryInBytes()); + assertEquals(18, manager.getUsedMemorySizeInBytes()); + owner.set(37); + assertEquals(37, view.getUsedMemoryInBytes()); + assertEquals(44, manager.getUsedMemorySizeInBytes()); + view.close(); + view.close(); + assertTrue(view.isReleased()); + assertEquals(37, owner.get()); + assertEquals(20, manager.getAllocatedMemorySizeInBytes()); + assertEquals(7, manager.getUsedMemorySizeInBytes()); + other.close(); + assertEquals(0, manager.getAllocatedMemorySizeInBytes()); + } + + @Test + public void writesCannotChangeAnExternallyOwnedUsage() throws Exception { + AtomicLong owner = new AtomicLong(11); + IMemoryBlock view = + new MemoryManager(100).exactAllocateReadOnly("view", MemoryBlockType.DYNAMIC, owner::get); + assertThrows(UnsupportedOperationException.class, () -> view.allocate(1)); + assertThrows(UnsupportedOperationException.class, () -> view.forceAllocateWithoutLimitation(1)); + assertThrows(UnsupportedOperationException.class, () -> view.allocateIfSufficient(1, .5)); + assertThrows(UnsupportedOperationException.class, () -> view.allocateUntilAvailable(1, 1)); + assertThrows(UnsupportedOperationException.class, () -> view.release(1)); + assertThrows(UnsupportedOperationException.class, () -> view.setUsedMemoryInBytes(1)); + assertEquals(11, view.getUsedMemoryInBytes()); + view.close(); + } + + @Test + public void mutableOwnerCannotSilentlyBecomeAReadOnlyView() { + MemoryManager manager = new MemoryManager(100); + IMemoryBlock existing = manager.exactAllocate("owner", 20, MemoryBlockType.DYNAMIC); + existing.allocate(7); + assertThrows( + MemoryException.class, + () -> manager.exactAllocateReadOnly("owner", MemoryBlockType.DYNAMIC, () -> 11)); + assertEquals(20, manager.getAllocatedMemorySizeInBytes()); + assertEquals(7, manager.getUsedMemorySizeInBytes()); + } +}