From bc22e3918a229e14119fa20d3270e9d37ed0b294 Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Tue, 3 Nov 2026 17:08:52 +0800 Subject: [PATCH] [Pipe] Improve fair TsFile parser admission (#19106) (cherry picked from commit c515756e93b77073151429605740aced6bf62c14) --- .../iotdb/db/i18n/DataNodePipeMessages.java | 3 + .../iotdb/db/i18n/DataNodePipeMessages.java | 3 + .../processor/PipeProcessorSubtask.java | 6 + .../processor/PipeProcessorSubtaskWorker.java | 6 +- .../tsfile/PipeTsFileInsertionEvent.java | 25 +- .../metric/overview/PipeResourceMetrics.java | 45 +- .../resource/memory/PipeMemoryManager.java | 190 ++++++-- ...ipeProcessorSubtaskExecutionGuardTest.java | 247 ++++++++++- .../PipeProcessorSubtaskWorkerTest.java | 181 ++++++++ .../overview/PipeResourceMetricsTest.java | 81 ++++ .../memory/PipeMemoryManagerTest.java | 409 +++++++++++++++++- .../commons/service/metric/enums/Metric.java | 1 + .../commons/service/metric/enums/Tag.java | 1 + 13 files changed, 1143 insertions(+), 55 deletions(-) create mode 100644 iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/metric/overview/PipeResourceMetricsTest.java diff --git a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java index db51ff09b0020..dd06f08864368 100644 --- a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java +++ b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java @@ -23,6 +23,9 @@ public final class DataNodePipeMessages { private DataNodePipeMessages() {} + public static final String LOG_DECREASE_REFERENCE_COUNT_FOR_TSFILE_ARG_ERROR_8EF0D114 = + "Decrease reference count for TsFile {} error."; + // ===================== LEGACY PIPE RECEIVER RESPONSES ===================== public static final String MESSAGE_PIPE_DATA_TRANSPORT_ERROR_ARG_ECE6A238 = diff --git a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java index 053fc9cf69f8b..86515a60b26ce 100644 --- a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java +++ b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java @@ -23,6 +23,9 @@ public final class DataNodePipeMessages { private DataNodePipeMessages() {} + public static final String LOG_DECREASE_REFERENCE_COUNT_FOR_TSFILE_ARG_ERROR_8EF0D114 = + "减少 TsFile {} 的引用计数时发生错误。"; + // ===================== LEGACY PIPE RECEIVER RESPONSES ===================== public static final String MESSAGE_PIPE_DATA_TRANSPORT_ERROR_ARG_ECE6A238 = "Pipe 数据传输错误：%s"; diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtask.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtask.java index 6e20498b7849c..814e2b9d61538 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtask.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtask.java @@ -293,6 +293,12 @@ public boolean isStoppedByException() { return lastEvent instanceof EnrichedEvent && retryCount.get() > MAX_RETRY_TIMES; } + boolean isWaitingForTsFileParserMemory() { + final Event event = lastEvent; + return event instanceof PipeTsFileInsertionEvent + && ((PipeTsFileInsertionEvent) event).isWaitingForTsFileParserMemory(); + } + @Override public void close() { // Always deregister the metrics to avoid the deletion of the data region diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtaskWorker.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtaskWorker.java index 4572d217e91f6..3daf348f0a0be 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtaskWorker.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtaskWorker.java @@ -88,9 +88,11 @@ boolean runSubtasks() { ++totalRoundInAdjustmentInterval; boolean canSleepBeforeNextRound = true; - for (final PipeProcessorSubtask subtask : subtasks) { - if (subtask.isClosed() || !subtask.isSubmittingSelf() || subtask.isStoppedByException()) { + if (subtask.isClosed() + || !subtask.isSubmittingSelf() + || subtask.isStoppedByException() + || subtask.isWaitingForTsFileParserMemory()) { continue; } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeTsFileInsertionEvent.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeTsFileInsertionEvent.java index 0d379db7831b5..802f2f5f3591e 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeTsFileInsertionEvent.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeTsFileInsertionEvent.java @@ -28,6 +28,7 @@ import org.apache.iotdb.commons.pipe.event.EnrichedEvent; import org.apache.iotdb.commons.pipe.resource.log.PipeLogger; import org.apache.iotdb.commons.pipe.resource.ref.PipePhantomReferenceManager.PipeEventResource; +import org.apache.iotdb.db.i18n.DataNodePipeMessages; import org.apache.iotdb.db.pipe.agent.task.subtask.processor.PipeProcessorSubtaskExecutionGuard; import org.apache.iotdb.db.pipe.agent.task.subtask.processor.PipeProcessorSubtaskYieldException; import org.apache.iotdb.db.pipe.event.ReferenceTrackableEvent; @@ -1060,6 +1061,10 @@ private void releaseTsFileParserMemoryIfReserved() { } } + public boolean isWaitingForTsFileParserMemory() { + return tsFileParserMemoryReservationKey.isWaitingForSignal(); + } + public void cancelTsFileParserMemoryReservationIfPending() { synchronized (isTsFileParserMemoryReserved) { if (!isTsFileParserMemoryReserved.get()) { @@ -1138,12 +1143,8 @@ public long count(final boolean skipReportOnCommit) throws Exception { /** Release the resource of {@link TsFileInsertionDataContainer}. */ @Override public void close() { - try { - cancelTsFileParserMemoryReservationIfPending(); - } catch (final Exception e) { - LOGGER.warn("Decrease reference count for TsFile {} error.", tsFile, e); - } - + // Every cleanup step is best effort and independent. A parser/reader close failure must not + // strand its reservation or the pending tablet event. final PipeRawTabletInsertionEvent detachedPendingEvent; final boolean pendingEventIsBeingConsumed; final TsFileInsertionDataContainer containerToClose; @@ -1164,6 +1165,18 @@ public void close() { } } + // Invalidate old invocations before cancelling: an invocation that passed its generation + // check can enqueue while close waits for dataContainer. It must either be removed here or + // observe the new generation and cancel its own request. + try { + cancelTsFileParserMemoryReservationIfPending(); + } catch (final Exception e) { + LOGGER.warn( + DataNodePipeMessages.LOG_DECREASE_REFERENCE_COUNT_FOR_TSFILE_ARG_ERROR_8EF0D114, + tsFile, + e); + } + if (detachedPendingEvent != null && !pendingEventIsBeingConsumed) { try { releaseParsedTabletEvent(detachedPendingEvent); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/metric/overview/PipeResourceMetrics.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/metric/overview/PipeResourceMetrics.java index 85ce5047b85ff..7aec113b0ea62 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/metric/overview/PipeResourceMetrics.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/metric/overview/PipeResourceMetrics.java @@ -33,6 +33,9 @@ import org.apache.iotdb.metrics.utils.MetricLevel; import org.apache.iotdb.metrics.utils.MetricType; +import java.util.Collections; +import java.util.EnumMap; +import java.util.Map; import java.util.concurrent.TimeUnit; public class PipeResourceMetrics implements IMetricSet { @@ -49,6 +52,21 @@ public class PipeResourceMetrics implements IMetricSet { private Counter diskIOCounter = DoNothingMetricManager.DO_NOTHING_COUNTER; private volatile Counter tsFileParserAdmissionRejectCounter = DoNothingMetricManager.DO_NOTHING_COUNTER; + + public enum TsFileParserAdmissionRejectReason { + CAP("cap"), + MEMORY("memory"), + FAIRNESS("fairness"); + + private final String tag; + + TsFileParserAdmissionRejectReason(final String tag) { + this.tag = tag; + } + } + + private volatile Map + tsFileParserAdmissionRejectCounters = Collections.emptyMap(); private volatile Counter tsFileParserAdmissionYieldCounter = DoNothingMetricManager.DO_NOTHING_COUNTER; private volatile Timer tsFileParserAdmissionWaitTimer = DoNothingMetricManager.DO_NOTHING_TIMER; @@ -110,9 +128,22 @@ public void bindTo(final AbstractMetricService metricService) { diskIOCounter = metricService.getOrCreateCounter( Metric.PIPE_TSFILE_SEND_DISK_IO.toString(), MetricLevel.IMPORTANT); + final Map rejectCounters = + new EnumMap<>(TsFileParserAdmissionRejectReason.class); tsFileParserAdmissionRejectCounter = metricService.getOrCreateCounter( Metric.PIPE_TSFILE_PARSER_ADMISSION_REJECT.toString(), MetricLevel.IMPORTANT); + for (final TsFileParserAdmissionRejectReason reason : + TsFileParserAdmissionRejectReason.values()) { + rejectCounters.put( + reason, + metricService.getOrCreateCounter( + Metric.PIPE_TSFILE_PARSER_ADMISSION_REJECT_REASON.toString(), + MetricLevel.IMPORTANT, + Tag.REASON.toString(), + reason.tag)); + } + tsFileParserAdmissionRejectCounters = rejectCounters; tsFileParserAdmissionYieldCounter = metricService.getOrCreateCounter( Metric.PIPE_TSFILE_PARSER_ADMISSION_YIELD.toString(), MetricLevel.IMPORTANT); @@ -166,10 +197,19 @@ public void unbindFrom(final AbstractMetricService metricService) { metricService.remove(MetricType.COUNTER, Metric.PIPE_TSFILE_SEND_DISK_IO.toString()); metricService.remove(MetricType.COUNTER, Metric.PIPE_TSFILE_PARSER_ADMISSION_REJECT.toString()); + for (final TsFileParserAdmissionRejectReason reason : + TsFileParserAdmissionRejectReason.values()) { + metricService.remove( + MetricType.COUNTER, + Metric.PIPE_TSFILE_PARSER_ADMISSION_REJECT_REASON.toString(), + Tag.REASON.toString(), + reason.tag); + } metricService.remove(MetricType.COUNTER, Metric.PIPE_TSFILE_PARSER_ADMISSION_YIELD.toString()); metricService.remove(MetricType.TIMER, Metric.PIPE_TSFILE_PARSER_ADMISSION_WAIT.toString()); metricService.remove(MetricType.AUTO_GAUGE, Metric.PIPE_TSFILE_PARSER_IN_FLIGHT.toString()); metricService.remove(MetricType.AUTO_GAUGE, Metric.PIPE_TSFILE_PARSER_WAITING.toString()); + tsFileParserAdmissionRejectCounters = Collections.emptyMap(); tsFileParserAdmissionRejectCounter = DoNothingMetricManager.DO_NOTHING_COUNTER; tsFileParserAdmissionYieldCounter = DoNothingMetricManager.DO_NOTHING_COUNTER; tsFileParserAdmissionWaitTimer = DoNothingMetricManager.DO_NOTHING_TIMER; @@ -179,8 +219,11 @@ public void recordDiskIO(final long bytes) { diskIOCounter.inc(bytes); } - public void recordTsFileParserAdmissionReject() { + public void recordTsFileParserAdmissionReject(final TsFileParserAdmissionRejectReason reason) { tsFileParserAdmissionRejectCounter.inc(); + tsFileParserAdmissionRejectCounters + .getOrDefault(reason, DoNothingMetricManager.DO_NOTHING_COUNTER) + .inc(); } public void recordTsFileParserAdmissionYield() { diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryManager.java index 8fc3d49536af1..8cf5dfd6cd74f 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryManager.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryManager.java @@ -24,6 +24,7 @@ import org.apache.iotdb.db.conf.IoTDBDescriptor; import org.apache.iotdb.db.pipe.agent.PipeDataNodeAgent; import org.apache.iotdb.db.pipe.metric.overview.PipeResourceMetrics; +import org.apache.iotdb.db.pipe.metric.overview.PipeResourceMetrics.TsFileParserAdmissionRejectReason; import org.apache.iotdb.db.pipe.resource.memory.strategy.ThresholdAllocationStrategy; import org.slf4j.Logger; @@ -93,6 +94,11 @@ public PipeMemoryManager() { "PipeMemoryManager#tryExpandAll()", this::tryExpandAllAndCheckConsistency, PipeConfig.getInstance().getPipeMemoryExpanderIntervalSeconds()); + PipeDataNodeAgent.runtime() + .registerPeriodicalJob( + "PipeMemoryManager#notifyTsFileParserMemoryReservation()", + this::notifyNextTsFileParserMemoryReservation, + PipeConfig.getInstance().getPipeSubtaskExecutorCronHeartbeatEventIntervalSeconds()); } PipeMemoryManager( @@ -208,7 +214,8 @@ public synchronized boolean tryReserveTsFileParserMemory( reservedTsFileParserCountByPipeRegion.getOrDefault(pipeRegionIdentity, 0); if (reservedTsFileParserCount >= globalLimit || reservedCountOfPipeRegion >= perPipeRegionLimit) { - PipeResourceMetrics.getInstance().recordTsFileParserAdmissionReject(); + PipeResourceMetrics.getInstance() + .recordTsFileParserAdmissionReject(TsFileParserAdmissionRejectReason.CAP); notifyNextTsFileParserMemoryReservationInternal(); return false; } @@ -225,25 +232,28 @@ public synchronized boolean tryReserveTsFileParserMemory( if (!isSoftMemoryEnough && !isHardEnough4TabletParsingWithReservedParserMemory( parserMemorySizeInBytes, totalNonFloatingMemorySizeInBytes)) { - PipeResourceMetrics.getInstance().recordTsFileParserAdmissionReject(); + PipeResourceMetrics.getInstance() + .recordTsFileParserAdmissionReject(TsFileParserAdmissionRejectReason.MEMORY); return false; } - final PipeRegionIdentity nextPipeRegion = - getNextEligibleTsFileParserPipeRegion(perPipeRegionLimit, !isSoftMemoryEnough); - final LinkedHashSet requestsOfPipeRegion = - waitingTsFileParserRequestsByPipeRegion.get(pipeRegionIdentity); - if (!pipeRegionIdentity.equals(nextPipeRegion) - || requestsOfPipeRegion == null - || !reservationKey.equals(requestsOfPipeRegion.iterator().next())) { - PipeResourceMetrics.getInstance().recordTsFileParserAdmissionReject(); - signalTsFileParserMemoryReservation(nextPipeRegion); + final List eligibleRequests = + getEligibleTsFileParserMemoryReservations( + globalLimit, perPipeRegionLimit, totalNonFloatingMemorySizeInBytes); + if (!eligibleRequests.contains(reservationKey)) { + PipeResourceMetrics.getInstance() + .recordTsFileParserAdmissionReject(TsFileParserAdmissionRejectReason.FAIRNESS); + eligibleRequests.forEach(TsFileParserMemoryReservation::signal); return false; } removeTsFileParserReservationRequest(pipeRegionIdentity, reservationKey, true); - if (wasRequestAlreadyWaiting) { + // First-attempt admissions also spend a fair turn when other requests are queued. Otherwise + // continuously arriving pipes can bypass the same waiting pipe without advancing the cursor. + if (wasRequestAlreadyWaiting || waitingTsFileParserCount > 0) { lastAdmittedWaitingTsFileParserPipe = pipeIdentity; + } + if (wasRequestAlreadyWaiting) { PipeResourceMetrics.getInstance() .recordTsFileParserAdmissionWait( System.nanoTime() - reservationKey.waitingStartTimeInNanos); @@ -313,7 +323,17 @@ private boolean enqueueTsFileParserReservationRequest( waitingTsFileParserRegionOrderByPipe.computeIfAbsent( key.pipeIdentity, pipe -> { - waitingTsFileParserPipeOrder.addLast(pipe); + // Keep the remembered cursor at the tail. A newly arriving pipe must not + // jump ahead of older, already notified requests by landing after it. + if (lastAdmittedWaitingTsFileParserPipe != null + && lastAdmittedWaitingTsFileParserPipe.equals( + waitingTsFileParserPipeOrder.peekLast())) { + final PipeIdentity cursor = waitingTsFileParserPipeOrder.removeLast(); + waitingTsFileParserPipeOrder.addLast(pipe); + waitingTsFileParserPipeOrder.addLast(cursor); + } else { + waitingTsFileParserPipeOrder.addLast(pipe); + } return new ArrayDeque<>(); }); regionOrder.addLast(key); @@ -322,8 +342,10 @@ private boolean enqueueTsFileParserReservationRequest( if (requestsOfPipeRegion.add(reservationKey)) { waitingTsFileParserCount++; reservationKey.waitingStartTimeInNanos = System.nanoTime(); + reservationKey.markWaiting(); return false; } + reservationKey.markWaiting(); return true; } @@ -337,45 +359,89 @@ private void notifyNextTsFileParserMemoryReservationInternal() { return; } - final long parserMemorySizeInBytes = getTsFileParserMemorySizeInBytes(); final long totalNonFloatingMemorySizeInBytes = PIPE_MEMORY_MANAGEMENT_ENABLED ? getTotalNonFloatingMemorySizeInBytes() : 0; - final boolean isSoftMemoryEnough = - !PIPE_MEMORY_MANAGEMENT_ENABLED - || isEnough4TabletParsingWithReservedParserMemory( - parserMemorySizeInBytes, totalNonFloatingMemorySizeInBytes); - if (!isSoftMemoryEnough - && !isHardEnough4TabletParsingWithReservedParserMemory( - parserMemorySizeInBytes, totalNonFloatingMemorySizeInBytes)) { - return; - } - final int perPipeRegionLimit = Math.max( 1, Math.min(globalLimit, PIPE_CONFIG.getPipeTsFileParserInFlightMaxNumPerPipeRegion())); - final PipeRegionIdentity nextPipeRegion = - getNextEligibleTsFileParserPipeRegion(perPipeRegionLimit, !isSoftMemoryEnough); - signalTsFileParserMemoryReservation(nextPipeRegion); + getEligibleTsFileParserMemoryReservations( + globalLimit, perPipeRegionLimit, totalNonFloatingMemorySizeInBytes) + .forEach(TsFileParserMemoryReservation::signal); } - private void signalTsFileParserMemoryReservation(final PipeRegionIdentity nextPipeRegion) { - if (nextPipeRegion == null) { - return; - } - - final LinkedHashSet requestsOfPipeRegion = - waitingTsFileParserRequestsByPipeRegion.get(nextPipeRegion); - if (requestsOfPipeRegion != null && !requestsOfPipeRegion.isEmpty()) { - requestsOfPipeRegion.iterator().next().signal(); + private List getEligibleTsFileParserMemoryReservations( + final int globalLimit, + final int perPipeRegionLimit, + final long totalNonFloatingMemorySizeInBytes) { + final List eligibleRequests = new ArrayList<>(); + final Map selectedRequestsByPipeRegion = new HashMap<>(); + final Set selectedPipes = new HashSet<>(); + PipeIdentity lastSelectedPipe = lastAdmittedWaitingTsFileParserPipe; + // Expose one fair turn per available slot, instead of serializing all free slots behind a + // single worker. Account for every earlier turn's memory even if its worker has not run yet. + // Thus later requests cannot take the capacity needed by an earlier waiting request. + while (eligibleRequests.size() < globalLimit - reservedTsFileParserCount) { + final long extraMemoryInBytes = + (eligibleRequests.size() + 1L) * getTsFileParserMemorySizeInBytes(); + final boolean isSoftMemoryEnough = + !PIPE_MEMORY_MANAGEMENT_ENABLED + || isEnough4TabletParsingWithReservedParserMemory( + extraMemoryInBytes, totalNonFloatingMemorySizeInBytes); + if (!isSoftMemoryEnough + && !isHardEnough4TabletParsingWithReservedParserMemory( + extraMemoryInBytes, totalNonFloatingMemorySizeInBytes)) { + break; + } + final PipeRegionIdentity nextPipeRegion = + getNextEligibleTsFileParserPipeRegion( + perPipeRegionLimit, + !isSoftMemoryEnough, + lastSelectedPipe, + selectedRequestsByPipeRegion, + selectedPipes, + true); + final PipeRegionIdentity nextRegionWhenNoOtherPipeIsEligible = + nextPipeRegion == null + ? getNextEligibleTsFileParserPipeRegion( + perPipeRegionLimit, + !isSoftMemoryEnough, + lastSelectedPipe, + selectedRequestsByPipeRegion, + selectedPipes, + false) + : nextPipeRegion; + if (nextRegionWhenNoOtherPipeIsEligible == null) { + break; + } + final java.util.Iterator requests = + waitingTsFileParserRequestsByPipeRegion + .get(nextRegionWhenNoOtherPipeIsEligible) + .iterator(); + for (int i = + selectedRequestsByPipeRegion.getOrDefault(nextRegionWhenNoOtherPipeIsEligible, 0); + i > 0; + i--) { + requests.next(); + } + eligibleRequests.add(requests.next()); + selectedRequestsByPipeRegion.merge(nextRegionWhenNoOtherPipeIsEligible, 1, Integer::sum); + selectedPipes.add(nextRegionWhenNoOtherPipeIsEligible.pipeIdentity); + lastSelectedPipe = nextRegionWhenNoOtherPipeIsEligible.pipeIdentity; } + return eligibleRequests; } private PipeRegionIdentity getNextEligibleTsFileParserPipeRegion( - final int perPipeRegionLimit, final boolean requirePipeWithoutReservedParser) { + final int perPipeRegionLimit, + final boolean requirePipeWithoutReservedParser, + final PipeIdentity lastSelectedPipe, + final Map selectedRequestsByPipeRegion, + final Set selectedPipes, + final boolean requireUnselectedPipe) { PipeRegionIdentity firstEligiblePipeRegion = null; - boolean hasVisitedLastAdmittedPipe = lastAdmittedWaitingTsFileParserPipe == null; + boolean hasVisitedLastAdmittedPipe = lastSelectedPipe == null; for (final PipeIdentity pipeIdentity : waitingTsFileParserPipeOrder) { - final boolean isLastAdmittedPipe = pipeIdentity.equals(lastAdmittedWaitingTsFileParserPipe); + final boolean isLastAdmittedPipe = pipeIdentity.equals(lastSelectedPipe); if (isLastAdmittedPipe) { hasVisitedLastAdmittedPipe = true; } @@ -383,7 +449,11 @@ private PipeRegionIdentity getNextEligibleTsFileParserPipeRegion( // Under soft memory pressure, reserve the hard-threshold headroom for a pipe that has no // parser yet. Otherwise a busy pipe at the queue head can block every pipe behind it. if (requirePipeWithoutReservedParser - && reservedTsFileParserCountByPipe.getOrDefault(pipeIdentity, 0) > 0) { + && (reservedTsFileParserCountByPipe.getOrDefault(pipeIdentity, 0) > 0 + || selectedPipes.contains(pipeIdentity))) { + continue; + } + if (requireUnselectedPipe && selectedPipes.contains(pipeIdentity)) { continue; } @@ -394,10 +464,20 @@ private PipeRegionIdentity getNextEligibleTsFileParserPipeRegion( } PipeRegionIdentity eligiblePipeRegion = null; for (final PipeRegionIdentity pipeRegionIdentity : regionOrder) { - if (reservedTsFileParserCountByPipeRegion.getOrDefault(pipeRegionIdentity, 0) - < perPipeRegionLimit) { - eligiblePipeRegion = pipeRegionIdentity; - break; + final int selectedCount = selectedRequestsByPipeRegion.getOrDefault(pipeRegionIdentity, 0); + if (selectedCount < waitingTsFileParserRequestsByPipeRegion.get(pipeRegionIdentity).size() + && reservedTsFileParserCountByPipeRegion.getOrDefault(pipeRegionIdentity, 0) + + selectedCount + < perPipeRegionLimit) { + if (eligiblePipeRegion == null) { + eligiblePipeRegion = pipeRegionIdentity; + } + // Give every eligible region of the pipe one turn before reusing a region's remaining + // per-pipe-region capacity in this admission window. + if (selectedCount == 0) { + eligiblePipeRegion = pipeRegionIdentity; + break; + } } } if (eligiblePipeRegion == null) { @@ -430,6 +510,7 @@ private void removeTsFileParserReservationRequest( return; } waitingTsFileParserCount--; + reservationKey.clearWaiting(); final PipeIdentity pipeIdentity = pipeRegionIdentity.pipeIdentity; final ArrayDeque regionOrder = @@ -463,6 +544,7 @@ private void removeTsFileParserReservationRequest( public static final class TsFileParserMemoryReservation { private boolean isSignaled; + private boolean isWaiting; private volatile Runnable wakeUpCallback; // Accessed only under the PipeMemoryManager monitor while this request is queued. private long waitingStartTimeInNanos; @@ -472,6 +554,20 @@ public void setWakeUpCallback(final Runnable wakeUpCallback) { this.wakeUpCallback = wakeUpCallback; } + public synchronized boolean isWaitingForSignal() { + return isWaiting && !isSignaled; + } + + private synchronized void markWaiting() { + isWaiting = true; + isSignaled = false; + } + + private synchronized void clearWaiting() { + isWaiting = false; + isSignaled = false; + } + public synchronized void await(final long timeoutInMs) throws InterruptedException { if (!isSignaled) { wait(timeoutInMs); @@ -480,6 +576,9 @@ public synchronized void await(final long timeoutInMs) throws InterruptedExcepti } private synchronized void signal() { + if (isSignaled) { + return; + } isSignaled = true; notify(); final Runnable callback = wakeUpCallback; @@ -1032,6 +1131,9 @@ void removeShrinkableBlock(final PipeMemoryBlock block) { public synchronized void tryExpandAllAndCheckConsistency() { expandableBlocks.forEach(PipeMemoryBlock::expand); + // Floating-memory usage and the parent pool capacity can change without releasing a pipe + // memory block. Re-evaluate blocked admissions even when no release notification was sent. + notifyNextTsFileParserMemoryReservationInternal(); if (LOGGER.isDebugEnabled()) { final long blockSum = diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtaskExecutionGuardTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtaskExecutionGuardTest.java index a9e49892e839e..02fb86c1538ff 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtaskExecutionGuardTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtaskExecutionGuardTest.java @@ -21,14 +21,25 @@ import org.apache.iotdb.commons.conf.CommonConfig; import org.apache.iotdb.commons.conf.CommonDescriptor; +import org.apache.iotdb.commons.pipe.agent.plugin.builtin.processor.donothing.DoNothingProcessor; +import org.apache.iotdb.commons.pipe.agent.task.connection.EventSupplier; +import org.apache.iotdb.commons.pipe.agent.task.connection.UnboundedBlockingPendingQueue; +import org.apache.iotdb.commons.pipe.agent.task.execution.PipeSubtaskExecutor; +import org.apache.iotdb.commons.pipe.agent.task.execution.PipeSubtaskScheduler; import org.apache.iotdb.commons.pipe.datastructure.pattern.PrefixPipePattern; import org.apache.iotdb.commons.utils.FileUtils; +import org.apache.iotdb.db.pipe.agent.task.connection.PipeEventCollector; +import org.apache.iotdb.db.pipe.event.common.tablet.PipeRawTabletInsertionEvent; import org.apache.iotdb.db.pipe.event.common.tsfile.PipeTsFileInsertionEvent; +import org.apache.iotdb.db.pipe.metric.source.PipeDataRegionEventCounter; 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.storageengine.dataregion.tsfile.TsFileResource; import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResourceStatus; +import org.apache.iotdb.pipe.api.collector.EventCollector; +import org.apache.iotdb.pipe.api.event.Event; +import org.apache.iotdb.pipe.api.event.dml.insertion.TsFileInsertionEvent; import org.apache.tsfile.file.metadata.IDeviceID; import org.apache.tsfile.file.metadata.PlainDeviceID; @@ -37,11 +48,19 @@ import org.junit.Test; import java.io.File; +import java.io.IOException; +import java.lang.reflect.Field; import java.nio.file.Files; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicReference; +import static org.awaitility.Awaitility.await; +import static org.mockito.Mockito.mock; + public class PipeProcessorSubtaskExecutionGuardTest { @Test @@ -109,12 +128,14 @@ public void testParserAdmissionYieldsWithoutBlockingAndResumes() throws Exceptio PipeProcessorSubtaskYieldException.Reason.PARSER_NOT_ADMITTED, admissionYield.getReason()); Assert.assertTrue(TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startTimeInNanos) < 1000); + Assert.assertTrue(event.isWaitingForTsFileParserMemory()); executionGuard.exit(); // A yielded invocation must keep its fair queue position and receive the release signal. memoryManager.releaseTsFileParserMemory("blocker", 0, "0"); isBlockerReserved = false; Assert.assertEquals(1, wakeUpCount.get()); + Assert.assertFalse(event.isWaitingForTsFileParserMemory()); executionGuard.stop(); event.cancelTsFileParserMemoryReservationIfPending(); isCompetitorReserved = @@ -156,6 +177,228 @@ public void testParserAdmissionYieldsWithoutBlockingAndResumes() throws Exceptio } } + @Test(timeout = 60000) + public void testCloseCancelsReservationEnqueuedByAWorkerInvocation() throws Exception { + final CommonConfig config = CommonDescriptor.getInstance().getConfig(); + final PipeMemoryManager manager = PipeDataNodeResourceManager.memory(); + final long originalParserMemory = config.getPipeTsFileParserMemory(); + final int originalGlobalLimit = config.getPipeTsFileParserInFlightMaxNum(); + final int originalRegionLimit = config.getPipeTsFileParserInFlightMaxNumPerPipeRegion(); + final File tempDir = Files.createTempDirectory("pipeProcessorCloseAdmission").toFile(); + final ExecutorService executor = Executors.newSingleThreadExecutor(); + final AtomicReference closingThread = new AtomicReference<>(); + PipeTsFileInsertionEvent event = null; + PipeProcessorSubtask subtask = null; + boolean blockerReserved = false; + try { + config.setPipeTsFileParserMemory(1); + config.setPipeTsFileParserInFlightMaxNum(1); + config.setPipeTsFileParserInFlightMaxNumPerPipeRegion(1); + blockerReserved = + manager.tryReserveTsFileParserMemory( + "blocker", 0, "0", new TsFileParserMemoryReservation()); + Assert.assertTrue(blockerReserved); + event = createEvent(tempDir, "close.tsfile", "closingPipe"); + Assert.assertTrue(event.increaseReferenceCount(getClass().getName())); + final PipeTsFileInsertionEvent eventToClose = event; + final AtomicReference suppliedEvent = new AtomicReference<>(event); + final UnboundedBlockingPendingQueue outputQueue = + new UnboundedBlockingPendingQueue<>(new PipeDataRegionEventCounter()); + subtask = + new ParserSubtask( + "closingPipe", + () -> suppliedEvent.getAndSet(null), + new DoNothingProcessor(), + outputQueue); + final PipeProcessorSubtaskWorker worker = new PipeProcessorSubtaskWorker(); + worker.schedule(subtask); + subtask.allowSubmittingSelf(); + final Field parserField = PipeTsFileInsertionEvent.class.getDeclaredField("dataContainer"); + parserField.setAccessible(true); + final Object parserMonitor = parserField.get(event); + final Future closing; + synchronized (parserMonitor) { + closing = + executor.submit( + () -> { + closingThread.set(Thread.currentThread()); + eventToClose.close(); + }); + await() + .atMost(5, TimeUnit.SECONDS) + .until( + () -> + closingThread.get() != null + && closingThread.get().getState() == Thread.State.BLOCKED); + // Delay only the closer. This thread re-enters the monitor through the real worker, + // subtask and collector, including both generation checks around admission. This models + // close being descheduled just before acquiring dataContainer, without bypassing parsing. + Assert.assertTrue(worker.runSubtasks()); + Assert.assertNull(suppliedEvent.get()); + Assert.assertTrue(event.isWaitingForTsFileParserMemory()); + Assert.assertTrue(subtask.isWaitingForTsFileParserMemory()); + Assert.assertEquals(1, manager.getWaitingTsFileParserCount()); + Assert.assertTrue(outputQueue.isEmpty()); + } + closing.get(5, TimeUnit.SECONDS); + Assert.assertFalse(event.isWaitingForTsFileParserMemory()); + Assert.assertEquals(0, manager.getWaitingTsFileParserCount()); + Assert.assertEquals(1, manager.getReservedTsFileParserCount()); + } finally { + executor.shutdownNow(); + Assert.assertTrue(executor.awaitTermination(5, TimeUnit.SECONDS)); + if (subtask != null) { + subtask.disallowSubmittingSelf(); + subtask.close(); + } + if (event != null) { + event.close(); + } + if (blockerReserved) { + manager.releaseTsFileParserMemory("blocker", 0, "0"); + } + config.setPipeTsFileParserMemory(originalParserMemory); + config.setPipeTsFileParserInFlightMaxNum(originalGlobalLimit); + config.setPipeTsFileParserInFlightMaxNumPerPipeRegion(originalRegionLimit); + FileUtils.deleteFileOrDirectory(tempDir); + } + } + + @Test(timeout = 60000) + public void testRealWorkerParserWaitAcrossStopStartAndDrop() throws Exception { + final CommonConfig config = CommonDescriptor.getInstance().getConfig(); + final PipeMemoryManager manager = PipeDataNodeResourceManager.memory(); + final long originalParserMemory = config.getPipeTsFileParserMemory(); + final int originalGlobalLimit = config.getPipeTsFileParserInFlightMaxNum(); + final int originalRegionLimit = config.getPipeTsFileParserInFlightMaxNumPerPipeRegion(); + final File tempDir = Files.createTempDirectory("pipeProcessorWorkerLifecycle").toFile(); + final AtomicReference suppliedEvent = new AtomicReference<>(); + final UnboundedBlockingPendingQueue outputQueue = + new UnboundedBlockingPendingQueue<>(new PipeDataRegionEventCounter()); + final AtomicInteger processorCalls = new AtomicInteger(); + final DoNothingProcessor processor = + new DoNothingProcessor() { + @Override + public void process(final TsFileInsertionEvent event, final EventCollector collector) + throws IOException { + processorCalls.incrementAndGet(); + super.process(event, collector); + } + }; + final PipeProcessorSubtask subtask = + new ParserSubtask( + "lifecyclePipe", () -> suppliedEvent.getAndSet(null), processor, outputQueue); + final PipeProcessorSubtaskWorker worker = new PipeProcessorSubtaskWorker(); + PipeTsFileInsertionEvent first = null; + PipeTsFileInsertionEvent dropped = null; + boolean blockerReserved = false; + try { + config.setPipeTsFileParserMemory(1); + config.setPipeTsFileParserInFlightMaxNum(1); + config.setPipeTsFileParserInFlightMaxNumPerPipeRegion(1); + blockerReserved = + manager.tryReserveTsFileParserMemory( + "blocker", 0, "0", new TsFileParserMemoryReservation()); + Assert.assertTrue(blockerReserved); + first = createEvent(tempDir, "resume.tsfile", "lifecyclePipe"); + Assert.assertTrue(first.increaseReferenceCount(getClass().getName())); + suppliedEvent.set(first); + worker.schedule(subtask); + subtask.allowSubmittingSelf(); + + Assert.assertTrue(worker.runSubtasks()); + Assert.assertTrue(subtask.isWaitingForTsFileParserMemory()); + Assert.assertTrue(worker.runSubtasks()); + Assert.assertEquals(1, processorCalls.get()); + Assert.assertTrue(subtask.disallowSubmittingSelf()); + Assert.assertFalse(first.isWaitingForTsFileParserMemory()); + Assert.assertEquals(0, manager.getWaitingTsFileParserCount()); + subtask.allowSubmittingSelf(); + Assert.assertTrue(worker.runSubtasks()); + Assert.assertTrue(subtask.isWaitingForTsFileParserMemory()); + Assert.assertEquals(2, processorCalls.get()); + Assert.assertTrue(outputQueue.isEmpty()); + + manager.releaseTsFileParserMemory("blocker", 0, "0"); + blockerReserved = false; + Assert.assertFalse(subtask.isWaitingForTsFileParserMemory()); + Assert.assertFalse(worker.runSubtasks()); + Assert.assertEquals(3, processorCalls.get()); + Assert.assertTrue(first.isReleased()); + Assert.assertTrue(outputQueue.getTabletInsertionEventCount() > 0); + Assert.assertEquals(0, manager.getWaitingTsFileParserCount()); + Assert.assertEquals(0, manager.getReservedTsFileParserCount()); + int rowCount = 0; + Event tablet; + while ((tablet = outputQueue.directPoll()) != null) { + Assert.assertTrue(tablet instanceof PipeRawTabletInsertionEvent); + rowCount += ((PipeRawTabletInsertionEvent) tablet).convertToTablet().rowSize; + ((PipeRawTabletInsertionEvent) tablet).clearReferenceCount(getClass().getName()); + } + Assert.assertEquals(10, rowCount); + + blockerReserved = + manager.tryReserveTsFileParserMemory( + "blocker", 0, "0", new TsFileParserMemoryReservation()); + Assert.assertTrue(blockerReserved); + dropped = createEvent(tempDir, "drop.tsfile", "lifecyclePipe"); + Assert.assertTrue(dropped.increaseReferenceCount(getClass().getName())); + suppliedEvent.set(dropped); + Assert.assertTrue(worker.runSubtasks()); + Assert.assertTrue(subtask.isWaitingForTsFileParserMemory()); + Assert.assertTrue(subtask.disallowSubmittingSelf()); + subtask.close(); + Assert.assertTrue(dropped.isReleased()); + Assert.assertFalse(dropped.isWaitingForTsFileParserMemory()); + Assert.assertEquals(0, manager.getWaitingTsFileParserCount()); + Assert.assertEquals(1, manager.getReservedTsFileParserCount()); + Assert.assertTrue(worker.runSubtasks()); + Assert.assertEquals(4, processorCalls.get()); + } finally { + subtask.disallowSubmittingSelf(); + subtask.close(); + if (first != null) { + if (!first.isReleased()) { + first.clearReferenceCount(getClass().getName()); + } + first.close(); + } + if (dropped != null) { + if (!dropped.isReleased()) { + dropped.clearReferenceCount(getClass().getName()); + } + dropped.close(); + } + outputQueue.discardAllEvents(); + if (blockerReserved) { + manager.releaseTsFileParserMemory("blocker", 0, "0"); + } + config.setPipeTsFileParserMemory(originalParserMemory); + config.setPipeTsFileParserInFlightMaxNum(originalGlobalLimit); + config.setPipeTsFileParserInFlightMaxNumPerPipeRegion(originalRegionLimit); + FileUtils.deleteFileOrDirectory(tempDir); + } + } + + private static final class ParserSubtask extends PipeProcessorSubtask { + private ParserSubtask( + final String pipeName, + final EventSupplier supplier, + final DoNothingProcessor processor, + final UnboundedBlockingPendingQueue outputQueue) { + super( + pipeName, + pipeName, + 0, + 0, + supplier, + processor, + new PipeEventCollector(outputQueue, 0, 0, true, false)); + // Use the production scheduler without creating the process-wide worker thread pool. + subtaskScheduler = new PipeSubtaskScheduler(mock(PipeSubtaskExecutor.class)); + } + } + @Test(timeout = 60000) public void testPauseAfterTabletResumesWithoutDuplicateConsumption() throws Exception { final CommonConfig commonConfig = CommonDescriptor.getInstance().getConfig(); @@ -221,9 +464,11 @@ public void testPauseAfterTabletResumesWithoutDuplicateConsumption() throws Exce private PipeTsFileInsertionEvent createEvent( final File tempDir, final String fileName, final String pipeName) throws Exception { + final File dataDir = new File(tempDir, "sequence/root.testsg/0/0"); + Files.createDirectories(dataDir.toPath()); final File tsFile = TsFileGeneratorUtils.generateNonAlignedTsFile( - new File(tempDir, fileName).getAbsolutePath(), 1, 1, 10, 0, 100, 10, 10); + new File(dataDir, fileName).getAbsolutePath(), 1, 1, 10, 0, 100, 10, 10); final TsFileResource resource = new TsFileResource(tsFile); resource.setStatusForTest(TsFileResourceStatus.NORMAL); final IDeviceID deviceID = new PlainDeviceID("root.testsg.d0"); diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtaskWorkerTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtaskWorkerTest.java index d1d24c1183739..2cc4e30b80ea6 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtaskWorkerTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtaskWorkerTest.java @@ -52,6 +52,7 @@ import static org.mockito.Mockito.inOrder; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.never; +import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; @@ -92,6 +93,8 @@ public void testOneHundredWaitingPipesDoNotWaitForWorkerPolling() throws Excepti final TsFileParserMemoryReservation request = new TsFileParserMemoryReservation(); requests.add(request); final PipeProcessorSubtask subtask = createRunnableSubtask(pipeName); + when(subtask.isWaitingForTsFileParserMemory()) + .thenAnswer(ignored -> request.isWaitingForSignal()); doAnswer( invocation -> { request.setWakeUpCallback(invocation.getArgument(0)); @@ -171,6 +174,59 @@ public void testOneHundredWaitingPipesDoNotWaitForWorkerPolling() throws Excepti } } + @Test(timeout = 10000) + public void testWaitingParserIsSkippedUntilAdmissionNotification() throws Exception { + final CommonConfig config = CommonDescriptor.getInstance().getConfig(); + final PipeMemoryManager memoryManager = PipeDataNodeResourceManager.memory(); + final long originalParserMemory = config.getPipeTsFileParserMemory(); + final int originalGlobalLimit = config.getPipeTsFileParserInFlightMaxNum(); + final int originalRegionLimit = config.getPipeTsFileParserInFlightMaxNumPerPipeRegion(); + final TsFileParserMemoryReservation blocker = new TsFileParserMemoryReservation(); + final TsFileParserMemoryReservation request = new TsFileParserMemoryReservation(); + boolean blockerReserved = false; + try { + config.setPipeTsFileParserMemory(1); + config.setPipeTsFileParserInFlightMaxNum(1); + config.setPipeTsFileParserInFlightMaxNumPerPipeRegion(1); + blockerReserved = memoryManager.tryReserveTsFileParserMemory("blocker", 0, "0", blocker); + Assert.assertTrue(blockerReserved); + final PipeProcessorSubtaskWorker worker = + new PipeProcessorSubtaskWorker(new LinkedHashSet<>()); + final PipeProcessorSubtask subtask = createRunnableSubtask("waitingPipe"); + when(subtask.isWaitingForTsFileParserMemory()) + .thenAnswer(ignored -> request.isWaitingForSignal()); + when(subtask.call()) + .thenAnswer( + ignored -> { + if (!memoryManager.tryReserveTsFileParserMemory("waitingPipe", 0, "0", request)) { + throw PipeProcessorSubtaskYieldException.parserNotAdmitted(); + } + memoryManager.releaseTsFileParserMemory("waitingPipe", 0, "0"); + return true; + }); + worker.schedule(subtask); + Assert.assertTrue(worker.runSubtasks()); + for (int i = 0; i < 100; i++) { + Assert.assertTrue(worker.runSubtasks()); + } + verify(subtask, times(1)).call(); + memoryManager.releaseTsFileParserMemory("blocker", 0, "0"); + blockerReserved = false; + Assert.assertFalse(worker.runSubtasks()); + verify(subtask, times(2)).call(); + Assert.assertEquals(0, memoryManager.getWaitingTsFileParserCount()); + Assert.assertEquals(0, memoryManager.getReservedTsFileParserCount()); + } finally { + memoryManager.cancelTsFileParserMemoryReservation("waitingPipe", 0, "0", request); + if (blockerReserved) { + memoryManager.releaseTsFileParserMemory("blocker", 0, "0"); + } + config.setPipeTsFileParserMemory(originalParserMemory); + config.setPipeTsFileParserInFlightMaxNum(originalGlobalLimit); + config.setPipeTsFileParserInFlightMaxNumPerPipeRegion(originalRegionLimit); + } + } + @Test(timeout = 10000) public void testNotificationBeforeIdleWaitIsNotLost() throws Exception { final PipeProcessorSubtaskWorker worker = new PipeProcessorSubtaskWorker(new LinkedHashSet<>()); @@ -196,6 +252,131 @@ public void testNotificationBeforeIdleWaitIsNotLost() throws Exception { } } + @Test(timeout = 15000) + public void testBusyFirstWorkerDoesNotSerializeOtherEligibleWorker() throws Exception { + final CommonConfig config = CommonDescriptor.getInstance().getConfig(); + final PipeMemoryManager manager = PipeDataNodeResourceManager.memory(); + final long originalParserMemory = config.getPipeTsFileParserMemory(); + final int originalGlobalLimit = config.getPipeTsFileParserInFlightMaxNum(); + final int originalRegionLimit = config.getPipeTsFileParserInFlightMaxNumPerPipeRegion(); + final boolean[] reservedBlockers = new boolean[2]; + final TsFileParserMemoryReservation first = new TsFileParserMemoryReservation(); + final TsFileParserMemoryReservation second = new TsFileParserMemoryReservation(); + final PipeProcessorSubtaskWorker firstWorker = + new PipeProcessorSubtaskWorker(new LinkedHashSet<>()); + final PipeProcessorSubtaskWorker secondWorker = + new PipeProcessorSubtaskWorker(new LinkedHashSet<>()); + final CountDownLatch firstWorkerBusy = new CountDownLatch(1); + final CountDownLatch unblockFirstWorker = new CountDownLatch(1); + final CountDownLatch secondAdmitted = new CountDownLatch(1); + final AtomicBoolean blockFirstWorker = new AtomicBoolean(); + final AtomicBoolean secondProcessed = new AtomicBoolean(); + final ExecutorService executor = Executors.newFixedThreadPool(2); + try { + config.setPipeTsFileParserMemory(1); + config.setPipeTsFileParserInFlightMaxNum(2); + config.setPipeTsFileParserInFlightMaxNumPerPipeRegion(1); + for (int i = 0; i < reservedBlockers.length; i++) { + reservedBlockers[i] = + manager.tryReserveTsFileParserMemory( + "blocker" + i, 0, "0", new TsFileParserMemoryReservation()); + Assert.assertTrue(reservedBlockers[i]); + } + final PipeProcessorSubtask busySubtask = createRunnableSubtask("busy"); + when(busySubtask.call()) + .thenAnswer( + ignored -> { + if (blockFirstWorker.get()) { + firstWorkerBusy.countDown(); + Assert.assertTrue(unblockFirstWorker.await(5, TimeUnit.SECONDS)); + } + return false; + }); + firstWorker.schedule(busySubtask); + final PipeProcessorSubtask firstSubtask = createRunnableSubtask("first"); + final PipeProcessorSubtask secondSubtask = createRunnableSubtask("second"); + when(firstSubtask.isWaitingForTsFileParserMemory()) + .thenAnswer(ignored -> first.isWaitingForSignal()); + when(secondSubtask.isWaitingForTsFileParserMemory()) + .thenAnswer(ignored -> second.isWaitingForSignal()); + doAnswer( + invocation -> { + first.setWakeUpCallback(invocation.getArgument(0)); + return null; + }) + .when(firstSubtask) + .setWorkerWakeUpCallback(any()); + doAnswer( + invocation -> { + second.setWakeUpCallback(invocation.getArgument(0)); + return null; + }) + .when(secondSubtask) + .setWorkerWakeUpCallback(any()); + when(firstSubtask.call()) + .thenAnswer( + ignored -> { + if (!manager.tryReserveTsFileParserMemory("first", 0, "0", first)) { + throw PipeProcessorSubtaskYieldException.parserNotAdmitted(); + } + manager.releaseTsFileParserMemory("first", 0, "0"); + return true; + }); + when(secondSubtask.call()) + .thenAnswer( + ignored -> { + if (secondProcessed.get()) { + return false; + } + if (!manager.tryReserveTsFileParserMemory("second", 0, "0", second)) { + throw PipeProcessorSubtaskYieldException.parserNotAdmitted(); + } + manager.releaseTsFileParserMemory("second", 0, "0"); + secondProcessed.set(true); + secondAdmitted.countDown(); + return true; + }); + firstWorker.schedule(firstSubtask); + secondWorker.schedule(secondSubtask); + Assert.assertTrue(firstWorker.runSubtasks()); + Assert.assertTrue(secondWorker.runSubtasks()); + blockFirstWorker.set(true); + final Future busyRound = executor.submit(firstWorker::runSubtasks); + Assert.assertTrue(firstWorkerBusy.await(3, TimeUnit.SECONDS)); + for (int i = 0; i < reservedBlockers.length; i++) { + manager.releaseTsFileParserMemory("blocker" + i, 0, "0"); + reservedBlockers[i] = false; + } + executor.submit( + () -> { + while (!Thread.currentThread().isInterrupted() && !secondProcessed.get()) { + secondWorker.sleepIfNecessary(secondWorker.runSubtasks()); + } + }); + Assert.assertTrue(secondAdmitted.await(3, TimeUnit.SECONDS)); + verify(firstSubtask, times(1)).call(); + Assert.assertFalse(busyRound.isDone()); + unblockFirstWorker.countDown(); + busyRound.get(3, TimeUnit.SECONDS); + Assert.assertEquals(0, manager.getWaitingTsFileParserCount()); + Assert.assertEquals(0, manager.getReservedTsFileParserCount()); + } finally { + unblockFirstWorker.countDown(); + executor.shutdownNow(); + Assert.assertTrue(executor.awaitTermination(3, TimeUnit.SECONDS)); + manager.cancelTsFileParserMemoryReservation("first", 0, "0", first); + manager.cancelTsFileParserMemoryReservation("second", 0, "0", second); + for (int i = 0; i < reservedBlockers.length; i++) { + if (reservedBlockers[i]) { + manager.releaseTsFileParserMemory("blocker" + i, 0, "0"); + } + } + config.setPipeTsFileParserMemory(originalParserMemory); + config.setPipeTsFileParserInFlightMaxNum(originalGlobalLimit); + config.setPipeTsFileParserInFlightMaxNumPerPipeRegion(originalRegionLimit); + } + } + @Test public void testYieldingPipesDoNotBlockAnotherPipeOnSameWorker() throws Exception { final PipeProcessorSubtaskWorker worker = new PipeProcessorSubtaskWorker(new LinkedHashSet<>()); diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/metric/overview/PipeResourceMetricsTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/metric/overview/PipeResourceMetricsTest.java new file mode 100644 index 0000000000000..962233b7faaf1 --- /dev/null +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/metric/overview/PipeResourceMetricsTest.java @@ -0,0 +1,81 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.iotdb.db.pipe.metric.overview; + +import org.apache.iotdb.commons.service.metric.enums.Metric; +import org.apache.iotdb.commons.service.metric.enums.Tag; +import org.apache.iotdb.db.pipe.metric.overview.PipeResourceMetrics.TsFileParserAdmissionRejectReason; +import org.apache.iotdb.metrics.AbstractMetricService; +import org.apache.iotdb.metrics.type.Counter; +import org.apache.iotdb.metrics.utils.MetricLevel; +import org.apache.iotdb.metrics.utils.MetricType; + +import org.junit.Test; + +import java.lang.reflect.Constructor; + +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +public class PipeResourceMetricsTest { + + @Test + public void testAdmissionRejectReasonsAreBoundAndRemoved() throws Exception { + final AbstractMetricService service = mock(AbstractMetricService.class); + final Constructor constructor = + PipeResourceMetrics.class.getDeclaredConstructor(); + constructor.setAccessible(true); + final PipeResourceMetrics metrics = constructor.newInstance(); + final Counter cap = mock(Counter.class); + final Counter memory = mock(Counter.class); + final Counter fairness = mock(Counter.class); + final String name = Metric.PIPE_TSFILE_PARSER_ADMISSION_REJECT.toString(); + final Counter aggregate = mock(Counter.class); + when(service.getOrCreateCounter(name, MetricLevel.IMPORTANT)).thenReturn(aggregate); + final String reasonMetricName = Metric.PIPE_TSFILE_PARSER_ADMISSION_REJECT_REASON.toString(); + when(service.getOrCreateCounter( + reasonMetricName, MetricLevel.IMPORTANT, Tag.REASON.toString(), "cap")) + .thenReturn(cap); + when(service.getOrCreateCounter( + reasonMetricName, MetricLevel.IMPORTANT, Tag.REASON.toString(), "memory")) + .thenReturn(memory); + when(service.getOrCreateCounter( + reasonMetricName, MetricLevel.IMPORTANT, Tag.REASON.toString(), "fairness")) + .thenReturn(fairness); + + metrics.bindTo(service); + metrics.recordTsFileParserAdmissionReject(TsFileParserAdmissionRejectReason.CAP); + metrics.recordTsFileParserAdmissionReject(TsFileParserAdmissionRejectReason.MEMORY); + metrics.recordTsFileParserAdmissionReject(TsFileParserAdmissionRejectReason.FAIRNESS); + metrics.unbindFrom(service); + verify(service).remove(MetricType.COUNTER, name); + for (final String reason : new String[] {"cap", "memory", "fairness"}) { + verify(service).remove(MetricType.COUNTER, reasonMetricName, Tag.REASON.toString(), reason); + } + // After unbinding, admissions must no longer update detached metric instances. + metrics.recordTsFileParserAdmissionReject(TsFileParserAdmissionRejectReason.CAP); + verify(aggregate, times(3)).inc(); + verify(cap, times(1)).inc(); + verify(memory, times(1)).inc(); + verify(fairness, times(1)).inc(); + } +} diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryManagerTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryManagerTest.java index 6b681c5dd49e1..09aec264340ad 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryManagerTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryManagerTest.java @@ -39,6 +39,7 @@ import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicLong; public class PipeMemoryManagerTest { @@ -104,6 +105,394 @@ public void testAdmissionNotificationOnlyWakesEligibleWaitingPipe() { Assert.assertEquals(0, memoryManager.getWaitingTsFileParserCount()); } + @Test + public void testFreeSlotsAreAvailableWithoutWaitingForTheFirstWorker() { + commonConfig.setPipeTsFileParserInFlightMaxNum(4); + commonConfig.setPipeTsFileParserInFlightMaxNumPerPipeRegion(1); + final List blockers = new ArrayList<>(); + final List waiting = new ArrayList<>(); + final AtomicInteger[] wakeUps = new AtomicInteger[100]; + for (int i = 0; i < 4; i++) { + final Reservation blocker = new Reservation("blocker" + i, 0); + blockers.add(blocker); + Assert.assertTrue(tryAcquire(blocker)); + } + for (int i = 0; i < 100; i++) { + final Reservation request = new Reservation("waiting" + i, 0); + waiting.add(request); + wakeUps[i] = new AtomicInteger(); + request.key.setWakeUpCallback(wakeUps[i]::incrementAndGet); + Assert.assertFalse(tryAcquire(request)); + } + blockers.forEach(this::release); + memoryManager.notifyNextTsFileParserMemoryReservation(); + for (int i = 0; i < 100; i++) { + Assert.assertEquals(i < 4 ? 1 : 0, wakeUps[i].get()); + } + + // The first worker is still busy. Other eligible workers can fill three slots, but must leave + // the remaining slot for the oldest request instead of allowing a fifth pipe to overtake it. + Assert.assertFalse(tryAcquire(waiting.get(4))); + Assert.assertTrue(tryAcquire(waiting.get(3))); + Assert.assertTrue(tryAcquire(waiting.get(2))); + Assert.assertTrue(tryAcquire(waiting.get(1))); + Assert.assertEquals(3, memoryManager.getReservedTsFileParserCount()); + Assert.assertFalse(tryAcquire(waiting.get(4))); + Assert.assertTrue(tryAcquire(waiting.get(0))); + Assert.assertEquals(4, memoryManager.getReservedTsFileParserCount()); + Assert.assertEquals(96, memoryManager.getWaitingTsFileParserCount()); + } + + @Test + public void testAdmissionWindowRespectsPipeFairnessAcrossRegions() { + commonConfig.setPipeTsFileParserInFlightMaxNum(2); + commonConfig.setPipeTsFileParserInFlightMaxNumPerPipeRegion(1); + final PipeMemoryManager manager = createTestMemoryManager(); + final Reservation blockerA = new Reservation("blockerA", 0); + final Reservation blockerB = new Reservation("blockerB", 0); + final Reservation pipeARegion1 = new Reservation("pipeA", 1, "1"); + final Reservation pipeARegion2 = new Reservation("pipeA", 1, "2"); + final Reservation pipeB = new Reservation("pipeB", 2); + final List testReservations = + List.of(blockerA, blockerB, pipeARegion1, pipeARegion2, pipeB); + try { + Assert.assertTrue(tryAcquire(manager, blockerA)); + Assert.assertTrue(tryAcquire(manager, blockerB)); + Assert.assertFalse(tryAcquire(manager, pipeARegion1)); + Assert.assertFalse(tryAcquire(manager, pipeARegion2)); + Assert.assertFalse(tryAcquire(manager, pipeB)); + release(manager, blockerA); + release(manager, blockerB); + + // Even with soft memory headroom for both slots, admit another pipe before a second region + // of the first pipe. This keeps the admission window fair independently of memory pressure. + Assert.assertFalse(tryAcquire(manager, pipeARegion2)); + Assert.assertTrue(tryAcquire(manager, pipeB)); + Assert.assertTrue(tryAcquire(manager, pipeARegion1)); + } finally { + for (final Reservation reservation : testReservations) { + manager.cancelTsFileParserMemoryReservation( + reservation.pipeName, + reservation.creationTime, + reservation.dataRegionId, + reservation.key); + release(manager, reservation); + } + } + } + + @Test + public void testAdmissionWindowUsesAvailableSlotsWhenOnlyOnePipeIsWaiting() { + commonConfig.setPipeTsFileParserInFlightMaxNum(2); + commonConfig.setPipeTsFileParserInFlightMaxNumPerPipeRegion(1); + final PipeMemoryManager manager = createTestMemoryManager(); + final Reservation blockerA = new Reservation("blockerA", 0); + final Reservation blockerB = new Reservation("blockerB", 0); + final Reservation pipeARegion1 = new Reservation("pipeA", 1, "1"); + final Reservation pipeARegion2 = new Reservation("pipeA", 1, "2"); + final List testReservations = + List.of(blockerA, blockerB, pipeARegion1, pipeARegion2); + try { + Assert.assertTrue(tryAcquire(manager, blockerA)); + Assert.assertTrue(tryAcquire(manager, blockerB)); + Assert.assertFalse(tryAcquire(manager, pipeARegion1)); + Assert.assertFalse(tryAcquire(manager, pipeARegion2)); + release(manager, blockerA); + release(manager, blockerB); + + Assert.assertTrue(tryAcquire(manager, pipeARegion1)); + Assert.assertTrue(tryAcquire(manager, pipeARegion2)); + Assert.assertEquals(2, manager.getReservedTsFileParserCount()); + } finally { + for (final Reservation reservation : testReservations) { + manager.cancelTsFileParserMemoryReservation( + reservation.pipeName, + reservation.creationTime, + reservation.dataRegionId, + reservation.key); + release(manager, reservation); + } + } + } + + @Test + public void testAdmissionWindowHonorsPerPipeRegionLimitAboveOne() { + commonConfig.setPipeTsFileParserInFlightMaxNum(4); + commonConfig.setPipeTsFileParserInFlightMaxNumPerPipeRegion(2); + Assert.assertEquals(2, commonConfig.getPipeTsFileParserInFlightMaxNumPerPipeRegion()); + final PipeMemoryManager manager = createTestMemoryManager(); + final List blockers = new ArrayList<>(); + final List waiting = new ArrayList<>(); + final AtomicInteger[] wakeUps = new AtomicInteger[3]; + for (int i = 0; i < 4; i++) { + final Reservation blocker = new Reservation("blocker" + i, 0); + blockers.add(blocker); + } + for (int i = 0; i < 3; i++) { + final Reservation request = new Reservation("pipeA", 1); + waiting.add(request); + wakeUps[i] = new AtomicInteger(); + request.key.setWakeUpCallback(wakeUps[i]::incrementAndGet); + } + final List testReservations = new ArrayList<>(blockers); + testReservations.addAll(waiting); + try { + for (final Reservation blocker : blockers) { + Assert.assertTrue(tryAcquire(manager, blocker)); + } + for (final Reservation request : waiting) { + Assert.assertFalse(tryAcquire(manager, request)); + } + blockers.forEach(blocker -> release(manager, blocker)); + manager.notifyNextTsFileParserMemoryReservation(); + + Assert.assertEquals(1, wakeUps[0].get()); + Assert.assertEquals(1, wakeUps[1].get()); + Assert.assertEquals(0, wakeUps[2].get()); + Assert.assertTrue(tryAcquire(manager, waiting.get(1))); + Assert.assertTrue(tryAcquire(manager, waiting.get(0))); + Assert.assertEquals(2, manager.getReservedTsFileParserCount()); + Assert.assertFalse(tryAcquire(manager, waiting.get(2))); + } finally { + for (final Reservation reservation : testReservations) { + manager.cancelTsFileParserMemoryReservation( + reservation.pipeName, + reservation.creationTime, + reservation.dataRegionId, + reservation.key); + release(manager, reservation); + } + } + } + + @Test + public void testNewPipesCannotOvertakeNotifiedRequestsAfterReverseAdmission() { + commonConfig.setPipeTsFileParserInFlightMaxNum(2); + commonConfig.setPipeTsFileParserInFlightMaxNumPerPipeRegion(1); + final PipeMemoryManager manager = createTestMemoryManager(); + final Reservation blockerA = new Reservation("blockerA", 0); + final Reservation blockerB = new Reservation("blockerB", 0); + final Reservation oldest = new Reservation("oldest", 1); + final Reservation multiRegion1 = new Reservation("multi", 2, "1"); + final Reservation multiRegion2 = new Reservation("multi", 2, "2"); + final List testReservations = new ArrayList<>(); + testReservations.addAll(List.of(blockerA, blockerB, oldest, multiRegion1, multiRegion2)); + try { + Assert.assertTrue(tryAcquire(manager, blockerA)); + Assert.assertTrue(tryAcquire(manager, blockerB)); + Assert.assertFalse(tryAcquire(manager, oldest)); + Assert.assertFalse(tryAcquire(manager, multiRegion1)); + Assert.assertFalse(tryAcquire(manager, multiRegion2)); + release(manager, blockerA); + release(manager, blockerB); + Assert.assertFalse(oldest.key.isWaitingForSignal()); + Assert.assertTrue(tryAcquire(manager, multiRegion1)); + + // A continuously replenished tail must not steal the remaining notified request's slot. + for (int i = 0; i < 10; i++) { + final Reservation newcomer = new Reservation("newcomer" + i, 3); + testReservations.add(newcomer); + Assert.assertFalse(tryAcquire(manager, newcomer)); + } + Assert.assertTrue(tryAcquire(manager, oldest)); + Assert.assertFalse(tryAcquire(manager, multiRegion2)); + } finally { + for (final Reservation reservation : testReservations) { + manager.cancelTsFileParserMemoryReservation( + reservation.pipeName, + reservation.creationTime, + reservation.dataRegionId, + reservation.key); + release(manager, reservation); + } + } + } + + @Test + public void testImmediateAdmissionsCannotContinuouslyOvertakeWaitingPipe() { + commonConfig.setPipeTsFileParserInFlightMaxNum(2); + commonConfig.setPipeTsFileParserInFlightMaxNumPerPipeRegion(2); + final PipeMemoryManager manager = createTestMemoryManager(); + final Reservation blockerA = new Reservation("blockerA", 0); + final Reservation blockerB = new Reservation("blockerB", 0); + final Reservation oldest = new Reservation("pipeA", 1); + final Reservation second = new Reservation("pipeA", 1); + final Reservation firstNewcomer = new Reservation("firstNewcomer", 2); + final List testReservations = + new ArrayList<>(List.of(blockerA, blockerB, oldest, second, firstNewcomer)); + try { + Assert.assertTrue(tryAcquire(manager, blockerA)); + Assert.assertTrue(tryAcquire(manager, blockerB)); + Assert.assertFalse(tryAcquire(manager, oldest)); + Assert.assertFalse(tryAcquire(manager, second)); + release(manager, blockerA); + release(manager, blockerB); + Assert.assertTrue(tryAcquire(manager, second)); + + // Another pipe can take a fair turn, even when its first attempt needs no retry. That turn + // must advance the cursor so a stream of new pipes cannot indefinitely bypass pipe A. + Assert.assertTrue(tryAcquire(manager, firstNewcomer)); + release(manager, firstNewcomer); + for (int i = 0; i < 10; i++) { + final Reservation newcomer = new Reservation("newcomer" + i, 3); + testReservations.add(newcomer); + Assert.assertFalse(tryAcquire(manager, newcomer)); + } + Assert.assertTrue(tryAcquire(manager, oldest)); + Assert.assertEquals(2, manager.getReservedTsFileParserCount()); + } finally { + for (final Reservation reservation : testReservations) { + manager.cancelTsFileParserMemoryReservation( + reservation.pipeName, + reservation.creationTime, + reservation.dataRegionId, + reservation.key); + release(manager, reservation); + } + } + } + + @Test + public void testSoftMemoryWindowAccountsForEarlierTurnsOfSamePipe() { + commonConfig.setPipeTsFileParserInFlightMaxNum(2); + commonConfig.setPipeTsFileParserInFlightMaxNumPerPipeRegion(1); + final PipeMemoryManager manager = createTestMemoryManager(); + final Reservation blockerA = new Reservation("blockerA", 0); + final Reservation blockerB = new Reservation("blockerB", 0); + final Reservation oldest = new Reservation("pipeA", 1, "1"); + final Reservation second = new Reservation("pipeA", 1, "2"); + final Reservation otherPipe = new Reservation("pipeB", 2); + final AtomicInteger secondWakeUps = new AtomicInteger(); + second.key.setWakeUpCallback(secondWakeUps::incrementAndGet); + final AtomicInteger otherPipeWakeUps = new AtomicInteger(); + otherPipe.key.setWakeUpCallback(otherPipeWakeUps::incrementAndGet); + final List testReservations = + List.of(blockerA, blockerB, oldest, second, otherPipe); + try { + Assert.assertTrue(tryAcquire(manager, blockerA)); + Assert.assertTrue(tryAcquire(manager, blockerB)); + Assert.assertFalse(tryAcquire(manager, oldest)); + Assert.assertFalse(tryAcquire(manager, second)); + Assert.assertFalse(tryAcquire(manager, otherPipe)); + final long totalMemory = manager.getTotalNonFloatingMemorySizeInBytes(); + final double tabletLimit = + (commonConfig.getPipeDataStructureTabletMemoryBlockAllocationRejectThreshold() + + commonConfig.getPipeDataStructureTsFileMemoryBlockAllocationRejectThreshold() + / 2) + * totalMemory; + final double combinedLimit = + (commonConfig.getPipeDataStructureTabletMemoryBlockAllocationRejectThreshold() + + commonConfig.getPipeDataStructureTsFileMemoryBlockAllocationRejectThreshold()) + * totalMemory; + // One turn fits below the soft threshold; two fit only below the hard threshold. + commonConfig.setPipeTsFileParserMemory((long) (Math.min(tabletLimit, combinedLimit) * 0.49)); + release(manager, blockerA); + release(manager, blockerB); + + Assert.assertFalse(oldest.key.isWaitingForSignal()); + Assert.assertEquals(0, secondWakeUps.get()); + Assert.assertEquals(1, otherPipeWakeUps.get()); + Assert.assertFalse(tryAcquire(manager, second)); + Assert.assertTrue(tryAcquire(manager, oldest)); + Assert.assertFalse(tryAcquire(manager, second)); + Assert.assertTrue(tryAcquire(manager, otherPipe)); + release(manager, oldest); + Assert.assertTrue(tryAcquire(manager, second)); + } finally { + for (final Reservation reservation : testReservations) { + manager.cancelTsFileParserMemoryReservation( + reservation.pipeName, + reservation.creationTime, + reservation.dataRegionId, + reservation.key); + release(manager, reservation); + } + } + } + + @Test + public void testAdmissionWindowIsBoundedByAvailableMemory() { + commonConfig.setPipeTsFileParserInFlightMaxNum(4); + commonConfig.setPipeTsFileParserInFlightMaxNumPerPipeRegion(1); + final List blockers = new ArrayList<>(); + for (int i = 0; i < 4; i++) { + final Reservation blocker = new Reservation("blocker" + i, 0); + blockers.add(blocker); + Assert.assertTrue(tryAcquire(blocker)); + } + final Reservation pipeA = new Reservation("pipeA", 1); + final Reservation pipeB = new Reservation("pipeB", 2); + final AtomicInteger pipeAWakeUps = new AtomicInteger(); + final AtomicInteger pipeBWakeUps = new AtomicInteger(); + pipeA.key.setWakeUpCallback(pipeAWakeUps::incrementAndGet); + pipeB.key.setWakeUpCallback(pipeBWakeUps::incrementAndGet); + Assert.assertFalse(tryAcquire(pipeA)); + Assert.assertFalse(tryAcquire(pipeB)); + final double tabletLimit = + (commonConfig.getPipeDataStructureTabletMemoryBlockAllocationRejectThreshold() + + commonConfig.getPipeDataStructureTsFileMemoryBlockAllocationRejectThreshold() / 2) + * memoryManager.getTotalNonFloatingMemorySizeInBytes(); + commonConfig.setPipeTsFileParserMemory((long) (tabletLimit * 0.6)); + blockers.forEach(this::release); + + Assert.assertEquals(1, pipeAWakeUps.get()); + Assert.assertEquals(0, pipeBWakeUps.get()); + Assert.assertFalse(tryAcquire(pipeB)); + Assert.assertTrue(tryAcquire(pipeA)); + Assert.assertFalse(tryAcquire(pipeB)); + release(pipeA); + Assert.assertTrue(tryAcquire(pipeB)); + } + + @Test + public void testCancellationClearsWaitStateAndWakesNextRequest() { + commonConfig.setPipeTsFileParserInFlightMaxNum(1); + commonConfig.setPipeTsFileParserInFlightMaxNumPerPipeRegion(1); + final Reservation blocker = new Reservation("blocker", 0); + final Reservation pipeA = new Reservation("pipeA", 1); + final Reservation pipeB = new Reservation("pipeB", 2); + Assert.assertTrue(tryAcquire(blocker)); + Assert.assertFalse(tryAcquire(pipeA)); + Assert.assertFalse(tryAcquire(pipeB)); + Assert.assertTrue(pipeA.key.isWaitingForSignal()); + release(blocker); + Assert.assertFalse(pipeA.key.isWaitingForSignal()); + Assert.assertTrue(pipeB.key.isWaitingForSignal()); + + memoryManager.cancelTsFileParserMemoryReservation("pipeA", 1, "0", pipeA.key); + Assert.assertFalse(pipeA.key.isWaitingForSignal()); + Assert.assertFalse(pipeB.key.isWaitingForSignal()); + Assert.assertTrue(tryAcquire(pipeB)); + } + + @Test + public void testFloatingMemoryReleaseWakesAdmissionDuringMaintenance() { + commonConfig.setPipeTsFileParserInFlightMaxNum(1); + commonConfig.setPipeTsFileParserInFlightMaxNumPerPipeRegion(1); + commonConfig.setPipeTsFileParserMemory(500); + final AtomicLong floatingUsage = new AtomicLong(1500); + final PipeMemoryManager manager = new PipeMemoryManager(2000, floatingUsage::get); + final TsFileParserMemoryReservation request = new TsFileParserMemoryReservation(); + final AtomicInteger wakeUps = new AtomicInteger(); + request.setWakeUpCallback(wakeUps::incrementAndGet); + boolean reserved = false; + try { + Assert.assertFalse(manager.tryReserveTsFileParserMemory("pipe", 0, "0", request)); + Assert.assertTrue(request.isWaitingForSignal()); + floatingUsage.set(0); + manager.tryExpandAllAndCheckConsistency(); + Assert.assertEquals(1, wakeUps.get()); + Assert.assertFalse(request.isWaitingForSignal()); + reserved = manager.tryReserveTsFileParserMemory("pipe", 0, "0", request); + Assert.assertTrue(reserved); + } finally { + manager.cancelTsFileParserMemoryReservation("pipe", 0, "0", request); + if (reserved) { + manager.releaseTsFileParserMemory("pipe", 0, "0"); + } + } + } + @Test public void testWaitingPipesAreAdmittedInRoundRobinOrder() { commonConfig.setPipeTsFileParserInFlightMaxNum(1); @@ -374,6 +763,16 @@ private boolean tryAcquire(final Reservation reservation) { return reservation.acquired; } + private boolean tryAcquire(final PipeMemoryManager manager, final Reservation reservation) { + reservation.acquired = + manager.tryReserveTsFileParserMemory( + reservation.pipeName, + reservation.creationTime, + reservation.dataRegionId, + reservation.key); + return reservation.acquired; + } + private boolean tryAcquireWithoutTracking(final Reservation reservation) { reservation.acquired = memoryManager.tryReserveTsFileParserMemory( @@ -385,14 +784,22 @@ private boolean tryAcquireWithoutTracking(final Reservation reservation) { } private void release(final Reservation reservation) { + release(memoryManager, reservation); + } + + private void release(final PipeMemoryManager manager, final Reservation reservation) { if (!reservation.acquired) { return; } - memoryManager.releaseTsFileParserMemory( + manager.releaseTsFileParserMemory( reservation.pipeName, reservation.creationTime, reservation.dataRegionId); reservation.acquired = false; } + private PipeMemoryManager createTestMemoryManager() { + return new PipeMemoryManager(2_000_000_000L, () -> 0); + } + private static class Reservation { private final String pipeName; diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/service/metric/enums/Metric.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/service/metric/enums/Metric.java index 2a53b1f940b09..92ee58b04cb45 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/service/metric/enums/Metric.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/service/metric/enums/Metric.java @@ -166,6 +166,7 @@ public enum Metric { PIPE_EXTRACTOR_TSFILE_EPOCH_STATE("pipe_extractor_tsfile_epoch_state"), PIPE_MEM("pipe_mem"), PIPE_TSFILE_PARSER_ADMISSION_REJECT("pipe_tsfile_parser_admission_reject"), + PIPE_TSFILE_PARSER_ADMISSION_REJECT_REASON("pipe_tsfile_parser_admission_reject_reason"), PIPE_TSFILE_PARSER_ADMISSION_YIELD("pipe_tsfile_parser_admission_yield"), PIPE_TSFILE_PARSER_ADMISSION_WAIT("pipe_tsfile_parser_admission_wait"), PIPE_TSFILE_PARSER_IN_FLIGHT("pipe_tsfile_parser_in_flight"), diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/service/metric/enums/Tag.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/service/metric/enums/Tag.java index 0bc05e2f30104..db0ec9b6f60db 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/service/metric/enums/Tag.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/service/metric/enums/Tag.java @@ -24,6 +24,7 @@ public enum Tag { NAME("name"), REGION("region"), STATUS("status"), + REASON("reason"), DATABASE("database"), FROM("from"), STAGE("stage"),