Skip to content

Commit b626c60

Browse files
authored
[Pipe] Unify TsFile parser allocation and resize quotas (#18984)
1 parent ca20794 commit b626c60

8 files changed

Lines changed: 305 additions & 23 deletions

File tree

‎iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2403,6 +2403,10 @@ private DataNodePipeMessages() {}
24032403
public static final String PIPE_EXCEPTION_FORCERESIZE_FAILED_TO_ALLOCATE_MEMORY_AFTER_D_RETRIES_TOTAL_8C6948BC =
24042404
"forceResize: failed to allocate memory after %d retries, total memory size %d bytes, used "
24052405
+ "memory size %d bytes, requested memory size %d bytes";
2406+
public static final String EXCEPTION_FORCERESIZE_FAILED_AFTER_ARG_RETRIES_TOTAL_ARG_BYTES_USED_ARG_BYTES_USEDTABLETS_ARG_BYTES_USEDTSFILES_ARG_BYTES_TABLETLIMIT_ARG_BYTES_TSFILELIMIT_ARG_BYTES_TABLETANDTSFILELIMIT_ARG_BYTES_PROTECTEDPARSERLIMIT_ARG_BYTES_REQUESTED_ARG_BYTES_311C8C7E =
2407+
"forceResize failed after %d retries: total=%d bytes, used=%d bytes, "
2408+
+ "usedTablets=%d bytes, usedTsFiles=%d bytes, tabletLimit=%d bytes, tsFileLimit=%d bytes, "
2409+
+ "tabletAndTsFileLimit=%d bytes, protectedParserLimit=%d bytes, requested=%d bytes";
24062410
public static final String
24072411
EXCEPTION_UNSUPPORTED_BATCH_TYPE_ARG_WHEN_TRANSFERRING_TABLET_INSERTION_EVENT_66153E12 =
24082412
"Unsupported batch type %s when transferring tablet insertion event.";

‎iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2233,6 +2233,10 @@ private DataNodePipeMessages() {}
22332233
"forceAllocate:重试 %d 次后仍无法分配内存,总内存大小 %d bytes,已用内存大小 %d bytes,请求内存大小 %d bytes";
22342234
public static final String PIPE_EXCEPTION_FORCERESIZE_FAILED_TO_ALLOCATE_MEMORY_AFTER_D_RETRIES_TOTAL_8C6948BC =
22352235
"forceResize:重试 %d 次后仍无法分配内存,总内存大小 %d bytes,已用内存大小 %d bytes,请求内存大小 %d bytes";
2236+
public static final String EXCEPTION_FORCERESIZE_FAILED_AFTER_ARG_RETRIES_TOTAL_ARG_BYTES_USED_ARG_BYTES_USEDTABLETS_ARG_BYTES_USEDTSFILES_ARG_BYTES_TABLETLIMIT_ARG_BYTES_TSFILELIMIT_ARG_BYTES_TABLETANDTSFILELIMIT_ARG_BYTES_PROTECTEDPARSERLIMIT_ARG_BYTES_REQUESTED_ARG_BYTES_311C8C7E =
2237+
"forceResize 重试 %d 次后仍无法分配内存:total=%d bytes,used=%d bytes,"
2238+
+ "usedTablets=%d bytes,usedTsFiles=%d bytes,tabletLimit=%d bytes,tsFileLimit=%d bytes,"
2239+
+ "tabletAndTsFileLimit=%d bytes,protectedParserLimit=%d bytes,requested=%d bytes";
22362240
public static final String
22372241
EXCEPTION_UNSUPPORTED_BATCH_TYPE_ARG_WHEN_TRANSFERRING_TABLET_INSERTION_EVENT_66153E12 =
22382242
"传输 tablet insertion event 时不支持 batch 类型 %s。";

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/parser/TsFileInsertionEventParserMemoryManager.java‎

Lines changed: 7 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -20,10 +20,10 @@
2020
package org.apache.iotdb.db.pipe.event.common.tsfile.parser;
2121

2222
import org.apache.iotdb.commons.exception.pipe.PipeRuntimeOutOfMemoryCriticalException;
23-
import org.apache.iotdb.db.i18n.DataNodePipeMessages;
2423
import org.apache.iotdb.db.pipe.resource.PipeDataNodeResourceManager;
2524
import org.apache.iotdb.db.pipe.resource.memory.PipeMemoryBlock;
2625
import org.apache.iotdb.db.pipe.resource.memory.PipeMemoryBlockCategory;
26+
import org.apache.iotdb.db.pipe.resource.memory.PipeMemoryManager;
2727

2828
/** Allocates parser working memory from the pool owned by the caller. */
2929
public interface TsFileInsertionEventParserMemoryManager {
@@ -63,11 +63,8 @@ public TsFileInsertionEventParserMemoryBlock forceAllocateForTabletWithRetry(
6363
final String name, final long sizeInBytes) {
6464
return new PipeBlock(
6565
PipeDataNodeResourceManager.memory()
66-
.forceAllocateForTabletWithRetry(
67-
name,
68-
sizeInBytes,
69-
PipeMemoryBlockCategory.TABLET,
70-
TsFileInsertionEventParser.class.getSimpleName()));
66+
.forceAllocateForTsFileParserWithRetry(
67+
name, sizeInBytes, TsFileInsertionEventParser.class.getSimpleName()));
7168
}
7269

7370
@Override
@@ -90,8 +87,7 @@ public TsFileInsertionEventParserMemoryBlock forceAllocateForTabletWithRetry(
9087
final Object assigner) {
9188
return new PipeBlock(
9289
PipeDataNodeResourceManager.memory()
93-
.forceAllocateForTabletWithRetry(
94-
name, sizeInBytes, PipeMemoryBlockCategory.TABLET, assigner, parent));
90+
.forceAllocateForTsFileParserWithRetry(name, sizeInBytes, assigner, parent));
9591
}
9692

