diff --git a/integration-test/src/test/java/org/apache/iotdb/relational/it/schema/IoTDBDatabaseIT.java b/integration-test/src/test/java/org/apache/iotdb/relational/it/schema/IoTDBDatabaseIT.java index 3d24915c1fed..6d5f4f48872c 100644 --- a/integration-test/src/test/java/org/apache/iotdb/relational/it/schema/IoTDBDatabaseIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/relational/it/schema/IoTDBDatabaseIT.java @@ -535,6 +535,7 @@ public void testInformationSchema() throws SQLException { "functions,INF,", "keywords,INF,", "nodes,INF,", + "pipe_memory,INF,", "pipe_plugins,INF,", "pipes,INF,", "queries,INF,", @@ -616,6 +617,21 @@ public void testInformationSchema() throws SQLException { "estimated_remaining_seconds,DOUBLE,ATTRIBUTE,", "is_degraded,BOOLEAN,ATTRIBUTE,", "recent_failures,STRING,ATTRIBUTE,"))); + TestUtils.assertResultSetEqual( + statement.executeQuery("desc pipe_memory"), + "ColumnName,DataType,Category,", + new HashSet<>( + Arrays.asList( + "block_id,INT64,TAG,", + "name,STRING,TAG,", + "category,STRING,TAG,", + "memory_usage_in_bytes,INT64,ATTRIBUTE,", + "max_memory_size_in_bytes,INT64,ATTRIBUTE,", + "allocation_time,TIMESTAMP,ATTRIBUTE,", + "assigner,STRING,ATTRIBUTE,", + "parent_block_id,INT64,ATTRIBUTE,", + "hierarchy_level,INT32,ATTRIBUTE,", + "accounted_memory_usage_in_bytes,INT64,ATTRIBUTE,"))); TestUtils.assertResultSetEqual( statement.executeQuery("desc pipe_plugins"), "ColumnName,DataType,Category,", @@ -731,6 +747,9 @@ public void testInformationSchema() throws SQLException { Assert.assertThrows(SQLException.class, () -> statement.execute("select * from data_nodes")); Assert.assertThrows( SQLException.class, () -> statement.executeQuery("select * from pipe_plugins")); + Assert.assertThrows( + SQLException.class, () -> statement.executeQuery("select * from pipe_memory")); + Assert.assertThrows(SQLException.class, () -> statement.executeQuery("SHOW PIPE MEMORY")); Assert.assertThrows( SQLException.class, () -> statement.executeQuery("select * from table_disk_usage")); @@ -765,6 +784,40 @@ public void testInformationSchema() throws SQLException { // Test table query statement.execute("use information_schema"); + try (final ResultSet resultSet = statement.executeQuery("SHOW PIPE MEMORY")) { + final ResultSetMetaData metaData = resultSet.getMetaData(); + assertEquals(10, metaData.getColumnCount()); + assertEquals("block_id", metaData.getColumnName(1)); + assertEquals("name", metaData.getColumnName(2)); + assertEquals("category", metaData.getColumnName(3)); + assertEquals("memory_usage_in_bytes", metaData.getColumnName(4)); + assertEquals("max_memory_size_in_bytes", metaData.getColumnName(5)); + assertEquals("allocation_time", metaData.getColumnName(6)); + assertEquals("assigner", metaData.getColumnName(7)); + assertEquals("parent_block_id", metaData.getColumnName(8)); + assertEquals("hierarchy_level", metaData.getColumnName(9)); + assertEquals("accounted_memory_usage_in_bytes", metaData.getColumnName(10)); + boolean hasFloatingMemory = false; + while (resultSet.next()) { + if ("FloatingMemory".equals(resultSet.getString(2))) { + assertTrue(resultSet.getLong(4) >= 0); + hasFloatingMemory = true; + } + } + assertTrue(hasFloatingMemory); + } + try (final ResultSet resultSet = + statement.executeQuery("select * from information_schema.pipe_memory")) { + boolean hasFloatingMemory = false; + while (resultSet.next()) { + if ("FloatingMemory".equals(resultSet.getString(2))) { + assertTrue(resultSet.getLong(4) >= 0); + hasFloatingMemory = true; + } + } + assertTrue(hasFloatingMemory); + } + statement.execute("create database test"); statement.execute( "create table test.test (a tag, b attribute, c int32 comment 'turbine') comment 'test'"); @@ -813,6 +866,7 @@ public void testInformationSchema() throws SQLException { "information_schema,columns,INF,USING,null,SYSTEM VIEW,false,", "information_schema,queries,INF,USING,null,SYSTEM VIEW,false,", "information_schema,regions,INF,USING,null,SYSTEM VIEW,false,", + "information_schema,pipe_memory,INF,USING,null,SYSTEM VIEW,false,", "information_schema,topics,INF,USING,null,SYSTEM VIEW,false,", "information_schema,pipe_plugins,INF,USING,null,SYSTEM VIEW,false,", "information_schema,pipes,INF,USING,null,SYSTEM VIEW,false,", @@ -834,7 +888,7 @@ public void testInformationSchema() throws SQLException { TestUtils.assertResultSetEqual( statement.executeQuery("count devices from tables where status = 'USING'"), "count(devices),", - Collections.singleton("23,")); + Collections.singleton("24,")); TestUtils.assertResultSetEqual( statement.executeQuery( "select * from columns where table_name = 'queries' or database = 'test'"), diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/runtime/PipeDataNodeRuntimeAgent.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/runtime/PipeDataNodeRuntimeAgent.java index 83c115a704e1..8ca2ee1f52f6 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/runtime/PipeDataNodeRuntimeAgent.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/runtime/PipeDataNodeRuntimeAgent.java @@ -45,6 +45,7 @@ import org.apache.iotdb.db.pipe.resource.PipeDataNodeHardlinkOrCopiedFileDirStartupCleaner; import org.apache.iotdb.db.pipe.resource.PipeDataNodeResourceManager; import org.apache.iotdb.db.pipe.resource.memory.PipeMemoryBlock; +import org.apache.iotdb.db.pipe.resource.memory.PipeMemoryBlockCategory; import org.apache.iotdb.db.pipe.source.schemaregion.SchemaRegionListeningQueue; import org.apache.iotdb.db.queryengine.plan.analyze.cache.schema.DataNodeDevicePathCache; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertNode; @@ -102,7 +103,11 @@ private void initLoggerPeriodicalLogReducer() { if (pipeLogReducerMemoryBlock == null) { pipeLogReducerMemoryBlock = PipeDataNodeResourceManager.memory() - .tryAllocate(PipeConfig.getInstance().getPipeLoggerCacheMaxSizeInBytes()); + .tryAllocate( + PipeDataNodeRuntimeAgent.class.getSimpleName() + "#logger", + PipeConfig.getInstance().getPipeLoggerCacheMaxSizeInBytes(), + PipeMemoryBlockCategory.CACHE, + PipeDataNodeRuntimeAgent.class.getSimpleName()); } LoggerPeriodicalLogReducer.setMemoryResizeFunction( diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/PipeInsertionEvent.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/PipeInsertionEvent.java index abef9cee9785..bab43ce55131 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/PipeInsertionEvent.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/PipeInsertionEvent.java @@ -26,6 +26,7 @@ import org.apache.iotdb.commons.utils.PathUtils; import org.apache.iotdb.db.i18n.DataNodePipeMessages; import org.apache.iotdb.db.pipe.event.common.tsfile.PipeTsFileInsertionEvent; +import org.apache.iotdb.db.pipe.resource.memory.PipeMemoryBlock; import jakarta.validation.constraints.NotNull; @@ -50,6 +51,14 @@ */ public abstract class PipeInsertionEvent extends EnrichedEvent { + /** + * Returns the event-level memory block used as the parent of parser/converted-data blocks. Events + * that do not retain a dedicated block return {@code null}. + */ + public PipeMemoryBlock getEventMemoryBlock() { + return null; + } + // Record the database name of the DataRegion corresponding to the SourceEvent private final String sourceDatabaseNameFromDataRegion; diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/statement/PipeStatementInsertionEvent.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/statement/PipeStatementInsertionEvent.java index 9f812d90b3c8..1e8f9a8ab91e 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/statement/PipeStatementInsertionEvent.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/statement/PipeStatementInsertionEvent.java @@ -31,6 +31,7 @@ import org.apache.iotdb.db.pipe.event.common.PipeInsertionEvent; import org.apache.iotdb.db.pipe.metric.overview.PipeDataNodeSinglePipeMetrics; import org.apache.iotdb.db.pipe.resource.PipeDataNodeResourceManager; +import org.apache.iotdb.db.pipe.resource.memory.PipeMemoryBlockCategory; import org.apache.iotdb.db.pipe.resource.memory.PipeTabletMemoryBlock; import org.apache.iotdb.db.queryengine.plan.statement.Statement; import org.apache.iotdb.db.queryengine.plan.statement.crud.InsertBaseStatement; @@ -86,7 +87,14 @@ public PipeStatementInsertionEvent( this.statement = statement; // Allocate empty memory block, will be resized later. this.allocatedMemoryBlock = - PipeDataNodeResourceManager.memory().forceAllocateForTabletWithRetry(0); + PipeDataNodeResourceManager.memory() + .forceAllocateForTabletWithRetry( + PipeStatementInsertionEvent.class.getSimpleName(), + 0, + PipeMemoryBlockCategory.EVENT, + this, + null); + this.allocatedMemoryBlock.setAssigner(this); } @Override @@ -170,6 +178,11 @@ public Statement getStatement() { return statement; } + @Override + public PipeTabletMemoryBlock getEventMemoryBlock() { + return allocatedMemoryBlock; + } + /////////////////////////// Object /////////////////////////// @Override diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tablet/PipeInsertNodeTabletInsertionEvent.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tablet/PipeInsertNodeTabletInsertionEvent.java index 3bebbc911530..7c346ac3bfd0 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tablet/PipeInsertNodeTabletInsertionEvent.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tablet/PipeInsertNodeTabletInsertionEvent.java @@ -44,6 +44,8 @@ import org.apache.iotdb.db.pipe.metric.overview.PipeDataNodeSinglePipeMetrics; import org.apache.iotdb.db.pipe.resource.PipeDataNodeResourceManager; import org.apache.iotdb.db.pipe.resource.memory.InsertNodeMemoryEstimator; +import org.apache.iotdb.db.pipe.resource.memory.PipeMemoryBlock; +import org.apache.iotdb.db.pipe.resource.memory.PipeMemoryBlockCategory; import org.apache.iotdb.db.pipe.resource.memory.PipeMemoryWeightUtil; import org.apache.iotdb.db.pipe.resource.memory.PipeTabletMemoryBlock; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertNode; @@ -93,6 +95,7 @@ public class PipeInsertNodeTabletInsertionEvent extends PipeInsertionEvent + RamUsageEstimator.shallowSizeOf(Boolean.class); private final AtomicReference allocatedMemoryBlock; + private final PipeMemoryBlock eventMemoryBlock; private volatile List tablets; // Calculated together with tablets so downstream batching does not rescan Tablet internals. private volatile long tabletsMemoryUsageInBytes; @@ -159,6 +162,14 @@ public PipeInsertNodeTabletInsertionEvent( this.insertNode = insertNode; this.progressIndex = insertNode.getProgressIndex(); + this.eventMemoryBlock = + PipeDataNodeResourceManager.memory() + .forceAllocate( + PipeInsertNodeTabletInsertionEvent.class.getSimpleName(), + 0, + PipeMemoryBlockCategory.EVENT, + this, + null); this.allocatedMemoryBlock = new AtomicReference<>(); } @@ -498,7 +509,12 @@ public synchronized List convertToTablets() { allocatedMemoryBlock.compareAndSet( null, PipeDataNodeResourceManager.memory() - .forceAllocateForTabletWithRetry(tabletMemoryUsageInBytes)); + .forceAllocateForTabletWithRetry( + PipeInsertNodeTabletInsertionEvent.class.getSimpleName(), + tabletMemoryUsageInBytes, + PipeMemoryBlockCategory.TABLET, + this, + eventMemoryBlock)); } return tablets; } @@ -508,6 +524,11 @@ public long getTabletsMemoryUsageInBytes() { return tabletsMemoryUsageInBytes; } + @Override + public PipeMemoryBlock getEventMemoryBlock() { + return eventMemoryBlock; + } + /////////////////////////// event parser /////////////////////////// private List initEventParsers() { @@ -653,7 +674,7 @@ protected void trackResource() { @Override public PipeEventResource eventResourceBuilder() { return new PipeInsertNodeTabletInsertionEventResource( - this.isReleased, this.referenceCount, this.allocatedMemoryBlock); + this.isReleased, this.referenceCount, this.allocatedMemoryBlock, this.eventMemoryBlock); } // Notes: @@ -677,13 +698,16 @@ public long ramBytesUsed() { private static class PipeInsertNodeTabletInsertionEventResource extends PipeEventResource { private final AtomicReference allocatedMemoryBlock; + private final PipeMemoryBlock eventMemoryBlock; private PipeInsertNodeTabletInsertionEventResource( final AtomicBoolean isReleased, final AtomicInteger referenceCount, - final AtomicReference allocatedMemoryBlock) { + final AtomicReference allocatedMemoryBlock, + final PipeMemoryBlock eventMemoryBlock) { super(isReleased, referenceCount); this.allocatedMemoryBlock = allocatedMemoryBlock; + this.eventMemoryBlock = eventMemoryBlock; } @Override @@ -696,6 +720,7 @@ protected void finalizeResource() { } return null; }); + eventMemoryBlock.close(); } catch (final Exception e) { LOGGER.warn(DataNodePipeMessages.DECREASE_REFERENCE_COUNT_ERROR, e); } @@ -713,6 +738,7 @@ public synchronized void close() { } return null; }); + eventMemoryBlock.close(); tablets = null; } } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tablet/PipeRawTabletInsertionEvent.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tablet/PipeRawTabletInsertionEvent.java index dc2ab1d381fd..06243e40ec50 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tablet/PipeRawTabletInsertionEvent.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tablet/PipeRawTabletInsertionEvent.java @@ -38,6 +38,8 @@ import org.apache.iotdb.db.pipe.event.common.tsfile.PipeTsFileInsertionEvent; import org.apache.iotdb.db.pipe.metric.overview.PipeDataNodeSinglePipeMetrics; import org.apache.iotdb.db.pipe.resource.PipeDataNodeResourceManager; +import org.apache.iotdb.db.pipe.resource.memory.PipeMemoryBlock; +import org.apache.iotdb.db.pipe.resource.memory.PipeMemoryBlockCategory; import org.apache.iotdb.db.pipe.resource.memory.PipeMemoryWeightUtil; import org.apache.iotdb.db.pipe.resource.memory.PipeTabletMemoryBlock; import org.apache.iotdb.pipe.api.access.Row; @@ -115,8 +117,21 @@ private PipeRawTabletInsertionEvent( inheritSourceEventReportSkippingIfNecessary(); // Allocate empty memory block, will be resized later. + final PipeMemoryBlock parentMemoryBlock = + sourceEvent instanceof PipeInsertionEvent + ? ((PipeInsertionEvent) sourceEvent).getEventMemoryBlock() + : null; this.allocatedMemoryBlock = - PipeDataNodeResourceManager.memory().forceAllocateForTabletWithRetry(0); + PipeDataNodeResourceManager.memory() + .forceAllocateForTabletWithRetry( + PipeRawTabletInsertionEvent.class.getSimpleName(), + 0, + parentMemoryBlock == null + ? PipeMemoryBlockCategory.EVENT + : PipeMemoryBlockCategory.TABLET, + this, + parentMemoryBlock); + this.allocatedMemoryBlock.setAssigner(this); if (needToReport) { addOnCommittedHook( @@ -269,6 +284,11 @@ public boolean internallyIncreaseResourceReferenceCount(final String holderMessa return true; } + @Override + public PipeTabletMemoryBlock getEventMemoryBlock() { + return allocatedMemoryBlock; + } + @Override public boolean internallyDecreaseResourceReferenceCount(final String holderMessage) { if (Objects.nonNull(pipeName)) { diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeTsFileInsertionEvent.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeTsFileInsertionEvent.java index 335278e2f379..d77eaeea5152 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeTsFileInsertionEvent.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeTsFileInsertionEvent.java @@ -48,6 +48,8 @@ import org.apache.iotdb.db.pipe.event.common.tsfile.parser.TsFileInsertionEventParserProvider; import org.apache.iotdb.db.pipe.metric.overview.PipeDataNodeSinglePipeMetrics; import org.apache.iotdb.db.pipe.resource.PipeDataNodeResourceManager; +import org.apache.iotdb.db.pipe.resource.memory.PipeMemoryBlock; +import org.apache.iotdb.db.pipe.resource.memory.PipeMemoryBlockCategory; import org.apache.iotdb.db.pipe.resource.memory.PipeMemoryManager; import org.apache.iotdb.db.pipe.resource.memory.PipeMemoryManager.TsFileParserMemoryReservation; import org.apache.iotdb.db.pipe.resource.tsfile.PipeTsFileResourceManager; @@ -83,6 +85,10 @@ public class PipeTsFileInsertionEvent extends PipeInsertionEvent private static final Logger LOGGER = LoggerFactory.getLogger(PipeTsFileInsertionEvent.class); private final TsFileResource resource; + + /** Event-level aggregate for parser and generated tablet blocks. */ + private final PipeMemoryBlock eventMemoryBlock; + private final String dataRegionId; private File tsFile; private long extractTime = 0; @@ -293,6 +299,17 @@ private PipeTsFileInsertionEvent( this.eventParser = new AtomicReference<>(null); + // Allocate the event aggregate after all immutable event fields have been initialized so its + // diagnostic snapshot contains the complete event identity. + this.eventMemoryBlock = + PipeDataNodeResourceManager.memory() + .forceAllocate( + PipeTsFileInsertionEvent.class.getSimpleName(), + 0, + PipeMemoryBlockCategory.EVENT, + this, + null); + addOnCommittedHook( () -> { if (shouldReportOnCommit) { @@ -1208,6 +1225,12 @@ public void close() { return null; }); releaseTsFileParserMemoryIfReserved(); + eventMemoryBlock.close(); + } + + @Override + public PipeMemoryBlock getEventMemoryBlock() { + return eventMemoryBlock; } /////////////////////////// Object /////////////////////////// @@ -1251,7 +1274,8 @@ public PipeEventResource eventResourceBuilder() { this.sharedModFile, this.eventParser, this.isTsFileParserMemoryReserved, - this.tsFileParserMemoryReservationKey); + this.tsFileParserMemoryReservationKey, + this.eventMemoryBlock); } private static class PipeTsFileInsertionEventResource extends PipeEventResource { @@ -1266,6 +1290,7 @@ private static class PipeTsFileInsertionEventResource extends PipeEventResource private final String dataRegionId; private final AtomicBoolean isTsFileParserMemoryReserved; private final TsFileParserMemoryReservation tsFileParserMemoryReservationKey; + private final PipeMemoryBlock eventMemoryBlock; private PipeTsFileInsertionEventResource( final AtomicBoolean isReleased, @@ -1279,7 +1304,8 @@ private PipeTsFileInsertionEventResource( final File sharedModFile, final AtomicReference eventParser, final AtomicBoolean isTsFileParserMemoryReserved, - final TsFileParserMemoryReservation tsFileParserMemoryReservationKey) { + final TsFileParserMemoryReservation tsFileParserMemoryReservationKey, + final PipeMemoryBlock eventMemoryBlock) { super(isReleased, referenceCount); this.pipeName = pipeName; this.creationTime = creationTime; @@ -1291,6 +1317,7 @@ private PipeTsFileInsertionEventResource( this.eventParser = eventParser; this.isTsFileParserMemoryReserved = isTsFileParserMemoryReserved; this.tsFileParserMemoryReservationKey = tsFileParserMemoryReservationKey; + this.eventMemoryBlock = eventMemoryBlock; } @Override @@ -1323,6 +1350,7 @@ protected void finalizeResource() { .releaseTsFileParserMemory(pipeName, creationTime, dataRegionId); } } + eventMemoryBlock.close(); } catch (final Exception e) { LOGGER.warn( DataNodePipeMessages.DECREASE_REFERENCE_COUNT_FOR_TSFILE_ERROR, tsFile.getPath(), e); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/parser/TsFileInsertionEventParser.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/parser/TsFileInsertionEventParser.java index eefcafec9c53..2fb1a30d49b0 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/parser/TsFileInsertionEventParser.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/parser/TsFileInsertionEventParser.java @@ -28,6 +28,7 @@ import org.apache.iotdb.db.pipe.event.common.PipeInsertionEvent; import org.apache.iotdb.db.pipe.event.common.tsfile.parser.table.TsFileInsertionEventTableParser; import org.apache.iotdb.db.pipe.metric.overview.PipeTsFileToTabletsMetrics; +import org.apache.iotdb.db.pipe.resource.memory.PipeMemoryBlock; import org.apache.iotdb.db.pipe.resource.memory.PipeMemoryWeightUtil; import org.apache.iotdb.db.storageengine.dataregion.modification.ModEntry; import org.apache.iotdb.db.utils.datastructure.PatternTreeMapFactory; @@ -138,7 +139,8 @@ protected TsFileInsertionEventParser( this.sourceEvent = sourceEvent; this.memoryManager = memoryManager; - this.allocatedMemoryBlockForTablet = memoryManager.forceAllocateForTabletWithRetry(0); + this.allocatedMemoryBlockForTablet = + allocateTabletMemory(TsFileInsertionEventParser.class.getSimpleName() + "#tablet", 0); LOGGER.debug( DataNodePipeMessages.TSFILE_HAS_INITIALIZED_PIPENAME_CREATION_TIME_PATTERN, @@ -152,6 +154,21 @@ protected TsFileInsertionEventParser( isWithMod); } + protected PipeMemoryBlock getParentMemoryBlock() { + return sourceEvent == null ? null : sourceEvent.getEventMemoryBlock(); + } + + protected TsFileInsertionEventParserMemoryBlock allocateTabletMemory( + final String name, final long sizeInBytes) { + return memoryManager.forceAllocateForTabletWithRetry( + name, sizeInBytes, getParentMemoryBlock(), sourceEvent); + } + + protected TsFileInsertionEventParserMemoryBlock allocateGenericMemory( + final String name, final long sizeInBytes) { + return memoryManager.forceAllocate(name, sizeInBytes, getParentMemoryBlock(), sourceEvent); + } + /** * @return {@link TabletInsertionEvent} in a streaming way */ diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/parser/TsFileInsertionEventParserMemoryManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/parser/TsFileInsertionEventParserMemoryManager.java index 68669ce4b55d..5913d11deee7 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/parser/TsFileInsertionEventParserMemoryManager.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/parser/TsFileInsertionEventParserMemoryManager.java @@ -21,13 +21,33 @@ import org.apache.iotdb.db.pipe.resource.PipeDataNodeResourceManager; import org.apache.iotdb.db.pipe.resource.memory.PipeMemoryBlock; +import org.apache.iotdb.db.pipe.resource.memory.PipeMemoryBlockCategory; /** Allocates parser working memory from the pool owned by the caller. */ public interface TsFileInsertionEventParserMemoryManager { - TsFileInsertionEventParserMemoryBlock forceAllocateForTabletWithRetry(long sizeInBytes); + TsFileInsertionEventParserMemoryBlock forceAllocateForTabletWithRetry( + String name, long sizeInBytes); - TsFileInsertionEventParserMemoryBlock forceAllocate(long sizeInBytes); + TsFileInsertionEventParserMemoryBlock forceAllocate(String name, long sizeInBytes); + + /** Allocates a parser child block when the caller has an event-level parent. */ + default TsFileInsertionEventParserMemoryBlock forceAllocateForTabletWithRetry( + final String name, + final long sizeInBytes, + final PipeMemoryBlock parent, + final Object assigner) { + return forceAllocateForTabletWithRetry(name, sizeInBytes); + } + + /** Allocates a generic parser child block when the caller has an event-level parent. */ + default TsFileInsertionEventParserMemoryBlock forceAllocate( + final String name, + final long sizeInBytes, + final PipeMemoryBlock parent, + final Object assigner) { + return forceAllocate(name, sizeInBytes); + } static TsFileInsertionEventParserMemoryManager pipe() { return PipeHolder.INSTANCE; @@ -38,14 +58,50 @@ final class PipeHolder { new TsFileInsertionEventParserMemoryManager() { @Override public TsFileInsertionEventParserMemoryBlock forceAllocateForTabletWithRetry( - final long sizeInBytes) { + final String name, final long sizeInBytes) { + return new PipeBlock( + PipeDataNodeResourceManager.memory() + .forceAllocateForTabletWithRetry( + name, + sizeInBytes, + PipeMemoryBlockCategory.TABLET, + TsFileInsertionEventParser.class.getSimpleName())); + } + + @Override + public TsFileInsertionEventParserMemoryBlock forceAllocate( + final String name, final long sizeInBytes) { + return new PipeBlock( + PipeDataNodeResourceManager.memory() + .forceAllocate( + name, + sizeInBytes, + PipeMemoryBlockCategory.PARSER, + TsFileInsertionEventParser.class.getSimpleName())); + } + + @Override + public TsFileInsertionEventParserMemoryBlock forceAllocateForTabletWithRetry( + final String name, + final long sizeInBytes, + final PipeMemoryBlock parent, + final Object assigner) { return new PipeBlock( - PipeDataNodeResourceManager.memory().forceAllocateForTabletWithRetry(sizeInBytes)); + PipeDataNodeResourceManager.memory() + .forceAllocateForTabletWithRetry( + name, sizeInBytes, PipeMemoryBlockCategory.TABLET, assigner, parent)); } @Override - public TsFileInsertionEventParserMemoryBlock forceAllocate(final long sizeInBytes) { - return new PipeBlock(PipeDataNodeResourceManager.memory().forceAllocate(sizeInBytes)); + public TsFileInsertionEventParserMemoryBlock forceAllocate( + final String name, + final long sizeInBytes, + final PipeMemoryBlock parent, + final Object assigner) { + return new PipeBlock( + PipeDataNodeResourceManager.memory() + .forceAllocate( + name, sizeInBytes, PipeMemoryBlockCategory.PARSER, assigner, parent)); } }; } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/parser/query/TsFileInsertionEventQueryParser.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/parser/query/TsFileInsertionEventQueryParser.java index c26910b426f1..9f1de12162e8 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/parser/query/TsFileInsertionEventQueryParser.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/parser/query/TsFileInsertionEventQueryParser.java @@ -221,7 +221,9 @@ public TsFileInsertionEventQueryParser( ? ModsOperationUtil.loadModificationsFromTsFile(tsFile) : PatternTreeMapFactory.getModsPatternTreeMap(); allocatedMemoryBlockForModifications = - memoryManager.forceAllocateForTabletWithRetry(currentModifications.ramBytesUsed()); + allocateTabletMemory( + TsFileInsertionEventQueryParser.class.getSimpleName() + "#modifications", + currentModifications.ramBytesUsed()); final PipeTsFileResourceManager tsFileResourceManager = PipeDataNodeResourceManager.tsfile(); final Map> deviceMeasurementsMap; @@ -282,7 +284,10 @@ public TsFileInsertionEventQueryParser( memoryRequiredInBytes += PipeMemoryWeightUtil.memoryOfIDeviceID2StrList(deviceMeasurementsMap); } - allocatedMemoryBlock = memoryManager.forceAllocate(memoryRequiredInBytes); + allocatedMemoryBlock = + allocateGenericMemory( + TsFileInsertionEventQueryParser.class.getSimpleName() + "#metadata", + memoryRequiredInBytes); final Iterator>> iterator = deviceMeasurementsMap.entrySet().iterator(); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/parser/scan/TsFileInsertionEventScanParser.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/parser/scan/TsFileInsertionEventScanParser.java index e18494ea492c..427117f3f4d5 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/parser/scan/TsFileInsertionEventScanParser.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/parser/scan/TsFileInsertionEventScanParser.java @@ -181,8 +181,11 @@ public TsFileInsertionEventScanParser( this.endTime = endTime; filter = Objects.nonNull(timeFilterExpression) ? timeFilterExpression.getFilter() : null; - this.allocatedMemoryBlockForBatchData = memoryManager.forceAllocateForTabletWithRetry(0); - this.allocatedMemoryBlockForChunk = memoryManager.forceAllocateForTabletWithRetry(0); + this.allocatedMemoryBlockForBatchData = + allocateTabletMemory( + TsFileInsertionEventScanParser.class.getSimpleName() + "#batchData", 0); + this.allocatedMemoryBlockForChunk = + allocateTabletMemory(TsFileInsertionEventScanParser.class.getSimpleName() + "#chunk", 0); try { currentModifications = @@ -190,7 +193,9 @@ public TsFileInsertionEventScanParser( ? ModsOperationUtil.loadModificationsFromTsFile(tsFile) : PatternTreeMapFactory.getModsPatternTreeMap(); allocatedMemoryBlockForModifications = - memoryManager.forceAllocateForTabletWithRetry(currentModifications.ramBytesUsed()); + allocateTabletMemory( + TsFileInsertionEventScanParser.class.getSimpleName() + "#modifications", + currentModifications.ramBytesUsed()); tsFileSequenceReader = createTsFileSequenceReader(tsFile, !currentModifications.isEmpty()); tsFileSequenceReader.position((long) TSFileConfig.MAGIC_STRING.getBytes().length + 1); @@ -257,7 +262,9 @@ private TsFileSequenceReader createTsFileSequenceReader( } allocatedMemoryBlockForTsFileInput = - memoryManager.forceAllocateForTabletWithRetry(TS_FILE_INPUT_BUFFER_SIZE_IN_BYTES); + allocateTabletMemory( + TsFileInsertionEventScanParser.class.getSimpleName() + "#tsFileInput", + TS_FILE_INPUT_BUFFER_SIZE_IN_BYTES); return new TsFileSequenceReader( new BufferedTsFileInput(tsFile.toPath(), TS_FILE_INPUT_BUFFER_SIZE_IN_BYTES), false, diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/parser/table/TsFileInsertionEventTableParser.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/parser/table/TsFileInsertionEventTableParser.java index 0fd7470e0e21..1f32af7dd78d 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/parser/table/TsFileInsertionEventTableParser.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/parser/table/TsFileInsertionEventTableParser.java @@ -119,11 +119,20 @@ public TsFileInsertionEventTableParser( ? ModsOperationUtil.loadModificationsFromTsFile(tsFile) : PatternTreeMapFactory.getModsPatternTreeMap(); allocatedMemoryBlockForModifications = - memoryManager.forceAllocateForTabletWithRetry(currentModifications.ramBytesUsed()); - this.allocatedMemoryBlockForChunk = memoryManager.forceAllocateForTabletWithRetry(0); - this.allocatedMemoryBlockForBatchData = memoryManager.forceAllocateForTabletWithRetry(0); - this.allocatedMemoryBlockForChunkMeta = memoryManager.forceAllocateForTabletWithRetry(0); - this.allocatedMemoryBlockForTableSchemas = memoryManager.forceAllocateForTabletWithRetry(0); + allocateTabletMemory( + TsFileInsertionEventTableParser.class.getSimpleName() + "#modifications", + currentModifications.ramBytesUsed()); + this.allocatedMemoryBlockForChunk = + allocateTabletMemory(TsFileInsertionEventTableParser.class.getSimpleName() + "#chunk", 0); + this.allocatedMemoryBlockForBatchData = + allocateTabletMemory( + TsFileInsertionEventTableParser.class.getSimpleName() + "#batchData", 0); + this.allocatedMemoryBlockForChunkMeta = + allocateTabletMemory( + TsFileInsertionEventTableParser.class.getSimpleName() + "#chunkMetadata", 0); + this.allocatedMemoryBlockForTableSchemas = + allocateTabletMemory( + TsFileInsertionEventTableParser.class.getSimpleName() + "#tableSchemas", 0); this.startTime = startTime; this.endTime = endTime; diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/processor/downsampling/PartialPathLastObjectCache.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/processor/downsampling/PartialPathLastObjectCache.java index 225e0c3e86eb..d569d686148a 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/processor/downsampling/PartialPathLastObjectCache.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/processor/downsampling/PartialPathLastObjectCache.java @@ -22,6 +22,7 @@ import org.apache.iotdb.db.i18n.DataNodePipeMessages; import org.apache.iotdb.db.pipe.resource.PipeDataNodeResourceManager; import org.apache.iotdb.db.pipe.resource.memory.PipeMemoryBlock; +import org.apache.iotdb.db.pipe.resource.memory.PipeMemoryBlockCategory; import org.apache.iotdb.db.utils.MemUtils; import com.github.benmanes.caffeine.cache.Cache; @@ -39,7 +40,13 @@ public abstract class PartialPathLastObjectCache implements AutoCloseable { private final Cache partialPath2ObjectCache; protected PartialPathLastObjectCache(final long memoryLimitInBytes) { - allocatedMemoryBlock = PipeDataNodeResourceManager.memory().tryAllocate(memoryLimitInBytes); + allocatedMemoryBlock = + PipeDataNodeResourceManager.memory() + .tryAllocate( + PartialPathLastObjectCache.class.getSimpleName(), + memoryLimitInBytes, + PipeMemoryBlockCategory.CACHE, + PartialPathLastObjectCache.class.getSimpleName()); // Currently disable the metric here because it's not a constant cache and the number may // fluctuate. In the future all the "processorCache"s may be recorded in single metric entry diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/thrift/IoTDBDataNodeReceiver.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/thrift/IoTDBDataNodeReceiver.java index e7d4053ea64a..8c3586551e7f 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/thrift/IoTDBDataNodeReceiver.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/thrift/IoTDBDataNodeReceiver.java @@ -66,6 +66,7 @@ import org.apache.iotdb.db.pipe.receiver.visitor.PipeTreeStatementToBatchVisitor; import org.apache.iotdb.db.pipe.resource.PipeDataNodeResourceManager; import org.apache.iotdb.db.pipe.resource.memory.PipeMemoryBlock; +import org.apache.iotdb.db.pipe.resource.memory.PipeMemoryBlockCategory; import org.apache.iotdb.db.pipe.sink.payload.evolvable.request.PipeTransferDataNodeHandshakeV2Req; import org.apache.iotdb.db.pipe.sink.payload.evolvable.request.PipeTransferPlanNodeReq; import org.apache.iotdb.db.pipe.sink.payload.evolvable.request.PipeTransferSchemaSnapshotPieceReq; @@ -1109,7 +1110,11 @@ private void closeMemoryBlock(final PipeMemoryBlock memoryBlock) { private PipeMemoryBlock tryAllocateReceiverMemory(final long requestedMemorySizeInBytes) throws PipeRuntimeOutOfMemoryCriticalException { return PipeDataNodeResourceManager.memory() - .forceAllocate(Math.max(requestedMemorySizeInBytes, 0)); + .forceAllocate( + IoTDBDataNodeReceiver.class.getSimpleName() + "#request", + Math.max(requestedMemorySizeInBytes, 0), + PipeMemoryBlockCategory.RECEIVER, + IoTDBDataNodeReceiver.class.getSimpleName()); } @Override @@ -1189,7 +1194,10 @@ private TSStatus executeStatementAndClassifyExceptions( allocatedMemoryBlock = PipeDataNodeResourceManager.memory() .forceAllocate( - (long) (estimatedMemory * pipeReceiverActualToEstimatedMemoryRatio)); + IoTDBDataNodeReceiver.class.getSimpleName() + "#statement", + (long) (estimatedMemory * pipeReceiverActualToEstimatedMemoryRatio), + PipeMemoryBlockCategory.RECEIVER, + IoTDBDataNodeReceiver.class.getSimpleName()); break; } catch (final PipeRuntimeOutOfMemoryCriticalException e) { if (i == tryCount - 1) { diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/PipeFixedMemoryBlock.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/PipeFixedMemoryBlock.java index 47073fbbedd1..15bcef56ac81 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/PipeFixedMemoryBlock.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/PipeFixedMemoryBlock.java @@ -26,8 +26,18 @@ public abstract class PipeFixedMemoryBlock extends PipeMemoryBlock { - public PipeFixedMemoryBlock(long memoryUsageInBytes) { - super(memoryUsageInBytes); + public PipeFixedMemoryBlock(final String name, final long memoryUsageInBytes) { + super(name, memoryUsageInBytes); + } + + PipeFixedMemoryBlock( + final PipeMemoryManager pipeMemoryManager, + final String name, + final long memoryUsageInBytes, + final PipeMemoryBlockCategory category, + final String assigner, + final PipeMemoryBlock parent) { + super(pipeMemoryManager, name, memoryUsageInBytes, category, assigner, parent); } @Override diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryBlock.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryBlock.java index 72ba8e35ea0f..6c1f215b1d0d 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryBlock.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryBlock.java @@ -19,13 +19,16 @@ package org.apache.iotdb.db.pipe.resource.memory; +import org.apache.iotdb.commons.pipe.event.EnrichedEvent; import org.apache.iotdb.db.i18n.DataNodePipeMessages; import org.apache.iotdb.db.pipe.resource.PipeDataNodeResourceManager; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import java.util.Collections; import java.util.Objects; +import java.util.Set; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicLong; import java.util.concurrent.atomic.AtomicReference; @@ -37,11 +40,31 @@ public class PipeMemoryBlock implements AutoCloseable { private static final Logger LOGGER = LoggerFactory.getLogger(PipeMemoryBlock.class); - private final PipeMemoryManager pipeMemoryManager = PipeDataNodeResourceManager.memory(); + /** Maximum number of characters retained in an assigner diagnostic snapshot. */ + private static final int MAX_ASSIGNER_LENGTH = 2048; + + private static final AtomicLong NEXT_BLOCK_ID = new AtomicLong(1); + + private final PipeMemoryManager pipeMemoryManager; private final ReentrantLock lock = new ReentrantLock(); + private final long blockId; + private final String name; + private final PipeMemoryBlockCategory category; + // The child keeps its parent alive while it is in use so the accounting chain cannot be + // truncated by GC. The parent only keeps weak references to children, avoiding a parent-child + // retention cycle and allowing forgotten zero-sized children to be collected. + private final PipeMemoryBlock parent; + private final Set children = + Collections.newSetFromMap(new java.util.WeakHashMap<>()); + private final int hierarchyLevel; + private final long allocationTime; + private final AtomicReference assigner = new AtomicReference<>(); + private final AtomicLong memoryUsageInBytes = new AtomicLong(0); + // This is a high-water mark for observability, not a hard allocation limit. + private final AtomicLong maxMemorySizeInBytes = new AtomicLong(0); private final AtomicReference shrinkMethod = new AtomicReference<>(); private final AtomicReference> shrinkCallback = new AtomicReference<>(); @@ -50,8 +73,138 @@ public class PipeMemoryBlock implements AutoCloseable { private volatile boolean isReleased = false; - public PipeMemoryBlock(final long memoryUsageInBytes) { - this.memoryUsageInBytes.set(memoryUsageInBytes); + public PipeMemoryBlock(final String name, final long memoryUsageInBytes) { + this( + PipeDataNodeResourceManager.memory(), + name, + memoryUsageInBytes, + PipeMemoryBlockCategory.OTHER, + null, + null); + } + + PipeMemoryBlock( + final PipeMemoryManager pipeMemoryManager, + final String name, + final long memoryUsageInBytes, + final PipeMemoryBlockCategory category, + final String assigner, + final PipeMemoryBlock parent) { + this.pipeMemoryManager = Objects.requireNonNull(pipeMemoryManager); + this.blockId = NEXT_BLOCK_ID.getAndIncrement(); + this.name = Objects.requireNonNull(name); + this.category = category == null ? PipeMemoryBlockCategory.OTHER : category; + this.parent = parent; + if (parent != null) { + synchronized (parent.children) { + parent.children.add(this); + } + } + this.hierarchyLevel = parent == null ? 0 : parent.getHierarchyLevel() + 1; + this.allocationTime = System.currentTimeMillis(); + this.assigner.set(truncateAssigner(assigner)); + this.memoryUsageInBytes.set(Math.max(0, memoryUsageInBytes)); + this.maxMemorySizeInBytes.set(Math.max(0, memoryUsageInBytes)); + } + + /** Returns the globally unique identifier of this block instance. */ + public long getBlockId() { + return blockId; + } + + public String getName() { + return name; + } + + public PipeMemoryBlockCategory getCategory() { + return category; + } + + public PipeMemoryBlock getParentBlock() { + return parent; + } + + public Long getParentBlockId() { + final PipeMemoryBlock parentBlock = getParentBlock(); + return parentBlock == null ? null : parentBlock.getBlockId(); + } + + public int getHierarchyLevel() { + return hierarchyLevel; + } + + public long getAllocationTime() { + return allocationTime; + } + + public long getAllocationTimeInMillis() { + return allocationTime; + } + + public String getAssigner() { + return assigner.get(); + } + + /** + * Replace the assigner diagnostic snapshot. The supplied object is converted immediately; no + * event object is retained by the memory block. + */ + public PipeMemoryBlock setAssigner(final Object assignerObject) { + String snapshot = null; + if (assignerObject != null) { + try { + if (assignerObject instanceof EnrichedEvent) { + snapshot = ((EnrichedEvent) assignerObject).coreReportMessage(); + // Some event implementations intentionally return no core message while they are being + // constructed. Keep a useful diagnostic value in that case instead of losing the + // assigner altogether. + if (snapshot == null) { + snapshot = String.valueOf(assignerObject); + } + } else { + snapshot = String.valueOf(assignerObject); + } + } catch (final Exception ignored) { + snapshot = assignerObject.getClass().getSimpleName(); + } + } + assigner.set(truncateAssigner(snapshot)); + return this; + } + + public boolean isRootBlock() { + return parent == null; + } + + Set getChildrenSnapshot() { + synchronized (children) { + return Set.copyOf(children); + } + } + + void removeFromParent() { + final PipeMemoryBlock parentBlock = getParentBlock(); + if (parentBlock != null) { + synchronized (parentBlock.children) { + parentBlock.children.remove(this); + } + } + } + + /** Returns bytes charged to the global pool by this row (children report zero). */ + public long getAccountedMemoryUsageInBytes() { + return isRootBlock() ? getMemoryUsageInBytes() : 0; + } + + PipeMemoryManager getPipeMemoryManager() { + return pipeMemoryManager; + } + + private static String truncateAssigner(final String value) { + if (value == null || value.length() <= MAX_ASSIGNER_LENGTH) { + return value; + } + return value.substring(0, MAX_ASSIGNER_LENGTH); } public long getMemoryUsageInBytes() { @@ -59,7 +212,13 @@ public long getMemoryUsageInBytes() { } public void setMemoryUsageInBytes(final long memoryUsageInBytes) { - this.memoryUsageInBytes.set(memoryUsageInBytes); + final long normalizedMemoryUsageInBytes = Math.max(0, memoryUsageInBytes); + this.memoryUsageInBytes.set(normalizedMemoryUsageInBytes); + maxMemorySizeInBytes.accumulateAndGet(normalizedMemoryUsageInBytes, Math::max); + } + + public long getMaxMemorySizeInBytes() { + return maxMemorySizeInBytes.get(); } public PipeMemoryBlock setShrinkMethod(final LongUnaryOperator shrinkMethod) { @@ -85,6 +244,9 @@ public PipeMemoryBlock setExpandCallback(final BiConsumer expandCall } boolean shrink() { + if (isReleased) { + return false; + } if (lock.tryLock()) { try { return doShrink(); @@ -120,6 +282,9 @@ private boolean doShrink() { } boolean expand() { + if (isReleased) { + return false; + } if (lock.tryLock()) { try { return doExpand(); @@ -165,8 +330,22 @@ void markAsReleased() { @Override public String toString() { return "PipeMemoryBlock{" - + "usedMemoryInBytes=" + + "blockId=" + + blockId + + ", category=" + + category + + ", name='" + + name + + '\'' + + ", usedMemoryInBytes=" + memoryUsageInBytes.get() + + ", maxMemorySizeInBytes=" + + maxMemorySizeInBytes.get() + + ", parentBlockId=" + + getParentBlockId() + + ", assigner='" + + assigner.get() + + '\'' + ", isReleased=" + isReleased + '}'; diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryBlockCategory.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryBlockCategory.java new file mode 100644 index 000000000000..06adfa90010d --- /dev/null +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryBlockCategory.java @@ -0,0 +1,61 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.iotdb.db.pipe.resource.memory; + +/** + * The diagnostic category of a Pipe memory block. + * + *

The category is intentionally independent from {@link PipeMemoryBlockType}. The latter + * controls allocation policy, while this enum describes the owner visible to operators. + */ +public enum PipeMemoryBlockCategory { + GLOBAL, + EVENT, + EVENT_CHILD, + PARSER, + TABLET, + TS_FILE, + BATCH, + WAL, + CACHE, + RECEIVER, + SINK, + SUBSCRIPTION, + FLOATING, + OTHER; + + public static PipeMemoryBlockCategory fromType(final PipeMemoryBlockType type) { + if (type == null) { + return OTHER; + } + switch (type) { + case TABLET: + return TABLET; + case TS_FILE: + return TS_FILE; + case BATCH: + return BATCH; + case WAL: + return WAL; + default: + return OTHER; + } + } +} diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryManager.java index 90a45c1542ec..f49f2c079132 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryManager.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryManager.java @@ -23,6 +23,7 @@ import org.apache.iotdb.commons.memory.IMemoryBlock; import org.apache.iotdb.commons.memory.MemoryBlockType; import org.apache.iotdb.commons.pipe.config.PipeConfig; +import org.apache.iotdb.commons.pipe.event.EnrichedEvent; import org.apache.iotdb.db.conf.IoTDBDescriptor; import org.apache.iotdb.db.i18n.DataNodePipeMessages; import org.apache.iotdb.db.pipe.agent.PipeDataNodeAgent; @@ -34,6 +35,7 @@ import java.util.ArrayDeque; import java.util.ArrayList; import java.util.Collections; +import java.util.Comparator; import java.util.HashMap; import java.util.HashSet; import java.util.LinkedHashSet; @@ -48,6 +50,8 @@ public class PipeMemoryManager { private static final Logger LOGGER = LoggerFactory.getLogger(PipeMemoryManager.class); + public static final String FLOATING_MEMORY_BLOCK_NAME = "FloatingMemory"; + private static final PipeConfig PIPE_CONFIG = PipeConfig.getInstance(); private static final boolean PIPE_MEMORY_MANAGEMENT_ENABLED = @@ -58,6 +62,9 @@ public class PipeMemoryManager { private final LongSupplier floatingMemoryUsageSupplier; + private final long floatingMemoryAllocationTime = System.currentTimeMillis(); + private volatile long floatingMemoryMaxUsageInBytes; + private static final double EXCEED_PROTECT_THRESHOLD = 0.95; private volatile long usedMemorySizeInBytesOfTablets; @@ -76,7 +83,12 @@ public class PipeMemoryManager { private final ArrayDeque waitingTsFileParserPipeOrder = new ArrayDeque<>(); private PipeIdentity lastAdmittedWaitingTsFileParserPipe; - // Only non-zero memory blocks will be added to this set. + // All unreleased memory blocks, including zero-sized blocks, are kept for inspection. The + // manager owns this registry until an explicit close, which also guarantees that parent close + // can cascade to every child and release the global charge exactly once. + private final Set memoryBlocks = new HashSet<>(); + + // Only non-zero memory blocks will be added to this set for memory accounting. private final Set allocatedBlocks = new HashSet<>(); private final Set shrinkableBlocks = new HashSet<>(); private final Set expandableBlocks = new HashSet<>(); @@ -503,29 +515,114 @@ < allowedMaxMemorySizeInBytesOfTabletsAndTsFiles() return true; } - public synchronized PipeMemoryBlock forceAllocate(long sizeInBytes) + public synchronized PipeMemoryBlock forceAllocate(final String name, final long sizeInBytes) + throws PipeRuntimeOutOfMemoryCriticalException { + return forceAllocate(name, sizeInBytes, PipeMemoryBlockCategory.OTHER, null, null); + } + + /** + * Backward-compatible allocation entry point for callers that do not provide a diagnostic name. + * Such blocks intentionally fall back to the generic category and a null assigner. + */ + @Deprecated + public synchronized PipeMemoryBlock forceAllocate(final long sizeInBytes) + throws PipeRuntimeOutOfMemoryCriticalException { + return forceAllocate(PipeMemoryBlock.class.getSimpleName(), sizeInBytes); + } + + /** + * Allocate a named block with explicit diagnostic metadata. The maximum size recorded for the + * block is its lifetime high-water mark; it is not a new hard limit. + */ + public synchronized PipeMemoryBlock forceAllocate( + final String name, + final long sizeInBytes, + final PipeMemoryBlockCategory category, + final Object assigner, + final PipeMemoryBlock parent) throws PipeRuntimeOutOfMemoryCriticalException { if (!PIPE_MEMORY_MANAGEMENT_ENABLED) { // No need to calculate the tablet size, skip it to save time - return new PipeMemoryBlock(0); + return registerMemoryBlock(name, 0, PipeMemoryBlockType.NORMAL, category, assigner, parent); } if (sizeInBytes == 0) { - return registerMemoryBlock(0); + return registerMemoryBlock(name, 0, PipeMemoryBlockType.NORMAL, category, assigner, parent); } - return forceAllocateWithRetry(sizeInBytes, PipeMemoryBlockType.NORMAL); + return forceAllocateWithRetry( + name, sizeInBytes, PipeMemoryBlockType.NORMAL, category, assigner, parent); } - public PipeTabletMemoryBlock forceAllocateForTabletWithRetry(long tabletSizeInBytes) + /** Convenience overload for callers that only need a category and an assigner. */ + public synchronized PipeMemoryBlock forceAllocate( + final String name, + final long sizeInBytes, + final PipeMemoryBlockCategory category, + final Object assigner) + throws PipeRuntimeOutOfMemoryCriticalException { + return forceAllocate(name, sizeInBytes, category, assigner, null); + } + + /** Allocate a child block using the supplied event block as its accounting parent. */ + public synchronized PipeMemoryBlock forceAllocate( + final String name, + final long sizeInBytes, + final PipeMemoryBlock parent, + final Object assigner) + throws PipeRuntimeOutOfMemoryCriticalException { + return forceAllocate( + name, + sizeInBytes, + parent == null ? PipeMemoryBlockCategory.OTHER : PipeMemoryBlockCategory.EVENT_CHILD, + assigner, + parent); + } + + public synchronized PipeMemoryBlock forceAllocate( + final PipeMemoryBlock parent, final String name, final long sizeInBytes) + throws PipeRuntimeOutOfMemoryCriticalException { + return forceAllocate(name, sizeInBytes, PipeMemoryBlockCategory.EVENT_CHILD, null, parent); + } + + /** Allocate a child block and charge its bytes through the parent to the global pool. */ + public synchronized PipeMemoryBlock forceAllocateChild( + final PipeMemoryBlock parent, final String name, final long sizeInBytes) + throws PipeRuntimeOutOfMemoryCriticalException { + return forceAllocate(parent, name, sizeInBytes); + } + + public PipeTabletMemoryBlock forceAllocateForTabletWithRetry( + final String name, final long tabletSizeInBytes) + throws PipeRuntimeOutOfMemoryCriticalException { + return forceAllocateForTabletWithRetry( + name, tabletSizeInBytes, PipeMemoryBlockCategory.TABLET, null, null); + } + + /** Backward-compatible tablet allocation entry point without diagnostic metadata. */ + @Deprecated + public PipeTabletMemoryBlock forceAllocateForTabletWithRetry(final long tabletSizeInBytes) + throws PipeRuntimeOutOfMemoryCriticalException { + return forceAllocateForTabletWithRetry( + PipeTabletMemoryBlock.class.getSimpleName(), tabletSizeInBytes); + } + + public PipeTabletMemoryBlock forceAllocateForTabletWithRetry( + final String name, + final long tabletSizeInBytes, + final PipeMemoryBlockCategory category, + final Object assigner, + final PipeMemoryBlock parent) throws PipeRuntimeOutOfMemoryCriticalException { if (!PIPE_MEMORY_MANAGEMENT_ENABLED) { // No need to calculate the tablet size, skip it to save time - return new PipeTabletMemoryBlock(0); + return (PipeTabletMemoryBlock) + registerMemoryBlock(name, 0, PipeMemoryBlockType.TABLET, category, assigner, parent); } if (tabletSizeInBytes == 0) { - return (PipeTabletMemoryBlock) registerMemoryBlock(0, PipeMemoryBlockType.TABLET); + return (PipeTabletMemoryBlock) + registerMemoryBlock(name, 0, PipeMemoryBlockType.TABLET, category, assigner, parent); } for (int i = 1, size = PIPE_CONFIG.getPipeMemoryAllocateMaxRetries(); i <= size; i++) { @@ -557,20 +654,74 @@ public PipeTabletMemoryBlock forceAllocateForTabletWithRetry(long tabletSizeInBy synchronized (this) { final PipeTabletMemoryBlock block = (PipeTabletMemoryBlock) - forceAllocateWithRetry(tabletSizeInBytes, PipeMemoryBlockType.TABLET); - usedMemorySizeInBytesOfTablets += block.getMemoryUsageInBytes(); + forceAllocateWithRetry( + name, tabletSizeInBytes, PipeMemoryBlockType.TABLET, category, assigner, parent); return block; } } - public PipeTsFileMemoryBlock forceAllocateForTsFileWithRetry(long tsFileSizeInBytes) + /** Convenience overload for a root tablet block with explicit diagnostic metadata. */ + public PipeTabletMemoryBlock forceAllocateForTabletWithRetry( + final String name, + final long tabletSizeInBytes, + final PipeMemoryBlockCategory category, + final Object assigner) + throws PipeRuntimeOutOfMemoryCriticalException { + return forceAllocateForTabletWithRetry(name, tabletSizeInBytes, category, assigner, null); + } + + /** Allocate a tablet/parser child block using the supplied event block as its parent. */ + public PipeTabletMemoryBlock forceAllocateForTabletWithRetry( + final String name, + final long tabletSizeInBytes, + final PipeMemoryBlock parent, + final Object assigner) + throws PipeRuntimeOutOfMemoryCriticalException { + return forceAllocateForTabletWithRetry( + name, + tabletSizeInBytes, + parent == null ? PipeMemoryBlockCategory.TABLET : PipeMemoryBlockCategory.EVENT_CHILD, + assigner, + parent); + } + + public PipeTabletMemoryBlock forceAllocateForTabletWithRetry( + final PipeMemoryBlock parent, final String name, final long tabletSizeInBytes) + throws PipeRuntimeOutOfMemoryCriticalException { + return forceAllocateForTabletWithRetry( + name, tabletSizeInBytes, PipeMemoryBlockCategory.EVENT_CHILD, null, parent); + } + + public PipeTsFileMemoryBlock forceAllocateForTsFileWithRetry( + final String name, final long tsFileSizeInBytes) + throws PipeRuntimeOutOfMemoryCriticalException { + return forceAllocateForTsFileWithRetry( + name, tsFileSizeInBytes, PipeMemoryBlockCategory.TS_FILE, null, null); + } + + /** Backward-compatible TsFile allocation entry point without diagnostic metadata. */ + @Deprecated + public PipeTsFileMemoryBlock forceAllocateForTsFileWithRetry(final long tsFileSizeInBytes) + throws PipeRuntimeOutOfMemoryCriticalException { + return forceAllocateForTsFileWithRetry( + PipeTsFileMemoryBlock.class.getSimpleName(), tsFileSizeInBytes); + } + + public PipeTsFileMemoryBlock forceAllocateForTsFileWithRetry( + final String name, + final long tsFileSizeInBytes, + final PipeMemoryBlockCategory category, + final Object assigner, + final PipeMemoryBlock parent) throws PipeRuntimeOutOfMemoryCriticalException { if (!PIPE_MEMORY_MANAGEMENT_ENABLED) { - return new PipeTsFileMemoryBlock(0); + return (PipeTsFileMemoryBlock) + registerMemoryBlock(name, 0, PipeMemoryBlockType.TS_FILE, category, assigner, parent); } if (tsFileSizeInBytes == 0) { - return (PipeTsFileMemoryBlock) registerMemoryBlock(0, PipeMemoryBlockType.TS_FILE); + return (PipeTsFileMemoryBlock) + registerMemoryBlock(name, 0, PipeMemoryBlockType.TS_FILE, category, assigner, parent); } for (int i = 1, size = PIPE_CONFIG.getPipeMemoryAllocateMaxRetries(); i <= size; i++) { @@ -602,21 +753,53 @@ public PipeTsFileMemoryBlock forceAllocateForTsFileWithRetry(long tsFileSizeInBy synchronized (this) { final PipeTsFileMemoryBlock block = (PipeTsFileMemoryBlock) - forceAllocateWithRetry(tsFileSizeInBytes, PipeMemoryBlockType.TS_FILE); - usedMemorySizeInBytesOfTsFiles += block.getMemoryUsageInBytes(); + forceAllocateWithRetry( + name, tsFileSizeInBytes, PipeMemoryBlockType.TS_FILE, category, assigner, parent); return block; } } + /** Convenience overload for a root TsFile block with explicit diagnostic metadata. */ + public PipeTsFileMemoryBlock forceAllocateForTsFileWithRetry( + final String name, + final long tsFileSizeInBytes, + final PipeMemoryBlockCategory category, + final Object assigner) + throws PipeRuntimeOutOfMemoryCriticalException { + return forceAllocateForTsFileWithRetry(name, tsFileSizeInBytes, category, assigner, null); + } + + /** Allocate a TsFile/parser child block using the supplied event block as its parent. */ + public PipeTsFileMemoryBlock forceAllocateForTsFileWithRetry( + final String name, + final long tsFileSizeInBytes, + final PipeMemoryBlock parent, + final Object assigner) + throws PipeRuntimeOutOfMemoryCriticalException { + return forceAllocateForTsFileWithRetry( + name, + tsFileSizeInBytes, + parent == null ? PipeMemoryBlockCategory.TS_FILE : PipeMemoryBlockCategory.EVENT_CHILD, + assigner, + parent); + } + + public PipeTsFileMemoryBlock forceAllocateForTsFileWithRetry( + final PipeMemoryBlock parent, final String name, final long tsFileSizeInBytes) + throws PipeRuntimeOutOfMemoryCriticalException { + return forceAllocateForTsFileWithRetry( + name, tsFileSizeInBytes, PipeMemoryBlockCategory.EVENT_CHILD, null, parent); + } + public PipeModelFixedMemoryBlock forceAllocateForModelFixedMemoryBlock( - long fixedSizeInBytes, PipeMemoryBlockType type) + final String name, final long fixedSizeInBytes, final PipeMemoryBlockType type) throws PipeRuntimeOutOfMemoryCriticalException { if (!PIPE_MEMORY_MANAGEMENT_ENABLED) { - return new PipeModelFixedMemoryBlock(Long.MAX_VALUE, new ThresholdAllocationStrategy()); + return (PipeModelFixedMemoryBlock) registerMemoryBlock(name, Long.MAX_VALUE, type); } if (fixedSizeInBytes == 0) { - return (PipeModelFixedMemoryBlock) registerMemoryBlock(0, type); + return (PipeModelFixedMemoryBlock) registerMemoryBlock(name, 0, type); } for (int i = 1, size = PIPE_CONFIG.getPipeMemoryAllocateMaxRetries(); i <= size; i++) { @@ -637,34 +820,47 @@ public PipeModelFixedMemoryBlock forceAllocateForModelFixedMemoryBlock( synchronized (this) { if (getFreeMemorySizeInBytes() < fixedSizeInBytes) { - return (PipeModelFixedMemoryBlock) forceAllocateWithRetry(getFreeMemorySizeInBytes(), type); + return (PipeModelFixedMemoryBlock) + forceAllocateWithRetry(name, getFreeMemorySizeInBytes(), type); } - return (PipeModelFixedMemoryBlock) forceAllocateWithRetry(fixedSizeInBytes, type); + return (PipeModelFixedMemoryBlock) forceAllocateWithRetry(name, fixedSizeInBytes, type); } } - private PipeMemoryBlock forceAllocateWithRetry(long sizeInBytes, PipeMemoryBlockType type) + /** Backward-compatible fixed-block allocation entry point without a diagnostic name. */ + @Deprecated + public PipeModelFixedMemoryBlock forceAllocateForModelFixedMemoryBlock( + final long fixedSizeInBytes, final PipeMemoryBlockType type) + throws PipeRuntimeOutOfMemoryCriticalException { + return forceAllocateForModelFixedMemoryBlock( + PipeModelFixedMemoryBlock.class.getSimpleName(), fixedSizeInBytes, type); + } + + private PipeMemoryBlock forceAllocateWithRetry( + final String name, final long sizeInBytes, final PipeMemoryBlockType type) + throws PipeRuntimeOutOfMemoryCriticalException { + return forceAllocateWithRetry( + name, sizeInBytes, type, PipeMemoryBlockCategory.fromType(type), null, null); + } + + private PipeMemoryBlock forceAllocateWithRetry( + final String name, + final long sizeInBytes, + final PipeMemoryBlockType type, + final PipeMemoryBlockCategory category, + final Object assigner, + final PipeMemoryBlock parent) throws PipeRuntimeOutOfMemoryCriticalException { if (!PIPE_MEMORY_MANAGEMENT_ENABLED) { - switch (type) { - case TABLET: - return new PipeTabletMemoryBlock(sizeInBytes); - case TS_FILE: - return new PipeTsFileMemoryBlock(sizeInBytes); - case BATCH: - case WAL: - return new PipeModelFixedMemoryBlock(sizeInBytes, new ThresholdAllocationStrategy()); - default: - return new PipeMemoryBlock(sizeInBytes); - } + return registerMemoryBlock(name, sizeInBytes, type, category, assigner, parent); } final int memoryAllocateMaxRetries = PIPE_CONFIG.getPipeMemoryAllocateMaxRetries(); for (int i = 1; i <= memoryAllocateMaxRetries; i++) { if (getTotalNonFloatingMemorySizeInBytes() - memoryBlock.getUsedMemoryInBytes() >= sizeInBytes) { - return registerMemoryBlock(sizeInBytes, type); + return registerMemoryBlock(name, sizeInBytes, type, category, assigner, parent); } try { @@ -698,34 +894,28 @@ public synchronized void resize( return; } + // Parent blocks expose an aggregate usage. Do not let a direct resize reduce that aggregate + // below the bytes still owned by live children. + final long normalizedTargetSize = + Math.max(Math.max(0, targetSize), getChildMemoryUsageInBytes(block)); + if (!PIPE_MEMORY_MANAGEMENT_ENABLED) { - block.setMemoryUsageInBytes(targetSize); + final long delta = normalizedTargetSize - block.getMemoryUsageInBytes(); + adjustMemoryUsageHierarchy(block, delta); return; } final long oldSize = block.getMemoryUsageInBytes(); - if (oldSize >= targetSize) { - memoryBlock.release(oldSize - targetSize); - if (block instanceof PipeTabletMemoryBlock) { - usedMemorySizeInBytesOfTablets -= oldSize - targetSize; - } - if (block instanceof PipeTsFileMemoryBlock) { - usedMemorySizeInBytesOfTsFiles -= oldSize - targetSize; - } - block.setMemoryUsageInBytes(targetSize); - - // If no memory is used in the block, we can remove it from the allocated blocks. - if (targetSize == 0) { - allocatedBlocks.remove(block); - } + if (oldSize >= normalizedTargetSize) { + releaseMemoryForBlock(block, oldSize - normalizedTargetSize); notifyNextTsFileParserMemoryReservationInternal(); this.notifyAll(); return; } - long sizeInBytes = targetSize - oldSize; + long sizeInBytes = normalizedTargetSize - oldSize; final int memoryAllocateMaxRetries = PIPE_CONFIG.getPipeMemoryAllocateMaxRetries(); for (int i = 1; i <= memoryAllocateMaxRetries; i++) { // Dynamically resized data-structure blocks must obey the same admission thresholds as @@ -735,19 +925,7 @@ public synchronized void resize( && getTotalNonFloatingMemorySizeInBytes() - memoryBlock.getUsedMemoryInBytes() >= sizeInBytes) { memoryBlock.forceAllocateWithoutLimitation(sizeInBytes); - if (oldSize == 0) { - // If the memory block is not registered, we need to register it first. - // Otherwise, the memory usage will be inconsistent. - // See registerMemoryBlock for more details. - allocatedBlocks.add(block); - } - if (block instanceof PipeTabletMemoryBlock) { - usedMemorySizeInBytesOfTablets += sizeInBytes; - } - if (block instanceof PipeTsFileMemoryBlock) { - usedMemorySizeInBytesOfTsFiles += sizeInBytes; - } - block.setMemoryUsageInBytes(targetSize); + adjustMemoryUsageHierarchy(block, sizeInBytes); return; } @@ -783,41 +961,110 @@ && getTotalNonFloatingMemorySizeInBytes() - memoryBlock.getUsedMemoryInBytes() * usedThreshold}. Will return a memory block otherwise. */ public synchronized PipeMemoryBlock forceAllocateIfSufficient( - long sizeInBytes, float usedThreshold) { + final String name, final long sizeInBytes, final float usedThreshold) { + return forceAllocateIfSufficient( + name, sizeInBytes, usedThreshold, PipeMemoryBlockCategory.OTHER, null, null); + } + + /** Backward-compatible threshold allocation entry point without a diagnostic name. */ + @Deprecated + public synchronized PipeMemoryBlock forceAllocateIfSufficient( + final long sizeInBytes, final float usedThreshold) { + return forceAllocateIfSufficient( + PipeMemoryBlock.class.getSimpleName(), sizeInBytes, usedThreshold); + } + + /** + * Allocate a block subject to a usage threshold while retaining diagnostic metadata. + * + *

The threshold applies to the actual global pool charge. A parent, when supplied, receives + * the same allocation as an aggregate value but is not charged a second time. + */ + public synchronized PipeMemoryBlock forceAllocateIfSufficient( + final String name, + final long sizeInBytes, + final float usedThreshold, + final PipeMemoryBlockCategory category, + final Object assigner, + final PipeMemoryBlock parent) { if (usedThreshold < 0.0f || usedThreshold > 1.0f) { return null; } if (!PIPE_MEMORY_MANAGEMENT_ENABLED) { - return new PipeMemoryBlock(sizeInBytes); + return registerMemoryBlock( + name, sizeInBytes, PipeMemoryBlockType.NORMAL, category, assigner, parent); } if (sizeInBytes == 0) { - return registerMemoryBlock(0); + return registerMemoryBlock(name, 0, PipeMemoryBlockType.NORMAL, category, assigner, parent); } if ((float) (memoryBlock.getUsedMemoryInBytes() + sizeInBytes) <= getTotalNonFloatingMemorySizeInBytes() * usedThreshold) { - return forceAllocate(sizeInBytes); + return forceAllocate(name, sizeInBytes, category, assigner, parent); } return null; } - public synchronized PipeMemoryBlock tryAllocate(long sizeInBytes) { - return tryAllocate(sizeInBytes, currentSize -> currentSize * 2 / 3); + public synchronized PipeMemoryBlock tryAllocate(final String name, final long sizeInBytes) { + return tryAllocate( + name, + sizeInBytes, + currentSize -> currentSize * 2 / 3, + PipeMemoryBlockCategory.OTHER, + null, + null); + } + + /** Backward-compatible gradual allocation entry point without a diagnostic name. */ + @Deprecated + public synchronized PipeMemoryBlock tryAllocate(final long sizeInBytes) { + return tryAllocate(PipeMemoryBlock.class.getSimpleName(), sizeInBytes); } public synchronized PipeMemoryBlock tryAllocate( - long sizeInBytes, LongUnaryOperator customAllocateStrategy) { + final String name, final long sizeInBytes, final LongUnaryOperator customAllocateStrategy) { + return tryAllocate( + name, sizeInBytes, customAllocateStrategy, PipeMemoryBlockCategory.OTHER, null, null); + } + + /** Backward-compatible gradual allocation entry point without a diagnostic name. */ + @Deprecated + public synchronized PipeMemoryBlock tryAllocate( + final long sizeInBytes, final LongUnaryOperator customAllocateStrategy) { + return tryAllocate(PipeMemoryBlock.class.getSimpleName(), sizeInBytes, customAllocateStrategy); + } + + /** Convenience overload using the default gradual-allocation strategy. */ + public synchronized PipeMemoryBlock tryAllocate( + final String name, + final long sizeInBytes, + final PipeMemoryBlockCategory category, + final Object assigner) { + return tryAllocate( + name, sizeInBytes, currentSize -> currentSize * 2 / 3, category, assigner, null); + } + + /** Try to allocate a block with explicit diagnostic metadata. */ + public synchronized PipeMemoryBlock tryAllocate( + final String name, + final long sizeInBytes, + final LongUnaryOperator customAllocateStrategy, + final PipeMemoryBlockCategory category, + final Object assigner, + final PipeMemoryBlock parent) { if (!PIPE_MEMORY_MANAGEMENT_ENABLED) { - return new PipeMemoryBlock(sizeInBytes); + return registerMemoryBlock( + name, sizeInBytes, PipeMemoryBlockType.NORMAL, category, assigner, parent); } if (sizeInBytes == 0 || getTotalNonFloatingMemorySizeInBytes() - memoryBlock.getUsedMemoryInBytes() >= sizeInBytes) { - return registerMemoryBlock(sizeInBytes); + return registerMemoryBlock( + name, sizeInBytes, PipeMemoryBlockType.NORMAL, category, assigner, parent); } long sizeToAllocateInBytes = sizeInBytes; @@ -832,7 +1079,8 @@ public synchronized PipeMemoryBlock tryAllocate( memoryBlock.getUsedMemoryInBytes(), sizeInBytes, sizeToAllocateInBytes); - return registerMemoryBlock(sizeToAllocateInBytes); + return registerMemoryBlock( + name, sizeToAllocateInBytes, PipeMemoryBlockType.NORMAL, category, assigner, parent); } sizeToAllocateInBytes = @@ -848,76 +1096,247 @@ public synchronized PipeMemoryBlock tryAllocate( memoryBlock.getUsedMemoryInBytes(), sizeInBytes, sizeToAllocateInBytes); - return registerMemoryBlock(sizeToAllocateInBytes); + return registerMemoryBlock( + name, sizeToAllocateInBytes, PipeMemoryBlockType.NORMAL, category, assigner, parent); } else { LOGGER.warn( DataNodePipeMessages.TRYALLOCATE_FAILED_TO_ALLOCATE_MEMORY_TOTAL_MEMORY, getTotalNonFloatingMemorySizeInBytes(), memoryBlock.getUsedMemoryInBytes(), sizeInBytes); - return registerMemoryBlock(0); + return registerMemoryBlock(name, 0, PipeMemoryBlockType.NORMAL, category, assigner, parent); } } public synchronized boolean tryAllocate( PipeMemoryBlock block, long memoryInBytesNeededToBeAllocated) { - if (!PIPE_MEMORY_MANAGEMENT_ENABLED || block == null || block.isReleased()) { + if (!PIPE_MEMORY_MANAGEMENT_ENABLED + || block == null + || block.isReleased() + || memoryInBytesNeededToBeAllocated <= 0) { return false; } - if (getTotalNonFloatingMemorySizeInBytes() - memoryBlock.getUsedMemoryInBytes() - >= memoryInBytesNeededToBeAllocated) { - memoryBlock.forceAllocateWithoutLimitation(memoryInBytesNeededToBeAllocated); - if (block.getMemoryUsageInBytes() == 0) { - allocatedBlocks.add(block); - } - if (block instanceof PipeTabletMemoryBlock) { - usedMemorySizeInBytesOfTablets += memoryInBytesNeededToBeAllocated; - } - if (block instanceof PipeTsFileMemoryBlock) { - usedMemorySizeInBytesOfTsFiles += memoryInBytesNeededToBeAllocated; - } - block.setMemoryUsageInBytes(block.getMemoryUsageInBytes() + memoryInBytesNeededToBeAllocated); - return true; - } + return reserveMemoryForBlock(block, memoryInBytesNeededToBeAllocated); + } - return false; + private PipeMemoryBlock registerMemoryBlock(final String name, final long sizeInBytes) { + return registerMemoryBlock(name, sizeInBytes, PipeMemoryBlockType.NORMAL); } - private PipeMemoryBlock registerMemoryBlock(long sizeInBytes) { - return registerMemoryBlock(sizeInBytes, PipeMemoryBlockType.NORMAL); + private synchronized PipeMemoryBlock registerMemoryBlock( + final String name, final long sizeInBytes, final PipeMemoryBlockType type) { + return registerMemoryBlock( + name, sizeInBytes, type, PipeMemoryBlockCategory.fromType(type), null, null); } - private PipeMemoryBlock registerMemoryBlock(long sizeInBytes, PipeMemoryBlockType type) { + private synchronized PipeMemoryBlock registerMemoryBlock( + final String name, + final long sizeInBytes, + final PipeMemoryBlockType type, + final PipeMemoryBlockCategory category, + final Object assigner, + final PipeMemoryBlock parent) { + // Never link blocks owned by another manager (or an already released parent). Such a link + // would make release charge the wrong global pool and could leave an orphaned accounting + // chain. Falling back to a root block keeps the allocation observable and safe. + final PipeMemoryBlock normalizedParent = + parent != null && parent.getPipeMemoryManager() == this && !parent.isReleased() + ? parent + : null; + final PipeMemoryBlockCategory normalizedCategory = + inferCategory(name, type, category, normalizedParent); final PipeMemoryBlock returnedMemoryBlock; switch (type) { case TABLET: - returnedMemoryBlock = new PipeTabletMemoryBlock(sizeInBytes); + returnedMemoryBlock = + new PipeTabletMemoryBlock( + this, name, 0, normalizedCategory, snapshotAssigner(assigner), normalizedParent); break; case TS_FILE: - returnedMemoryBlock = new PipeTsFileMemoryBlock(sizeInBytes); + returnedMemoryBlock = + new PipeTsFileMemoryBlock( + this, name, 0, normalizedCategory, snapshotAssigner(assigner), normalizedParent); break; case BATCH: case WAL: returnedMemoryBlock = - new PipeModelFixedMemoryBlock(sizeInBytes, new ThresholdAllocationStrategy()); + new PipeModelFixedMemoryBlock( + this, + name, + 0, + new ThresholdAllocationStrategy(), + normalizedCategory, + snapshotAssigner(assigner), + normalizedParent); break; default: - returnedMemoryBlock = new PipeMemoryBlock(sizeInBytes); + returnedMemoryBlock = + new PipeMemoryBlock( + this, name, 0, normalizedCategory, snapshotAssigner(assigner), normalizedParent); break; } - // For memory block whose size is 0, we do not need to add it to the allocated blocks now. - // It's good for performance and will not trigger concurrent issues. - // If forceResize is called on it, we will add it to the allocated blocks. + memoryBlocks.add(returnedMemoryBlock); + + // Zero-sized blocks do not participate in memory accounting until they are resized. For a + // child block, the same bytes are charged once through the root block while every ancestor is + // updated for diagnostics. if (sizeInBytes > 0) { - memoryBlock.forceAllocateWithoutLimitation(sizeInBytes); - allocatedBlocks.add(returnedMemoryBlock); + if (!reserveMemoryForBlock(returnedMemoryBlock, sizeInBytes)) { + // Callers normally check availability before registering. Keep the block observable even + // if a concurrent allocation wins the race; it starts at zero and can be resized later. + returnedMemoryBlock.setMemoryUsageInBytes(0); + } } return returnedMemoryBlock; } + private static String snapshotAssigner(final Object assigner) { + if (assigner == null) { + return null; + } + try { + if (assigner instanceof EnrichedEvent) { + final String coreReportMessage = ((EnrichedEvent) assigner).coreReportMessage(); + return coreReportMessage == null ? String.valueOf(assigner) : coreReportMessage; + } + return assigner instanceof String ? (String) assigner : String.valueOf(assigner); + } catch (final Exception ignored) { + return assigner.getClass().getSimpleName(); + } + } + + private static PipeMemoryBlockCategory inferCategory( + final String name, + final PipeMemoryBlockType type, + final PipeMemoryBlockCategory requestedCategory, + final PipeMemoryBlock parent) { + if (parent != null) { + return requestedCategory == null || requestedCategory == PipeMemoryBlockCategory.OTHER + ? PipeMemoryBlockCategory.EVENT_CHILD + : requestedCategory; + } + if (requestedCategory != null && requestedCategory != PipeMemoryBlockCategory.OTHER) { + return requestedCategory; + } + final String normalizedName = name == null ? "" : name.toLowerCase(java.util.Locale.ROOT); + if (normalizedName.contains("parser")) { + return PipeMemoryBlockCategory.PARSER; + } + if (normalizedName.contains("receiver")) { + return PipeMemoryBlockCategory.RECEIVER; + } + if (normalizedName.contains("sink")) { + return PipeMemoryBlockCategory.SINK; + } + if (normalizedName.contains("cache") || normalizedName.contains("logger")) { + return PipeMemoryBlockCategory.CACHE; + } + if (normalizedName.contains("subscription")) { + return PipeMemoryBlockCategory.SUBSCRIPTION; + } + if (normalizedName.contains("event")) { + return PipeMemoryBlockCategory.EVENT; + } + return PipeMemoryBlockCategory.fromType(type); + } + + private boolean reserveMemoryForBlock(final PipeMemoryBlock block, final long sizeInBytes) { + if (sizeInBytes <= 0 || block == null || block.isReleased()) { + return sizeInBytes == 0 && block != null && !block.isReleased(); + } + if (PIPE_MEMORY_MANAGEMENT_ENABLED + && getTotalNonFloatingMemorySizeInBytes() - memoryBlock.getUsedMemoryInBytes() + < sizeInBytes) { + return false; + } + if (PIPE_MEMORY_MANAGEMENT_ENABLED) { + memoryBlock.forceAllocateWithoutLimitation(sizeInBytes); + } + adjustMemoryUsageHierarchy(block, sizeInBytes); + return true; + } + + private void adjustMemoryUsageHierarchy(final PipeMemoryBlock block, final long delta) { + PipeMemoryBlock current = block; + while (current != null) { + current.setMemoryUsageInBytes(safeAdd(current.getMemoryUsageInBytes(), delta)); + current = current.getParentBlock(); + } + + if (PIPE_MEMORY_MANAGEMENT_ENABLED && delta > 0) { + final PipeMemoryBlock root = getRootBlock(block); + allocatedBlocks.add(root); + } else if (PIPE_MEMORY_MANAGEMENT_ENABLED + && delta < 0 + && getRootBlock(block).getMemoryUsageInBytes() == 0) { + allocatedBlocks.remove(getRootBlock(block)); + } + + if (block instanceof PipeTabletMemoryBlock) { + usedMemorySizeInBytesOfTablets = safeAdd(usedMemorySizeInBytesOfTablets, delta); + } + if (block instanceof PipeTsFileMemoryBlock) { + usedMemorySizeInBytesOfTsFiles = safeAdd(usedMemorySizeInBytesOfTsFiles, delta); + } + } + + /** Saturating add keeps diagnostic counters valid even for the unbounded disabled-mode block. */ + private static long safeAdd(final long value, final long delta) { + if (delta > 0 && value > Long.MAX_VALUE - delta) { + return Long.MAX_VALUE; + } + if (delta < 0 && value < Long.MIN_VALUE - delta) { + return Long.MIN_VALUE; + } + return value + delta; + } + + private static PipeMemoryBlock getRootBlock(final PipeMemoryBlock block) { + PipeMemoryBlock root = block; + PipeMemoryBlock parent = root.getParentBlock(); + while (parent != null) { + root = parent; + parent = root.getParentBlock(); + } + return root; + } + + private boolean releaseMemoryForBlock( + final PipeMemoryBlock block, final long requestedSizeInBytes) { + if (block == null || block.isReleased()) { + return false; + } + // A parent row reports an aggregate usage. Only bytes owned directly by that row may be + // released here; descendants are released through their own blocks (or by release(parent)'s + // cascade). This prevents a parent resize from stealing a child's global reservation. + final long directMemoryUsageInBytes = getDirectMemoryUsageInBytes(block); + final long sizeInBytes = Math.min(Math.max(0, requestedSizeInBytes), directMemoryUsageInBytes); + if (sizeInBytes <= 0) { + return false; + } + if (PIPE_MEMORY_MANAGEMENT_ENABLED) { + memoryBlock.release(sizeInBytes); + } + adjustMemoryUsageHierarchy(block, -sizeInBytes); + return true; + } + + private static long getDirectMemoryUsageInBytes(final PipeMemoryBlock block) { + final long childUsageInBytes = getChildMemoryUsageInBytes(block); + return Math.max(0, block.getMemoryUsageInBytes() - childUsageInBytes); + } + + private static long getChildMemoryUsageInBytes(final PipeMemoryBlock block) { + long childUsageInBytes = 0; + for (final PipeMemoryBlock child : block.getChildrenSnapshot()) { + childUsageInBytes = safeAdd(childUsageInBytes, child.getMemoryUsageInBytes()); + } + return Math.max(0, childUsageInBytes); + } + // Single-threaded logic private boolean tryShrinkUntilFreeMemorySatisfy(long sizeInBytes) { final List shuffledBlocks = new ArrayList<>(shrinkableBlocks); @@ -963,9 +1382,9 @@ public synchronized void tryExpandAllAndCheckConsistency() { } final long tabletBlockSum = - allocatedBlocks.stream() + memoryBlocks.stream() .filter(PipeTabletMemoryBlock.class::isInstance) - .mapToLong(PipeMemoryBlock::getMemoryUsageInBytes) + .mapToLong(PipeMemoryManager::getDirectMemoryUsageInBytes) .sum(); if (tabletBlockSum != usedMemorySizeInBytesOfTablets) { LOGGER.debug( @@ -975,9 +1394,9 @@ public synchronized void tryExpandAllAndCheckConsistency() { } final long tsFileBlockSum = - allocatedBlocks.stream() + memoryBlocks.stream() .filter(PipeTsFileMemoryBlock.class::isInstance) - .mapToLong(PipeMemoryBlock::getMemoryUsageInBytes) + .mapToLong(PipeMemoryManager::getDirectMemoryUsageInBytes) .sum(); if (tsFileBlockSum != usedMemorySizeInBytesOfTsFiles) { LOGGER.debug( @@ -997,18 +1416,25 @@ void removeExpandableBlock(final PipeMemoryBlock block) { } public synchronized void release(PipeMemoryBlock block) { - if (!PIPE_MEMORY_MANAGEMENT_ENABLED || block == null || block.isReleased()) { + if (block == null || block.isReleased()) { return; } - allocatedBlocks.remove(block); - memoryBlock.release(block.getMemoryUsageInBytes()); - if (block instanceof PipeTabletMemoryBlock) { - usedMemorySizeInBytesOfTablets -= block.getMemoryUsageInBytes(); + // A parent owns the lifetime of its descendants. Release children first so their bytes are + // removed from the parent aggregate before the parent itself is released. + for (final PipeMemoryBlock child : block.getChildrenSnapshot()) { + release(child); } - if (block instanceof PipeTsFileMemoryBlock) { - usedMemorySizeInBytesOfTsFiles -= block.getMemoryUsageInBytes(); + // A cascaded release does not invoke each child's close() method. Remove every released block + // from the shrink/expand registries here so the periodic maintenance task cannot touch it. + shrinkableBlocks.remove(block); + expandableBlocks.remove(block); + memoryBlocks.remove(block); + releaseMemoryForBlock(block, block.getMemoryUsageInBytes()); + if (PIPE_MEMORY_MANAGEMENT_ENABLED && getRootBlock(block).getMemoryUsageInBytes() == 0) { + allocatedBlocks.remove(getRootBlock(block)); } + block.removeFromParent(); block.markAsReleased(); notifyNextTsFileParserMemoryReservationInternal(); @@ -1016,19 +1442,12 @@ public synchronized void release(PipeMemoryBlock block) { } public synchronized boolean release(PipeMemoryBlock block, long sizeInBytes) { - if (!PIPE_MEMORY_MANAGEMENT_ENABLED || block == null || block.isReleased()) { + // Keep the historical disabled-mode behavior: dynamic shrink callbacks do not participate in + // memory management when the feature is turned off. A full close still removes diagnostics. + if (!PIPE_MEMORY_MANAGEMENT_ENABLED || !releaseMemoryForBlock(block, sizeInBytes)) { return false; } - memoryBlock.release(sizeInBytes); - if (block instanceof PipeTabletMemoryBlock) { - usedMemorySizeInBytesOfTablets -= sizeInBytes; - } - if (block instanceof PipeTsFileMemoryBlock) { - usedMemorySizeInBytesOfTsFiles -= sizeInBytes; - } - block.setMemoryUsageInBytes(block.getMemoryUsageInBytes() - sizeInBytes); - notifyNextTsFileParserMemoryReservationInternal(); this.notifyAll(); @@ -1072,13 +1491,148 @@ public long getTotalFloatingMemorySizeInBytes() { } private long getUsedFloatingMemorySizeInBytes() { - return Math.max(0, floatingMemoryUsageSupplier.getAsLong()); + final long usageInBytes = Math.max(0, floatingMemoryUsageSupplier.getAsLong()); + floatingMemoryMaxUsageInBytes = Math.max(floatingMemoryMaxUsageInBytes, usageInBytes); + return usageInBytes; } public long getTotalMemorySizeInBytes() { return memoryBlock.getTotalMemorySizeInBytes(); } + public synchronized List getPipeMemoryBlockInfoList() { + final List memoryBlockInfoList = new ArrayList<>(); + memoryBlocks.forEach( + block -> + memoryBlockInfoList.add( + new PipeMemoryBlockInfo( + block.getBlockId(), + block.getName(), + block.getCategory().name(), + block.getMemoryUsageInBytes(), + block.getMaxMemorySizeInBytes(), + block.getAllocationTimeInMillis(), + block.getAssigner(), + block.getParentBlockId(), + block.getHierarchyLevel(), + block.getAccountedMemoryUsageInBytes()))); + final long floatingMemoryUsageInBytes = getUsedFloatingMemorySizeInBytes(); + floatingMemoryMaxUsageInBytes = + Math.max(floatingMemoryMaxUsageInBytes, floatingMemoryUsageInBytes); + memoryBlockInfoList.add( + new PipeMemoryBlockInfo( + 0, + FLOATING_MEMORY_BLOCK_NAME, + PipeMemoryBlockCategory.FLOATING.name(), + floatingMemoryUsageInBytes, + floatingMemoryMaxUsageInBytes, + floatingMemoryAllocationTime, + "PipeMemoryManager", + null, + 0, + floatingMemoryUsageInBytes)); + memoryBlockInfoList.sort( + Comparator.comparing(PipeMemoryBlockInfo::getName) + .thenComparingLong(PipeMemoryBlockInfo::getMemoryUsageInBytes) + .thenComparingLong(PipeMemoryBlockInfo::getBlockId)); + return memoryBlockInfoList; + } + + public static final class PipeMemoryBlockInfo { + + private final long blockId; + private final String name; + private final String category; + private final long memoryUsageInBytes; + private final long maxMemorySizeInBytes; + private final long allocationTime; + private final String assigner; + private final Long parentBlockId; + private final int hierarchyLevel; + private final long accountedMemoryUsageInBytes; + + private PipeMemoryBlockInfo( + final long blockId, + final String name, + final String category, + final long memoryUsageInBytes, + final long maxMemorySizeInBytes, + final long allocationTime, + final String assigner, + final Long parentBlockId, + final int hierarchyLevel, + final long accountedMemoryUsageInBytes) { + this.blockId = blockId; + this.name = name; + this.category = category; + this.memoryUsageInBytes = memoryUsageInBytes; + this.maxMemorySizeInBytes = maxMemorySizeInBytes; + this.allocationTime = allocationTime; + this.assigner = assigner; + this.parentBlockId = parentBlockId; + this.hierarchyLevel = hierarchyLevel; + this.accountedMemoryUsageInBytes = accountedMemoryUsageInBytes; + } + + private PipeMemoryBlockInfo(final String name, final long memoryUsageInBytes) { + this( + 0, + name, + PipeMemoryBlockCategory.OTHER.name(), + memoryUsageInBytes, + memoryUsageInBytes, + System.currentTimeMillis(), + null, + null, + 0, + memoryUsageInBytes); + } + + public long getBlockId() { + return blockId; + } + + public String getName() { + return name; + } + + public String getCategory() { + return category; + } + + public long getMemoryUsageInBytes() { + return memoryUsageInBytes; + } + + public long getMaxMemorySizeInBytes() { + return maxMemorySizeInBytes; + } + + public long getAllocationTime() { + return allocationTime; + } + + public long getAllocationTimeInMillis() { + return allocationTime; + } + + public String getAssigner() { + return assigner; + } + + public Long getParentBlockId() { + return parentBlockId; + } + + public int getHierarchyLevel() { + return hierarchyLevel; + } + + public long getAccountedMemoryUsageInBytes() { + return accountedMemoryUsageInBytes; + } + } + private static class PipeIdentity { private final String pipeName; diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/PipeModelFixedMemoryBlock.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/PipeModelFixedMemoryBlock.java index 90b3d0329f15..11b84af8d6f0 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/PipeModelFixedMemoryBlock.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/PipeModelFixedMemoryBlock.java @@ -37,8 +37,23 @@ public class PipeModelFixedMemoryBlock extends PipeFixedMemoryBlock { private volatile long memoryAllocatedInBytes; public PipeModelFixedMemoryBlock( - final long memoryUsageInBytes, final DynamicMemoryAllocationStrategy allocationStrategy) { - super(memoryUsageInBytes); + final String name, + final long memoryUsageInBytes, + final DynamicMemoryAllocationStrategy allocationStrategy) { + super(name, memoryUsageInBytes); + this.memoryAllocatedInBytes = 0; + this.allocationStrategy = allocationStrategy; + } + + PipeModelFixedMemoryBlock( + final PipeMemoryManager pipeMemoryManager, + final String name, + final long memoryUsageInBytes, + final DynamicMemoryAllocationStrategy allocationStrategy, + final PipeMemoryBlockCategory category, + final String assigner, + final PipeMemoryBlock parent) { + super(pipeMemoryManager, name, memoryUsageInBytes, category, assigner, parent); this.memoryAllocatedInBytes = 0; this.allocationStrategy = allocationStrategy; } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/PipeTabletMemoryBlock.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/PipeTabletMemoryBlock.java index 529a2e1ac5c5..a87d58e66b39 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/PipeTabletMemoryBlock.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/PipeTabletMemoryBlock.java @@ -21,7 +21,17 @@ public class PipeTabletMemoryBlock extends PipeFixedMemoryBlock { - public PipeTabletMemoryBlock(long memoryUsageInBytes) { - super(memoryUsageInBytes); + public PipeTabletMemoryBlock(final String name, final long memoryUsageInBytes) { + super(name, memoryUsageInBytes); + } + + PipeTabletMemoryBlock( + final PipeMemoryManager pipeMemoryManager, + final String name, + final long memoryUsageInBytes, + final PipeMemoryBlockCategory category, + final String assigner, + final PipeMemoryBlock parent) { + super(pipeMemoryManager, name, memoryUsageInBytes, category, assigner, parent); } } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/PipeTsFileMemoryBlock.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/PipeTsFileMemoryBlock.java index 268388d08000..58ecc51d504d 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/PipeTsFileMemoryBlock.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/PipeTsFileMemoryBlock.java @@ -21,7 +21,17 @@ public class PipeTsFileMemoryBlock extends PipeFixedMemoryBlock { - public PipeTsFileMemoryBlock(long memoryUsageInBytes) { - super(memoryUsageInBytes); + public PipeTsFileMemoryBlock(final String name, final long memoryUsageInBytes) { + super(name, memoryUsageInBytes); + } + + PipeTsFileMemoryBlock( + final PipeMemoryManager pipeMemoryManager, + final String name, + final long memoryUsageInBytes, + final PipeMemoryBlockCategory category, + final String assigner, + final PipeMemoryBlock parent) { + super(pipeMemoryManager, name, memoryUsageInBytes, category, assigner, parent); } } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/tsfile/PipeTsFilePublicResource.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/tsfile/PipeTsFilePublicResource.java index fe54ee48b547..5f7279d074af 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/tsfile/PipeTsFilePublicResource.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/tsfile/PipeTsFilePublicResource.java @@ -23,6 +23,7 @@ import org.apache.iotdb.db.i18n.DataNodePipeMessages; import org.apache.iotdb.db.pipe.resource.PipeDataNodeResourceManager; import org.apache.iotdb.db.pipe.resource.memory.PipeMemoryBlock; +import org.apache.iotdb.db.pipe.resource.memory.PipeMemoryBlockCategory; import org.apache.iotdb.db.pipe.resource.memory.PipeMemoryWeightUtil; import org.apache.tsfile.enums.TSDataType; @@ -117,8 +118,12 @@ synchronized boolean cacheDeviceIsAlignedMapIfAbsent(final File tsFile) throws I allocatedMemoryBlock = PipeDataNodeResourceManager.memory() .forceAllocateIfSufficient( + PipeTsFilePublicResource.class.getSimpleName() + "#sequenceReader", PipeConfig.getInstance().getPipeMemoryAllocateForTsFileSequenceReaderInBytes(), - MEMORY_SUFFICIENT_THRESHOLD); + MEMORY_SUFFICIENT_THRESHOLD, + PipeMemoryBlockCategory.TS_FILE, + PipeTsFilePublicResource.class.getSimpleName(), + null); if (allocatedMemoryBlock == null) { LOGGER.info( DataNodePipeMessages.FAILED_TO_CACHEDEVICEISALIGNEDMAPIFABSENT_FOR_TSFILE_BECAUSE_MEMORY, @@ -145,7 +150,13 @@ synchronized boolean cacheDeviceIsAlignedMapIfAbsent(final File tsFile) throws I // Allocate again for the cached objects. allocatedMemoryBlock = PipeDataNodeResourceManager.memory() - .forceAllocateIfSufficient(memoryRequiredInBytes, MEMORY_SUFFICIENT_THRESHOLD); + .forceAllocateIfSufficient( + PipeTsFilePublicResource.class.getSimpleName() + "#metadata", + memoryRequiredInBytes, + MEMORY_SUFFICIENT_THRESHOLD, + PipeMemoryBlockCategory.CACHE, + PipeTsFilePublicResource.class.getSimpleName(), + null); if (allocatedMemoryBlock == null) { LOGGER.info( DataNodePipeMessages.PIPETSFILERESOURCE_FAILED_TO_CACHE_OBJECTS_FOR_TSFILE, @@ -177,8 +188,12 @@ synchronized boolean cacheObjectsIfAbsent(final File tsFile) throws IOException allocatedMemoryBlock = PipeDataNodeResourceManager.memory() .forceAllocateIfSufficient( + PipeTsFilePublicResource.class.getSimpleName() + "#sequenceReader", PipeConfig.getInstance().getPipeMemoryAllocateForTsFileSequenceReaderInBytes(), - MEMORY_SUFFICIENT_THRESHOLD); + MEMORY_SUFFICIENT_THRESHOLD, + PipeMemoryBlockCategory.TS_FILE, + PipeTsFilePublicResource.class.getSimpleName(), + null); if (allocatedMemoryBlock == null) { LOGGER.info( DataNodePipeMessages.FAILED_TO_CACHEOBJECTSIFABSENT_FOR_TSFILE_BECAUSE_MEMORY, @@ -214,7 +229,13 @@ synchronized boolean cacheObjectsIfAbsent(final File tsFile) throws IOException // Allocate again for the cached objects. allocatedMemoryBlock = PipeDataNodeResourceManager.memory() - .forceAllocateIfSufficient(memoryRequiredInBytes, MEMORY_SUFFICIENT_THRESHOLD); + .forceAllocateIfSufficient( + PipeTsFilePublicResource.class.getSimpleName() + "#metadata", + memoryRequiredInBytes, + MEMORY_SUFFICIENT_THRESHOLD, + PipeMemoryBlockCategory.CACHE, + PipeTsFilePublicResource.class.getSimpleName(), + null); if (allocatedMemoryBlock == null) { LOGGER.info( DataNodePipeMessages.PIPETSFILERESOURCE_FAILED_TO_CACHE_OBJECTS_FOR_TSFILE, diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/client/IoTDBDataNodeCacheLeaderClientManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/client/IoTDBDataNodeCacheLeaderClientManager.java index f32c8cb72bbd..8ea2131395a8 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/client/IoTDBDataNodeCacheLeaderClientManager.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/client/IoTDBDataNodeCacheLeaderClientManager.java @@ -24,6 +24,7 @@ import org.apache.iotdb.db.i18n.DataNodePipeMessages; import org.apache.iotdb.db.pipe.resource.PipeDataNodeResourceManager; import org.apache.iotdb.db.pipe.resource.memory.PipeMemoryBlock; +import org.apache.iotdb.db.pipe.resource.memory.PipeMemoryBlockCategory; import com.github.benmanes.caffeine.cache.Cache; import com.github.benmanes.caffeine.cache.Caffeine; @@ -53,7 +54,12 @@ public LeaderCacheManager() { // properties required by pipe memory control framework final PipeMemoryBlock allocatedMemoryBlock = - PipeDataNodeResourceManager.memory().tryAllocate(initMemorySizeInBytes); + PipeDataNodeResourceManager.memory() + .tryAllocate( + IoTDBDataNodeCacheLeaderClientManager.class.getSimpleName(), + initMemorySizeInBytes, + PipeMemoryBlockCategory.CACHE, + IoTDBDataNodeCacheLeaderClientManager.class.getSimpleName()); device2endpoint = Caffeine.newBuilder() diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/batch/PipeSchemaRegionWritePlanEventBatch.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/batch/PipeSchemaRegionWritePlanEventBatch.java index 39a861c01b82..6d14d1aa50e9 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/batch/PipeSchemaRegionWritePlanEventBatch.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/batch/PipeSchemaRegionWritePlanEventBatch.java @@ -27,6 +27,7 @@ import org.apache.iotdb.db.pipe.event.common.schema.PipeSchemaRegionWritePlanEvent; import org.apache.iotdb.db.pipe.resource.PipeDataNodeResourceManager; import org.apache.iotdb.db.pipe.resource.memory.PipeMemoryBlock; +import org.apache.iotdb.db.pipe.resource.memory.PipeMemoryBlockCategory; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.metadata.write.ActivateTemplateNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.metadata.write.BatchActivateTemplateNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.metadata.write.CreateAlignedTimeSeriesNode; @@ -112,7 +113,13 @@ public PipeSchemaRegionWritePlanEventBatch(final PipeParameters parameters) { parameters.getLongOrDefault( Arrays.asList(CONNECTOR_IOTDB_BATCH_SIZE_KEY, SINK_IOTDB_BATCH_SIZE_KEY), CONNECTOR_IOTDB_PLAIN_BATCH_SIZE_DEFAULT_VALUE); - allocatedMemoryBlock = PipeDataNodeResourceManager.memory().forceAllocate(maxBatchSizeInBytes); + allocatedMemoryBlock = + PipeDataNodeResourceManager.memory() + .forceAllocate( + PipeSchemaRegionWritePlanEventBatch.class.getSimpleName(), + maxBatchSizeInBytes, + PipeMemoryBlockCategory.BATCH, + PipeSchemaRegionWritePlanEventBatch.class.getSimpleName()); } public synchronized boolean onEvent(final PipeSchemaRegionWritePlanEvent event) { diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/batch/PipeTabletEventBatch.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/batch/PipeTabletEventBatch.java index 39d3bea1dbe1..afc4464b10c1 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/batch/PipeTabletEventBatch.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/batch/PipeTabletEventBatch.java @@ -24,6 +24,7 @@ import org.apache.iotdb.db.i18n.DataNodePipeMessages; import org.apache.iotdb.db.pipe.resource.PipeDataNodeResourceManager; import org.apache.iotdb.db.pipe.resource.memory.PipeMemoryBlock; +import org.apache.iotdb.db.pipe.resource.memory.PipeMemoryBlockCategory; import org.apache.iotdb.db.pipe.sink.protocol.thrift.async.IoTDBDataRegionAsyncSink; import org.apache.iotdb.db.storageengine.dataregion.wal.exception.WALPipeException; import org.apache.iotdb.pipe.api.event.Event; @@ -61,7 +62,13 @@ protected PipeTabletEventBatch( // limit in buffer size this.maxBatchSizeInBytes = requestMaxBatchSizeInBytes; - this.allocatedMemoryBlock = PipeDataNodeResourceManager.memory().forceAllocate(0); + this.allocatedMemoryBlock = + PipeDataNodeResourceManager.memory() + .forceAllocate( + PipeTabletEventBatch.class.getSimpleName(), + 0, + PipeMemoryBlockCategory.BATCH, + PipeTabletEventBatch.class.getSimpleName()); if (recordMetric != null) { this.recordMetric = recordMetric; } else { diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/airgap/IoTDBDataRegionAirGapSink.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/airgap/IoTDBDataRegionAirGapSink.java index ff624204e247..811e357ad40f 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/airgap/IoTDBDataRegionAirGapSink.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/airgap/IoTDBDataRegionAirGapSink.java @@ -34,6 +34,7 @@ import org.apache.iotdb.db.pipe.metric.overview.PipeResourceMetrics; import org.apache.iotdb.db.pipe.metric.sink.PipeDataRegionSinkMetrics; import org.apache.iotdb.db.pipe.resource.PipeDataNodeResourceManager; +import org.apache.iotdb.db.pipe.resource.memory.PipeMemoryBlockCategory; import org.apache.iotdb.db.pipe.resource.memory.PipeTsFileMemoryBlock; import org.apache.iotdb.db.pipe.sink.payload.evolvable.batch.PipeTabletEventBatch; import org.apache.iotdb.db.pipe.sink.payload.evolvable.batch.PipeTabletEventPlainBatch; @@ -542,7 +543,11 @@ private void transferFilePieces( final int readFileBufferSize = getReadFileBufferSize(file); try (final PipeTsFileMemoryBlock ignored = PipeDataNodeResourceManager.memory() - .forceAllocateForTsFileWithRetry(readFileBufferSize); + .forceAllocateForTsFileWithRetry( + IoTDBDataRegionAirGapSink.class.getSimpleName(), + readFileBufferSize, + PipeMemoryBlockCategory.SINK, + IoTDBDataRegionAirGapSink.class.getSimpleName()); final RandomAccessFile reader = new RandomAccessFile(file, "r")) { final byte[] readBuffer = new byte[readFileBufferSize]; long position = 0; diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/iotconsensusv2/IoTConsensusV2SyncSink.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/iotconsensusv2/IoTConsensusV2SyncSink.java index 2f2741898094..3e89b5a93289 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/iotconsensusv2/IoTConsensusV2SyncSink.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/iotconsensusv2/IoTConsensusV2SyncSink.java @@ -42,6 +42,7 @@ import org.apache.iotdb.db.pipe.event.common.tablet.PipeInsertNodeTabletInsertionEvent; import org.apache.iotdb.db.pipe.event.common.tsfile.PipeTsFileInsertionEvent; import org.apache.iotdb.db.pipe.resource.PipeDataNodeResourceManager; +import org.apache.iotdb.db.pipe.resource.memory.PipeMemoryBlockCategory; import org.apache.iotdb.db.pipe.resource.memory.PipeTsFileMemoryBlock; import org.apache.iotdb.db.pipe.sink.protocol.iotconsensusv2.payload.builder.IoTConsensusV2SyncBatchReqBuilder; import org.apache.iotdb.db.pipe.sink.protocol.iotconsensusv2.payload.request.IoTConsensusV2DeleteNodeReq; @@ -452,7 +453,11 @@ protected void transferFilePieces( final int readFileBufferSize = getReadFileBufferSize(file); try (final PipeTsFileMemoryBlock ignored = PipeDataNodeResourceManager.memory() - .forceAllocateForTsFileWithRetry(readFileBufferSize); + .forceAllocateForTsFileWithRetry( + IoTConsensusV2SyncSink.class.getSimpleName(), + readFileBufferSize, + PipeMemoryBlockCategory.SINK, + IoTConsensusV2SyncSink.class.getSimpleName()); final RandomAccessFile reader = new RandomAccessFile(file, "r")) { final byte[] readBuffer = new byte[readFileBufferSize]; long position = 0; diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/iotconsensusv2/handler/IoTConsensusV2TsFileInsertionEventHandler.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/iotconsensusv2/handler/IoTConsensusV2TsFileInsertionEventHandler.java index 8ebff3745319..4fa621de08bf 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/iotconsensusv2/handler/IoTConsensusV2TsFileInsertionEventHandler.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/iotconsensusv2/handler/IoTConsensusV2TsFileInsertionEventHandler.java @@ -32,6 +32,7 @@ import org.apache.iotdb.db.pipe.consensus.metric.IoTConsensusV2SinkMetrics; import org.apache.iotdb.db.pipe.event.common.tsfile.PipeTsFileInsertionEvent; import org.apache.iotdb.db.pipe.resource.PipeDataNodeResourceManager; +import org.apache.iotdb.db.pipe.resource.memory.PipeMemoryBlockCategory; import org.apache.iotdb.db.pipe.resource.memory.PipeTsFileMemoryBlock; import org.apache.iotdb.db.pipe.sink.protocol.iotconsensusv2.IoTConsensusV2AsyncSink; import org.apache.iotdb.db.pipe.sink.protocol.iotconsensusv2.payload.request.IoTConsensusV2TsFilePieceReq; @@ -140,7 +141,12 @@ public void transfer(final AsyncIoTConsensusV2ServiceClient client) if (readBuffer == null) { memoryBlock = - PipeDataNodeResourceManager.memory().forceAllocateForTsFileWithRetry(readFileBufferSize); + PipeDataNodeResourceManager.memory() + .forceAllocateForTsFileWithRetry( + IoTConsensusV2TsFileInsertionEventHandler.class.getSimpleName(), + readFileBufferSize, + PipeMemoryBlockCategory.SINK, + IoTConsensusV2TsFileInsertionEventHandler.class.getSimpleName()); readBuffer = new byte[readFileBufferSize]; } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/iotconsensusv2/payload/builder/IoTConsensusV2TransferBatchReqBuilder.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/iotconsensusv2/payload/builder/IoTConsensusV2TransferBatchReqBuilder.java index 8c9e0299f31f..e24490dd5178 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/iotconsensusv2/payload/builder/IoTConsensusV2TransferBatchReqBuilder.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/iotconsensusv2/payload/builder/IoTConsensusV2TransferBatchReqBuilder.java @@ -28,6 +28,7 @@ import org.apache.iotdb.db.pipe.event.common.tablet.PipeInsertNodeTabletInsertionEvent; import org.apache.iotdb.db.pipe.resource.PipeDataNodeResourceManager; import org.apache.iotdb.db.pipe.resource.memory.PipeMemoryBlock; +import org.apache.iotdb.db.pipe.resource.memory.PipeMemoryBlockCategory; import org.apache.iotdb.db.pipe.sink.protocol.iotconsensusv2.payload.request.IoTConsensusV2TabletBatchReq; import org.apache.iotdb.db.pipe.sink.protocol.iotconsensusv2.payload.request.IoTConsensusV2TabletInsertNodeReq; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertNode; @@ -97,7 +98,13 @@ protected IoTConsensusV2TransferBatchReqBuilder( Arrays.asList(CONNECTOR_IOTDB_BATCH_SIZE_KEY, SINK_IOTDB_BATCH_SIZE_KEY), CONNECTOR_IOTDB_PLAIN_BATCH_SIZE_DEFAULT_VALUE); - allocatedMemoryBlock = PipeDataNodeResourceManager.memory().forceAllocate(0); + allocatedMemoryBlock = + PipeDataNodeResourceManager.memory() + .forceAllocate( + IoTConsensusV2TransferBatchReqBuilder.class.getSimpleName(), + 0, + PipeMemoryBlockCategory.BATCH, + IoTConsensusV2TransferBatchReqBuilder.class.getSimpleName()); } /** diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/legacy/IoTDBLegacyPipeSink.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/legacy/IoTDBLegacyPipeSink.java index a74cfcbd2d8b..14a10c391e32 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/legacy/IoTDBLegacyPipeSink.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/legacy/IoTDBLegacyPipeSink.java @@ -38,6 +38,7 @@ import org.apache.iotdb.db.pipe.event.common.terminate.PipeTerminateEvent; import org.apache.iotdb.db.pipe.event.common.tsfile.PipeTsFileInsertionEvent; import org.apache.iotdb.db.pipe.resource.PipeDataNodeResourceManager; +import org.apache.iotdb.db.pipe.resource.memory.PipeMemoryBlockCategory; import org.apache.iotdb.db.pipe.resource.memory.PipeTsFileMemoryBlock; import org.apache.iotdb.db.pipe.sink.payload.legacy.TsFilePipeData; import org.apache.iotdb.db.storageengine.StorageEngine; @@ -527,7 +528,11 @@ private void transportSingleFilePieceByPiece(final File file) throws IOException final int readFileBufferSize = getReadFileBufferSize(file); try (final PipeTsFileMemoryBlock ignored = PipeDataNodeResourceManager.memory() - .forceAllocateForTsFileWithRetry(readFileBufferSize); + .forceAllocateForTsFileWithRetry( + IoTDBLegacyPipeSink.class.getSimpleName(), + readFileBufferSize, + PipeMemoryBlockCategory.SINK, + IoTDBLegacyPipeSink.class.getSimpleName()); final RandomAccessFile randomAccessFile = new RandomAccessFile(file, "r")) { final byte[] buffer = new byte[readFileBufferSize]; while (true) { diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/handler/PipeTransferTsFileHandler.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/handler/PipeTransferTsFileHandler.java index b79e7d6cb2bc..7782f3e0f4c7 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/handler/PipeTransferTsFileHandler.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/handler/PipeTransferTsFileHandler.java @@ -31,6 +31,7 @@ import org.apache.iotdb.db.pipe.event.common.tsfile.PipeTsFileInsertionEvent; import org.apache.iotdb.db.pipe.metric.overview.PipeResourceMetrics; import org.apache.iotdb.db.pipe.resource.PipeDataNodeResourceManager; +import org.apache.iotdb.db.pipe.resource.memory.PipeMemoryBlockCategory; import org.apache.iotdb.db.pipe.resource.memory.PipeMemoryManager; import org.apache.iotdb.db.pipe.resource.memory.PipeTsFileMemoryBlock; import org.apache.iotdb.db.pipe.sink.client.IoTDBDataNodeAsyncClientManager; @@ -199,7 +200,12 @@ public void transfer( // Delay creation of resources to avoid OOM or too many open files if (readBuffer == null) { memoryBlock = - PipeDataNodeResourceManager.memory().forceAllocateForTsFileWithRetry(readFileBufferSize); + PipeDataNodeResourceManager.memory() + .forceAllocateForTsFileWithRetry( + PipeTransferTsFileHandler.class.getSimpleName(), + readFileBufferSize, + PipeMemoryBlockCategory.SINK, + PipeTransferTsFileHandler.class.getSimpleName()); readBuffer = new byte[readFileBufferSize]; } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/sync/IoTDBDataRegionSyncSink.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/sync/IoTDBDataRegionSyncSink.java index 16c044b491d4..154d0af5a41d 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/sync/IoTDBDataRegionSyncSink.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/sync/IoTDBDataRegionSyncSink.java @@ -39,6 +39,7 @@ import org.apache.iotdb.db.pipe.metric.overview.PipeResourceMetrics; import org.apache.iotdb.db.pipe.metric.sink.PipeDataRegionSinkMetrics; import org.apache.iotdb.db.pipe.resource.PipeDataNodeResourceManager; +import org.apache.iotdb.db.pipe.resource.memory.PipeMemoryBlockCategory; import org.apache.iotdb.db.pipe.resource.memory.PipeTsFileMemoryBlock; import org.apache.iotdb.db.pipe.sink.client.IoTDBDataNodeSyncClientManager; import org.apache.iotdb.db.pipe.sink.payload.evolvable.batch.PipeTabletEventBatch; @@ -651,7 +652,11 @@ protected void transferFilePieces( final int readFileBufferSize = getReadFileBufferSize(file); try (final PipeTsFileMemoryBlock ignored = PipeDataNodeResourceManager.memory() - .forceAllocateForTsFileWithRetry(readFileBufferSize); + .forceAllocateForTsFileWithRetry( + IoTDBDataRegionSyncSink.class.getSimpleName(), + readFileBufferSize, + PipeMemoryBlockCategory.SINK, + IoTDBDataRegionSyncSink.class.getSimpleName()); final RandomAccessFile reader = new RandomAccessFile(file, "r")) { final byte[] readBuffer = new byte[readFileBufferSize]; long position = 0; diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/assigner/DisruptorQueue.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/assigner/DisruptorQueue.java index 3ef29dbc9001..7f346f7441ac 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/assigner/DisruptorQueue.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/assigner/DisruptorQueue.java @@ -27,6 +27,7 @@ import org.apache.iotdb.db.pipe.event.realtime.PipeRealtimeEvent; import org.apache.iotdb.db.pipe.resource.PipeDataNodeResourceManager; import org.apache.iotdb.db.pipe.resource.memory.PipeMemoryBlock; +import org.apache.iotdb.db.pipe.resource.memory.PipeMemoryBlockCategory; import org.apache.iotdb.db.pipe.source.dataregion.realtime.disruptor.Disruptor; import org.apache.iotdb.db.pipe.source.dataregion.realtime.disruptor.EventHandler; import org.apache.iotdb.db.pipe.source.dataregion.realtime.disruptor.RingBuffer; @@ -65,7 +66,12 @@ public DisruptorQueue( allocatedMemoryBlock = PipeDataNodeResourceManager.memory() .tryAllocate( - ringBufferSize * ringBufferEntrySizeInBytes, currentSize -> currentSize / 2); + DisruptorQueue.class.getSimpleName(), + ringBufferSize * ringBufferEntrySizeInBytes, + currentSize -> currentSize / 2, + PipeMemoryBlockCategory.EVENT, + DisruptorQueue.class.getSimpleName(), + null); disruptor = new Disruptor<>( diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/InformationSchemaContentSupplierFactory.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/InformationSchemaContentSupplierFactory.java index c0e7ca6d91a5..e68f58f59754 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/InformationSchemaContentSupplierFactory.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/InformationSchemaContentSupplierFactory.java @@ -75,6 +75,8 @@ import org.apache.iotdb.db.conf.IoTDBDescriptor; import org.apache.iotdb.db.i18n.DataNodeQueryMessages; import org.apache.iotdb.db.pipe.metric.overview.PipeDataNodeSinglePipeMetrics; +import org.apache.iotdb.db.pipe.resource.PipeDataNodeResourceManager; +import org.apache.iotdb.db.pipe.resource.memory.PipeMemoryManager.PipeMemoryBlockInfo; import org.apache.iotdb.db.protocol.client.ConfigNodeClient; import org.apache.iotdb.db.protocol.client.ConfigNodeClientManager; import org.apache.iotdb.db.protocol.client.ConfigNodeInfo; @@ -186,6 +188,8 @@ public static IInformationSchemaContentSupplier getSupplier( return new RegionSupplier(dataTypes, userEntity); case InformationSchema.PIPES: return new PipeSupplier(dataTypes, userEntity.getUsername()); + case InformationSchema.PIPE_MEMORY: + return new PipeMemorySupplier(dataTypes, userEntity); case InformationSchema.PIPE_PLUGINS: return new PipePluginSupplier(dataTypes, userEntity); case InformationSchema.TOPICS: @@ -728,6 +732,48 @@ public boolean hasNext() { } } + private static class PipeMemorySupplier extends TsBlockSupplier { + + private final Iterator iterator; + + private PipeMemorySupplier(final List dataTypes, final UserEntity userEntity) { + super(dataTypes); + accessControl.checkUserGlobalSysPrivilege(userEntity); + iterator = PipeDataNodeResourceManager.memory().getPipeMemoryBlockInfoList().iterator(); + } + + @Override + protected void constructLine() { + final PipeMemoryBlockInfo memoryBlockInfo = iterator.next(); + columnBuilders[0].writeLong(memoryBlockInfo.getBlockId()); + columnBuilders[1].writeBinary(BytesUtils.valueOf(memoryBlockInfo.getName())); + columnBuilders[2].writeBinary(BytesUtils.valueOf(memoryBlockInfo.getCategory())); + columnBuilders[3].writeLong(memoryBlockInfo.getMemoryUsageInBytes()); + columnBuilders[4].writeLong(memoryBlockInfo.getMaxMemorySizeInBytes()); + columnBuilders[5].writeLong( + TimestampPrecisionUtils.convertToCurrPrecision( + memoryBlockInfo.getAllocationTime(), TimeUnit.MILLISECONDS)); + if (memoryBlockInfo.getAssigner() == null) { + columnBuilders[6].appendNull(); + } else { + columnBuilders[6].writeBinary(BytesUtils.valueOf(memoryBlockInfo.getAssigner())); + } + if (memoryBlockInfo.getParentBlockId() == null) { + columnBuilders[7].appendNull(); + } else { + columnBuilders[7].writeLong(memoryBlockInfo.getParentBlockId()); + } + columnBuilders[8].writeInt(memoryBlockInfo.getHierarchyLevel()); + columnBuilders[9].writeLong(memoryBlockInfo.getAccountedMemoryUsageInBytes()); + resultBuilder.declarePosition(); + } + + @Override + public boolean hasNext() { + return iterator.hasNext(); + } + } + private static class PipePluginSupplier extends TsBlockSupplier { private final Iterator iterator; diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/DataNodeLocationSupplierFactory.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/DataNodeLocationSupplierFactory.java index 7df20573d991..4c5259c5f76f 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/DataNodeLocationSupplierFactory.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/DataNodeLocationSupplierFactory.java @@ -150,6 +150,7 @@ public List getDataNodeLocations(final String tableName) { case InformationSchema.CONFIG_NODES: case InformationSchema.DATA_NODES: case InformationSchema.SERVICES: + case InformationSchema.PIPE_MEMORY: return Collections.singletonList(DataNodeEndPoints.getLocalDataNodeLocation()); default: throw new UnsupportedOperationException(DataNodeQueryMessages.UNKNOWN_TABLE + tableName); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/parser/AstBuilder.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/parser/AstBuilder.java index f7a99ab3485c..b602d38a52e1 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/parser/AstBuilder.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/parser/AstBuilder.java @@ -1356,6 +1356,18 @@ public Node visitShowPipesStatement(RelationalSqlParser.ShowPipesStatementContex return new ShowPipes(pipeName, hasWhereClause); } + @Override + public Node visitShowPipeMemoryStatement( + final RelationalSqlParser.ShowPipeMemoryStatementContext ctx) { + return new ShowStatement( + getLocation(ctx), + InformationSchema.PIPE_MEMORY, + Optional.empty(), + Optional.empty(), + Optional.empty(), + Optional.empty()); + } + @Override public Node visitShowCreatePipeStatement(RelationalSqlParser.ShowCreatePipeStatementContext ctx) { return new ShowCreatePipe(((Identifier) visit(ctx.pipeName)).getValue()); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/memory/LoadTsFileParserMemoryManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/memory/LoadTsFileParserMemoryManager.java index a7430a3f819f..ae2d12b94ad7 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/memory/LoadTsFileParserMemoryManager.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/memory/LoadTsFileParserMemoryManager.java @@ -39,12 +39,13 @@ public static LoadTsFileParserMemoryManager getInstance() { @Override public TsFileInsertionEventParserMemoryBlock forceAllocateForTabletWithRetry( - final long sizeInBytes) { + final String name, final long sizeInBytes) { return new LoadParserMemoryBlock(sizeInBytes); } @Override - public TsFileInsertionEventParserMemoryBlock forceAllocate(final long sizeInBytes) { + public TsFileInsertionEventParserMemoryBlock forceAllocate( + final String name, final long sizeInBytes) { return new LoadParserMemoryBlock(sizeInBytes); } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/event/cache/SubscriptionPollResponseCache.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/event/cache/SubscriptionPollResponseCache.java index c4b37b88263d..a993559d8258 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/event/cache/SubscriptionPollResponseCache.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/event/cache/SubscriptionPollResponseCache.java @@ -24,6 +24,7 @@ import org.apache.iotdb.db.i18n.DataNodePipeMessages; import org.apache.iotdb.db.pipe.resource.PipeDataNodeResourceManager; import org.apache.iotdb.db.pipe.resource.memory.PipeMemoryBlock; +import org.apache.iotdb.db.pipe.resource.memory.PipeMemoryBlockCategory; import org.apache.iotdb.rpc.subscription.payload.poll.SubscriptionPollResponse; import com.github.benmanes.caffeine.cache.Caffeine; @@ -114,7 +115,12 @@ private SubscriptionPollResponseCache() { // properties required by pipe memory control framework final PipeMemoryBlock allocatedMemoryBlock = - PipeDataNodeResourceManager.memory().tryAllocate(initMemorySizeInBytes); + PipeDataNodeResourceManager.memory() + .tryAllocate( + SubscriptionPollResponseCache.class.getSimpleName(), + initMemorySizeInBytes, + PipeMemoryBlockCategory.SUBSCRIPTION, + SubscriptionPollResponseCache.class.getSimpleName()); this.cache = Caffeine.newBuilder() diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/event/response/SubscriptionEventTabletResponse.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/event/response/SubscriptionEventTabletResponse.java index 1392c0cbb286..5854365fbf13 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/event/response/SubscriptionEventTabletResponse.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/event/response/SubscriptionEventTabletResponse.java @@ -23,6 +23,7 @@ import org.apache.iotdb.commons.subscription.config.SubscriptionConfig; import org.apache.iotdb.db.i18n.DataNodePipeMessages; import org.apache.iotdb.db.pipe.resource.PipeDataNodeResourceManager; +import org.apache.iotdb.db.pipe.resource.memory.PipeMemoryBlockCategory; import org.apache.iotdb.db.pipe.resource.memory.PipeMemoryManager; import org.apache.iotdb.db.pipe.resource.memory.PipeMemoryWeightUtil; import org.apache.iotdb.db.pipe.resource.memory.PipeTabletMemoryBlock; @@ -278,7 +279,12 @@ private synchronized CachedSubscriptionPollResponse generateNextTabletResponse() final List tablets = ((TabletsPayload) response.getPayload()).getTablets(); if (Objects.nonNull(tablets) && !tablets.isEmpty()) { final PipeTabletMemoryBlock memoryBlock = - PipeDataNodeResourceManager.memory().forceAllocateForTabletWithRetry(currentBufferSize); + PipeDataNodeResourceManager.memory() + .forceAllocateForTabletWithRetry( + SubscriptionEventTabletResponse.class.getSimpleName(), + currentBufferSize, + PipeMemoryBlockCategory.SUBSCRIPTION, + SubscriptionEventTabletResponse.class.getSimpleName()); response.setMemoryBlock(memoryBlock); } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/event/response/SubscriptionEventTsFileResponse.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/event/response/SubscriptionEventTsFileResponse.java index 9ddeca25c4cc..309c7a5eccf5 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/event/response/SubscriptionEventTsFileResponse.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/event/response/SubscriptionEventTsFileResponse.java @@ -24,6 +24,7 @@ import org.apache.iotdb.db.i18n.DataNodeMiscMessages; import org.apache.iotdb.db.i18n.DataNodePipeMessages; import org.apache.iotdb.db.pipe.resource.PipeDataNodeResourceManager; +import org.apache.iotdb.db.pipe.resource.memory.PipeMemoryBlockCategory; import org.apache.iotdb.db.pipe.resource.memory.PipeMemoryManager; import org.apache.iotdb.db.pipe.resource.memory.PipeTsFileMemoryBlock; import org.apache.iotdb.db.subscription.agent.SubscriptionAgent; @@ -208,7 +209,12 @@ private CachedSubscriptionPollResponse generateResponseWithPieceOrSealPayload( reader.seek(writingOffset); final PipeTsFileMemoryBlock memoryBlock = - PipeDataNodeResourceManager.memory().forceAllocateForTsFileWithRetry(bufferSize); + PipeDataNodeResourceManager.memory() + .forceAllocateForTsFileWithRetry( + SubscriptionEventTsFileResponse.class.getSimpleName(), + bufferSize, + PipeMemoryBlockCategory.SUBSCRIPTION, + SubscriptionEventTsFileResponse.class.getSimpleName()); final byte[] readBuffer = new byte[(int) bufferSize]; reader.readFully(readBuffer); diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryManagerResizeTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryManagerResizeTest.java index d63fe1c6e830..33359105a53c 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryManagerResizeTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryManagerResizeTest.java @@ -30,6 +30,8 @@ import org.junit.Before; import org.junit.Test; +import java.util.List; +import java.util.Optional; import java.util.concurrent.atomic.AtomicLong; public class PipeMemoryManagerResizeTest { @@ -87,7 +89,7 @@ public void testTabletResizeCannotCrossTabletHardLimit() { null, TOTAL_MEMORY_SIZE_IN_BYTES, MemoryBlockType.DYNAMIC)); - final PipeTabletMemoryBlock tablet = manager.forceAllocateForTabletWithRetry(0); + final PipeTabletMemoryBlock tablet = manager.forceAllocateForTabletWithRetry("tablet", 0); try { Assert.assertThrows( @@ -111,9 +113,10 @@ public void testTabletResizeLeavesMemoryForSinkForwardProgress() { TOTAL_MEMORY_SIZE_IN_BYTES, MemoryBlockType.DYNAMIC)); final PipeTabletMemoryBlock retainedTablet = - manager.forceAllocateForTabletWithRetry(TABLET_MEMORY_SIZE_IN_BYTES); - final PipeTabletMemoryBlock pendingTablet = manager.forceAllocateForTabletWithRetry(0); - final PipeMemoryBlock sinkBatch = manager.forceAllocate(0); + manager.forceAllocateForTabletWithRetry("retainedTablet", TABLET_MEMORY_SIZE_IN_BYTES); + final PipeTabletMemoryBlock pendingTablet = + manager.forceAllocateForTabletWithRetry("pendingTablet", 0); + final PipeMemoryBlock sinkBatch = manager.forceAllocate("sinkBatch", 0); try { Assert.assertThrows( @@ -155,7 +158,8 @@ public void testFloatingAndNonFloatingMemoryShareTheSamePool() { Assert.assertEquals( TOTAL_MEMORY_SIZE_IN_BYTES / 2, manager.getTotalFloatingMemorySizeInBytes()); - final PipeTsFileMemoryBlock nonFloatingMemory = manager.forceAllocateForTsFileWithRetry(1200); + final PipeTsFileMemoryBlock nonFloatingMemory = + manager.forceAllocateForTsFileWithRetry("tsFile", 1200); try { // Non-floating memory can borrow the unused half that was previously reserved for InsertNode // queues. Its usage also reduces the current floating-memory limit symmetrically. @@ -167,9 +171,171 @@ public void testFloatingAndNonFloatingMemoryShareTheSamePool() { Assert.assertEquals(300, manager.getFreeMemorySizeInBytes()); Assert.assertThrows( - PipeRuntimeOutOfMemoryCriticalException.class, () -> manager.forceAllocate(301)); + PipeRuntimeOutOfMemoryCriticalException.class, + () -> manager.forceAllocate("normal", 301)); } finally { manager.release(nonFloatingMemory); } } + + @Test + public void testMemoryBlockInfoIncludesNamesAndSeparatesFloatingMemory() { + final AtomicLong floatingMemoryUsageInBytes = new AtomicLong(0); + final PipeMemoryManager manager = + new PipeMemoryManager( + new AtomicLongMemoryBlock( + "PipeMemoryManagerResizeTest", + null, + TOTAL_MEMORY_SIZE_IN_BYTES, + MemoryBlockType.DYNAMIC), + floatingMemoryUsageInBytes::get); + final PipeMemoryBlock normalMemory = manager.forceAllocate("normal", 100); + final PipeMemoryBlock zeroSizedMemory = manager.forceAllocate("zero", 0); + + try { + floatingMemoryUsageInBytes.set(250); + final List memoryBlockInfoList = + manager.getPipeMemoryBlockInfoList(); + + Assert.assertEquals(3, memoryBlockInfoList.size()); + Assert.assertEquals("FloatingMemory", memoryBlockInfoList.get(0).getName()); + Assert.assertEquals(250, memoryBlockInfoList.get(0).getMemoryUsageInBytes()); + Assert.assertEquals("normal", memoryBlockInfoList.get(1).getName()); + Assert.assertEquals(100, memoryBlockInfoList.get(1).getMemoryUsageInBytes()); + Assert.assertEquals("zero", memoryBlockInfoList.get(2).getName()); + Assert.assertEquals(0, memoryBlockInfoList.get(2).getMemoryUsageInBytes()); + Assert.assertEquals(100, manager.getUsedMemorySizeInBytes()); + + manager.forceResize(normalMemory, 0); + final List memoryBlockInfoListAfterResize = + manager.getPipeMemoryBlockInfoList(); + Assert.assertEquals(3, memoryBlockInfoListAfterResize.size()); + Assert.assertEquals("normal", memoryBlockInfoListAfterResize.get(1).getName()); + Assert.assertEquals(0, memoryBlockInfoListAfterResize.get(1).getMemoryUsageInBytes()); + Assert.assertEquals(0, manager.getUsedMemorySizeInBytes()); + } finally { + manager.release(normalMemory); + manager.release(zeroSizedMemory); + } + + final List memoryBlockInfoListAfterRelease = + manager.getPipeMemoryBlockInfoList(); + Assert.assertEquals(1, memoryBlockInfoListAfterRelease.size()); + Assert.assertEquals("FloatingMemory", memoryBlockInfoListAfterRelease.get(0).getName()); + Assert.assertEquals(250, memoryBlockInfoListAfterRelease.get(0).getMemoryUsageInBytes()); + } + + @Test + public void testHierarchicalAccountingMetadataAndCascadeRelease() { + final PipeMemoryManager manager = + new PipeMemoryManager( + new AtomicLongMemoryBlock( + "PipeMemoryManagerHierarchyTest", + null, + TOTAL_MEMORY_SIZE_IN_BYTES, + MemoryBlockType.DYNAMIC)); + final long allocationStart = System.currentTimeMillis(); + final PipeMemoryBlock eventBlock = + manager.forceAllocate("event", 0, PipeMemoryBlockCategory.EVENT, "event-assigner", null); + final PipeMemoryBlock parserBlock = + manager.forceAllocate( + "parser", 100, PipeMemoryBlockCategory.PARSER, "parser-assigner", eventBlock); + + try { + Assert.assertNotEquals(eventBlock.getBlockId(), parserBlock.getBlockId()); + Assert.assertEquals(PipeMemoryBlockCategory.EVENT, eventBlock.getCategory()); + Assert.assertEquals(PipeMemoryBlockCategory.PARSER, parserBlock.getCategory()); + Assert.assertEquals(0, eventBlock.getHierarchyLevel()); + Assert.assertEquals(1, parserBlock.getHierarchyLevel()); + Assert.assertEquals(eventBlock.getBlockId(), parserBlock.getParentBlockId().longValue()); + Assert.assertEquals("event-assigner", eventBlock.getAssigner()); + Assert.assertEquals("parser-assigner", parserBlock.getAssigner()); + Assert.assertTrue(eventBlock.getAllocationTimeInMillis() >= allocationStart); + Assert.assertTrue(parserBlock.getAllocationTimeInMillis() >= allocationStart); + + // A child reserves the global pool once, while both rows expose the aggregate usage. + Assert.assertEquals(100, manager.getUsedMemorySizeInBytes()); + Assert.assertEquals(100, eventBlock.getMemoryUsageInBytes()); + Assert.assertEquals(100, parserBlock.getMemoryUsageInBytes()); + Assert.assertEquals(100, eventBlock.getAccountedMemoryUsageInBytes()); + Assert.assertEquals(0, parserBlock.getAccountedMemoryUsageInBytes()); + Assert.assertEquals(100, eventBlock.getMaxMemorySizeInBytes()); + Assert.assertEquals(100, parserBlock.getMaxMemorySizeInBytes()); + + manager.forceResize(parserBlock, 160); + manager.forceResize(parserBlock, 40); + Assert.assertEquals(40, manager.getUsedMemorySizeInBytes()); + Assert.assertEquals(40, eventBlock.getMemoryUsageInBytes()); + Assert.assertEquals(40, parserBlock.getMemoryUsageInBytes()); + Assert.assertEquals(160, eventBlock.getMaxMemorySizeInBytes()); + Assert.assertEquals(160, parserBlock.getMaxMemorySizeInBytes()); + + // A parent resize cannot release bytes still owned by a live child. + manager.forceResize(eventBlock, 0); + Assert.assertEquals(40, manager.getUsedMemorySizeInBytes()); + Assert.assertEquals(40, eventBlock.getMemoryUsageInBytes()); + + final Optional parserInfo = + manager.getPipeMemoryBlockInfoList().stream() + .filter(info -> info.getBlockId() == parserBlock.getBlockId()) + .findFirst(); + Assert.assertTrue(parserInfo.isPresent()); + Assert.assertEquals("PARSER", parserInfo.get().getCategory()); + Assert.assertEquals(eventBlock.getBlockId(), parserInfo.get().getParentBlockId().longValue()); + Assert.assertEquals(0, parserInfo.get().getAccountedMemoryUsageInBytes()); + } finally { + // Closing the aggregate must recursively release all descendants and the global reservation. + manager.release(eventBlock); + parserBlock.close(); + } + + Assert.assertTrue(eventBlock.isReleased()); + Assert.assertTrue(parserBlock.isReleased()); + Assert.assertEquals(0, manager.getUsedMemorySizeInBytes()); + Assert.assertEquals(1, manager.getPipeMemoryBlockInfoList().size()); + } + + @Test + public void testAssignerSnapshotIsBoundedAndFloatingPeakIsRetained() { + final AtomicLong floatingMemoryUsageInBytes = new AtomicLong(0); + final PipeMemoryManager manager = + new PipeMemoryManager( + new AtomicLongMemoryBlock( + "PipeMemoryManagerMetadataTest", + null, + TOTAL_MEMORY_SIZE_IN_BYTES, + MemoryBlockType.DYNAMIC), + floatingMemoryUsageInBytes::get); + final PipeMemoryBlock block = manager.forceAllocate("metadata", 0); + final String longAssigner = "x".repeat(4096); + + try { + block.setAssigner(longAssigner); + Assert.assertEquals(2048, block.getAssigner().length()); + + floatingMemoryUsageInBytes.set(300); + PipeMemoryManager.PipeMemoryBlockInfo floatingInfo = + manager.getPipeMemoryBlockInfoList().stream() + .filter(info -> PipeMemoryManager.FLOATING_MEMORY_BLOCK_NAME.equals(info.getName())) + .findFirst() + .orElseThrow(); + Assert.assertEquals(0, floatingInfo.getBlockId()); + Assert.assertEquals("FLOATING", floatingInfo.getCategory()); + Assert.assertEquals(300, floatingInfo.getMemoryUsageInBytes()); + Assert.assertEquals(300, floatingInfo.getMaxMemorySizeInBytes()); + Assert.assertTrue(floatingInfo.getAllocationTime() > 0); + Assert.assertEquals("PipeMemoryManager", floatingInfo.getAssigner()); + + floatingMemoryUsageInBytes.set(10); + floatingInfo = + manager.getPipeMemoryBlockInfoList().stream() + .filter(info -> PipeMemoryManager.FLOATING_MEMORY_BLOCK_NAME.equals(info.getName())) + .findFirst() + .orElseThrow(); + Assert.assertEquals(10, floatingInfo.getMemoryUsageInBytes()); + Assert.assertEquals(300, floatingInfo.getMaxMemorySizeInBytes()); + } finally { + block.close(); + } + } } diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/sql/parser/ShowPipeMemoryTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/sql/parser/ShowPipeMemoryTest.java new file mode 100644 index 000000000000..d0722c940a2d --- /dev/null +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/sql/parser/ShowPipeMemoryTest.java @@ -0,0 +1,39 @@ +/* + * 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.queryengine.plan.relational.sql.parser; + +import org.apache.iotdb.commons.queryengine.plan.relational.sql.ast.Statement; +import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.ShowStatement; + +import org.junit.Assert; +import org.junit.Test; + +import java.time.ZoneId; + +public class ShowPipeMemoryTest { + + @Test + public void testShowPipeMemoryStatement() { + final Statement statement = + new SqlParser().createStatement("SHOW PIPE MEMORY", ZoneId.systemDefault(), null); + + Assert.assertTrue(statement instanceof ShowStatement); + Assert.assertEquals("pipe_memory", ((ShowStatement) statement).getTableName()); + } +} diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/memory/LoadTsFileMemoryManagerTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/memory/LoadTsFileMemoryManagerTest.java index f0ffe1d5f998..967552b75448 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/memory/LoadTsFileMemoryManagerTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/memory/LoadTsFileMemoryManagerTest.java @@ -89,7 +89,7 @@ public void testParserMemoryBlockGrowsAndReleasesFromQueryPool() throws Exceptio final LoadTsFileMemoryManager manager = LoadTsFileMemoryManager.getInstance(); final long usedMemoryBefore = manager.getUsedMemorySizeInBytes(); final TsFileInsertionEventParserMemoryBlock block = - LoadTsFileParserMemoryManager.getInstance().forceAllocate(0); + LoadTsFileParserMemoryManager.getInstance().forceAllocate("test", 0); Assert.assertEquals(0L, block.getMemoryUsageInBytes()); block.forceResize(1024); diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/schema/column/ColumnHeaderConstant.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/schema/column/ColumnHeaderConstant.java index d66b5a8a301e..d97c8d3ceec4 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/schema/column/ColumnHeaderConstant.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/schema/column/ColumnHeaderConstant.java @@ -295,6 +295,17 @@ private ColumnHeaderConstant() { "estimated_remaining_seconds"; public static final String IS_DEGRADED_TABLE_MODEL = "is_degraded"; public static final String RECENT_FAILURES_TABLE_MODEL = "recent_failures"; + public static final String NAME_TABLE_MODEL = "name"; + public static final String BLOCK_ID_TABLE_MODEL = "block_id"; + public static final String CATEGORY_TABLE_MODEL = "category"; + public static final String MEMORY_USAGE_IN_BYTES_TABLE_MODEL = "memory_usage_in_bytes"; + public static final String MAX_MEMORY_SIZE_IN_BYTES_TABLE_MODEL = "max_memory_size_in_bytes"; + public static final String ALLOCATION_TIME_TABLE_MODEL = "allocation_time"; + public static final String ASSIGNER_TABLE_MODEL = "assigner"; + public static final String PARENT_BLOCK_ID_TABLE_MODEL = "parent_block_id"; + public static final String HIERARCHY_LEVEL_TABLE_MODEL = "hierarchy_level"; + public static final String ACCOUNTED_MEMORY_USAGE_IN_BYTES_TABLE_MODEL = + "accounted_memory_usage_in_bytes"; public static final String PLUGIN_NAME_TABLE_MODEL = "plugin_name"; public static final String PLUGIN_TYPE_TABLE_MODEL = "plugin_type"; diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/schema/table/InformationSchema.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/schema/table/InformationSchema.java index 048ce8f8763b..9d69ff9951ac 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/schema/table/InformationSchema.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/schema/table/InformationSchema.java @@ -47,6 +47,7 @@ public class InformationSchema { public static final String COLUMNS = "columns"; public static final String REGIONS = "regions"; public static final String PIPES = "pipes"; + public static final String PIPE_MEMORY = "pipe_memory"; public static final String PIPE_PLUGINS = "pipe_plugins"; public static final String TOPICS = "topics"; public static final String SUBSCRIPTIONS = "subscriptions"; @@ -241,6 +242,35 @@ public class InformationSchema { ColumnHeaderConstant.RECENT_FAILURES_TABLE_MODEL, TSDataType.STRING)); schemaTables.put(PIPES, pipeTable); + final TsTable pipeMemoryTable = new TsTable(PIPE_MEMORY); + pipeMemoryTable.addColumnSchema( + new TagColumnSchema(ColumnHeaderConstant.BLOCK_ID_TABLE_MODEL, TSDataType.INT64)); + pipeMemoryTable.addColumnSchema( + new TagColumnSchema(ColumnHeaderConstant.NAME_TABLE_MODEL, TSDataType.STRING)); + pipeMemoryTable.addColumnSchema( + new TagColumnSchema(ColumnHeaderConstant.CATEGORY_TABLE_MODEL, TSDataType.STRING)); + pipeMemoryTable.addColumnSchema( + new AttributeColumnSchema( + ColumnHeaderConstant.MEMORY_USAGE_IN_BYTES_TABLE_MODEL, TSDataType.INT64)); + pipeMemoryTable.addColumnSchema( + new AttributeColumnSchema( + ColumnHeaderConstant.MAX_MEMORY_SIZE_IN_BYTES_TABLE_MODEL, TSDataType.INT64)); + pipeMemoryTable.addColumnSchema( + new AttributeColumnSchema( + ColumnHeaderConstant.ALLOCATION_TIME_TABLE_MODEL, TSDataType.TIMESTAMP)); + pipeMemoryTable.addColumnSchema( + new AttributeColumnSchema(ColumnHeaderConstant.ASSIGNER_TABLE_MODEL, TSDataType.STRING)); + pipeMemoryTable.addColumnSchema( + new AttributeColumnSchema( + ColumnHeaderConstant.PARENT_BLOCK_ID_TABLE_MODEL, TSDataType.INT64)); + pipeMemoryTable.addColumnSchema( + new AttributeColumnSchema( + ColumnHeaderConstant.HIERARCHY_LEVEL_TABLE_MODEL, TSDataType.INT32)); + pipeMemoryTable.addColumnSchema( + new AttributeColumnSchema( + ColumnHeaderConstant.ACCOUNTED_MEMORY_USAGE_IN_BYTES_TABLE_MODEL, TSDataType.INT64)); + schemaTables.put(PIPE_MEMORY, pipeMemoryTable); + final TsTable pipePluginTable = new TsTable(PIPE_PLUGINS); pipePluginTable.addColumnSchema( new TagColumnSchema(ColumnHeaderConstant.PLUGIN_NAME_TABLE_MODEL, TSDataType.STRING)); diff --git a/iotdb-core/relational-grammar/src/main/antlr4/org/apache/iotdb/db/relational/grammar/sql/RelationalSql.g4 b/iotdb-core/relational-grammar/src/main/antlr4/org/apache/iotdb/db/relational/grammar/sql/RelationalSql.g4 index 037778597d49..3b2452eb100d 100644 --- a/iotdb-core/relational-grammar/src/main/antlr4/org/apache/iotdb/db/relational/grammar/sql/RelationalSql.g4 +++ b/iotdb-core/relational-grammar/src/main/antlr4/org/apache/iotdb/db/relational/grammar/sql/RelationalSql.g4 @@ -102,6 +102,7 @@ statement | dropPipeStatement | startPipeStatement | stopPipeStatement + | showPipeMemoryStatement | showPipesStatement | showCreatePipeStatement | createPipePluginStatement @@ -527,6 +528,10 @@ showPipesStatement : SHOW ((PIPE pipeName=identifier) | PIPES (WHERE (CONNECTOR | SINK) USED BY pipeName=identifier)?) ; +showPipeMemoryStatement + : SHOW PIPE MEMORY + ; + showCreatePipeStatement : SHOW CREATE PIPE pipeName=identifier ; @@ -1521,7 +1526,7 @@ nonReserved | JSON | KEEP | KEY | KEYS | KILL | LANGUAGE | LAST | LATERAL | LEADING | LEAVE | LEVEL | LIMIT | LINEAR | LOAD | LOCAL | LOGICAL | LOOP - | MANAGE_ROLE | MANAGE_USER | MAP | MATCH | MATCHED | MATCHES | MATCH_RECOGNIZE | MATERIALIZED | MEASURES | MEMORY_THRESHOLD | METHOD | MERGE | MICROSECOND | MIGRATE | MILLISECOND | MINUTE | MODEL | MODELS | MODIFY | MONTH + | MANAGE_ROLE | MANAGE_USER | MAP | MATCH | MATCHED | MATCHES | MATCH_RECOGNIZE | MATERIALIZED | MEASURES | MEMORY | MEMORY_THRESHOLD | METHOD | MERGE | MICROSECOND | MIGRATE | MILLISECOND | MINUTE | MODEL | MODELS | MODIFY | MONTH | NANOSECOND | NESTED | NEXT | NFC | NFD | NFKC | NFKD | NO | NODEID | NONE | NULLIF | NULLS | OBJECT | OF | OFFSET | OMIT | ONE | ONLY | OPTION | ORDINALITY | OUTPUT | OVER | OVERFLOW | PARTITION | PARTITIONS | PASSING | PAST | PATH | PATTERN | PER | PERIOD | PERMUTE | PIPE | PIPEPLUGIN | PIPEPLUGINS | PIPES | PLAN | POSITION | PRECEDING | PRECISION | PRIVILEGES | PREVIOUS | PROCESSLIST | PROCESSOR | PROPERTIES | PRUNE @@ -1742,6 +1747,7 @@ MATCHES: 'MATCHES'; MATCH_RECOGNIZE: 'MATCH_RECOGNIZE'; MATERIALIZED: 'MATERIALIZED'; MEASURES: 'MEASURES'; +MEMORY: 'MEMORY'; MEMORY_THRESHOLD: 'MEMORY_THRESHOLD'; METHOD: 'METHOD'; MERGE: 'MERGE';