Skip to content

[Pipe] Fairly limit concurrent TsFile parsers (#18260) #18295

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

Merged
Caideyipi merged 1 commit into from
Jul 25, 2026
Merged

[Pipe] Fairly limit concurrent TsFile parsers (#18260) #18295

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
2 changes: 2 additions & 0 deletions iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -35,6 +35,7 @@
import org.apache.iotdb.confignode.rpc.thrift.TRatisConfig;
import org.apache.iotdb.db.consensus.DataRegionConsensusImpl;
import org.apache.iotdb.db.exception.query.QueryProcessException;
import org.apache.iotdb.db.pipe.resource.PipeDataNodeResourceManager;
import org.apache.iotdb.db.queryengine.plan.relational.metadata.fetcher.cache.LastCacheLoadStrategy;
import org.apache.iotdb.db.service.metrics.IoTDBInternalLocalReporter;
import org.apache.iotdb.db.storageengine.StorageEngine;
Expand Down Expand Up @@ -2647,6 +2648,7 @@ private void loadLoadTsFileHotModifiedProp(TrimProperties properties) throws IOE

private void loadPipeHotModifiedProp(TrimProperties properties) throws IOException {
PipeDescriptor.loadPipeProps(commonDescriptor.getConfig(), properties, true);
PipeDataNodeResourceManager.memory().notifyNextTsFileParserMemoryReservation();
LoggerPeriodicalLogReducer.update();
}

Expand Down
124 changes: 69 additions & 55 deletions .../src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeTsFileInsertionEvent.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -35,6 +35,7 @@
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;
import org.apache.iotdb.db.pipe.resource.memory.PipeMemoryManager.TsFileParserMemoryReservation;
import org.apache.iotdb.db.pipe.resource.tsfile.PipeTsFileResourceManager;
import org.apache.iotdb.db.pipe.source.dataregion.realtime.assigner.PipeTsFileEpochProgressIndexKeeper;
import org.apache.iotdb.db.storageengine.dataregion.memtable.TsFileProcessor;
Expand All @@ -56,7 +57,6 @@
import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
Expand All @@ -68,6 +68,7 @@ public class PipeTsFileInsertionEvent extends EnrichedEvent
private static final Logger LOGGER = LoggerFactory.getLogger(PipeTsFileInsertionEvent.class);

protected final TsFileResource resource;
private final String dataRegionId;
protected File tsFile;
protected long extractTime = 0;

Expand All @@ -87,6 +88,8 @@ public class PipeTsFileInsertionEvent extends EnrichedEvent
protected final AtomicBoolean isClosed;
protected final AtomicReference<TsFileInsertionDataContainer> dataContainer;
private final AtomicBoolean isTsFileParserMemoryReserved = new AtomicBoolean(false);
private final TsFileParserMemoryReservation tsFileParserMemoryReservationKey =
new TsFileParserMemoryReservation();
private final AtomicReference<Iterator<TabletInsertionEvent>> tabletInsertionEventIterator =
new AtomicReference<>();
private final AtomicReference<PipeRawTabletInsertionEvent> pendingTabletInsertionEvent =
Expand Down Expand Up @@ -162,6 +165,7 @@ private PipeTsFileInsertionEvent(
super(pipeName, creationTime, pipeTaskMeta, pipePattern, startTime, endTime);

this.resource = resource;
this.dataRegionId = getDataRegionId(resource);

// For events created at assigner or historical extractor, the tsFile is get from the resource
// For events created for source, the tsFile is inherited from the assigner, because the
Expand Down Expand Up @@ -226,6 +230,17 @@ private PipeTsFileInsertionEvent(
});
}

private static String getDataRegionId(final TsFileResource resource) {
// TsFileResource#getDataRegionId assumes the storage-engine directory structure, while a
// synthetic resource may wrap a standalone file.
final File resourceTsFile = resource.getTsFile();
final File timePartitionDir =
Objects.isNull(resourceTsFile) ? null : resourceTsFile.getParentFile();
final File dataRegionDir =
Objects.isNull(timePartitionDir) ? null : timePartitionDir.getParentFile();
return Objects.isNull(dataRegionDir) ? "" : dataRegionDir.getName();
}

/**
* @return {@code false} if this file can't be sent by pipe because it is empty. {@code true}
* otherwise.
Expand Down Expand Up @@ -677,31 +692,12 @@ private void waitForResourceEnough4Parsing(final long timeoutMs) throws Interrup
final long startTime = System.currentTimeMillis();
long lastRecordTime = startTime;

final long initialMemoryCheckIntervalMs =
Math.max(1, PipeConfig.getInstance().getPipeCheckMemoryEnoughIntervalMs());
final long maxMemoryCheckIntervalMs =
getMaxMemoryCheckIntervalMs(
initialMemoryCheckIntervalMs,
PipeConfig.getInstance().getPipeMemoryAllocateMaxRetries());
long memoryCheckIntervalMs = initialMemoryCheckIntervalMs;
while (true) {
final long elapsedTimeMs = Math.max(0, System.currentTimeMillis() - startTime);
if (elapsedTimeMs >= timeoutMs) {
// should contain 'TimeoutException' in exception message
throw new PipeRuntimeOutOfMemoryCriticalException(
String.format(
"TimeoutException: Waited %s seconds for memory to parse TsFile",
elapsedTimeMs / 1000.0));
}

memoryManager.waitForTsFileParserMemory(
Math.min(
getMemoryCheckIntervalWithJitter(memoryCheckIntervalMs), timeoutMs - elapsedTimeMs));

while (!tryReserveTsFileParserMemory(memoryManager)) {
final long currentTime = System.currentTimeMillis();
final double elapsedRecordTimeSeconds = (currentTime - lastRecordTime) / 1000.0;
final double waitTimeSeconds = (currentTime - startTime) / 1000.0;
if (elapsedRecordTimeSeconds > 10.0) {
final long elapsedRecordTimeInMs = currentTime - lastRecordTime;
final long waitTimeInMs = currentTime - startTime;
final double waitTimeSeconds = waitTimeInMs / 1000.0;
if (elapsedRecordTimeInMs > 10_000) {
LOGGER.info(
"Wait for memory enough for parsing {} for {} seconds.",
resource != null ? resource.getTsFilePath() : "tsfile",
Expand All @@ -714,35 +710,26 @@ private void waitForResourceEnough4Parsing(final long timeoutMs) throws Interrup
waitTimeSeconds);
}

if (tryReserveTsFileParserMemory(memoryManager)) {
LOGGER.info(
"Wait for memory enough for parsing {} for {} seconds.",
resource != null ? resource.getTsFilePath() : "tsfile",
waitTimeSeconds);
return;
if (waitTimeInMs > timeoutMs) {
// should contain 'TimeoutException' in exception message
throw new PipeRuntimeOutOfMemoryCriticalException(
String.format(
"TimeoutException: Waited %s seconds for memory to parse TsFile", waitTimeSeconds));
}

memoryCheckIntervalMs =
getNextMemoryCheckIntervalMs(memoryCheckIntervalMs, maxMemoryCheckIntervalMs);
tsFileParserMemoryReservationKey.await(
Math.max(
1,
Math.min(
timeoutMs - waitTimeInMs, 10_000 - Math.min(10_000, elapsedRecordTimeInMs))));
}
}

static long getMaxMemoryCheckIntervalMs(final long initialIntervalMs, final int maxRetries) {
final long multiplier = Math.max(1, maxRetries);
return initialIntervalMs > Long.MAX_VALUE / multiplier
? Long.MAX_VALUE
: initialIntervalMs * multiplier;
}

static long getNextMemoryCheckIntervalMs(final long currentIntervalMs, final long maxIntervalMs) {
return currentIntervalMs >= maxIntervalMs - currentIntervalMs
? maxIntervalMs
: currentIntervalMs << 1;
}

static long getMemoryCheckIntervalWithJitter(final long intervalMs) {
return Math.max(
1, (long) (intervalMs * (0.5 + ThreadLocalRandom.current().nextDouble() * 0.5)));
final long currentTime = System.currentTimeMillis();
final double waitTimeSeconds = (currentTime - startTime) / 1000.0;
LOGGER.info(
"Wait for memory enough for parsing {} for {} seconds.",
resource != null ? resource.getTsFilePath() : "tsfile",
waitTimeSeconds);
}

private boolean tryReserveTsFileParserMemory(final PipeMemoryManager memoryManager) {
Expand All @@ -751,7 +738,8 @@ private boolean tryReserveTsFileParserMemory(final PipeMemoryManager memoryManag
return true;
}

if (!memoryManager.tryReserveTsFileParserMemory()) {
if (!memoryManager.tryReserveTsFileParserMemory(
pipeName, creationTime, dataRegionId, tsFileParserMemoryReservationKey)) {
return false;
}

Expand All @@ -763,11 +751,20 @@ private boolean tryReserveTsFileParserMemory(final PipeMemoryManager memoryManag
private void releaseTsFileParserMemoryIfReserved() {
synchronized (isTsFileParserMemoryReserved) {
if (isTsFileParserMemoryReserved.compareAndSet(true, false)) {
PipeDataNodeResourceManager.memory().releaseTsFileParserMemory();
PipeDataNodeResourceManager.memory()
.releaseTsFileParserMemory(pipeName, creationTime, dataRegionId);
}
}
}

private void cancelTsFileParserMemoryReservationIfPending() {
if (!isTsFileParserMemoryReserved.get()) {
PipeDataNodeResourceManager.memory()
.cancelTsFileParserMemoryReservation(
pipeName, creationTime, dataRegionId, tsFileParserMemoryReservationKey);
}
}

/** The method is used to prevent circular replication in PipeConsensus */
public boolean isGeneratedByPipeConsensus() {
return isGeneratedByPipeConsensus;
Expand Down Expand Up @@ -829,6 +826,7 @@ public long count(final boolean skipReportOnCommit) throws Exception {
/** Release the resource of {@link TsFileInsertionDataContainer}. */
@Override
public void close() {
cancelTsFileParserMemoryReservationIfPending();
tabletInsertionEventIterator.set(null);
releaseParsedTabletEvent(pendingTabletInsertionEvent.getAndSet(null));
parsedTabletInsertionEventCount.set(0);
Expand Down Expand Up @@ -877,11 +875,14 @@ public PipeEventResource eventResourceBuilder() {
this.isReleased,
this.referenceCount,
this.pipeName,
this.creationTime,
this.dataRegionId,
this.tsFile,
this.isWithMod,
this.modFile,
this.dataContainer,
this.isTsFileParserMemoryReserved);
this.isTsFileParserMemoryReserved,
this.tsFileParserMemoryReservationKey);
}

private static class PipeTsFileInsertionEventResource extends PipeEventResource {
Expand All @@ -891,29 +892,41 @@ private static class PipeTsFileInsertionEventResource extends PipeEventResource
private final File modFile;
private final AtomicReference<TsFileInsertionDataContainer> dataContainer;
private final String pipeName;
private final long creationTime;
private final String dataRegionId;
private final AtomicBoolean isTsFileParserMemoryReserved;
private final TsFileParserMemoryReservation tsFileParserMemoryReservationKey;

private PipeTsFileInsertionEventResource(
final AtomicBoolean isReleased,
final AtomicInteger referenceCount,
final String pipeName,
final long creationTime,
final String dataRegionId,
final File tsFile,
final boolean isWithMod,
final File modFile,
final AtomicReference<TsFileInsertionDataContainer> dataContainer,
final AtomicBoolean isTsFileParserMemoryReserved) {
final AtomicBoolean isTsFileParserMemoryReserved,
final TsFileParserMemoryReservation tsFileParserMemoryReservationKey) {
super(isReleased, referenceCount);
this.pipeName = pipeName;
this.creationTime = creationTime;
this.dataRegionId = dataRegionId;
this.tsFile = tsFile;
this.isWithMod = isWithMod;
this.modFile = modFile;
this.dataContainer = dataContainer;
this.isTsFileParserMemoryReserved = isTsFileParserMemoryReserved;
this.tsFileParserMemoryReservationKey = tsFileParserMemoryReservationKey;
}

@Override
protected void finalizeResource() {
try {
PipeDataNodeResourceManager.memory()
.cancelTsFileParserMemoryReservation(
pipeName, creationTime, dataRegionId, tsFileParserMemoryReservationKey);
// decrease reference count
PipeDataNodeResourceManager.tsfile().decreaseFileReference(tsFile, pipeName);
if (isWithMod) {
Expand All @@ -930,7 +943,8 @@ protected void finalizeResource() {
});
synchronized (isTsFileParserMemoryReserved) {
if (isTsFileParserMemoryReserved.compareAndSet(true, false)) {
PipeDataNodeResourceManager.memory().releaseTsFileParserMemory();
PipeDataNodeResourceManager.memory()
.releaseTsFileParserMemory(pipeName, creationTime, dataRegionId);
}
}
} catch (final Exception e) {
Expand Down
Loading
Loading