9793
@Override
@@ -123,15 +119,10 @@ public long getMemoryUsageInBytes() {
123119
@Override
124120
public void forceResize(final long newSizeInBytes) {
125121
final long oldSize = delegate.getMemoryUsageInBytes();
126-
if (!PipeDataNodeResourceManager.memory().tryResize(delegate, newSizeInBytes)) {
122+
final PipeMemoryManager memoryManager = PipeDataNodeResourceManager.memory();
123+
if (!memoryManager.tryResize(delegate, newSizeInBytes)) {
127124
throw new PipeRuntimeOutOfMemoryCriticalException(
128-
String.format(
129-
DataNodePipeMessages
130-
.PIPE_EXCEPTION_FORCERESIZE_FAILED_TO_ALLOCATE_MEMORY_AFTER_D_RETRIES_TOTAL_8C6948BC,
131-
0,
132-
PipeDataNodeResourceManager.memory().getTotalNonFloatingMemorySizeInBytes(),
133-
PipeDataNodeResourceManager.memory().getUsedMemorySizeInBytes(),
134-
newSizeInBytes - oldSize));
125+
memoryManager.getResizeFailureMessage(0, newSizeInBytes - oldSize));
135126
}
136127
}
137128

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryBlockCategory.java‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -48,6 +48,7 @@ public static PipeMemoryBlockCategory fromType(final PipeMemoryBlockType type) {
4848
return switch (type) {
4949
case NORMAL -> OTHER;
5050
case TABLET -> TABLET;
51+
case TS_FILE_PARSER -> PARSER;
5152
case TS_FILE -> TS_FILE;
5253
case BATCH -> BATCH;
5354
case WAL -> WAL;

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryBlockType.java‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -22,6 +22,7 @@
2222
public enum PipeMemoryBlockType {
2323
NORMAL,
2424
TABLET,
25+
TS_FILE_PARSER,
2526
TS_FILE,
2627
BATCH,
2728
WAL

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryManager.java‎

Lines changed: 75 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -152,6 +152,35 @@ private double allowedMaxMemorySizeInBytesOfTsTiles(
152152
* totalNonFloatingMemorySizeInBytes;
153153
}
154154

155+
private boolean isEnoughForTsFileParserMemory(
156+
final long extraMemoryInBytes, final long totalNonFloatingMemorySizeInBytes) {
157+
return (double) usedMemorySizeInBytesOfTablets
158+
+ (double) usedMemorySizeInBytesOfTsFiles
159+
+ (double) extraMemoryInBytes
160+
< EXCEED_PROTECT_THRESHOLD
161+
* allowedMaxMemorySizeInBytesOfTabletsAndTsFiles(totalNonFloatingMemorySizeInBytes);
162+
}
163+
164+
/** Reports both global availability and the category limits that can reject a resize. */
165+
public synchronized String getResizeFailureMessage(final int retries, final long requestedSize) {
166+
final long totalNonFloatingMemorySizeInBytes = getTotalNonFloatingMemorySizeInBytes();
167+
final double combinedLimit =
168+
allowedMaxMemorySizeInBytesOfTabletsAndTsFiles(totalNonFloatingMemorySizeInBytes);
169+
return String.format(
170+
DataNodePipeMessages
171+
.EXCEPTION_FORCERESIZE_FAILED_AFTER_ARG_RETRIES_TOTAL_ARG_BYTES_USED_ARG_BYTES_USEDTABLETS_ARG_BYTES_USEDTSFILES_ARG_BYTES_TABLETLIMIT_ARG_BYTES_TSFILELIMIT_ARG_BYTES_TABLETANDTSFILELIMIT_ARG_BYTES_PROTECTEDPARSERLIMIT_ARG_BYTES_REQUESTED_ARG_BYTES_311C8C7E,
172+
retries,
173+
totalNonFloatingMemorySizeInBytes,
174+
memoryBlock.getUsedMemoryInBytes(),
175+
usedMemorySizeInBytesOfTablets,
176+
usedMemorySizeInBytesOfTsFiles,
177+
Math.round(allowedMaxMemorySizeInBytesOfTablets(totalNonFloatingMemorySizeInBytes)),
178+
Math.round(allowedMaxMemorySizeInBytesOfTsTiles(totalNonFloatingMemorySizeInBytes)),
179+
Math.round(combinedLimit),
180+
Math.round(EXCEED_PROTECT_THRESHOLD * combinedLimit),
181+
requestedSize);
182+
}
183+
155184
private static long getTsFileParserMemorySizeInBytes() {
156185
return Math.max(
157186
PIPE_CONFIG.getTsFileParserMemory(), PIPE_CONFIG.getPipeMemoryAllocateMinSizeInBytes());
@@ -556,6 +585,10 @@ < allowedMaxMemorySizeInBytesOfTabletsAndTsFiles(totalNonFloatingMemorySizeInByt
556585

557586
private boolean isHardEnoughForResizing(
558587
final PipeMemoryBlock block, final long extraMemoryInBytes) {
588+
if (block instanceof PipeTsFileParserMemoryBlock) {
589+
return isEnoughForTsFileParserMemory(
590+
extraMemoryInBytes, getTotalNonFloatingMemorySizeInBytes());
591+
}
559592
if (!(block instanceof PipeTabletMemoryBlock) && !(block instanceof PipeTsFileMemoryBlock)) {
560593
return true;
561594
}
@@ -757,6 +790,39 @@ public PipeTabletMemoryBlock forceAllocateForTabletWithRetry(
757790
name, tabletSizeInBytes, PipeMemoryBlockCategory.EVENT_CHILD, null, parent);
758791
}
759792

793+
/** Allocates TsFile parser memory against the protected tablet + TsFile combined quota. */
794+
public PipeTsFileParserMemoryBlock forceAllocateForTsFileParserWithRetry(
795+
final String name, final long parserMemorySizeInBytes, final Object assigner) {
796+
return forceAllocateForTsFileParserWithRetry(name, parserMemorySizeInBytes, assigner, null);
797+
}
798+
799+
/** Allocates TsFile parser memory as an event child against the combined quota. */
800+
public synchronized PipeTsFileParserMemoryBlock forceAllocateForTsFileParserWithRetry(
801+
final String name,
802+
final long parserMemorySizeInBytes,
803+
final Object assigner,
804+
final PipeMemoryBlock parent) {
805+
final PipeMemoryBlockCategory category =
806+
parent == null ? PipeMemoryBlockCategory.PARSER : PipeMemoryBlockCategory.EVENT_CHILD;
807+
final PipeTsFileParserMemoryBlock block =
808+
(PipeTsFileParserMemoryBlock)
809+
registerMemoryBlock(
810+
name, 0, PipeMemoryBlockType.TS_FILE_PARSER, category, assigner, parent);
811+
if (!PIPE_MEMORY_MANAGEMENT_ENABLED || parserMemorySizeInBytes == 0) {
812+
return block;
813+
}
814+
815+
try {
816+
// Initial allocation and subsequent growth must check the same post-allocation usage
817+
// under the manager lock, including after a wait releases that lock.
818+
forceResize(block, parserMemorySizeInBytes);
819+
return block;
820+
} catch (final RuntimeException e) {
821+
release(block);
822+
throw e;
823+
}
824+
}
825+
760826
public PipeTsFileMemoryBlock forceAllocateForTsFileWithRetry(
761827
final String name, final long tsFileSizeInBytes)
762828
throws PipeRuntimeOutOfMemoryCriticalException {
@@ -1001,17 +1067,14 @@ public synchronized void resize(
10011067

10021068
if (force) {
10031069
throw new PipeRuntimeOutOfMemoryCriticalException(
1004-
String.format(
1005-
DataNodePipeMessages
1006-
.PIPE_EXCEPTION_FORCERESIZE_FAILED_TO_ALLOCATE_MEMORY_AFTER_D_RETRIES_TOTAL_8C6948BC,
1007-
memoryAllocateMaxRetries,
1008-
getTotalNonFloatingMemorySizeInBytes(),
1009-
memoryBlock.getUsedMemoryInBytes(),
1010-
sizeInBytes));
1070+
getResizeFailureMessage(memoryAllocateMaxRetries, sizeInBytes));
10111071
}
10121072
}
10131073

10141074
private boolean tryResizeInternal(final PipeMemoryBlock block, final long targetSize) {
1075+
if (block.isReleased()) {
1076+
return false;
1077+
}
10151078
// Parent blocks expose an aggregate usage. Do not let a direct resize reduce that aggregate
10161079
// below the bytes still owned by live children.
10171080
final long normalizedTargetSize =
@@ -1249,6 +1312,11 @@ private synchronized PipeMemoryBlock registerMemoryBlock(
12491312
returnedMemoryBlock =
12501313
new PipeTabletMemoryBlock(this, name, 0, normalizedCategory, null, normalizedParent);
12511314
break;
1315+
case TS_FILE_PARSER:
1316+
returnedMemoryBlock =
1317+
new PipeTsFileParserMemoryBlock(
1318+
this, name, 0, normalizedCategory, null, normalizedParent);
1319+
break;
12521320
case TS_FILE:
12531321
returnedMemoryBlock =
12541322
new PipeTsFileMemoryBlock(this, name, 0, normalizedCategory, null, normalizedParent);
Lines changed: 33 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,33 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one
3+
* or more contributor license agreements. See the NOTICE file
4+
* distributed with this work for additional information
5+
* regarding copyright ownership. The ASF licenses this file
6+
* to you under the Apache License, Version 2.0 (the
7+
* "License"); you may not use this file except in compliance
8+
* with the License. You may obtain a copy of the License at
9+
*
10+
* http://www.apache.org/licenses/LICENSE-2.0
11+
*
12+
* Unless required by applicable law or agreed to in writing, software
13+
* distributed under the License is distributed on an "AS IS" BASIS,
14+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15+
* See the License for the specific language governing permissions and
16+
* limitations under the License.
17+
*/
18+
19+
package org.apache.iotdb.db.pipe.resource.memory;
20+
21+
/** Parser working memory charged to TsFiles and admitted against the protected combined quota. */
22+
public class PipeTsFileParserMemoryBlock extends PipeTsFileMemoryBlock {
23+
24+
PipeTsFileParserMemoryBlock(
25+
final PipeMemoryManager pipeMemoryManager,
26+
final String name,
27+
final long memoryUsageInBytes,
28+
final PipeMemoryBlockCategory category,
29+
final String assigner,
30+
final PipeMemoryBlock parent) {
31+
super(pipeMemoryManager, name, memoryUsageInBytes, category, assigner, parent);
32+
}
33+
}

0 commit comments

Comments
 (0)