Skip to content

[Pipe] Fix batch and async resource lifecycle leaks #18582

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
Caideyipi wants to merge 1 commit into
base: master
Choose a base branch
from
Open

[Pipe] Fix batch and async resource lifecycle leaks #18582

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 9 additions & 0 deletions iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,9 @@

public final class DataNodePipeMessages {

public static final String LOG_FAILED_TO_RESOLVE_TRANSFER_EXCEPTION_A4F5397A =
"Failed to resolve transfer exception.";

// ===================== CONSENSUS =====================

public static final String CLOSING_DELETION_RESOURCE_MANAGER_FOR =
Expand Down Expand Up @@ -2314,6 +2317,12 @@ private DataNodePipeMessages() {}
public static final String PIPE_EXCEPTION_FORCERESIZE_FAILED_TO_ALLOCATE_MEMORY_AFTER_D_RETRIES_TOTAL_8C6948BC =
"forceResize: failed to allocate memory after %d retries, total memory size %d bytes, used "
+ "memory size %d bytes, requested memory size %d bytes";
public static final String
EXCEPTION_UNSUPPORTED_BATCH_TYPE_ARG_WHEN_TRANSFERRING_TABLET_INSERTION_EVENT_66153E12 =
"Unsupported batch type %s when transferring tablet insertion event.";
public static final String
EXCEPTION_FAILED_TO_TRANSFER_TSFILE_BATCH_BECAUSE_NO_TSFILE_WAS_GENERATED_FOR_ARG_CC60CCEB =
"Failed to transfer TsFile batch because no TsFile was generated for %s.";
public static final String PIPE_EXCEPTION_FAILED_TO_GET_HARDLINK_OR_COPIED_FILE_IN_PIPE_DIR_FOR_FILE_F009D86E =
"failed to get hardlink or copied file in pipe dir for file %s, it is not a tsfile, mod file "
+ "or resource file";
Expand Down
9 changes: 9 additions & 0 deletions iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,9 @@

public final class DataNodePipeMessages {

public static final String LOG_FAILED_TO_RESOLVE_TRANSFER_EXCEPTION_A4F5397A =
"解析 transfer exception 失败。";

// ===================== CONSENSUS =====================

public static final String CLOSING_DELETION_RESOURCE_MANAGER_FOR = "正在关闭 {} 的删除资源管理器...";
Expand Down Expand Up @@ -2145,6 +2148,12 @@ private DataNodePipeMessages() {}
"forceAllocate：重试 %d 次后仍无法分配内存，总内存大小 %d bytes，已用内存大小 %d bytes，请求内存大小 %d bytes";
public static final String PIPE_EXCEPTION_FORCERESIZE_FAILED_TO_ALLOCATE_MEMORY_AFTER_D_RETRIES_TOTAL_8C6948BC =
"forceResize：重试 %d 次后仍无法分配内存，总内存大小 %d bytes，已用内存大小 %d bytes，请求内存大小 %d bytes";
public static final String
EXCEPTION_UNSUPPORTED_BATCH_TYPE_ARG_WHEN_TRANSFERRING_TABLET_INSERTION_EVENT_66153E12 =
"传输 tablet insertion event 时不支持 batch 类型 %s。";
public static final String
EXCEPTION_FAILED_TO_TRANSFER_TSFILE_BATCH_BECAUSE_NO_TSFILE_WAS_GENERATED_FOR_ARG_CC60CCEB =
"无法传输 TsFile batch，因为没有为 %s 生成 TsFile。";
public static final String PIPE_EXCEPTION_FAILED_TO_GET_HARDLINK_OR_COPIED_FILE_IN_PIPE_DIR_FOR_FILE_F009D86E =
"获取 pipe 目录中文件 %s 的 hardlink 或复制文件失败；该文件不是 tsfile、mod 文件或 resource 文件";
public static final String PIPE_EXCEPTION_PIPEPLANTOSTATEMENTVISITOR_DOES_NOT_SUPPORT_VISITING_GENERAL_452AAA60 =
Expand Down
14 changes: 12 additions & 2 deletions ...ain/java/org/apache/iotdb/db/pipe/event/common/statement/PipeStatementInsertionEvent.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,7 @@

import org.apache.iotdb.commons.consensus.index.ProgressIndex;
import org.apache.iotdb.commons.consensus.index.impl.MinimumProgressIndex;
import org.apache.iotdb.commons.exception.pipe.PipeRuntimeOutOfMemoryCriticalException;
import org.apache.iotdb.commons.pipe.agent.task.meta.PipeTaskMeta;
import org.apache.iotdb.commons.pipe.datastructure.pattern.TablePattern;
import org.apache.iotdb.commons.pipe.datastructure.pattern.TreePattern;
Expand Down Expand Up @@ -91,8 +92,17 @@ public PipeStatementInsertionEvent(

@Override
public boolean internallyIncreaseResourceReferenceCount(String holderMessage) {
PipeDataNodeResourceManager.memory()
.forceResize(allocatedMemoryBlock, statement.ramBytesUsed() + INSTANCE_SIZE);
final long targetSize = statement.ramBytesUsed() + INSTANCE_SIZE;
if (!PipeDataNodeResourceManager.memory().tryResize(allocatedMemoryBlock, targetSize)) {
throw new PipeRuntimeOutOfMemoryCriticalException(
String.format(
DataNodePipeMessages
.PIPE_EXCEPTION_FORCERESIZE_FAILED_TO_ALLOCATE_MEMORY_AFTER_D_RETRIES_TOTAL_8C6948BC,
0,
PipeDataNodeResourceManager.memory().getTotalNonFloatingMemorySizeInBytes(),
PipeDataNodeResourceManager.memory().getUsedMemorySizeInBytes(),
targetSize - allocatedMemoryBlock.getMemoryUsageInBytes()));
}
if (Objects.nonNull(pipeName)) {
PipeDataNodeSinglePipeMetrics.getInstance()
.increaseRawTabletEventCount(pipeName, creationTime);
Expand Down
16 changes: 12 additions & 4 deletions ...c/main/java/org/apache/iotdb/db/pipe/event/common/tablet/PipeRawTabletInsertionEvent.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,7 @@
import org.apache.iotdb.commons.consensus.index.ProgressIndex;
import org.apache.iotdb.commons.consensus.index.impl.MinimumProgressIndex;
import org.apache.iotdb.commons.exception.IllegalPathException;
import org.apache.iotdb.commons.exception.pipe.PipeRuntimeOutOfMemoryCriticalException;
import org.apache.iotdb.commons.pipe.agent.task.meta.PipeTaskMeta;
import org.apache.iotdb.commons.pipe.datastructure.pattern.TablePattern;
import org.apache.iotdb.commons.pipe.datastructure.pattern.TreePattern;
Expand Down Expand Up @@ -258,10 +259,17 @@ public PipeRawTabletInsertionEvent(

@Override
public boolean internallyIncreaseResourceReferenceCount(final String holderMessage) {
PipeDataNodeResourceManager.memory()
.forceResize(
allocatedMemoryBlock,
PipeMemoryWeightUtil.calculateTabletSizeInBytes(tablet) + INSTANCE_SIZE);
final long targetSize = PipeMemoryWeightUtil.calculateTabletSizeInBytes(tablet) + INSTANCE_SIZE;
if (!PipeDataNodeResourceManager.memory().tryResize(allocatedMemoryBlock, targetSize)) {
throw new PipeRuntimeOutOfMemoryCriticalException(
String.format(
DataNodePipeMessages
.PIPE_EXCEPTION_FORCERESIZE_FAILED_TO_ALLOCATE_MEMORY_AFTER_D_RETRIES_TOTAL_8C6948BC,
0,
PipeDataNodeResourceManager.memory().getTotalNonFloatingMemorySizeInBytes(),
PipeDataNodeResourceManager.memory().getUsedMemorySizeInBytes(),
targetSize - allocatedMemoryBlock.getMemoryUsageInBytes()));
}
if (Objects.nonNull(pipeName)) {
PipeDataNodeSinglePipeMetrics.getInstance()
.increaseRawTabletEventCount(pipeName, creationTime);
Expand Down
Loading