Caideyipi commented on code in PR #18260:
URL: https://github.com/apache/iotdb/pull/18260#discussion_r3627248064
##########
iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/PipeDescriptor.java:
##########
@@ -238,6 +238,16 @@ public static void loadPipeInternalConfig(CommonConfig
config, TrimProperties pr
Long.parseLong(
properties.getProperty(
"pipe_tsfile_parser_memory",
String.valueOf(config.getPipeTsFileParserMemory()))));
+ config.setPipeTsFileParserInFlightMaxNum(
+ Integer.parseInt(
+ properties.getProperty(
+ "pipe_tsfile_parser_in_flight_max_num",
+ String.valueOf(config.getPipeTsFileParserInFlightMaxNum()))));
+ config.setPipeTsFileParserInFlightMaxNumPerPipe(
+ Integer.parseInt(
+ properties.getProperty(
+ "pipe_tsfile_parser_in_flight_max_num_per_pipe",
+
String.valueOf(config.getPipeTsFileParserInFlightMaxNumPerPipe()))));
Review Comment:
Applied in 043f79892f7. Both properties are now documented in
`iotdb-system.properties.template` with `effectiveMode: hot_reload`. The
existing Pipe hot-reload path updates the values, and it now also wakes the
next eligible queued reservation so an increased limit takes effect immediately.
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryManager.java:
##########
@@ -144,29 +157,118 @@ < allowedMaxMemorySizeInBytesOfTabletsAndTsFiles()
&& (double) usedMemorySizeInBytesOfTablets <
allowedMaxMemorySizeInBytesOfTablets();
}
- public synchronized boolean tryReserveTsFileParserMemory() {
- if (!PIPE_MEMORY_MANAGEMENT_ENABLED) {
- return true;
+ public synchronized boolean tryReserveTsFileParserMemory(
+ final String pipeName, final long creationTime, final Object
reservationKey) {
+ if (reservationKey == null) {
+ return false;
+ }
+
+ final PipeIdentity pipeIdentity = new PipeIdentity(pipeName, creationTime);
+ enqueueTsFileParserReservationRequest(pipeIdentity, reservationKey);
+
+ final int globalLimit = Math.max(1,
PIPE_CONFIG.getPipeTsFileParserInFlightMaxNum());
+ final int perPipeLimit =
+ Math.max(1, Math.min(globalLimit,
PIPE_CONFIG.getPipeTsFileParserInFlightMaxNumPerPipe()));
+ final int reservedCountOfPipe =
reservedTsFileParserCountByPipe.getOrDefault(pipeIdentity, 0);
+ if (reservedTsFileParserCount >= globalLimit || reservedCountOfPipe >=
perPipeLimit) {
+ return false;
}
final long parserMemorySizeInBytes = getTsFileParserMemorySizeInBytes();
- if
(isEnough4TabletParsingWithReservedParserMemory(parserMemorySizeInBytes)) {
- reservedTsFileParserCount++;
- return true;
+ final boolean isSoftMemoryEnough =
+ !PIPE_MEMORY_MANAGEMENT_ENABLED
+ ||
isEnough4TabletParsingWithReservedParserMemory(parserMemorySizeInBytes);
+ if (!isSoftMemoryEnough
+ &&
!isHardEnough4TabletParsingWithReservedParserMemory(parserMemorySizeInBytes)) {
+ return false;
}
- return false;
+ final PipeIdentity nextPipe =
+ getNextEligibleTsFileParserPipe(perPipeLimit, !isSoftMemoryEnough);
+ final LinkedHashSet<Object> requestsOfPipe =
+ waitingTsFileParserRequestsByPipe.get(pipeIdentity);
+ if (!pipeIdentity.equals(nextPipe)
+ || requestsOfPipe == null
+ || !reservationKey.equals(requestsOfPipe.iterator().next())) {
+ return false;
+ }
+
+ removeTsFileParserReservationRequest(pipeIdentity, reservationKey, true);
+ reservedTsFileParserCount++;
+ reservedTsFileParserCountByPipe.put(pipeIdentity, reservedCountOfPipe + 1);
+ return true;
}
- public synchronized void releaseTsFileParserMemory() {
- if (!PIPE_MEMORY_MANAGEMENT_ENABLED) {
+ public synchronized void cancelTsFileParserMemoryReservation(
+ final String pipeName, final long creationTime, final Object
reservationKey) {
+ if (reservationKey == null) {
+ return;
+ }
+ removeTsFileParserReservationRequest(
+ new PipeIdentity(pipeName, creationTime), reservationKey, false);
+ this.notifyAll();
+ }
+
+ public synchronized void releaseTsFileParserMemory(
+ final String pipeName, final long creationTime) {
+ final PipeIdentity pipeIdentity = new PipeIdentity(pipeName, creationTime);
+ final int reservedCountOfPipe =
reservedTsFileParserCountByPipe.getOrDefault(pipeIdentity, 0);
+ if (reservedCountOfPipe <= 0) {
return;
}
Review Comment:
Agreed, this is not expected. Applied in 043f79892f7: an unmatched release
now emits a WARN with the Pipe name, creation time, and DataRegion. The message
was added to both English and Chinese locale sources.
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryManager.java:
##########
@@ -144,29 +157,118 @@ < allowedMaxMemorySizeInBytesOfTabletsAndTsFiles()
&& (double) usedMemorySizeInBytesOfTablets <
allowedMaxMemorySizeInBytesOfTablets();
}
- public synchronized boolean tryReserveTsFileParserMemory() {
- if (!PIPE_MEMORY_MANAGEMENT_ENABLED) {
- return true;
+ public synchronized boolean tryReserveTsFileParserMemory(
+ final String pipeName, final long creationTime, final Object
reservationKey) {
+ if (reservationKey == null) {
+ return false;
+ }
Review Comment:
Applied in 043f79892f7. Each waiting event now blocks on a dedicated sticky
reservation key instead of polling every 10 ms. The manager signals only the
next eligible key after admission, parser release, cancellation, memory
release, or a hot-reloaded limit increase. The concurrent fairness test now
waits on these directed notifications without polling.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
