jt2594838 commented on code in PR #18260:
URL: https://github.com/apache/iotdb/pull/18260#discussion_r3626887937
##########
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:
Is this normal? If not, may print a warn log to help debugging.
##########
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:
I wonder if it would be better to let the caller wait on reservationKey when
allocation fails.
And this method notify the next-to-wait reservationKey to reduce cases where
the callers call this method only to find that the caller does not have the
priority.
##########
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:
Add to the template. Better to support hot-reload, in case that a single
pipe is too slow due to the limitation.
--
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]
