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 84485bc0c04f0..6e20498b7849c 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 @@ -271,6 +271,10 @@ public void submitSelf() { // and the worker will be submitted to the executor } + void setWorkerWakeUpCallback(final Runnable wakeUpCallback) { + executionGuard.setWorkerWakeUpCallback(wakeUpCallback); + } + @Override protected void onAllowSubmittingSelf() { executionGuard.start(); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtaskExecutionGuard.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtaskExecutionGuard.java index a4eeab3482d0d..8510f58ab0879 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtaskExecutionGuard.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtaskExecutionGuard.java @@ -19,6 +19,8 @@ package org.apache.iotdb.db.pipe.agent.task.subtask.processor; +import org.apache.iotdb.db.pipe.metric.overview.PipeResourceMetrics; + import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicLong; @@ -38,6 +40,7 @@ public class PipeProcessorSubtaskExecutionGuard { private final AtomicBoolean isRunning = new AtomicBoolean(false); private final AtomicLong executionEpoch = new AtomicLong(0); private final ThreadLocal invocationEpoch = new ThreadLocal<>(); + private volatile Runnable workerWakeUpCallback; public PipeProcessorSubtaskExecutionGuard() { this(true); @@ -55,9 +58,21 @@ public boolean isEnabled() { return enabled; } + void setWorkerWakeUpCallback(final Runnable workerWakeUpCallback) { + this.workerWakeUpCallback = workerWakeUpCallback; + } + + public Runnable getWorkerWakeUpCallback() { + return workerWakeUpCallback; + } + void start() { if (enabled) { isRunning.set(true); + final Runnable wakeUpCallback = workerWakeUpCallback; + if (wakeUpCallback != null) { + wakeUpCallback.run(); + } } } @@ -105,6 +120,7 @@ public boolean isCurrentInvocationValid() { } public void yieldIfParserNotAdmitted() { + PipeResourceMetrics.getInstance().recordTsFileParserAdmissionYield(); throw PipeProcessorSubtaskYieldException.parserNotAdmitted(); } } 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 ad1548d9432be..4572d217e91f6 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 @@ -49,6 +49,9 @@ public class PipeProcessorSubtaskWorker extends WrappedRunnable { private long sleepingTimeInMilliSecond = 50; private final Set subtasks; + private final Object wakeUpMonitor = new Object(); + private final Runnable wakeUpCallback = this::wakeUp; + private boolean isWakeUpRequested; private volatile Thread workerThread; private volatile PipeProcessorSubtask currentSubtask; @@ -115,10 +118,18 @@ boolean runSubtasks() { return canSleepBeforeNextRound; } - private void sleepIfNecessary(final boolean canSleepBeforeNextRound) { + @VisibleForTesting + void sleepIfNecessary(final boolean canSleepBeforeNextRound) { if (canSleepBeforeNextRound) { try { - Thread.sleep(sleepingTimeInMilliSecond); + synchronized (wakeUpMonitor) { + // A parser can become eligible before this worker starts waiting. Keep that notification + // until the idle round consumes it, instead of sleeping with the fair queue head ready. + if (!isWakeUpRequested) { + wakeUpMonitor.wait(sleepingTimeInMilliSecond); + } + isWakeUpRequested = false; + } } catch (final InterruptedException e) { LOGGER.warn("subtask worker is interrupted", e); Thread.currentThread().interrupt(); @@ -128,6 +139,11 @@ private void sleepIfNecessary(final boolean canSleepBeforeNextRound) { } } + @VisibleForTesting + void setSleepingTimeInMilliSecond(final long sleepingTimeInMilliSecond) { + this.sleepingTimeInMilliSecond = sleepingTimeInMilliSecond; + } + private void adjustSleepingTimeIfNecessary() { if (totalRoundInAdjustmentInterval % SLEEP_INTERVAL_ADJUSTMENT_ROUND_INTERVAL == 0) { final double workingRatioInAdjustmentInterval = @@ -147,7 +163,16 @@ private void adjustSleepingTimeIfNecessary() { } public void schedule(final PipeProcessorSubtask pipeProcessorSubtask) { + pipeProcessorSubtask.setWorkerWakeUpCallback(wakeUpCallback); subtasks.add(pipeProcessorSubtask); + wakeUp(); + } + + private void wakeUp() { + synchronized (wakeUpMonitor) { + isWakeUpRequested = true; + wakeUpMonitor.notifyAll(); + } } void watchLongRunningEvent() { 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 6258dfee1fd8e..0d379db7831b5 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 @@ -923,6 +923,10 @@ private boolean reserveResource4Parsing( final PipeProcessorSubtaskExecutionGuard processorExecutionGuard, final long expectedParserStateGeneration) throws InterruptedException { + // Register before enqueueing: memory may be released as soon as admission fails, before this + // invocation yields or its worker enters the idle wait. + tsFileParserMemoryReservationKey.setWakeUpCallback( + processorExecutionGuard.getWorkerWakeUpCallback()); if (!processorExecutionGuard.isEnabled()) { return waitForResourceEnough4Parsing( (long) ((1 + Math.random()) * 20 * 1000), expectedParserStateGeneration); 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 dd41d032f80e6..85ce5047b85ff 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 @@ -29,9 +29,12 @@ import org.apache.iotdb.metrics.impl.DoNothingMetricManager; import org.apache.iotdb.metrics.metricsets.IMetricSet; import org.apache.iotdb.metrics.type.Counter; +import org.apache.iotdb.metrics.type.Timer; import org.apache.iotdb.metrics.utils.MetricLevel; import org.apache.iotdb.metrics.utils.MetricType; +import java.util.concurrent.TimeUnit; + public class PipeResourceMetrics implements IMetricSet { private static final String PIPE_USED_MEMORY = "PipeUsedMemory"; @@ -44,6 +47,11 @@ public class PipeResourceMetrics implements IMetricSet { private static final String PIPE_TOTAL_MEMORY = "PipeTotalMemory"; private static final String PIPE_FLOATING_MEMORY = "PipeFloatingMemory"; private Counter diskIOCounter = DoNothingMetricManager.DO_NOTHING_COUNTER; + private volatile Counter tsFileParserAdmissionRejectCounter = + DoNothingMetricManager.DO_NOTHING_COUNTER; + private volatile Counter tsFileParserAdmissionYieldCounter = + DoNothingMetricManager.DO_NOTHING_COUNTER; + private volatile Timer tsFileParserAdmissionWaitTimer = DoNothingMetricManager.DO_NOTHING_TIMER; //////////////////////////// bindTo & unbindFrom (metric framework) //////////////////////////// @@ -102,6 +110,25 @@ public void bindTo(final AbstractMetricService metricService) { diskIOCounter = metricService.getOrCreateCounter( Metric.PIPE_TSFILE_SEND_DISK_IO.toString(), MetricLevel.IMPORTANT); + tsFileParserAdmissionRejectCounter = + metricService.getOrCreateCounter( + Metric.PIPE_TSFILE_PARSER_ADMISSION_REJECT.toString(), MetricLevel.IMPORTANT); + tsFileParserAdmissionYieldCounter = + metricService.getOrCreateCounter( + Metric.PIPE_TSFILE_PARSER_ADMISSION_YIELD.toString(), MetricLevel.IMPORTANT); + tsFileParserAdmissionWaitTimer = + metricService.getOrCreateTimer( + Metric.PIPE_TSFILE_PARSER_ADMISSION_WAIT.toString(), MetricLevel.IMPORTANT); + metricService.createAutoGauge( + Metric.PIPE_TSFILE_PARSER_IN_FLIGHT.toString(), + MetricLevel.IMPORTANT, + PipeDataNodeResourceManager.memory(), + PipeMemoryManager::getReservedTsFileParserCount); + metricService.createAutoGauge( + Metric.PIPE_TSFILE_PARSER_WAITING.toString(), + MetricLevel.IMPORTANT, + PipeDataNodeResourceManager.memory(), + PipeMemoryManager::getWaitingTsFileParserCount); } @Override @@ -138,12 +165,32 @@ public void unbindFrom(final AbstractMetricService metricService) { metricService.remove(MetricType.AUTO_GAUGE, Metric.PIPE_PHANTOM_REFERENCE_COUNT.toString()); metricService.remove(MetricType.COUNTER, Metric.PIPE_TSFILE_SEND_DISK_IO.toString()); + metricService.remove(MetricType.COUNTER, Metric.PIPE_TSFILE_PARSER_ADMISSION_REJECT.toString()); + 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()); + tsFileParserAdmissionRejectCounter = DoNothingMetricManager.DO_NOTHING_COUNTER; + tsFileParserAdmissionYieldCounter = DoNothingMetricManager.DO_NOTHING_COUNTER; + tsFileParserAdmissionWaitTimer = DoNothingMetricManager.DO_NOTHING_TIMER; } public void recordDiskIO(final long bytes) { diskIOCounter.inc(bytes); } + public void recordTsFileParserAdmissionReject() { + tsFileParserAdmissionRejectCounter.inc(); + } + + public void recordTsFileParserAdmissionYield() { + tsFileParserAdmissionYieldCounter.inc(); + } + + public void recordTsFileParserAdmissionWait(final long waitTimeInNanos) { + tsFileParserAdmissionWaitTimer.update(waitTimeInNanos, TimeUnit.NANOSECONDS); + } + //////////////////////////// singleton //////////////////////////// private static class PipeResourceMetricsHolder { 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 5f03991d3501d..8fc3d49536af1 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 @@ -23,6 +23,7 @@ import org.apache.iotdb.commons.pipe.config.PipeConfig; 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.resource.memory.strategy.ThresholdAllocationStrategy; import org.slf4j.Logger; @@ -66,6 +67,7 @@ public class PipeMemoryManager { private volatile long usedMemorySizeInBytesOfTsFiles; private volatile long reservedTsFileParserCount; + private volatile long waitingTsFileParserCount; private final Map reservedTsFileParserCountByPipe = new HashMap<>(); private final Map reservedTsFileParserCountByPipeRegion = @@ -109,30 +111,25 @@ public PipeMemoryManager() { // 3. The sum of the memory proportion occupied by the tablet memory block and the tsfile memory // block does not exceed TABLET_MEMORY_REJECT_THRESHOLD + TS_FILE_MEMORY_REJECT_THRESHOLD - private double allowedMaxMemorySizeInBytesOfTabletsAndTsFiles() { - return (PipeConfig.getInstance() - .getPipeDataStructureTabletMemoryBlockAllocationRejectThreshold() - + PipeConfig.getInstance() - .getPipeDataStructureTsFileMemoryBlockAllocationRejectThreshold()) - * getTotalNonFloatingMemorySizeInBytes(); + private double allowedMaxMemorySizeInBytesOfTabletsAndTsFiles( + final long totalNonFloatingMemorySizeInBytes) { + return (PIPE_CONFIG.getPipeDataStructureTabletMemoryBlockAllocationRejectThreshold() + + PIPE_CONFIG.getPipeDataStructureTsFileMemoryBlockAllocationRejectThreshold()) + * totalNonFloatingMemorySizeInBytes; } - private double allowedMaxMemorySizeInBytesOfTablets() { - return (PipeConfig.getInstance() - .getPipeDataStructureTabletMemoryBlockAllocationRejectThreshold() - + PipeConfig.getInstance() - .getPipeDataStructureTsFileMemoryBlockAllocationRejectThreshold() - / 2) - * getTotalNonFloatingMemorySizeInBytes(); + private double allowedMaxMemorySizeInBytesOfTablets( + final long totalNonFloatingMemorySizeInBytes) { + return (PIPE_CONFIG.getPipeDataStructureTabletMemoryBlockAllocationRejectThreshold() + + PIPE_CONFIG.getPipeDataStructureTsFileMemoryBlockAllocationRejectThreshold() / 2) + * totalNonFloatingMemorySizeInBytes; } - private double allowedMaxMemorySizeInBytesOfTsTiles() { - return (PipeConfig.getInstance() - .getPipeDataStructureTsFileMemoryBlockAllocationRejectThreshold() - + PipeConfig.getInstance() - .getPipeDataStructureTabletMemoryBlockAllocationRejectThreshold() - / 2) - * getTotalNonFloatingMemorySizeInBytes(); + private double allowedMaxMemorySizeInBytesOfTsTiles( + final long totalNonFloatingMemorySizeInBytes) { + return (PIPE_CONFIG.getPipeDataStructureTsFileMemoryBlockAllocationRejectThreshold() + + PIPE_CONFIG.getPipeDataStructureTabletMemoryBlockAllocationRejectThreshold() / 2) + * totalNonFloatingMemorySizeInBytes; } private static long getTsFileParserMemorySizeInBytes() { @@ -144,39 +141,48 @@ private long getReservedTsFileParserMemorySizeInBytes() { return reservedTsFileParserCount * getTsFileParserMemorySizeInBytes(); } - private boolean isEnough4TabletParsingWithReservedParserMemory(final long extraMemoryInBytes) { + private boolean isEnough4TabletParsingWithReservedParserMemory( + final long extraMemoryInBytes, final long totalNonFloatingMemorySizeInBytes) { final double tabletMemoryWithParserMemory = (double) usedMemorySizeInBytesOfTablets + getReservedTsFileParserMemorySizeInBytes() + extraMemoryInBytes; return tabletMemoryWithParserMemory + (double) usedMemorySizeInBytesOfTsFiles - < EXCEED_PROTECT_THRESHOLD * allowedMaxMemorySizeInBytesOfTabletsAndTsFiles() + < EXCEED_PROTECT_THRESHOLD + * allowedMaxMemorySizeInBytesOfTabletsAndTsFiles(totalNonFloatingMemorySizeInBytes) && tabletMemoryWithParserMemory - < EXCEED_PROTECT_THRESHOLD * allowedMaxMemorySizeInBytesOfTablets(); + < EXCEED_PROTECT_THRESHOLD + * allowedMaxMemorySizeInBytesOfTablets(totalNonFloatingMemorySizeInBytes); } private boolean isHardEnough4TabletParsingWithReservedParserMemory( - final long extraMemoryInBytes) { + final long extraMemoryInBytes, final long totalNonFloatingMemorySizeInBytes) { final double tabletMemoryWithParserMemory = (double) usedMemorySizeInBytesOfTablets + getReservedTsFileParserMemorySizeInBytes() + extraMemoryInBytes; return tabletMemoryWithParserMemory + (double) usedMemorySizeInBytesOfTsFiles - < allowedMaxMemorySizeInBytesOfTabletsAndTsFiles() - && tabletMemoryWithParserMemory < allowedMaxMemorySizeInBytesOfTablets(); + < allowedMaxMemorySizeInBytesOfTabletsAndTsFiles(totalNonFloatingMemorySizeInBytes) + && tabletMemoryWithParserMemory + < allowedMaxMemorySizeInBytesOfTablets(totalNonFloatingMemorySizeInBytes); } public boolean isEnough4TabletParsing() { + final long totalNonFloatingMemorySizeInBytes = getTotalNonFloatingMemorySizeInBytes(); return (double) usedMemorySizeInBytesOfTablets + (double) usedMemorySizeInBytesOfTsFiles - < EXCEED_PROTECT_THRESHOLD * allowedMaxMemorySizeInBytesOfTabletsAndTsFiles() + < EXCEED_PROTECT_THRESHOLD + * allowedMaxMemorySizeInBytesOfTabletsAndTsFiles(totalNonFloatingMemorySizeInBytes) && (double) usedMemorySizeInBytesOfTablets - < EXCEED_PROTECT_THRESHOLD * allowedMaxMemorySizeInBytesOfTablets(); + < EXCEED_PROTECT_THRESHOLD + * allowedMaxMemorySizeInBytesOfTablets(totalNonFloatingMemorySizeInBytes); } private boolean isHardEnough4TabletParsing() { + final long totalNonFloatingMemorySizeInBytes = getTotalNonFloatingMemorySizeInBytes(); return (double) usedMemorySizeInBytesOfTablets + (double) usedMemorySizeInBytesOfTsFiles - < allowedMaxMemorySizeInBytesOfTabletsAndTsFiles() - && (double) usedMemorySizeInBytesOfTablets < allowedMaxMemorySizeInBytesOfTablets(); + < allowedMaxMemorySizeInBytesOfTabletsAndTsFiles(totalNonFloatingMemorySizeInBytes) + && (double) usedMemorySizeInBytesOfTablets + < allowedMaxMemorySizeInBytesOfTablets(totalNonFloatingMemorySizeInBytes); } public synchronized boolean tryReserveTsFileParserMemory( @@ -202,16 +208,24 @@ public synchronized boolean tryReserveTsFileParserMemory( reservedTsFileParserCountByPipeRegion.getOrDefault(pipeRegionIdentity, 0); if (reservedTsFileParserCount >= globalLimit || reservedCountOfPipeRegion >= perPipeRegionLimit) { + PipeResourceMetrics.getInstance().recordTsFileParserAdmissionReject(); notifyNextTsFileParserMemoryReservationInternal(); return false; } final long parserMemorySizeInBytes = getTsFileParserMemorySizeInBytes(); + // Reuse one live snapshot for both thresholds. The supplier traverses all pipe metadata, so + // repeatedly sampling it while holding this monitor amplifies contention with many pipes. + final long totalNonFloatingMemorySizeInBytes = + PIPE_MEMORY_MANAGEMENT_ENABLED ? getTotalNonFloatingMemorySizeInBytes() : 0; final boolean isSoftMemoryEnough = !PIPE_MEMORY_MANAGEMENT_ENABLED - || isEnough4TabletParsingWithReservedParserMemory(parserMemorySizeInBytes); + || isEnough4TabletParsingWithReservedParserMemory( + parserMemorySizeInBytes, totalNonFloatingMemorySizeInBytes); if (!isSoftMemoryEnough - && !isHardEnough4TabletParsingWithReservedParserMemory(parserMemorySizeInBytes)) { + && !isHardEnough4TabletParsingWithReservedParserMemory( + parserMemorySizeInBytes, totalNonFloatingMemorySizeInBytes)) { + PipeResourceMetrics.getInstance().recordTsFileParserAdmissionReject(); return false; } @@ -222,13 +236,17 @@ public synchronized boolean tryReserveTsFileParserMemory( if (!pipeRegionIdentity.equals(nextPipeRegion) || requestsOfPipeRegion == null || !reservationKey.equals(requestsOfPipeRegion.iterator().next())) { - notifyNextTsFileParserMemoryReservationInternal(); + PipeResourceMetrics.getInstance().recordTsFileParserAdmissionReject(); + signalTsFileParserMemoryReservation(nextPipeRegion); return false; } removeTsFileParserReservationRequest(pipeRegionIdentity, reservationKey, true); if (wasRequestAlreadyWaiting) { lastAdmittedWaitingTsFileParserPipe = pipeIdentity; + PipeResourceMetrics.getInstance() + .recordTsFileParserAdmissionWait( + System.nanoTime() - reservationKey.waitingStartTimeInNanos); } reservedTsFileParserCount++; reservedTsFileParserCountByPipe.merge(pipeIdentity, 1, Integer::sum); @@ -301,7 +319,12 @@ private boolean enqueueTsFileParserReservationRequest( regionOrder.addLast(key); return new LinkedHashSet<>(); }); - return !requestsOfPipeRegion.add(reservationKey); + if (requestsOfPipeRegion.add(reservationKey)) { + waitingTsFileParserCount++; + reservationKey.waitingStartTimeInNanos = System.nanoTime(); + return false; + } + return true; } public synchronized void notifyNextTsFileParserMemoryReservation() { @@ -310,16 +333,20 @@ public synchronized void notifyNextTsFileParserMemoryReservation() { private void notifyNextTsFileParserMemoryReservationInternal() { final int globalLimit = Math.max(1, PIPE_CONFIG.getPipeTsFileParserInFlightMaxNum()); - if (reservedTsFileParserCount >= globalLimit) { + if (waitingTsFileParserPipeOrder.isEmpty() || reservedTsFileParserCount >= globalLimit) { return; } final long parserMemorySizeInBytes = getTsFileParserMemorySizeInBytes(); + final long totalNonFloatingMemorySizeInBytes = + PIPE_MEMORY_MANAGEMENT_ENABLED ? getTotalNonFloatingMemorySizeInBytes() : 0; final boolean isSoftMemoryEnough = !PIPE_MEMORY_MANAGEMENT_ENABLED - || isEnough4TabletParsingWithReservedParserMemory(parserMemorySizeInBytes); + || isEnough4TabletParsingWithReservedParserMemory( + parserMemorySizeInBytes, totalNonFloatingMemorySizeInBytes); if (!isSoftMemoryEnough - && !isHardEnough4TabletParsingWithReservedParserMemory(parserMemorySizeInBytes)) { + && !isHardEnough4TabletParsingWithReservedParserMemory( + parserMemorySizeInBytes, totalNonFloatingMemorySizeInBytes)) { return; } @@ -328,6 +355,10 @@ private void notifyNextTsFileParserMemoryReservationInternal() { 1, Math.min(globalLimit, PIPE_CONFIG.getPipeTsFileParserInFlightMaxNumPerPipeRegion())); final PipeRegionIdentity nextPipeRegion = getNextEligibleTsFileParserPipeRegion(perPipeRegionLimit, !isSoftMemoryEnough); + signalTsFileParserMemoryReservation(nextPipeRegion); + } + + private void signalTsFileParserMemoryReservation(final PipeRegionIdentity nextPipeRegion) { if (nextPipeRegion == null) { return; } @@ -398,6 +429,7 @@ private void removeTsFileParserReservationRequest( if (requestsOfPipeRegion == null || !requestsOfPipeRegion.remove(reservationKey)) { return; } + waitingTsFileParserCount--; final PipeIdentity pipeIdentity = pipeRegionIdentity.pipeIdentity; final ArrayDeque regionOrder = @@ -431,6 +463,14 @@ private void removeTsFileParserReservationRequest( public static final class TsFileParserMemoryReservation { private boolean isSignaled; + private volatile Runnable wakeUpCallback; + // Accessed only under the PipeMemoryManager monitor while this request is queued. + private long waitingStartTimeInNanos; + + /** The callback must only wake the worker; it runs while the memory manager monitor is held. */ + public void setWakeUpCallback(final Runnable wakeUpCallback) { + this.wakeUpCallback = wakeUpCallback; + } public synchronized void await(final long timeoutInMs) throws InterruptedException { if (!isSignaled) { @@ -442,6 +482,10 @@ public synchronized void await(final long timeoutInMs) throws InterruptedExcepti private synchronized void signal() { isSignaled = true; notify(); + final Runnable callback = wakeUpCallback; + if (callback != null) { + callback.run(); + } } } @@ -464,7 +508,8 @@ public boolean shouldReleaseTsFileParserOnOutOfMemory( return elapsedTimeInMs >= maxRetryTimeInMs; } - if (!isHardEnough4TabletParsingWithReservedParserMemory(0)) { + if (!isHardEnough4TabletParsingWithReservedParserMemory( + 0, getTotalNonFloatingMemorySizeInBytes())) { return true; } @@ -473,35 +518,44 @@ public boolean shouldReleaseTsFileParserOnOutOfMemory( } public boolean isEnough4TsFileSlicing() { + final long totalNonFloatingMemorySizeInBytes = getTotalNonFloatingMemorySizeInBytes(); return (double) usedMemorySizeInBytesOfTablets + (double) usedMemorySizeInBytesOfTsFiles - < EXCEED_PROTECT_THRESHOLD * allowedMaxMemorySizeInBytesOfTabletsAndTsFiles() + < EXCEED_PROTECT_THRESHOLD + * allowedMaxMemorySizeInBytesOfTabletsAndTsFiles(totalNonFloatingMemorySizeInBytes) && (double) usedMemorySizeInBytesOfTsFiles - < EXCEED_PROTECT_THRESHOLD * allowedMaxMemorySizeInBytesOfTsTiles(); + < EXCEED_PROTECT_THRESHOLD + * allowedMaxMemorySizeInBytesOfTsTiles(totalNonFloatingMemorySizeInBytes); } private boolean isHardEnough4TsFileSlicing() { + final long totalNonFloatingMemorySizeInBytes = getTotalNonFloatingMemorySizeInBytes(); return (double) usedMemorySizeInBytesOfTablets + (double) usedMemorySizeInBytesOfTsFiles - < allowedMaxMemorySizeInBytesOfTabletsAndTsFiles() - && (double) usedMemorySizeInBytesOfTsFiles < allowedMaxMemorySizeInBytesOfTsTiles(); + < allowedMaxMemorySizeInBytesOfTabletsAndTsFiles(totalNonFloatingMemorySizeInBytes) + && (double) usedMemorySizeInBytesOfTsFiles + < allowedMaxMemorySizeInBytesOfTsTiles(totalNonFloatingMemorySizeInBytes); } private boolean isHardEnoughForResizing( final PipeMemoryBlock block, final long extraMemoryInBytes) { + if (!(block instanceof PipeTabletMemoryBlock) && !(block instanceof PipeTsFileMemoryBlock)) { + return true; + } + final long totalNonFloatingMemorySizeInBytes = getTotalNonFloatingMemorySizeInBytes(); if (block instanceof PipeTabletMemoryBlock) { return (double) usedMemorySizeInBytesOfTablets + (double) extraMemoryInBytes + (double) usedMemorySizeInBytesOfTsFiles - < allowedMaxMemorySizeInBytesOfTabletsAndTsFiles() + < allowedMaxMemorySizeInBytesOfTabletsAndTsFiles(totalNonFloatingMemorySizeInBytes) && (double) usedMemorySizeInBytesOfTablets + (double) extraMemoryInBytes - < allowedMaxMemorySizeInBytesOfTablets(); + < allowedMaxMemorySizeInBytesOfTablets(totalNonFloatingMemorySizeInBytes); } if (block instanceof PipeTsFileMemoryBlock) { return (double) usedMemorySizeInBytesOfTablets + (double) usedMemorySizeInBytesOfTsFiles + (double) extraMemoryInBytes - < allowedMaxMemorySizeInBytesOfTabletsAndTsFiles() + < allowedMaxMemorySizeInBytesOfTabletsAndTsFiles(totalNonFloatingMemorySizeInBytes) && (double) usedMemorySizeInBytesOfTsFiles + (double) extraMemoryInBytes - < allowedMaxMemorySizeInBytesOfTsTiles(); + < allowedMaxMemorySizeInBytesOfTsTiles(totalNonFloatingMemorySizeInBytes); } return true; } @@ -1077,6 +1131,14 @@ public long getUsedMemorySizeInBytesOfTsFiles() { return usedMemorySizeInBytesOfTsFiles; } + public long getReservedTsFileParserCount() { + return reservedTsFileParserCount; + } + + public long getWaitingTsFileParserCount() { + return waitingTsFileParserCount; + } + public long getFreeMemorySizeInBytes() { return Math.max(0, getTotalNonFloatingMemorySizeInBytes() - usedMemorySizeInBytes); } 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 b7c5fdc554203..a9e49892e839e 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 @@ -90,7 +90,10 @@ public void testParserAdmissionYieldsWithoutBlockingAndResumes() throws Exceptio final PipeTsFileInsertionEvent eventToConsume = event; final PipeProcessorSubtaskExecutionGuard executionGuard = new PipeProcessorSubtaskExecutionGuard(); + final AtomicInteger wakeUpCount = new AtomicInteger(); + executionGuard.setWorkerWakeUpCallback(wakeUpCount::incrementAndGet); executionGuard.start(); + wakeUpCount.set(0); executionGuard.enter(); final long startTimeInNanos = System.nanoTime(); @@ -108,10 +111,12 @@ public void testParserAdmissionYieldsWithoutBlockingAndResumes() throws Exceptio Assert.assertTrue(TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startTimeInNanos) < 1000); executionGuard.exit(); - executionGuard.stop(); - event.cancelTsFileParserMemoryReservationIfPending(); + // 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()); + executionGuard.stop(); + event.cancelTsFileParserMemoryReservationIfPending(); isCompetitorReserved = memoryManager.tryReserveTsFileParserMemory("competitor", 0, "0", competitorReservation); Assert.assertTrue(isCompetitorReserved); @@ -120,6 +125,7 @@ public void testParserAdmissionYieldsWithoutBlockingAndResumes() throws Exceptio final AtomicInteger consumedTabletCount = new AtomicInteger(0); executionGuard.start(); + Assert.assertEquals(2, wakeUpCount.get()); executionGuard.enter(); event.consumeTabletInsertionEventsWithRetry( parsedEvent -> { 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 d1c34ccf47160..d1d24c1183739 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 @@ -19,7 +19,12 @@ package org.apache.iotdb.db.pipe.agent.task.subtask.processor; +import org.apache.iotdb.commons.conf.CommonConfig; +import org.apache.iotdb.commons.conf.CommonDescriptor; import org.apache.iotdb.commons.pipe.event.EnrichedEvent; +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 com.google.common.util.concurrent.ListeningExecutorService; import com.google.common.util.concurrent.ListeningScheduledExecutorService; @@ -28,12 +33,22 @@ import org.junit.Test; import org.mockito.InOrder; +import java.util.ArrayList; import java.util.LinkedHashSet; +import java.util.List; +import java.util.concurrent.CountDownLatch; +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.AtomicBoolean; +import java.util.concurrent.atomic.AtomicReference; +import static org.awaitility.Awaitility.await; import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.eq; import static org.mockito.Mockito.atLeastOnce; +import static org.mockito.Mockito.doAnswer; import static org.mockito.Mockito.inOrder; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.never; @@ -42,6 +57,145 @@ public class PipeProcessorSubtaskWorkerTest { + @Test(timeout = 30000) + public void testOneHundredWaitingPipesDoNotWaitForWorkerPolling() 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 List workers = new ArrayList<>(); + final List> workerThreads = new ArrayList<>(); + final List requests = new ArrayList<>(); + final List> futures = new ArrayList<>(); + final CountDownLatch processed = new CountDownLatch(100); + final ExecutorService executor = Executors.newFixedThreadPool(5); + final boolean[] reservedBlockers = new boolean[4]; + try { + config.setPipeTsFileParserMemory(1); + config.setPipeTsFileParserInFlightMaxNum(4); + config.setPipeTsFileParserInFlightMaxNumPerPipeRegion(1); + for (int i = 0; i < 4; i++) { + Assert.assertTrue( + memoryManager.tryReserveTsFileParserMemory( + "blocker" + i, 0, "0", new TsFileParserMemoryReservation())); + reservedBlockers[i] = true; + } + for (int i = 0; i < 5; i++) { + final PipeProcessorSubtaskWorker worker = + new PipeProcessorSubtaskWorker(new LinkedHashSet<>()); + workers.add(worker); + workerThreads.add(new AtomicReference<>()); + } + for (int i = 0; i < 100; i++) { + final String pipeName = "waitingPipe" + i; + final TsFileParserMemoryReservation request = new TsFileParserMemoryReservation(); + requests.add(request); + final PipeProcessorSubtask subtask = createRunnableSubtask(pipeName); + doAnswer( + invocation -> { + request.setWakeUpCallback(invocation.getArgument(0)); + return null; + }) + .when(subtask) + .setWorkerWakeUpCallback(any()); + final AtomicBoolean hasProcessed = new AtomicBoolean(); + when(subtask.call()) + .thenAnswer( + invocation -> { + if (hasProcessed.get()) { + return false; + } + if (!memoryManager.tryReserveTsFileParserMemory(pipeName, 0, "0", request)) { + throw PipeProcessorSubtaskYieldException.parserNotAdmitted(); + } + hasProcessed.set(true); + memoryManager.releaseTsFileParserMemory(pipeName, 0, "0"); + processed.countDown(); + return true; + }); + workers.get(i % workers.size()).schedule(subtask); + Assert.assertFalse(memoryManager.tryReserveTsFileParserMemory(pipeName, 0, "0", request)); + } + Assert.assertEquals(100, memoryManager.getWaitingTsFileParserCount()); + for (int i = 0; i < workers.size(); i++) { + final PipeProcessorSubtaskWorker worker = workers.get(i); + final AtomicReference workerThread = workerThreads.get(i); + // Consume the scheduling notification before all workers become idle on admission. + worker.sleepIfNecessary(true); + // Make a missed wake-up observable without relying on a tight performance assertion. + worker.setSleepingTimeInMilliSecond(60000L); + futures.add( + executor.submit( + () -> { + workerThread.set(Thread.currentThread()); + while (!Thread.currentThread().isInterrupted()) { + worker.sleepIfNecessary(worker.runSubtasks()); + } + })); + } + await() + .atMost(5, TimeUnit.SECONDS) + .until( + () -> + workerThreads.stream() + .allMatch( + thread -> + thread.get() != null + && thread.get().getState() == Thread.State.TIMED_WAITING)); + for (int i = 0; i < 4; i++) { + memoryManager.releaseTsFileParserMemory("blocker" + i, 0, "0"); + reservedBlockers[i] = false; + } + Assert.assertTrue( + "Admission notifications must wake every queue head", + processed.await(5, TimeUnit.SECONDS)); + Assert.assertEquals(0, memoryManager.getWaitingTsFileParserCount()); + Assert.assertEquals(0, memoryManager.getReservedTsFileParserCount()); + } finally { + futures.forEach(future -> future.cancel(true)); + executor.shutdownNow(); + Assert.assertTrue(executor.awaitTermination(5, TimeUnit.SECONDS)); + for (int i = 0; i < requests.size(); i++) { + memoryManager.cancelTsFileParserMemoryReservation( + "waitingPipe" + i, 0, "0", requests.get(i)); + } + for (int i = 0; i < reservedBlockers.length; i++) { + if (reservedBlockers[i]) { + memoryManager.releaseTsFileParserMemory("blocker" + i, 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<>()); + worker.setSleepingTimeInMilliSecond(60000L); + final PipeProcessorSubtask subtask = createRunnableSubtask("waitingPipe"); + final AtomicReference wakeUpCallback = new AtomicReference<>(); + doAnswer( + invocation -> { + wakeUpCallback.set(invocation.getArgument(0)); + return null; + }) + .when(subtask) + .setWorkerWakeUpCallback(any()); + worker.schedule(subtask); + worker.sleepIfNecessary(true); + final ExecutorService executor = Executors.newSingleThreadExecutor(); + try { + wakeUpCallback.get().run(); + executor.submit(() -> worker.sleepIfNecessary(true)).get(3, TimeUnit.SECONDS); + } finally { + executor.shutdownNow(); + Assert.assertTrue(executor.awaitTermination(3, TimeUnit.SECONDS)); + } + } + @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/resource/memory/PipeMemoryManagerTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryManagerTest.java index 3a50002667e37..6b681c5dd49e1 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 @@ -38,6 +38,7 @@ import java.util.concurrent.Executors; import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; public class PipeMemoryManagerTest { @@ -74,6 +75,35 @@ public void tearDown() { commonConfig.setPipeTsFileParserMemory(originalParserMemoryInBytes); } + @Test + public void testAdmissionNotificationOnlyWakesEligibleWaitingPipe() { + 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); + final AtomicInteger pipeAWakeUps = new AtomicInteger(); + final AtomicInteger pipeBWakeUps = new AtomicInteger(); + pipeA.key.setWakeUpCallback(pipeAWakeUps::incrementAndGet); + pipeB.key.setWakeUpCallback(pipeBWakeUps::incrementAndGet); + Assert.assertTrue(tryAcquire(blocker)); + Assert.assertFalse(tryAcquire(pipeA)); + Assert.assertFalse(tryAcquire(pipeB)); + Assert.assertEquals(2, memoryManager.getWaitingTsFileParserCount()); + Assert.assertEquals(0, pipeAWakeUps.get()); + Assert.assertEquals(0, pipeBWakeUps.get()); + + release(blocker); + Assert.assertEquals(1, pipeAWakeUps.get()); + Assert.assertEquals(0, pipeBWakeUps.get()); + Assert.assertTrue(tryAcquire(pipeA)); + Assert.assertEquals(1, memoryManager.getWaitingTsFileParserCount()); + release(pipeA); + Assert.assertEquals(1, pipeBWakeUps.get()); + Assert.assertTrue(tryAcquire(pipeB)); + Assert.assertEquals(0, memoryManager.getWaitingTsFileParserCount()); + } + @Test public void testWaitingPipesAreAdmittedInRoundRobinOrder() { commonConfig.setPipeTsFileParserInFlightMaxNum(1); 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 8a01eb3a1bf7d..2a53b1f940b09 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 @@ -165,6 +165,11 @@ public enum Metric { PIPE_HEARTBEAT_EVENT("pipe_heartbeat_event"), 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_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"), + PIPE_TSFILE_PARSER_WAITING("pipe_tsfile_parser_waiting"), PIPE_LINKED_TSFILE_COUNT("pipe_linked_tsfile_count"), PIPE_LINKED_TSFILE_SIZE("pipe_linked_tsfile_size"), PIPE_PHANTOM_REFERENCE_COUNT("pipe_phantom_reference_count"),