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 e6268fbe01e7..836ac88d84f2 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 @@ -21,6 +21,46 @@ public final class DataNodePipeMessages { + public static final String + LOG_SUBSCRIPTION_PREFETCH_ROUND_STARTING_FOR_ARG_ACTIVE_ARG_INITIALIZED_ARG_CLOSED_ARG_PENDINGENTRIES_ARG_PREFETCHEDEVENTS_ARG_INFLIGHTEVENTS_ARG_NEXTREADSEARCHINDEX_ARG_F75FF2CD = + "Subscription prefetch round starting for {}: active={}, initialized={}, closed={}, pendingEntries={}, prefetchedEvents={}, inFlightEvents={}, nextReadSearchIndex={}."; + + public static final String + LOG_SUBSCRIPTION_PREFETCH_ROUND_COMPLETED_FOR_ARG_DURATIONMS_ARG_RESULT_ARG_DELAYMS_ARG_READSEARCHINDEX_ARG_ARG_PENDINGACCEPTED_ARG_WALACCEPTED_ARG_PENDINGENTRIES_ARG_PREFETCHEDEVENTS_ARG_INFLIGHTEVENTS_ARG_MEMORYBLOCKREASON_ARG_ADMISSIONBLOCKREASON_ARG_0F0A3D57 = + "Subscription prefetch round completed for {}: durationMs={}, result={}, delayMs={}, readSearchIndex={} -> {}, pendingAccepted={}, walAccepted={}, pendingEntries={}, prefetchedEvents={}, inFlightEvents={}, memoryBlockReason={}, admissionBlockReason={}."; + + public static final String + LOG_SUBSCRIPTION_PREFETCH_ACQUIRED_QUEUE_READ_LOCK_FOR_ARG_AFTER_ARG_MS_425E9A20 = + "Subscription prefetch acquired queue read lock for {} after {} ms."; + + public static final String + LOG_SUBSCRIPTION_PREFETCH_SCANNING_WAL_FOR_ARG_NEXTREADSEARCHINDEX_ARG_MAXWALENTRIES_ARG_PREFETCHEDEVENTS_ARG_AA16EEA1 = + "Subscription prefetch scanning WAL for {}: nextReadSearchIndex={}, maxWalEntries={}, prefetchedEvents={}."; + + public static final String + LOG_SUBSCRIPTION_PREFETCH_PREPARING_ENTRY_FOR_ARG_SEARCHINDEX_ARG_WRITERNODEID_ARG_LOCALSEQ_ARG_EE485CC7 = + "Subscription prefetch preparing entry for {}: searchIndex={}, writerNodeId={}, localSeq={}."; + + public static final String + LOG_SUBSCRIPTION_PREFETCH_PREPARED_ENTRY_FOR_ARG_SEARCHINDEX_ARG_WRITERNODEID_ARG_LOCALSEQ_ARG_TABLETS_ARG_ESTIMATEDBYTES_ARG_DURATIONMS_ARG_E91BA0CD = + "Subscription prefetch prepared entry for {}: searchIndex={}, writerNodeId={}, localSeq={}, tablets={}, estimatedBytes={}, durationMs={}."; + + public static final String + LOG_SUBSCRIPTION_PREFETCH_MEMORY_RESERVATION_REJECTED_FOR_ARG_ENTRYBYTES_ARG_REASON_ARG_QUOTABYTES_ARG_FREEBYTES_ARG_98903D3B = + "Subscription prefetch memory reservation rejected for {}: entryBytes={}, reason={}, quotaBytes={}, freeBytes={}."; + + public static final String + LOG_SUBSCRIPTION_PREFETCH_TASK_ARG_SCHEDULING_AFTER_ROUND_RESULT_ARG_WAKEUPPENDING_ARG_ENQUEUENOW_ARG_DELAYMS_ARG_602EA4A4 = + "Subscription prefetch task {} scheduling after round: result={}, wakeupPending={}, enqueueNow={}, delayMs={}."; + + public static final String + LOG_SUBSCRIPTION_MEMORY_OVERCOMMIT_ENTRYBYTES_ARG_BUDGETBYTES_ARG_OVERCOMMITBYTES_ARG_OTHER_QUEUES_MAY_BE_BLOCKED_UNTIL_RELEASE_DF9B914E = + "Subscription memory overcommit: entryBytes={}, budgetBytes={}, overcommitBytes={}; other queues may be blocked until release."; + + public static final String + LOG_SUBSCRIPTION_PREFETCH_ROUND_TOOK_ARG_MS_FOR_ARG_DELIVERYIDLETIMEMS_ARG_MEMORYUSEDBYTES_ARG_MEMORYLIMITBYTES_ARG_INSPECT_WORKER_STACKS_WAL_I_O_AND_JVM_PAUSES_D399CEDD = + "Subscription prefetch round took {} ms for {}. deliveryIdleTimeMs={}, memoryUsedBytes={}, memoryLimitBytes={}; inspect worker stacks, WAL I/O and JVM pauses."; + public static final String LOG_FAILED_TO_RESOLVE_TRANSFER_EXCEPTION_A4F5397A = "Failed to resolve transfer exception."; @@ -2124,7 +2164,7 @@ private DataNodePipeMessages() {} "ConsensusPrefetchingQueue {}: periodic stats, lag={}, pendingDelta={}, walDelta={}, " + "pendingTotal={}, walTotal={}, walGapSkippedEntries={}, pendingQueueSize={}, " + "prefetchingQueueSize={}, inFlightEventsSize={}, realtimeWriterCount={}, " - + "walHasNext={}, isActive={}, subtaskScheduled={}"; + + "walNextBuffered={}, isActive={}, subtaskScheduled={}"; public static final String PIPE_LOG_CONSENSUSPREFETCHINGQUEUE_WAITING_MS_FOR_WAL_GAP_TO_BECOME_7D91C6C5 = "ConsensusPrefetchingQueue {}: waiting {}ms for WAL gap [{}, {}) to become visible, " + "currentNextExpected={}, currentWalIndex={}, seekGeneration={}"; 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 64cd9c7a2d67..5840da12dafa 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 @@ -21,6 +21,46 @@ public final class DataNodePipeMessages { + public static final String + LOG_SUBSCRIPTION_PREFETCH_ROUND_STARTING_FOR_ARG_ACTIVE_ARG_INITIALIZED_ARG_CLOSED_ARG_PENDINGENTRIES_ARG_PREFETCHEDEVENTS_ARG_INFLIGHTEVENTS_ARG_NEXTREADSEARCHINDEX_ARG_F75FF2CD = + "订阅预取轮次开始，队列 {}：active={}，initialized={}，closed={}，pendingEntries={}，prefetchedEvents={}，inFlightEvents={}，nextReadSearchIndex={}。"; + + public static final String + LOG_SUBSCRIPTION_PREFETCH_ROUND_COMPLETED_FOR_ARG_DURATIONMS_ARG_RESULT_ARG_DELAYMS_ARG_READSEARCHINDEX_ARG_ARG_PENDINGACCEPTED_ARG_WALACCEPTED_ARG_PENDINGENTRIES_ARG_PREFETCHEDEVENTS_ARG_INFLIGHTEVENTS_ARG_MEMORYBLOCKREASON_ARG_ADMISSIONBLOCKREASON_ARG_0F0A3D57 = + "订阅预取轮次结束，队列 {}：durationMs={}，result={}，delayMs={}，readSearchIndex={} -> {}，pendingAccepted={}，walAccepted={}，pendingEntries={}，prefetchedEvents={}，inFlightEvents={}，memoryBlockReason={}，admissionBlockReason={}。"; + + public static final String + LOG_SUBSCRIPTION_PREFETCH_ACQUIRED_QUEUE_READ_LOCK_FOR_ARG_AFTER_ARG_MS_425E9A20 = + "订阅预取已获取队列 {} 的读锁，距离轮次开始已过 {} ms。"; + + public static final String + LOG_SUBSCRIPTION_PREFETCH_SCANNING_WAL_FOR_ARG_NEXTREADSEARCHINDEX_ARG_MAXWALENTRIES_ARG_PREFETCHEDEVENTS_ARG_AA16EEA1 = + "订阅预取开始扫描队列 {} 的 WAL：nextReadSearchIndex={}，maxWalEntries={}，prefetchedEvents={}。"; + + public static final String + LOG_SUBSCRIPTION_PREFETCH_PREPARING_ENTRY_FOR_ARG_SEARCHINDEX_ARG_WRITERNODEID_ARG_LOCALSEQ_ARG_EE485CC7 = + "订阅预取开始解析和转换队列 {} 的 entry：searchIndex={}，writerNodeId={}，localSeq={}。"; + + public static final String + LOG_SUBSCRIPTION_PREFETCH_PREPARED_ENTRY_FOR_ARG_SEARCHINDEX_ARG_WRITERNODEID_ARG_LOCALSEQ_ARG_TABLETS_ARG_ESTIMATEDBYTES_ARG_DURATIONMS_ARG_E91BA0CD = + "订阅预取完成队列 {} 的 entry 解析和转换：searchIndex={}，writerNodeId={}，localSeq={}，tablets={}，estimatedBytes={}，durationMs={}。"; + + public static final String + LOG_SUBSCRIPTION_PREFETCH_MEMORY_RESERVATION_REJECTED_FOR_ARG_ENTRYBYTES_ARG_REASON_ARG_QUOTABYTES_ARG_FREEBYTES_ARG_98903D3B = + "订阅预取队列 {} 的内存预留被拒绝：entryBytes={}，reason={}，quotaBytes={}，freeBytes={}。"; + + public static final String + LOG_SUBSCRIPTION_PREFETCH_TASK_ARG_SCHEDULING_AFTER_ROUND_RESULT_ARG_WAKEUPPENDING_ARG_ENQUEUENOW_ARG_DELAYMS_ARG_602EA4A4 = + "订阅预取任务 {} 的轮次后调度：result={}，wakeupPending={}，enqueueNow={}，delayMs={}。"; + + public static final String + LOG_SUBSCRIPTION_MEMORY_OVERCOMMIT_ENTRYBYTES_ARG_BUDGETBYTES_ARG_OVERCOMMITBYTES_ARG_OTHER_QUEUES_MAY_BE_BLOCKED_UNTIL_RELEASE_DF9B914E = + "订阅内存超出预算：entryBytes={}，budgetBytes={}，overcommitBytes={}；其它队列可能阻塞，直到该 entry 释放。"; + + public static final String + LOG_SUBSCRIPTION_PREFETCH_ROUND_TOOK_ARG_MS_FOR_ARG_DELIVERYIDLETIMEMS_ARG_MEMORYUSEDBYTES_ARG_MEMORYLIMITBYTES_ARG_INSPECT_WORKER_STACKS_WAL_I_O_AND_JVM_PAUSES_D399CEDD = + "订阅预取轮次耗时 {} ms，队列 {}。deliveryIdleTimeMs={}，memoryUsedBytes={}，memoryLimitBytes={}；请检查工作线程栈、WAL I/O 和 JVM 停顿。"; + public static final String LOG_FAILED_TO_RESOLVE_TRANSFER_EXCEPTION_A4F5397A = "解析 transfer exception 失败。"; @@ -1970,7 +2010,7 @@ private DataNodePipeMessages() {} public static final String PIPE_LOG_CONSENSUSPREFETCHINGQUEUE_PERIODIC_STATS_LAG_PENDINGDELTA_WALGAPSKIPPEDENTRIES_9A4E6608 = "ConsensusPrefetchingQueue {}：周期统计，lag={}，pendingDelta={}，walDelta={}，pendingTotal={}，" + "walTotal={}，walGapSkippedEntries={}，pendingQueueSize={}，prefetchingQueueSize={}，" - + "inFlightEventsSize={}，realtimeWriterCount={}，walHasNext={}，isActive={}，" + + "inFlightEventsSize={}，realtimeWriterCount={}，walNextBuffered={}，isActive={}，" + "subtaskScheduled={}"; public static final String PIPE_LOG_CONSENSUSPREFETCHINGQUEUE_WAITING_MS_FOR_WAL_GAP_TO_BECOME_7D91C6C5 = "ConsensusPrefetchingQueue {}：等待 {}ms，使 WAL 缺口 [{}, {}) 可见，currentNextExpected={}，" diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusLogToTabletConverter.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusLogToTabletConverter.java index abd4c5cc381a..882860cebbc7 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusLogToTabletConverter.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusLogToTabletConverter.java @@ -134,7 +134,9 @@ static InsertNode deserializeToInsertNode(final IndexedConsensusRequest indexedR try { if (req instanceof IoTConsensusRequest) { // WAL entries read from file are wrapped as IoTConsensusRequest (ByteBuffer). - planNode = WALEntry.deserializeForConsensus(req.serializeToByteBuffer()); + // Preserve the request's buffer position so memory backpressure can retry every + // fragment. duplicate() shares the bytes without retaining a second payload copy. + planNode = WALEntry.deserializeForConsensus(req.serializeToByteBuffer().duplicate()); } else if (req instanceof InsertNode) { // In-memory entries that are not yet flushed to WAL file may already be PlanNode. planNode = (PlanNode) req; diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueue.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueue.java index e24748edc0b1..dd5d20400097 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueue.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueue.java @@ -319,6 +319,8 @@ public class ConsensusPrefetchingQueue { private volatile long lastProgressTimeMs = 0L; + private final SubscriptionQueueTimeTracker timeTracker = new SubscriptionQueueTimeTracker(); + private volatile String lastConsumerId = ""; private volatile long lastPendingAcceptedEntries = 0L; @@ -954,6 +956,7 @@ private boolean initPrefetchUnderInitializationLock(final RegionProgress regionP this.subscriptionWALIterator = createSubscriptionWALIterator(resolvedStart.getStartSearchIndex()); this.prefetchInitialized = true; + timeTracker.start(); this.observedSeekGeneration = seekGeneration.get(); discardBatch(this.lingerBatch); resetBatchWriterProgress(); @@ -1443,6 +1446,7 @@ private SubscriptionEvent pollInternal(final String consumerId) { // Mark as polled before updating inFlightEvents event.recordLastPolledTimestamp(); + timeTracker.recordDelivery(); lastPollTimeMs = System.currentTimeMillis(); lastConsumerId = consumerId; inFlightEvents.put(new InFlightEventKey(consumerId, event.getCommitContext()), event); @@ -1528,7 +1532,69 @@ public SubscriptionEvent pollTablets( private static final long PREFETCH_STATS_LOG_INTERVAL_MS = 5_000L; + private static final long SLOW_PREFETCH_ROUND_THRESHOLD_MS = 10_000L; + + private String getPrefetchDebugId() { + return getPrefetchingQueueId() + "/" + consensusGroupId; + } + public PrefetchRoundResult drivePrefetchOnce() { + timeTracker.beginPrefetch(); + final boolean debugEnabled = LOGGER.isDebugEnabled(); + final long startReadSearchIndex = debugEnabled ? nextExpectedSearchIndex.get() : 0L; + final long startPendingAccepted = debugEnabled ? pendingPathAcceptedEntries.get() : 0L; + final long startWalAccepted = debugEnabled ? walPathAcceptedEntries.get() : 0L; + PrefetchRoundResult result = null; + try { + if (debugEnabled) { + LOGGER.debug( + DataNodePipeMessages + .LOG_SUBSCRIPTION_PREFETCH_ROUND_STARTING_FOR_ARG_ACTIVE_ARG_INITIALIZED_ARG_CLOSED_ARG_PENDINGENTRIES_ARG_PREFETCHEDEVENTS_ARG_INFLIGHTEVENTS_ARG_NEXTREADSEARCHINDEX_ARG_F75FF2CD, + getPrefetchDebugId(), + isActive, + prefetchInitialized, + closeRequested || isClosed, + pendingEntries.size(), + prefetchingQueue.size(), + inFlightEvents.size(), + startReadSearchIndex); + } + result = drivePrefetchOnceInternal(); + return result; + } finally { + final long durationMs = timeTracker.endPrefetch(); + if (debugEnabled) { + LOGGER.debug( + DataNodePipeMessages + .LOG_SUBSCRIPTION_PREFETCH_ROUND_COMPLETED_FOR_ARG_DURATIONMS_ARG_RESULT_ARG_DELAYMS_ARG_READSEARCHINDEX_ARG_ARG_PENDINGACCEPTED_ARG_WALACCEPTED_ARG_PENDINGENTRIES_ARG_PREFETCHEDEVENTS_ARG_INFLIGHTEVENTS_ARG_MEMORYBLOCKREASON_ARG_ADMISSIONBLOCKREASON_ARG_0F0A3D57, + getPrefetchDebugId(), + durationMs, + Objects.nonNull(result) ? result.getType() : null, + Objects.nonNull(result) ? result.getDelayMs() : null, + startReadSearchIndex, + nextExpectedSearchIndex.get(), + pendingPathAcceptedEntries.get() - startPendingAccepted, + walPathAcceptedEntries.get() - startWalAccepted, + pendingEntries.size(), + prefetchingQueue.size(), + inFlightEvents.size(), + memoryBlockReason, + realtimeAdmissionBlockReason); + } + if (durationMs >= SLOW_PREFETCH_ROUND_THRESHOLD_MS) { + LOGGER.warn( + DataNodePipeMessages + .LOG_SUBSCRIPTION_PREFETCH_ROUND_TOOK_ARG_MS_FOR_ARG_DELIVERYIDLETIMEMS_ARG_MEMORYUSEDBYTES_ARG_MEMORYLIMITBYTES_ARG_INSPECT_WORKER_STACKS_WAL_I_O_AND_JVM_PAUSES_D399CEDD, + durationMs, + this, + getDeliveryIdleTimeMs(), + subscriptionMemoryManager.getUsedMemorySizeInBytes(), + subscriptionMemoryManager.getTotalMemorySizeInBytes()); + } + } + } + + private PrefetchRoundResult drivePrefetchOnceInternal() { if (applyPendingSeekRequestIfNecessary()) { return closeRequested ? PrefetchRoundResult.dormant() : PrefetchRoundResult.rescheduleNow(); } @@ -1539,6 +1605,13 @@ public PrefetchRoundResult drivePrefetchOnce() { return PrefetchRoundResult.dormant(); } + if (LOGGER.isDebugEnabled()) { + LOGGER.debug( + DataNodePipeMessages + .LOG_SUBSCRIPTION_PREFETCH_ACQUIRED_QUEUE_READ_LOCK_FOR_ARG_AFTER_ARG_MS_425E9A20, + getPrefetchDebugId(), + getPrefetchDurationMs()); + } logPeriodicStatsIfNecessary(); final long currentSeekGeneration = seekGeneration.get(); @@ -1714,7 +1787,7 @@ private void logPeriodicStatsIfNecessary() { prefetchingQueue.size(), inFlightEvents.size(), realtimeEntriesByWriter.size(), - hasReadableWalEntries(), + Objects.nonNull(subscriptionWALIterator) && subscriptionWALIterator.hasBufferedNext(), isActive, Objects.nonNull(prefetchSubtask) && prefetchSubtask.isScheduledOrRunning()); lastStatsLogTimeMs = nowMs; @@ -1819,24 +1892,9 @@ private boolean isBeforeLocalCursor(final IndexedConsensusRequest request) { return hasLocalSearchIndex(request) && request.getSearchIndex() < nextExpectedSearchIndex.get(); } - private boolean advanceLocalCursorIfPresent(final IndexedConsensusRequest request) { + private void advanceLocalCursorIfPresent(final IndexedConsensusRequest request) { if (hasLocalSearchIndex(request)) { nextExpectedSearchIndex.set(request.getSearchIndex() + 1); - return true; - } - return false; - } - - private void advanceLocalCursorFromPendingIfPresent( - final IndexedConsensusRequest request, final long expectedSeekGeneration) { - if (advanceLocalCursorIfPresent(request)) { - // Pending delivery advances independently of the WAL reader. Raise its local lower bound in - // place so stale local requests are filtered without rebuilding and rescanning retained WAL. - final ProgressWALIterator iterator = subscriptionWALIterator; - if (Objects.nonNull(iterator) && seekGeneration.get() == expectedSeekGeneration) { - iterator.advanceTo( - nextExpectedSearchIndex.get(), this::isWriterProgressCoveredForWalFastForward); - } } } @@ -1895,76 +1953,90 @@ private MaterializationResult accumulateFromPending( int processedCount = 0; int skippedCount = 0; + final long startSearchIndex = nextExpectedSearchIndex.get(); - for (int index = 0; index < batch.size(); index++) { - final IndexedConsensusRequest request = batch.get(index); - final long searchIndex = request.getSearchIndex(); + try { + for (int index = 0; index < batch.size(); index++) { + final IndexedConsensusRequest request = batch.get(index); + final long searchIndex = request.getSearchIndex(); - // Only local-indexed requests participate in the internal WAL read cursor. - final long expected = nextExpectedSearchIndex.get(); - if (hasLocalSearchIndex(request) && searchIndex > expected) { - LOGGER.debug( - DataNodePipeMessages - .PIPE_LOG_CONSENSUSPREFETCHINGQUEUE_GAP_DETECTED_EXPECTED_GOT_FILLING_70DD08B3, - this, - expected, - searchIndex, - searchIndex - expected); - final MaterializationResult gapFillResult = - fillGapFromWAL( - expected, - searchIndex, - lingerBatch, - expectedSeekGeneration, - maxWalEntries, - maxTablets, - maxBatchBytes); - if (gapFillResult != MaterializationResult.SUCCESS) { - return gapFillResult; + // Only local-indexed requests participate in the internal WAL read cursor. + final long expected = nextExpectedSearchIndex.get(); + if (hasLocalSearchIndex(request) && searchIndex > expected) { + LOGGER.debug( + DataNodePipeMessages + .PIPE_LOG_CONSENSUSPREFETCHINGQUEUE_GAP_DETECTED_EXPECTED_GOT_FILLING_70DD08B3, + this, + expected, + searchIndex, + searchIndex - expected); + final MaterializationResult gapFillResult = + fillGapFromWAL( + expected, + searchIndex, + lingerBatch, + expectedSeekGeneration, + maxWalEntries, + maxTablets, + maxBatchBytes); + if (gapFillResult != MaterializationResult.SUCCESS) { + return gapFillResult; + } } - } - if (isBeforeLocalCursor(request)) { - skippedCount++; - continue; - } + if (isBeforeLocalCursor(request)) { + skippedCount++; + continue; + } - if (shouldSkipForRecoveryProgress(request)) { - skippedCount++; - advanceLocalCursorFromPendingIfPresent(request, expectedSeekGeneration); - continue; - } - if (shouldSkipForMaterializedProgress(request)) { - skippedCount++; - advanceLocalCursorFromPendingIfPresent(request, expectedSeekGeneration); - continue; - } + if (shouldSkipForRecoveryProgress(request)) { + skippedCount++; + advanceLocalCursorIfPresent(request); + continue; + } + if (shouldSkipForMaterializedProgress(request)) { + skippedCount++; + advanceLocalCursorIfPresent(request); + continue; + } - final MaterializationResult appendResult = - appendRealtimeRequest( - request, lingerBatch, expectedSeekGeneration, maxTablets, maxBatchBytes, true); - if (appendResult != MaterializationResult.SUCCESS) { - return appendResult; - } - markMaterializedProgress(request); - processedCount++; - advanceLocalCursorFromPendingIfPresent(request, expectedSeekGeneration); - if (prefetchingQueue.size() >= MAX_PREFETCHING_QUEUE_SIZE) { - break; + final MaterializationResult appendResult = + appendRealtimeRequest( + request, lingerBatch, expectedSeekGeneration, maxTablets, maxBatchBytes, true); + if (appendResult != MaterializationResult.SUCCESS) { + return appendResult; + } + markMaterializedProgress(request); + processedCount++; + advanceLocalCursorIfPresent(request); + if (prefetchingQueue.size() >= MAX_PREFETCHING_QUEUE_SIZE) { + break; + } } - } - LOGGER.debug( - DataNodePipeMessages - .PIPE_LOG_CONSENSUSPREFETCHINGQUEUE_ACCUMULATE_COMPLETE_BATCHSIZE_FA3F3B41, - this, - batch.size(), - processedCount, - skippedCount, - lingerBatch.tablets.size(), - nextExpectedSearchIndex.get()); + LOGGER.debug( + DataNodePipeMessages + .PIPE_LOG_CONSENSUSPREFETCHINGQUEUE_ACCUMULATE_COMPLETE_BATCHSIZE_FA3F3B41, + this, + batch.size(), + processedCount, + skippedCount, + lingerBatch.tablets.size(), + nextExpectedSearchIndex.get()); - return MaterializationResult.SUCCESS; + return MaterializationResult.SUCCESS; + } finally { + // Advance once for the consumed prefix, including when a gap or memory block stops the + // batch. Per-entry fast-forward can repeatedly inspect the same uncovered WAL file footer. + // Gap recovery creates its own iterator at the current cursor before reading WAL. + final ProgressWALIterator iterator = subscriptionWALIterator; + if (nextExpectedSearchIndex.get() > startSearchIndex + && Objects.nonNull(iterator) + && seekGeneration.get() == expectedSeekGeneration) { + iterator.advanceTo( + nextExpectedSearchIndex.get(), this::isWriterProgressCoveredForWalFastForward); + } + } } /** @@ -2080,6 +2152,15 @@ private MaterializationResult pumpFromSubscriptionWAL( return MaterializationResult.SUCCESS; } + if (LOGGER.isDebugEnabled()) { + LOGGER.debug( + DataNodePipeMessages + .LOG_SUBSCRIPTION_PREFETCH_SCANNING_WAL_FOR_ARG_NEXTREADSEARCHINDEX_ARG_MAXWALENTRIES_ARG_PREFETCHEDEVENTS_ARG_AA16EEA1, + getPrefetchDebugId(), + nextExpectedSearchIndex.get(), + maxWalEntries, + prefetchingQueue.size()); + } ensureSubscriptionWalReadable(); reportUnreadableWalReplayIfNecessary(); @@ -2109,7 +2190,10 @@ private MaterializationResult pumpFromSubscriptionWAL( walEntry, batchState, expectedSeekGeneration, maxTablets, maxBatchBytes, false); if (appendResult != MaterializationResult.SUCCESS) { if (appendResult == MaterializationResult.MEMORY_BLOCKED) { - resetSubscriptionWALPosition(nextExpectedSearchIndex.get()); + // next() has removed this request from the iterator, but replay progress has not + // advanced. Keep it buffered so the next round retries it without reopening the WAL + // and rescanning the prefix before the current reader position. + subscriptionWALIterator.retry(walEntry); } return appendResult; } @@ -2304,6 +2388,17 @@ private void closeSubscriptionWALIterator() { } private PreparedEntry prepareEntry(final IndexedConsensusRequest indexedRequest) { + final boolean debugEnabled = LOGGER.isDebugEnabled(); + final long prepareStartNs = debugEnabled ? System.nanoTime() : 0L; + if (debugEnabled) { + LOGGER.debug( + DataNodePipeMessages + .LOG_SUBSCRIPTION_PREFETCH_PREPARING_ENTRY_FOR_ARG_SEARCHINDEX_ARG_WRITERNODEID_ARG_LOCALSEQ_ARG_EE485CC7, + getPrefetchDebugId(), + indexedRequest.getSearchIndex(), + indexedRequest.getNodeId(), + indexedRequest.getProgressLocalSeq()); + } final InsertNode insertNode = ConsensusLogToTabletConverter.deserializeToInsertNode(indexedRequest); if (Objects.isNull(insertNode)) { @@ -2330,6 +2425,18 @@ private PreparedEntry prepareEntry(final IndexedConsensusRequest indexedRequest) } final List tablets = converter.convert(insertNode); final long estimatedBytes = estimateTabletsBytes(tablets); + if (debugEnabled) { + LOGGER.debug( + DataNodePipeMessages + .LOG_SUBSCRIPTION_PREFETCH_PREPARED_ENTRY_FOR_ARG_SEARCHINDEX_ARG_WRITERNODEID_ARG_LOCALSEQ_ARG_TABLETS_ARG_ESTIMATEDBYTES_ARG_DURATIONMS_ARG_E91BA0CD, + getPrefetchDebugId(), + searchIndex, + writerNodeId, + localSeq, + tablets.size(), + estimatedBytes, + TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - prepareStartNs)); + } if (!tryReserveTabletMemory(estimatedBytes)) { return null; } @@ -2472,6 +2579,16 @@ private boolean tryReserveTabletMemory(final long bytes) { final SubscriptionMemoryManager.AllocationResult allocationResult = subscriptionMemoryHandle.tryAllocate(bytes); if (!allocationResult.isAccepted()) { + if (LOGGER.isDebugEnabled()) { + LOGGER.debug( + DataNodePipeMessages + .LOG_SUBSCRIPTION_PREFETCH_MEMORY_RESERVATION_REJECTED_FOR_ARG_ENTRYBYTES_ARG_REASON_ARG_QUOTABYTES_ARG_FREEBYTES_ARG_98903D3B, + getPrefetchDebugId(), + bytes, + allocationResult.getRejectionReason(), + allocationResult.getQuotaInBytes(), + allocationResult.getFreeMemoryInBytes()); + } memoryBlockedEntryBytes = bytes; final SubscriptionQueueRejectionReason rejectionReason = toQueueRejectionReason(allocationResult.getRejectionReason()); @@ -4378,6 +4495,30 @@ public long getLastProgressTimeMs() { return lastProgressTimeMs; } + public long getLastDeliveryIntervalMs() { + return timeTracker.getLastDeliveryIntervalMs(); + } + + public long getMaxDeliveryIntervalMs() { + return timeTracker.getMaxDeliveryIntervalMs(); + } + + public long getDeliveryIdleTimeMs() { + return timeTracker.getDeliveryIdleTimeMs(); + } + + public long getPrefetchDurationMs() { + return timeTracker.getPrefetchDurationMs(); + } + + public long getMaxPrefetchDurationMs() { + return timeTracker.getMaxPrefetchDurationMs(); + } + + public long getPrefetchIdleTimeMs() { + return timeTracker.getPrefetchIdleTimeMs(); + } + /** Returns 0=uninitialized, 1=inactive, 2=caught up, 3=catching up, 4=stalled. */ public long getProgressStatus() { switch (getProgressStatusName()) { @@ -4743,6 +4884,21 @@ public Map coreReportMessage() { result.put("retainedRequestBytes", String.valueOf(getRetainedRequestBytes())); result.put("retainedTabletBytes", String.valueOf(retainedTabletBytes.get())); result.put("memoryBlockedEntryBytes", String.valueOf(memoryBlockedEntryBytes)); + result.put( + "dataNodeSubscriptionMemoryUsedInBytes", + String.valueOf(subscriptionMemoryManager.getUsedMemorySizeInBytes())); + result.put( + "subscriptionMemoryOvercommitBytes", + String.valueOf(subscriptionMemoryManager.getOvercommitSizeInBytes())); + result.put( + "subscriptionMemoryOversizedEntryCount", + String.valueOf(subscriptionMemoryManager.getOversizedEntryCount())); + result.put("lastDeliveryIntervalMs", String.valueOf(getLastDeliveryIntervalMs())); + result.put("maxDeliveryIntervalMs", String.valueOf(getMaxDeliveryIntervalMs())); + result.put("deliveryIdleTimeMs", String.valueOf(getDeliveryIdleTimeMs())); + result.put("prefetchDurationMs", String.valueOf(getPrefetchDurationMs())); + result.put("maxPrefetchDurationMs", String.valueOf(getMaxPrefetchDurationMs())); + result.put("prefetchIdleTimeMs", String.valueOf(getPrefetchIdleTimeMs())); result.put("realtimeAdmissionBlocked", String.valueOf(realtimeAdmissionBlocked.get())); result.put( "subscriptionMemoryLimitInBytes", diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ProgressWALIterator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ProgressWALIterator.java index cbf0649f2243..3636829274f7 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ProgressWALIterator.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ProgressWALIterator.java @@ -49,6 +49,7 @@ import java.util.Iterator; import java.util.List; import java.util.NoSuchElementException; +import java.util.Objects; import java.util.Set; /** @@ -205,6 +206,23 @@ public boolean hasNext() { } } + /** + * Returns whether a usable request is already buffered, without reading or advancing WAL. A false + * result does not imply that the iterator is exhausted. + */ + boolean hasBufferedNext() { + return nextReady != null && !shouldSkip(nextReady); + } + + /** + * Keeps a request returned by {@link #next()} available for a retry without reopening the WAL. + * The caller uses this when materialization is temporarily blocked and replay progress has not + * advanced yet. + */ + void retry(final IndexedConsensusRequest request) { + nextReady = Objects.requireNonNull(request); + } + /** * Advances the local search-index lower bound without rebuilding the iterator. Whole WAL files * are skipped only when every writer progress tuple in them is already covered by queue state. diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/SubscriptionQueueTimeTracker.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/SubscriptionQueueTimeTracker.java new file mode 100644 index 000000000000..0a8187e4e2ec --- /dev/null +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/SubscriptionQueueTimeTracker.java @@ -0,0 +1,119 @@ +/* + * 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.subscription.broker.consensus; + +import org.apache.iotdb.commons.utils.TestOnly; + +import java.util.concurrent.TimeUnit; +import java.util.function.LongSupplier; + +/** Monotonic timing that remains readable while a queue's prefetch worker is blocked. */ +class SubscriptionQueueTimeTracker { + + private final LongSupplier nanoTime; + + private volatile boolean started; + private boolean delivered; + private volatile long lastDeliveryTimeNs; + private volatile long lastDeliveryIntervalMs; + private volatile long maxDeliveryIntervalMs; + private volatile long lastPrefetchCompletionTimeNs; + private volatile long prefetchStartTimeNs; + private volatile boolean prefetchRunning; + private volatile long maxPrefetchDurationMs; + + SubscriptionQueueTimeTracker() { + this(System::nanoTime); + } + + @TestOnly + SubscriptionQueueTimeTracker(final LongSupplier nanoTime) { + this.nanoTime = nanoTime; + } + + synchronized void start() { + if (!started) { + final long nowNs = nanoTime.getAsLong(); + lastDeliveryTimeNs = nowNs; + lastPrefetchCompletionTimeNs = nowNs; + started = true; + } + } + + // Multiple consumers may poll the queue concurrently under its read lock, so delivery interval + // updates must remain atomic even though prefetch rounds are serialized. + synchronized void recordDelivery() { + start(); + final long nowNs = nanoTime.getAsLong(); + if (delivered) { + lastDeliveryIntervalMs = elapsedMs(nowNs, lastDeliveryTimeNs); + maxDeliveryIntervalMs = Math.max(maxDeliveryIntervalMs, lastDeliveryIntervalMs); + } + lastDeliveryTimeNs = nowNs; + delivered = true; + } + + // The prefetch subtask serializes rounds, so this path does not need synchronization beyond the + // one-time initialization performed by start(). + void beginPrefetch() { + start(); + prefetchStartTimeNs = nanoTime.getAsLong(); + prefetchRunning = true; + } + + long endPrefetch() { + final long nowNs = nanoTime.getAsLong(); + final long durationMs = elapsedMs(nowNs, prefetchStartTimeNs); + maxPrefetchDurationMs = Math.max(maxPrefetchDurationMs, durationMs); + lastPrefetchCompletionTimeNs = nowNs; + prefetchRunning = false; + return durationMs; + } + + long getLastDeliveryIntervalMs() { + return lastDeliveryIntervalMs; + } + + long getMaxDeliveryIntervalMs() { + return maxDeliveryIntervalMs; + } + + long getDeliveryIdleTimeMs() { + return started ? elapsedMs(nanoTime.getAsLong(), lastDeliveryTimeNs) : 0L; + } + + /** Elapsed time of an ongoing round, including waiting to acquire the queue lock. */ + long getPrefetchDurationMs() { + return prefetchRunning ? elapsedMs(nanoTime.getAsLong(), prefetchStartTimeNs) : 0L; + } + + long getMaxPrefetchDurationMs() { + return Math.max(maxPrefetchDurationMs, getPrefetchDurationMs()); + } + + /** Time since the last completed round, or activation before the first completed round. */ + long getPrefetchIdleTimeMs() { + return started ? elapsedMs(nanoTime.getAsLong(), lastPrefetchCompletionTimeNs) : 0L; + } + + private static long elapsedMs(final long nowNs, final long thenNs) { + return Math.max(0L, TimeUnit.NANOSECONDS.toMillis(nowNs - thenNs)); + } +} diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/metric/ConsensusSubscriptionPrefetchingQueueMetrics.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/metric/ConsensusSubscriptionPrefetchingQueueMetrics.java index 359da8f92397..f0c3cfed01db 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/metric/ConsensusSubscriptionPrefetchingQueueMetrics.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/metric/ConsensusSubscriptionPrefetchingQueueMetrics.java @@ -206,6 +206,42 @@ private void createAutoGauge(final QueueMetricsKey key) { queue, ConsensusPrefetchingQueue::getLastProgressTimeMs, key.getTags()); + metricService.createAutoGauge( + Metric.SUBSCRIPTION_CONSENSUS_LAST_DELIVERY_INTERVAL_MS.toString(), + MetricLevel.IMPORTANT, + queue, + ConsensusPrefetchingQueue::getLastDeliveryIntervalMs, + key.getTags()); + metricService.createAutoGauge( + Metric.SUBSCRIPTION_CONSENSUS_MAX_DELIVERY_INTERVAL_MS.toString(), + MetricLevel.IMPORTANT, + queue, + ConsensusPrefetchingQueue::getMaxDeliveryIntervalMs, + key.getTags()); + metricService.createAutoGauge( + Metric.SUBSCRIPTION_CONSENSUS_DELIVERY_IDLE_TIME_MS.toString(), + MetricLevel.IMPORTANT, + queue, + ConsensusPrefetchingQueue::getDeliveryIdleTimeMs, + key.getTags()); + metricService.createAutoGauge( + Metric.SUBSCRIPTION_CONSENSUS_PREFETCH_DURATION_MS.toString(), + MetricLevel.IMPORTANT, + queue, + ConsensusPrefetchingQueue::getPrefetchDurationMs, + key.getTags()); + metricService.createAutoGauge( + Metric.SUBSCRIPTION_CONSENSUS_MAX_PREFETCH_DURATION_MS.toString(), + MetricLevel.IMPORTANT, + queue, + ConsensusPrefetchingQueue::getMaxPrefetchDurationMs, + key.getTags()); + metricService.createAutoGauge( + Metric.SUBSCRIPTION_CONSENSUS_PREFETCH_IDLE_TIME_MS.toString(), + MetricLevel.IMPORTANT, + queue, + ConsensusPrefetchingQueue::getPrefetchIdleTimeMs, + key.getTags()); metricService.createAutoGauge( Metric.SUBSCRIPTION_CONSENSUS_PROGRESS_STATUS.toString(), MetricLevel.IMPORTANT, @@ -373,6 +409,30 @@ private void removeAutoGauge(final QueueMetricsKey key) { MetricType.AUTO_GAUGE, Metric.SUBSCRIPTION_CONSENSUS_LAST_PROGRESS_TIME.toString(), key.getTags()); + metricService.remove( + MetricType.AUTO_GAUGE, + Metric.SUBSCRIPTION_CONSENSUS_LAST_DELIVERY_INTERVAL_MS.toString(), + key.getTags()); + metricService.remove( + MetricType.AUTO_GAUGE, + Metric.SUBSCRIPTION_CONSENSUS_MAX_DELIVERY_INTERVAL_MS.toString(), + key.getTags()); + metricService.remove( + MetricType.AUTO_GAUGE, + Metric.SUBSCRIPTION_CONSENSUS_DELIVERY_IDLE_TIME_MS.toString(), + key.getTags()); + metricService.remove( + MetricType.AUTO_GAUGE, + Metric.SUBSCRIPTION_CONSENSUS_PREFETCH_DURATION_MS.toString(), + key.getTags()); + metricService.remove( + MetricType.AUTO_GAUGE, + Metric.SUBSCRIPTION_CONSENSUS_MAX_PREFETCH_DURATION_MS.toString(), + key.getTags()); + metricService.remove( + MetricType.AUTO_GAUGE, + Metric.SUBSCRIPTION_CONSENSUS_PREFETCH_IDLE_TIME_MS.toString(), + key.getTags()); metricService.remove( MetricType.AUTO_GAUGE, Metric.SUBSCRIPTION_CONSENSUS_PROGRESS_STATUS.toString(), diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/metric/SubscriptionMetrics.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/metric/SubscriptionMetrics.java index 29de59ddf326..8e8657bb3d55 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/metric/SubscriptionMetrics.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/metric/SubscriptionMetrics.java @@ -19,8 +19,13 @@ package org.apache.iotdb.db.subscription.metric; +import org.apache.iotdb.commons.service.metric.enums.Metric; +import org.apache.iotdb.db.subscription.resource.SubscriptionDataNodeResourceManager; +import org.apache.iotdb.db.subscription.resource.SubscriptionMemoryManager; import org.apache.iotdb.metrics.AbstractMetricService; import org.apache.iotdb.metrics.metricsets.IMetricSet; +import org.apache.iotdb.metrics.utils.MetricLevel; +import org.apache.iotdb.metrics.utils.MetricType; public class SubscriptionMetrics implements IMetricSet { @@ -30,12 +35,41 @@ public class SubscriptionMetrics implements IMetricSet { public void bindTo(final AbstractMetricService metricService) { SubscriptionPrefetchingQueueMetrics.getInstance().bindTo(metricService); ConsensusSubscriptionPrefetchingQueueMetrics.getInstance().bindTo(metricService); + // Read the shared allocator directly; queue snapshots taken at different times cannot be + // summed to establish whether a DataNode exceeded its subscription memory budget. + final SubscriptionMemoryManager memoryManager = SubscriptionDataNodeResourceManager.memory(); + metricService.createAutoGauge( + Metric.SUBSCRIPTION_MEMORY_USED_BYTES.toString(), + MetricLevel.IMPORTANT, + memoryManager, + SubscriptionMemoryManager::getUsedMemorySizeInBytes); + metricService.createAutoGauge( + Metric.SUBSCRIPTION_MEMORY_LIMIT_BYTES.toString(), + MetricLevel.IMPORTANT, + memoryManager, + SubscriptionMemoryManager::getTotalMemorySizeInBytes); + metricService.createAutoGauge( + Metric.SUBSCRIPTION_MEMORY_OVERCOMMIT_BYTES.toString(), + MetricLevel.IMPORTANT, + memoryManager, + SubscriptionMemoryManager::getOvercommitSizeInBytes); + metricService.createAutoGauge( + Metric.SUBSCRIPTION_MEMORY_OVERSIZED_ENTRY_COUNT.toString(), + MetricLevel.IMPORTANT, + memoryManager, + SubscriptionMemoryManager::getOversizedEntryCount); } @Override public void unbindFrom(final AbstractMetricService metricService) { SubscriptionPrefetchingQueueMetrics.getInstance().unbindFrom(metricService); ConsensusSubscriptionPrefetchingQueueMetrics.getInstance().unbindFrom(metricService); + metricService.remove(MetricType.AUTO_GAUGE, Metric.SUBSCRIPTION_MEMORY_USED_BYTES.toString()); + metricService.remove(MetricType.AUTO_GAUGE, Metric.SUBSCRIPTION_MEMORY_LIMIT_BYTES.toString()); + metricService.remove( + MetricType.AUTO_GAUGE, Metric.SUBSCRIPTION_MEMORY_OVERCOMMIT_BYTES.toString()); + metricService.remove( + MetricType.AUTO_GAUGE, Metric.SUBSCRIPTION_MEMORY_OVERSIZED_ENTRY_COUNT.toString()); } //////////////////////////// singleton //////////////////////////// diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/resource/SubscriptionMemoryManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/resource/SubscriptionMemoryManager.java index d250cd0ba167..f13ef2b1329d 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/resource/SubscriptionMemoryManager.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/resource/SubscriptionMemoryManager.java @@ -24,9 +24,14 @@ import org.apache.iotdb.commons.memory.MemoryBlockType; import org.apache.iotdb.commons.utils.TestOnly; import org.apache.iotdb.db.conf.IoTDBDescriptor; +import org.apache.iotdb.db.i18n.DataNodePipeMessages; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; import java.util.HashMap; import java.util.Map; +import java.util.concurrent.TimeUnit; /** * DataNode-wide memory manager for materialized subscription data. @@ -39,8 +44,12 @@ */ public class SubscriptionMemoryManager { + private static final Logger LOGGER = LoggerFactory.getLogger(SubscriptionMemoryManager.class); + private static final String MEMORY_BLOCK_NAME = "Subscription"; + private static final long OVERCOMMIT_WARNING_INTERVAL_NS = TimeUnit.SECONDS.toNanos(30); + private final IMemoryBlock memoryBlock; private static final int PROTECTED_SHARE_DIVISOR = 2; private final Map handles = new HashMap<>(); @@ -49,6 +58,10 @@ public class SubscriptionMemoryManager { // Kept for callers of the original manager-level API. It is deliberately not a fair-share owner. private long legacyUsedMemoryInBytes; + private volatile long oversizedEntryCount = 0L; + + private long lastOvercommitWarningTimeNs; + SubscriptionMemoryManager() { memoryBlock = IoTDBDescriptor.getInstance() @@ -71,7 +84,17 @@ public synchronized MemoryHandle registerQueue() { return handle; } - /** Reserves memory through the compatibility owner. New queues should use a handle. */ + /** + * Reserves memory through the compatibility owner. Consensus queues use a handle, which rejects + * oversized entries instead of overcommitting the node budget. + * + *

A single entry larger than the whole budget is allowed only while the budget is otherwise + * empty. This avoids permanently blocking progress while keeping the overrun bounded by one + * entry. + * + *

This is a soft limit for such entries. Other queues can be blocked until the oversized entry + * is released; the exception is exposed through metrics and a rate-limited warning. + */ public synchronized boolean tryAllocate(final long sizeInBytes) { return tryAllocateLegacy(sizeInBytes).isAccepted(); } @@ -93,6 +116,18 @@ private AllocationResult tryAllocateLegacy(final long sizeInBytes) { if (total > 0L && used == 0L && sizeInBytes > total) { memoryBlock.forceAllocateWithoutLimitation(sizeInBytes); legacyUsedMemoryInBytes += sizeInBytes; + oversizedEntryCount++; + final long nowNs = System.nanoTime(); + if (oversizedEntryCount == 1L + || nowNs - lastOvercommitWarningTimeNs >= OVERCOMMIT_WARNING_INTERVAL_NS) { + lastOvercommitWarningTimeNs = nowNs; + LOGGER.warn( + DataNodePipeMessages + .LOG_SUBSCRIPTION_MEMORY_OVERCOMMIT_ENTRYBYTES_ARG_BUDGETBYTES_ARG_OVERCOMMITBYTES_ARG_OTHER_QUEUES_MAY_BE_BLOCKED_UNTIL_RELEASE_DF9B914E, + sizeInBytes, + memoryBlock.getTotalMemorySizeInBytes(), + getOvercommitSizeInBytes()); + } return AllocationResult.accepted(total, total - sizeInBytes); } return AllocationResult.rejected( @@ -126,6 +161,14 @@ public synchronized long getFreeMemorySizeInBytes() { return memoryBlock.getFreeMemoryInBytes(); } + public synchronized long getOvercommitSizeInBytes() { + return Math.max(0L, getUsedMemorySizeInBytes() - getTotalMemorySizeInBytes()); + } + + public long getOversizedEntryCount() { + return oversizedEntryCount; + } + private synchronized AllocationResult tryAllocate( final MemoryHandle handle, final long sizeInBytes) { final AllocationResult decision = inspectAllocationUnderLock(handle, sizeInBytes); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/task/subtask/ConsensusPrefetchSubtask.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/task/subtask/ConsensusPrefetchSubtask.java index 703289b736b8..5f91fecd3d12 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/task/subtask/ConsensusPrefetchSubtask.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/task/subtask/ConsensusPrefetchSubtask.java @@ -124,6 +124,7 @@ public void runOneRound() { boolean shouldEnqueue = false; Long delayedWakeMs = null; long delayedToken = 0L; + boolean wakeupRequested; synchronized (monitor) { running = false; if (closed) { @@ -132,6 +133,7 @@ public void runOneRound() { return; } + wakeupRequested = wakeupPending; if (wakeupPending) { wakeupPending = false; shouldEnqueue = true; @@ -158,6 +160,15 @@ public void runOneRound() { monitor.notifyAll(); } + LOGGER.debug( + DataNodePipeMessages + .LOG_SUBSCRIPTION_PREFETCH_TASK_ARG_SCHEDULING_AFTER_ROUND_RESULT_ARG_WAKEUPPENDING_ARG_ENQUEUENOW_ARG_DELAYMS_ARG_602EA4A4, + taskId, + result.getType(), + wakeupRequested, + shouldEnqueue, + delayedWakeMs); + final ConsensusSubscriptionPrefetchExecutor currentExecutor = executor; if (currentExecutor == null) { return; diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusLogToTabletConverterTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusLogToTabletConverterTest.java index b06ccf453fee..c4af27f764c9 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusLogToTabletConverterTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusLogToTabletConverterTest.java @@ -24,7 +24,11 @@ import org.apache.iotdb.commons.pipe.datastructure.pattern.IoTDBTreePattern; import org.apache.iotdb.commons.pipe.datastructure.pattern.TablePattern; import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNodeId; +import org.apache.iotdb.commons.request.IConsensusRequest; import org.apache.iotdb.commons.schema.table.column.TsTableColumnCategory; +import org.apache.iotdb.consensus.common.request.IndexedConsensusRequest; +import org.apache.iotdb.consensus.common.request.IoTConsensusRequest; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowsNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowsOfOneDeviceNode; @@ -32,6 +36,8 @@ import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalInsertRowNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalInsertTabletNode; import org.apache.iotdb.db.queryengine.plan.statement.StatementTestUtils; +import org.apache.iotdb.db.storageengine.dataregion.wal.buffer.WALInfoEntry; +import org.apache.iotdb.db.storageengine.dataregion.wal.utils.WALByteBufferForTest; import org.apache.iotdb.db.subscription.columnfilter.ColumnFilterMatcher; import org.apache.tsfile.enums.ColumnCategory; @@ -44,6 +50,7 @@ import org.junit.Assert; import org.junit.Test; +import java.nio.ByteBuffer; import java.nio.charset.StandardCharsets; import java.time.LocalDate; import java.util.ArrayList; @@ -55,6 +62,44 @@ public class ConsensusLogToTabletConverterTest { private static final String DATABASE_NAME = "db"; + @Test + public void testRepeatedWalDeserializationPreservesAllFragments() { + final List fragments = new ArrayList<>(); + for (int row = 1; row <= 2; row++) { + final RelationalInsertRowNode node = StatementTestUtils.genInsertRowNode(row); + node.setSearchIndex(4L); + node.setLastFragment(row == 2); + final WALInfoEntry walEntry = new WALInfoEntry(1L, node); + final ByteBuffer buffer = ByteBuffer.allocate(walEntry.serializedSize()); + walEntry.serialize(new WALByteBufferForTest(buffer)); + buffer.flip(); + fragments.add(new IoTConsensusRequest(buffer)); + } + final IndexedConsensusRequest request = + new IndexedConsensusRequest(4L, 4L, fragments).setPhysicalTime(1000L).setNodeId(7); + final ConsensusLogToTabletConverter converter = createConverter("id1", "m1"); + + for (int attempt = 0; attempt < 3; attempt++) { + final InsertNode node = ConsensusLogToTabletConverter.deserializeToInsertNode(request); + Assert.assertNotNull(node); + Assert.assertEquals(4L, node.getSearchIndex()); + Assert.assertEquals(4L, node.getSyncIndex()); + Assert.assertEquals(1000L, node.getPhysicalTime()); + Assert.assertEquals(7, node.getNodeId()); + final List tablets = converter.convert(node); + Assert.assertEquals(2, tablets.size()); + for (int row = 1; row <= 2; row++) { + final Tablet tablet = tablets.get(row - 1); + Assert.assertEquals(1, tablet.getRowSize()); + Assert.assertEquals("id:" + row, toUtf8(((Binary[]) tablet.getValues()[0])[0])); + Assert.assertEquals((double) row, ((double[]) tablet.getValues()[1])[0], 0.0); + final ByteBuffer buffer = fragments.get(row - 1).serializeToByteBuffer(); + Assert.assertEquals(0, buffer.position()); + Assert.assertEquals(buffer.capacity(), buffer.limit()); + } + } + } + @Test public void testDataModelIsDerivedFromImmutableConversionPattern() { final ConsensusLogToTabletConverter tableConverter = createConverter("id1"); diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueDataNodeMemoryTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueDataNodeMemoryTest.java index c7ebf06ab055..16f675a4da26 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueDataNodeMemoryTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueDataNodeMemoryTest.java @@ -36,6 +36,7 @@ import org.apache.iotdb.rpc.subscription.config.TopicConstant; import org.apache.iotdb.rpc.subscription.payload.poll.ErrorPayload; import org.apache.iotdb.rpc.subscription.payload.poll.RegionProgress; +import org.apache.iotdb.rpc.subscription.payload.poll.SubscriptionPollResponseType; import org.apache.tsfile.enums.ColumnCategory; import org.apache.tsfile.enums.TSDataType; @@ -48,12 +49,19 @@ import java.io.File; import java.lang.reflect.Constructor; import java.lang.reflect.Field; +import java.lang.reflect.Method; import java.util.Arrays; import java.util.Collections; import java.util.List; import java.util.Set; import java.util.concurrent.BlockingQueue; +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.AtomicInteger; +import java.util.concurrent.atomic.AtomicLong; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; @@ -226,6 +234,12 @@ public void testOversizedEntryReportsErrorWhileOtherQueueHoldsMemory() throws Ex assertEquals(1L, largeQueue.getOversizedEntryRejectionCount()); assertEquals(0L, largeQueue.getRealtimeAdmissionRejectionCount()); assertEquals(smallBytes, memoryManager.getUsedMemorySizeInBytes()); + assertEquals(0L, memoryManager.getOvercommitSizeInBytes()); + assertEquals(0L, memoryManager.getOversizedEntryCount()); + assertEquals("0", largeQueue.coreReportMessage().get("subscriptionMemoryUsedInBytes")); + assertEquals( + String.valueOf(smallBytes), + largeQueue.coreReportMessage().get("dataNodeSubscriptionMemoryUsedInBytes")); final SubscriptionEvent error = largeQueue.poll("largeConsumer"); assertNotNull(error); final ErrorPayload payload = (ErrorPayload) error.getCurrentResponse().getPayload(); @@ -306,9 +320,10 @@ public void testQueuesShareDataNodeMemoryBudget() throws Exception { assertEquals(oneTabletBytes, queueA.getRetainedTabletBytes()); assertEquals(0L, queueB.getRetainedTabletBytes()); assertEquals(oneTabletBytes, memoryManager.getUsedMemorySizeInBytes()); - assertTrue( - queueA.getRetainedTabletBytes() + queueB.getRetainedTabletBytes() - <= memoryManager.getTotalMemorySizeInBytes()); + assertEquals(0L, memoryManager.getOvercommitSizeInBytes()); + assertEquals(0L, memoryManager.getOversizedEntryCount()); + assertEquals( + oneTabletBytes, queueA.getRetainedTabletBytes() + queueB.getRetainedTabletBytes()); readerB.currentSearchIndex = 1L; assertTrue(pendingEntries(queueB).offer(createRequest(1L))); @@ -323,6 +338,7 @@ public void testQueuesShareDataNodeMemoryBudget() throws Exception { assertTrue(queueA.ack("consumerA", eventA.getCommitContext())); assertEquals(0L, queueA.getRetainedTabletBytes()); assertEquals(oneTabletBytes, memoryManager.getUsedMemorySizeInBytes()); + assertEquals(0L, memoryManager.getOvercommitSizeInBytes()); queueB.drivePrefetchOnce(); assertEquals("false", queueB.coreReportMessage().get("realtimeAdmissionBlocked")); @@ -331,6 +347,7 @@ public void testQueuesShareDataNodeMemoryBudget() throws Exception { assertNotNull(eventB); assertTrue(queueB.ack("consumerB", eventB.getCommitContext())); assertEquals(0L, memoryManager.getUsedMemorySizeInBytes()); + assertEquals(0L, memoryManager.getOvercommitSizeInBytes()); queueB.drivePrefetchOnce(); assertEquals("false", queueB.coreReportMessage().get("realtimeAdmissionBlocked")); } finally { @@ -347,6 +364,151 @@ public void testQueuesShareDataNodeMemoryBudget() throws Exception { } } + @Test + public void testWatermarkAndAckDoNotHideDataDeliveryGap() throws Exception { + final String originalSystemDir = IoTDBDescriptor.getInstance().getConfig().getSystemDir(); + final int originalBatchMaxDelay = + CommonDescriptor.getInstance().getConfig().getSubscriptionConsensusBatchMaxDelayInMs(); + final File systemDir = temporaryFolder.newFolder("delivery-gap"); + ConsensusPrefetchingQueue queue = null; + try { + CommonDescriptor.getInstance().getConfig().setSubscriptionConsensusBatchMaxDelayInMs(0); + final FakeConsensusReqReader reader = new FakeConsensusReqReader(); + queue = + newQueue( + "consumerGroup", + new DataRegionId(1), + reader, + newConverter(new AtomicInteger()), + newCommitManager(systemDir), + TopicConstant.ORDER_MODE_LEADER_ONLY_VALUE); + final AtomicLong clockNs = new AtomicLong(-1_000_000L); + setTimeTracker(queue, new SubscriptionQueueTimeTracker(clockNs::get)); + queue.setSubscriptionMemoryManager(new SubscriptionMemoryManager(1_000_000L)); + + assertNull(queue.poll("consumer")); + reader.currentSearchIndex = 1L; + assertTrue(pendingEntries(queue).offer(createRequest(1L))); + queue.drivePrefetchOnce(); + final SubscriptionEvent first = queue.poll("consumer"); + assertNotNull(first); + + clockNs.addAndGet(TimeUnit.MILLISECONDS.toNanos(60_000L)); + final Method injectWatermark = + ConsensusPrefetchingQueue.class.getDeclaredMethod("injectWatermark", long.class); + injectWatermark.setAccessible(true); + injectWatermark.invoke(queue, 1L); + final SubscriptionEvent watermark = queue.poll("consumer"); + assertNotNull(watermark); + assertEquals( + SubscriptionPollResponseType.WATERMARK.getType(), + watermark.getCurrentResponse().getResponseType()); + assertEquals(60_000L, queue.getDeliveryIdleTimeMs()); + assertTrue(queue.ack("consumer", first.getCommitContext())); + assertEquals(60_000L, queue.getDeliveryIdleTimeMs()); + assertEquals(0L, queue.getLastDeliveryIntervalMs()); + + clockNs.addAndGet(TimeUnit.MILLISECONDS.toNanos(1L)); + reader.currentSearchIndex = 2L; + assertTrue(pendingEntries(queue).offer(createRequest(2L))); + queue.drivePrefetchOnce(); + final SubscriptionEvent second = queue.poll("consumer"); + assertNotNull(second); + assertEquals(60_001L, queue.getLastDeliveryIntervalMs()); + assertEquals(60_001L, queue.getMaxDeliveryIntervalMs()); + assertEquals(0L, queue.getDeliveryIdleTimeMs()); + assertTrue(queue.ack("consumer", second.getCommitContext())); + + clockNs.addAndGet(TimeUnit.MILLISECONDS.toNanos(1L)); + reader.currentSearchIndex = 3L; + assertTrue(pendingEntries(queue).offer(createRequest(3L))); + queue.drivePrefetchOnce(); + assertNotNull(queue.poll("consumer")); + assertEquals(1L, queue.getLastDeliveryIntervalMs()); + assertEquals(60_001L, queue.getMaxDeliveryIntervalMs()); + } finally { + if (queue != null) { + queue.close(); + } + CommonDescriptor.getInstance() + .getConfig() + .setSubscriptionConsensusBatchMaxDelayInMs(originalBatchMaxDelay); + IoTDBDescriptor.getInstance().getConfig().setSystemDir(originalSystemDir); + } + } + + @Test + public void testTimingRemainsReadableDuringBlockedPrefetch() throws Exception { + final String originalSystemDir = IoTDBDescriptor.getInstance().getConfig().getSystemDir(); + final File systemDir = temporaryFolder.newFolder("blocked-prefetch-timing"); + final CountDownLatch conversionStarted = new CountDownLatch(1); + final CountDownLatch releaseConversion = new CountDownLatch(1); + final ExecutorService executor = Executors.newSingleThreadExecutor(); + ConsensusPrefetchingQueue queue = null; + try { + final FakeConsensusReqReader reader = new FakeConsensusReqReader(); + final ConsensusLogToTabletConverter converter = mock(ConsensusLogToTabletConverter.class); + when(converter.getDatabaseName()).thenReturn("db"); + when(converter.convert(any())) + .thenAnswer( + ignored -> { + conversionStarted.countDown(); + assertTrue(releaseConversion.await(5L, TimeUnit.SECONDS)); + return Collections.singletonList(createTablet()); + }); + queue = + newQueue( + "consumerGroup", + new DataRegionId(1), + reader, + converter, + newCommitManager(systemDir), + TopicConstant.ORDER_MODE_LEADER_ONLY_VALUE); + final AtomicLong clockNs = new AtomicLong(Long.MAX_VALUE - 1_000_000L); + setTimeTracker(queue, new SubscriptionQueueTimeTracker(clockNs::get)); + queue.setSubscriptionMemoryManager(new SubscriptionMemoryManager(1_000_000L)); + + assertNull(queue.poll("consumer")); + reader.currentSearchIndex = 1L; + assertTrue(pendingEntries(queue).offer(createRequest(1L))); + final ConsensusPrefetchingQueue blockedQueue = queue; + final Future round = executor.submit(blockedQueue::drivePrefetchOnce); + assertTrue(conversionStarted.await(5L, TimeUnit.SECONDS)); + clockNs.addAndGet(TimeUnit.MILLISECONDS.toNanos(65_000L)); + + assertEquals(65_000L, queue.getPrefetchDurationMs()); + assertEquals(65_000L, queue.getMaxPrefetchDurationMs()); + assertEquals(65_000L, queue.getPrefetchIdleTimeMs()); + assertEquals(65_000L, queue.getDeliveryIdleTimeMs()); + assertEquals("65000", queue.coreReportMessage().get("prefetchDurationMs")); + assertEquals("0", queue.coreReportMessage().get("subscriptionMemoryUsedInBytes")); + assertEquals("0", queue.coreReportMessage().get("dataNodeSubscriptionMemoryUsedInBytes")); + + releaseConversion.countDown(); + round.get(5L, TimeUnit.SECONDS); + assertEquals(0L, queue.getPrefetchDurationMs()); + assertEquals(65_000L, queue.getMaxPrefetchDurationMs()); + assertEquals(0L, queue.getPrefetchIdleTimeMs()); + queue.drivePrefetchOnce(); + assertEquals(65_000L, queue.getMaxPrefetchDurationMs()); + } finally { + releaseConversion.countDown(); + executor.shutdownNow(); + if (queue != null) { + queue.close(); + } + IoTDBDescriptor.getInstance().getConfig().setSystemDir(originalSystemDir); + } + } + + private static void setTimeTracker( + final ConsensusPrefetchingQueue queue, final SubscriptionQueueTimeTracker tracker) + throws Exception { + final Field field = ConsensusPrefetchingQueue.class.getDeclaredField("timeTracker"); + field.setAccessible(true); + field.set(queue, tracker); + } + @Test public void testRealtimeBacklogIsDrainedBeforeMorePendingEntriesAreConverted() throws Exception { final String originalSystemDir = IoTDBDescriptor.getInstance().getConfig().getSystemDir(); diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueTest.java index 600073d7e3d6..0b9e12eda48a 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueTest.java @@ -94,7 +94,7 @@ import static org.junit.Assert.assertSame; import static org.junit.Assert.assertTrue; import static org.mockito.ArgumentMatchers.any; -import static org.mockito.ArgumentMatchers.anyLong; +import static org.mockito.ArgumentMatchers.eq; import static org.mockito.Mockito.doAnswer; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.never; @@ -2388,6 +2388,57 @@ protected ProgressWALIterator createSubscriptionWALIterator( } } + @Test + public void testPeriodicStatsDoNotReadWal() throws Exception { + final String originalSystemDir = IoTDBDescriptor.getInstance().getConfig().getSystemDir(); + final File systemDir = temporaryFolder.newFolder("stats-without-wal-read"); + ConsensusPrefetchingQueue queue = null; + try { + final IoTConsensusServerImpl serverImpl = mock(IoTConsensusServerImpl.class); + when(serverImpl.getConsensusReqReader()).thenReturn(new FakeConsensusReqReader()); + when(serverImpl.getWriterSafeFrontierTracker()).thenReturn(new WriterSafeFrontierTracker()); + queue = + new ConsensusPrefetchingQueue( + "consumerGroup", + "topic", + TopicConstant.ORDER_MODE_LEADER_ONLY_VALUE, + new DataRegionId(1), + serverImpl, + new SubscriptionWalRetentionPolicy( + "topic", + SubscriptionWalRetentionPolicy.UNBOUNDED, + SubscriptionWalRetentionPolicy.UNBOUNDED), + mock(ConsensusLogToTabletConverter.class), + newCommitManager(systemDir), + new RegionProgress(Collections.emptyMap()), + 1L, + 1L, + true); + + final ProgressWALIterator iterator = mock(ProgressWALIterator.class); + setSubscriptionWalIterator(queue, iterator); + final Field lastStatsLogTime = + ConsensusPrefetchingQueue.class.getDeclaredField("lastStatsLogTimeMs"); + lastStatsLogTime.setAccessible(true); + lastStatsLogTime.setLong(queue, 0L); + final Method logStats = + ConsensusPrefetchingQueue.class.getDeclaredMethod("logPeriodicStatsIfNecessary"); + logStats.setAccessible(true); + logStats.invoke(queue); + + verify(iterator).hasBufferedNext(); + verify(iterator, never()).hasNext(); + verify(iterator, never()).next(); + verify(iterator, never()).refresh(); + assertEquals(1L, queue.getCurrentReadSearchIndex()); + } finally { + if (queue != null) { + queue.close(); + } + IoTDBDescriptor.getInstance().getConfig().setSystemDir(originalSystemDir); + } + } + @Test public void testPendingCursorAdvanceFastForwardsWalIteratorInPlace() throws Exception { final String originalSystemDir = IoTDBDescriptor.getInstance().getConfig().getSystemDir(); @@ -2398,6 +2449,8 @@ public void testPendingCursorAdvanceFastForwardsWalIteratorInPlace() throws Exce final IoTConsensusServerImpl serverImpl = mock(IoTConsensusServerImpl.class); when(serverImpl.getConsensusReqReader()).thenReturn(reader); when(serverImpl.getWriterSafeFrontierTracker()).thenReturn(new WriterSafeFrontierTracker()); + final ConsensusLogToTabletConverter converter = mock(ConsensusLogToTabletConverter.class); + when(converter.convert(any())).thenReturn(Collections.emptyList()); queue = new ConsensusPrefetchingQueue( "consumerGroup", @@ -2409,28 +2462,26 @@ public void testPendingCursorAdvanceFastForwardsWalIteratorInPlace() throws Exce "topic", SubscriptionWalRetentionPolicy.UNBOUNDED, SubscriptionWalRetentionPolicy.UNBOUNDED), - mock(ConsensusLogToTabletConverter.class), + converter, newCommitManager(systemDir), new RegionProgress(Collections.emptyMap()), 1L, 1L, true); + for (long searchIndex = 1L; searchIndex <= 7L; searchIndex++) { + assertTrue(pendingEntries(queue).offer(createRequest(searchIndex))); + } + reader.currentSearchIndex = 7L; + assertNull(queue.poll("consumer")); final ProgressWALIterator staleIterator = mock(ProgressWALIterator.class); setSubscriptionWalIterator(queue, staleIterator); - - final Method advancePendingCursor = - ConsensusPrefetchingQueue.class.getDeclaredMethod( - "advanceLocalCursorFromPendingIfPresent", IndexedConsensusRequest.class, long.class); - advancePendingCursor.setAccessible(true); - advancePendingCursor.invoke( - queue, - new IndexedConsensusRequest(7L, Collections.emptyList()), - queue.getCurrentSeekGeneration()); + queue.drivePrefetchOnce(); assertEquals(8L, queue.getCurrentReadSearchIndex()); + verify(converter, times(7)).convert(any()); verify(staleIterator) - .advanceTo(anyLong(), any(ProgressWALIterator.WriterProgressCoverage.class)); + .advanceTo(eq(8L), any(ProgressWALIterator.WriterProgressCoverage.class)); verify(staleIterator, never()).close(); assertSame(staleIterator, subscriptionWalIterator(queue)); } finally { @@ -2545,10 +2596,14 @@ public void testMemoryBackpressureStopsFurtherMaterializationUntilAck() throws E reader.currentSearchIndex = 3L; assertNull(queue.poll("consumer")); + final ProgressWALIterator iterator = mock(ProgressWALIterator.class); + setSubscriptionWalIterator(queue, iterator); queue.drivePrefetchOnce(); assertEquals(2, conversionCount.get()); assertEquals(2L, queue.getCurrentReadSearchIndex()); + verify(iterator).advanceTo(eq(2L), any(ProgressWALIterator.WriterProgressCoverage.class)); + verify(iterator, never()).close(); assertEquals(oneTabletBytes, queue.getRetainedTabletBytes()); assertTrue(memoryManager.getFreeMemorySizeInBytes() > 0L); assertEquals(1, queue.getPrefetchedEventCount()); diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueWalBackpressureTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueWalBackpressureTest.java index e25c77803cff..54ea59fbb0a1 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueWalBackpressureTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueWalBackpressureTest.java @@ -66,6 +66,7 @@ import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertSame; import static org.junit.Assert.assertTrue; import static org.mockito.ArgumentMatchers.any; import static org.mockito.Mockito.mock; @@ -210,6 +211,127 @@ public void testAckRecoversDrainedSuffixFromWalWithoutReoffer() throws Exception } } + @Test + public void testWalMemoryBlockRetriesWithoutRecreatingIterator() throws Exception { + final String originalSystemDir = IoTDBDescriptor.getInstance().getConfig().getSystemDir(); + final CommonConfig config = CommonDescriptor.getInstance().getConfig(); + final int originalBatchMaxWalEntries = config.getSubscriptionConsensusBatchMaxWalEntries(); + final int originalBatchMaxTabletCount = config.getSubscriptionConsensusBatchMaxTabletCount(); + final long originalBatchMaxSize = config.getSubscriptionConsensusBatchMaxSizeInBytes(); + final int originalBatchMaxDelay = config.getSubscriptionConsensusBatchMaxDelayInMs(); + final File systemDir = temporaryFolder.newFolder("system-wal-memory-retry"); + final File walDirectory = temporaryFolder.newFolder("wal-memory-retry"); + ConsensusPrefetchingQueue queue = null; + try { + final Tablet sampleTablet = createTablet(); + final long oneTabletBytes = sampleTablet.ramBytesUsed(); + config.setSubscriptionConsensusBatchMaxWalEntries(128); + config.setSubscriptionConsensusBatchMaxTabletCount(128); + config.setSubscriptionConsensusBatchMaxSizeInBytes(oneTabletBytes * 16L); + config.setSubscriptionConsensusBatchMaxDelayInMs(0); + + writeSealedWal(walDirectory); + + final WALNode walNode = mock(WALNode.class); + when(walNode.getLogDirectory()).thenReturn(walDirectory); + when(walNode.getCurrentSearchIndex()).thenReturn((long) REQUEST_COUNT); + when(walNode.getCurrentWALFileVersion()).thenReturn(1L); + when(walNode.getCurrentWALMetaDataSnapshot()).thenReturn(new WALMetaData()); + + final IoTConsensusServerImpl serverImpl = mock(IoTConsensusServerImpl.class); + when(serverImpl.getConsensusReqReader()).thenReturn(walNode); + when(serverImpl.getWriterSafeFrontierTracker()).thenReturn(new WriterSafeFrontierTracker()); + + final List conversionAttempts = new ArrayList<>(); + final ConsensusLogToTabletConverter converter = mock(ConsensusLogToTabletConverter.class); + when(converter.convert(any())) + .thenAnswer( + invocation -> { + final InsertNode insertNode = (InsertNode) invocation.getArgument(0); + conversionAttempts.add(insertNode.getSearchIndex()); + return Collections.singletonList(createTablet()); + }); + when(converter.getDatabaseName()).thenReturn("db"); + + final DataRegionId regionId = new DataRegionId(1); + final ConsensusSubscriptionCommitManager commitManager = newCommitManager(systemDir); + queue = + new ConsensusPrefetchingQueue( + "consumerGroup", + "topic", + TopicConstant.ORDER_MODE_LEADER_ONLY_VALUE, + regionId, + serverImpl, + new SubscriptionWalRetentionPolicy( + "topic", + SubscriptionWalRetentionPolicy.UNBOUNDED, + SubscriptionWalRetentionPolicy.UNBOUNDED), + converter, + commitManager, + new RegionProgress(Collections.emptyMap()), + 1L, + 1L, + true); + final TrackingSubscriptionMemoryManager memoryManager = + new TrackingSubscriptionMemoryManager(oneTabletBytes + Math.max(1L, oneTabletBytes / 2L)); + queue.setSubscriptionMemoryManager(memoryManager); + + assertNull(queue.poll("consumer")); + final ProgressWALIterator iterator = subscriptionWalIterator(queue); + + queue.drivePrefetchOnce(); + assertEquals(Arrays.asList(1L, 2L), conversionAttempts); + assertEquals(2L, queue.getCurrentReadSearchIndex()); + assertTrue(iterator.hasBufferedNext()); + assertSame(iterator, subscriptionWalIterator(queue)); + assertMemoryBounded(queue, memoryManager); + + // No memory has been released yet: another round must neither convert nor lose the request. + queue.drivePrefetchOnce(); + assertEquals(Arrays.asList(1L, 2L), conversionAttempts); + assertEquals(2L, queue.getCurrentReadSearchIndex()); + assertSame(iterator, subscriptionWalIterator(queue)); + + for (long expectedLocalSeq = 1L; expectedLocalSeq <= REQUEST_COUNT; expectedLocalSeq++) { + final SubscriptionEvent event = queue.poll("consumer"); + assertNotNull(event); + assertEquals(expectedLocalSeq, event.getCommitContext().getWriterProgress().getLocalSeq()); + assertEquals( + 1000L + expectedLocalSeq, + event.getCommitContext().getWriterProgress().getPhysicalTime()); + assertTrue(queue.ack("consumer", event.getCommitContext())); + assertEquals(0L, memoryManager.getUsedMemorySizeInBytes()); + + queue.drivePrefetchOnce(); + assertSame(iterator, subscriptionWalIterator(queue)); + assertMemoryBounded(queue, memoryManager); + } + + assertEquals(Arrays.asList(1L, 2L, 2L, 3L, 3L), conversionAttempts); + assertEquals(3L, queue.getWalPathAcceptedEntries()); + assertEquals(4L, queue.getCurrentReadSearchIndex()); + assertEquals(0, queue.getPrefetchedEventCount()); + assertEquals(0L, queue.getSubscriptionUncommittedEventCount()); + assertEquals(0L, queue.getRetainedTabletBytes()); + assertNull(queue.poll("consumer")); + assertEquals( + new WriterProgress(1000L + REQUEST_COUNT, REQUEST_COUNT), + commitManager + .getCommittedRegionProgress("consumerGroup", "topic", regionId) + .getWriterPositions() + .get(new WriterId(regionId.toString(), WRITER_NODE_ID))); + } finally { + if (queue != null) { + queue.close(); + } + config.setSubscriptionConsensusBatchMaxWalEntries(originalBatchMaxWalEntries); + config.setSubscriptionConsensusBatchMaxTabletCount(originalBatchMaxTabletCount); + config.setSubscriptionConsensusBatchMaxSizeInBytes(originalBatchMaxSize); + config.setSubscriptionConsensusBatchMaxDelayInMs(originalBatchMaxDelay); + IoTDBDescriptor.getInstance().getConfig().setSystemDir(originalSystemDir); + } + } + private static void writeSealedWal(final File walDirectory) throws Exception { final File historicalWal = new File( @@ -288,6 +410,13 @@ private static BlockingQueue pendingEntries( return (BlockingQueue) field.get(queue); } + private static ProgressWALIterator subscriptionWalIterator(final ConsensusPrefetchingQueue queue) + throws Exception { + final Field field = ConsensusPrefetchingQueue.class.getDeclaredField("subscriptionWALIterator"); + field.setAccessible(true); + return (ProgressWALIterator) field.get(queue); + } + private static ConsensusSubscriptionCommitManager newCommitManager(final File systemDir) throws Exception { IoTDBDescriptor.getInstance().getConfig().setSystemDir(systemDir.getAbsolutePath()); diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ProgressWALIteratorTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ProgressWALIteratorTest.java index 8af97b319644..d6159e6c692a 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ProgressWALIteratorTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ProgressWALIteratorTest.java @@ -22,6 +22,7 @@ import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNodeType; import org.apache.iotdb.consensus.common.request.IndexedConsensusRequest; import org.apache.iotdb.db.conf.IoTDBDescriptor; +import org.apache.iotdb.db.i18n.DataNodePipeMessages; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.SearchNode; import org.apache.iotdb.db.storageengine.dataregion.wal.buffer.WALEntryType; import org.apache.iotdb.db.storageengine.dataregion.wal.buffer.WALInfoEntry; @@ -397,7 +398,8 @@ public void testIteratorAggregatesUnreadableRetainedWalFiles() throws Exception final String detail = iterator.getSkippedBrokenWalFileDetails(0); assertTrue(detail.contains(firstBrokenWal.getName())); assertTrue(detail.contains(secondBrokenWal.getName())); - assertTrue(detail.contains("skippedEntries=unknown")); + assertTrue( + detail.contains("skippedEntries=" + DataNodePipeMessages.MESSAGE_UNKNOWN_AD921D60)); assertEquals(2L, iterator.next().getSearchIndex()); assertFalse(iterator.hasNext()); assertTrue(iterator.hasSkippedBrokenWalFiles()); @@ -735,6 +737,47 @@ public void testIteratorMarksIncompleteScanWhenNearLiveWalCannotBeOpened() throw } } + @Test + public void testBufferedNextDoesNotScanPastStaleLocalRequest() throws Exception { + final Path dir = Files.createTempDirectory("progress-wal-iterator-buffered-next"); + final File dataWal = + dir.resolve(WALFileUtils.getLogFileName(0, 0, WALFileStatus.CONTAINS_SEARCH_INDEX)) + .toFile(); + final File successorWal = + dir.resolve(WALFileUtils.getLogFileName(1, 2, WALFileStatus.CONTAINS_SEARCH_INDEX)) + .toFile(); + try { + try (WALWriter writer = new WALWriter(dataWal, WALFileVersion.V3)) { + writer.write(searchableEntry(1L), singleEntryMeta(19, 1L, 1L, 100L, 7, 1L)); + writer.write(searchableEntry(-1L), singleEntryMeta(19, -1L, 1L, 200L, 8, 20L)); + writer.write(searchableEntry(2L), singleEntryMeta(19, 2L, 1L, 300L, 7, 2L)); + } + try (WALWriter ignored = new WALWriter(successorWal, WALFileVersion.V3)) { + // Seal the data file so all requests can be read from retained WAL. + } + try (ProgressWALIterator iterator = new ProgressWALIterator(dir.toFile(), 1L)) { + assertFalse(iterator.hasBufferedNext()); + assertTrue(iterator.hasNext()); + assertTrue(iterator.hasBufferedNext()); + + iterator.advanceTo(2L, null); + assertFalse(iterator.hasBufferedNext()); + assertTrue(iterator.hasNext()); + assertTrue(iterator.hasBufferedNext()); + + iterator.advanceTo(3L, null); + assertTrue(iterator.hasBufferedNext()); + assertEquals(-1L, iterator.next().getSearchIndex()); + assertFalse(iterator.hasBufferedNext()); + assertFalse(iterator.hasNext()); + } + } finally { + Files.deleteIfExists(dataWal.toPath()); + Files.deleteIfExists(successorWal.toPath()); + Files.deleteIfExists(dir); + } + } + @Test public void testAdvanceToKeepsUncoveredFollowerRequestInCurrentFile() throws Exception { final Path dir = Files.createTempDirectory("progress-wal-iterator-current-file-advance"); diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/SubscriptionWalPrefetchPerformanceTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/SubscriptionWalPrefetchPerformanceTest.java new file mode 100644 index 000000000000..b268bfd4249f --- /dev/null +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/SubscriptionWalPrefetchPerformanceTest.java @@ -0,0 +1,380 @@ +/* + * 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.subscription.broker.consensus; + +import org.apache.iotdb.commons.request.IConsensusRequest; +import org.apache.iotdb.consensus.common.request.IndexedConsensusRequest; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertNode; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowNode; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowsNode; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalInsertRowNode; +import org.apache.iotdb.db.queryengine.plan.statement.StatementTestUtils; +import org.apache.iotdb.db.storageengine.dataregion.wal.buffer.WALInfoEntry; +import org.apache.iotdb.db.storageengine.dataregion.wal.io.WALFileVersion; +import org.apache.iotdb.db.storageengine.dataregion.wal.io.WALMetaData; +import org.apache.iotdb.db.storageengine.dataregion.wal.io.WALWriter; +import org.apache.iotdb.db.storageengine.dataregion.wal.utils.WALByteBufferForTest; +import org.apache.iotdb.db.storageengine.dataregion.wal.utils.WALFileStatus; +import org.apache.iotdb.db.storageengine.dataregion.wal.utils.WALFileUtils; + +import org.junit.Assume; +import org.junit.Rule; +import org.junit.Test; +import org.junit.rules.TemporaryFolder; + +import java.io.File; +import java.lang.reflect.Field; +import java.nio.ByteBuffer; +import java.util.Arrays; +import java.util.Locale; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertSame; +import static org.junit.Assert.assertTrue; + +/** + * Manual comparisons of the iterator operations changed by subscription prefetch optimizations. + * Both paths use the current implementation and the same sealed, uncompressed V3 WAL. Wall time is + * diagnostic only; assertions cover work counts, writer progress and retry payload integrity. + */ +public class SubscriptionWalPrefetchPerformanceTest { + + private static final String PREFIX = "iotdb.test.subscription.performance."; + private static final int FRAGMENTS = 2; + + @Rule public final TemporaryFolder temporaryFolder = new TemporaryFolder(); + + /** + * Enable with -Diotdb.test.subscription.performance.prefetch-overhead=true. Models a lagging + * iterator and a pending batch that advances past a retained file with an uncovered follower. + */ + @Test + public void testNoIoStatsProbeAndCoalescedAdvancePerformance() throws Exception { + assumeEnabled("prefetch-overhead"); + final int entryCount = positiveProperty("entries", 4096); + final int batchSize = positiveProperty("batch-size", 64); + final int iterations = positiveProperty("iterations", 5); + final int warmups = positiveProperty("warmups", 2); + assertTrue(entryCount >= 2); + final File walDirectory = createWal(entryCount, true, batchSize); + final long probeIndex = Math.max(1L, entryCount * 3L / 4L); + + final Sample[] probing = new Sample[iterations]; + final Sample[] peeking = new Sample[iterations]; + final Sample[] perEntry = new Sample[iterations]; + final Sample[] coalesced = new Sample[iterations]; + for (int round = 0; round < warmups + iterations; round++) { + // Alternate order so the optimized path does not always follow the baseline. + final boolean optimizedFirst = (round & 1) == 1; + final Sample firstProbe = measureStatsProbe(walDirectory, probeIndex, optimizedFirst); + final Sample secondProbe = measureStatsProbe(walDirectory, probeIndex, !optimizedFirst); + final Sample firstAdvance = + measureAdvance(walDirectory, entryCount, batchSize, optimizedFirst); + final Sample secondAdvance = + measureAdvance(walDirectory, entryCount, batchSize, !optimizedFirst); + if (round >= warmups) { + final int index = round - warmups; + probing[index] = optimizedFirst ? secondProbe : firstProbe; + peeking[index] = optimizedFirst ? firstProbe : secondProbe; + perEntry[index] = optimizedFirst ? secondAdvance : firstAdvance; + coalesced[index] = optimizedFirst ? firstAdvance : secondAdvance; + } + } + + printWorkload(entryCount, warmups, iterations); + System.out.printf(Locale.ROOT, "batchSize=%d, probeSearchIndex=%d%n", batchSize, probeIndex); + printComparison("stats probe (work=entry bodies read)", probing, peeking); + printComparison("cursor advance (work=footer tuples checked)", perEntry, coalesced); + } + + /** + * Enable with -Diotdb.test.subscription.performance.memory-retry=true. Replays the queue's old + * reset and new cache-restoration operations after each rejected materialization attempt. + * Includes deserialization; excludes Tablet conversion, allocation and scheduler wait time. + */ + @Test + public void testMemoryBlockedRetryPerformance() throws Exception { + assumeEnabled("memory-retry"); + final int entryCount = positiveProperty("entries", 4096); + final int retryCount = positiveProperty("retries", 8); + final int iterations = positiveProperty("iterations", 5); + final int warmups = positiveProperty("warmups", 2); + assertTrue(entryCount >= 2); + final File walDirectory = createWal(entryCount, false, 0); + final long retryIndex = Math.max(1L, entryCount * 3L / 4L); + + final Sample[] recreated = new Sample[iterations]; + final Sample[] cached = new Sample[iterations]; + for (int round = 0; round < warmups + iterations; round++) { + final boolean optimizedFirst = (round & 1) == 1; + final Sample first = measureRetries(walDirectory, retryIndex, retryCount, optimizedFirst); + final Sample second = measureRetries(walDirectory, retryIndex, retryCount, !optimizedFirst); + if (round >= warmups) { + final int index = round - warmups; + recreated[index] = optimizedFirst ? second : first; + cached[index] = optimizedFirst ? first : second; + } + } + + printWorkload(entryCount, warmups, iterations); + System.out.printf(Locale.ROOT, "attempts=%d, retrySearchIndex=%d%n", retryCount, retryIndex); + printComparison("memory retry (work=entry bodies read)", recreated, cached); + } + + private File createWal(final int entryCount, final boolean includeFollower, final int batchSize) + throws Exception { + final File directory = temporaryFolder.newFolder(); + final File historical = + new File( + directory, WALFileUtils.getLogFileName(0L, 0L, WALFileStatus.CONTAINS_SEARCH_INDEX)); + try (WALWriter writer = new WALWriter(historical, WALFileVersion.V3)) { + for (long index = 1L; index <= entryCount; index++) { + writeRequest(writer, index, 7, index); + } + if (includeFollower) { + writeRequest(writer, -1L, 8, 1L); + } + } + final File successor = + new File( + directory, + WALFileUtils.getLogFileName(1L, entryCount, WALFileStatus.CONTAINS_SEARCH_INDEX)); + try (WALWriter writer = new WALWriter(successor, WALFileVersion.V3)) { + if (includeFollower) { + for (long index = entryCount + 1L; index <= entryCount + batchSize + 1L; index++) { + writeRequest(writer, index, 7, index); + } + } + // Even an empty successor makes the first file historical and readable. + } + return directory; + } + + private static void writeRequest( + final WALWriter writer, final long searchIndex, final int nodeId, final long localSeq) + throws Exception { + for (int fragment = 0; fragment < FRAGMENTS; fragment++) { + final RelationalInsertRowNode node = + StatementTestUtils.genInsertRowNode(Math.toIntExact(localSeq * FRAGMENTS + fragment)); + node.setSearchIndex(searchIndex); + node.setLastFragment(fragment == FRAGMENTS - 1); + final WALInfoEntry entry = new WALInfoEntry(1L, node); + final ByteBuffer buffer = ByteBuffer.allocate(entry.serializedSize()); + entry.serialize(new WALByteBufferForTest(buffer)); + final WALMetaData metadata = new WALMetaData(); + metadata.add(entry.serializedSize(), searchIndex, 1L, 1000L + localSeq, nodeId, localSeq); + // Small individual segments intentionally avoid changing global WAL compression settings. + writer.write(buffer, metadata); + } + } + + private static Sample measureStatsProbe( + final File directory, final long probeIndex, final boolean optimized) throws Exception { + final Field bodyCount = bodyCountField(); + try (ProgressWALIterator iterator = new ProgressWALIterator(directory, probeIndex)) { + final long startNanos = System.nanoTime(); + final boolean ready = optimized ? iterator.hasBufferedNext() : iterator.hasNext(); + final long elapsedNanos = System.nanoTime() - startNanos; + assertEquals(!optimized, ready); + final long readBodies = bodyCount.getInt(iterator); + // One lookahead fragment is read to establish the end of a logical request. + assertEquals(optimized ? 0L : probeIndex * FRAGMENTS + 1L, readBodies); + assertRequest(iterator.next(), probeIndex, 7, probeIndex); + return new Sample(elapsedNanos, readBodies); + } + } + + private static Sample measureAdvance( + final File directory, final int entryCount, final int batchSize, final boolean optimized) + throws Exception { + final long[] footerTuples = {0L}; + final ProgressWALIterator.WriterProgressCoverage coverage = + (physicalTime, nodeId, localSeq) -> { + footerTuples[0]++; + return nodeId == 7 && localSeq <= entryCount; + }; + try (ProgressWALIterator iterator = new ProgressWALIterator(directory, 1L)) { + final long startNanos = System.nanoTime(); + if (optimized) { + iterator.advanceTo(entryCount + batchSize + 1L, coverage); + } else { + for (int index = 1; index <= batchSize; index++) { + iterator.advanceTo(entryCount + index + 1L, coverage); + } + } + final long elapsedNanos = System.nanoTime() - startNanos; + // The first follower tuple prevents skipping the retained file on every coverage check. + assertEquals( + (entryCount * (long) FRAGMENTS + 1L) * (optimized ? 1L : batchSize), footerTuples[0]); + assertRequest(iterator.next(), -1L, 8, 1L); + final long nextIndex = entryCount + batchSize + 1L; + assertRequest(iterator.next(), nextIndex, 7, nextIndex); + assertFalse(iterator.hasNext()); + return new Sample(elapsedNanos, footerTuples[0]); + } + } + + private static Sample measureRetries( + final File directory, final long retryIndex, final int attempts, final boolean optimized) + throws Exception { + final Field bodyCount = bodyCountField(); + ProgressWALIterator iterator = null; + IndexedConsensusRequest previousRequest = null; + long readBodies = 0L; + int previousBodyCount = 0; + try { + final long startNanos = System.nanoTime(); + for (int attempt = 0; attempt < attempts; attempt++) { + if (iterator == null) { + iterator = new ProgressWALIterator(directory, retryIndex); + } + final IndexedConsensusRequest request = iterator.next(); + assertRequest(request, retryIndex, 7, retryIndex); + final int currentBodyCount = bodyCount.getInt(iterator); + readBodies += optimized ? currentBodyCount - previousBodyCount : currentBodyCount; + previousBodyCount = currentBodyCount; + if (optimized && previousRequest != null) { + assertSame(previousRequest, request); + } + previousRequest = request; + if (attempt + 1 < attempts) { + if (optimized) { + iterator.retry(request); + } else { + iterator.close(); + iterator = null; + } + } + } + final long elapsedNanos = System.nanoTime() - startNanos; + assertEquals((retryIndex * FRAGMENTS + 1L) * (optimized ? 1L : attempts), readBodies); + // The last attempt succeeds: the pending lookahead must still lead to the next request. + assertRequest(iterator.next(), retryIndex + 1L, 7, retryIndex + 1L); + return new Sample(elapsedNanos, readBodies); + } finally { + if (iterator != null) { + iterator.close(); + } + } + } + + private static void assertRequest( + final IndexedConsensusRequest request, + final long searchIndex, + final int nodeId, + final long localSeq) { + assertEquals(searchIndex, request.getSearchIndex()); + assertEquals(nodeId, request.getNodeId()); + assertEquals(localSeq, request.getProgressLocalSeq()); + assertEquals(1000L + localSeq, request.getPhysicalTime()); + assertEquals(FRAGMENTS, request.getRequests().size()); + final InsertNode insert = ConsensusLogToTabletConverter.deserializeToInsertNode(request); + assertNotNull(insert); + assertEquals(searchIndex, insert.getSearchIndex()); + assertEquals(nodeId, insert.getNodeId()); + assertEquals(localSeq, insert.getSyncIndex()); + assertEquals(1000L + localSeq, insert.getPhysicalTime()); + assertTrue(insert instanceof InsertRowsNode); + final InsertRowsNode rows = (InsertRowsNode) insert; + assertEquals(FRAGMENTS, rows.getInsertRowNodeList().size()); + for (int fragment = 0; fragment < FRAGMENTS; fragment++) { + final InsertRowNode row = rows.getInsertRowNodeList().get(fragment); + final long expectedValue = localSeq * FRAGMENTS + fragment; + assertEquals(expectedValue, row.getTime()); + assertEquals((double) expectedValue, (Double) row.getValues()[2], 0.0); + } + for (final IConsensusRequest fragment : request.getRequests()) { + assertEquals(0, fragment.serializeToByteBuffer().position()); + } + } + + private static Field bodyCountField() throws Exception { + // All counted reads remain in one historical file; inspect existing state without a + // production instrumentation hook or overhead in WALReader.next(). + final Field field = + ProgressWALIterator.class.getDeclaredField("consumedEntryCountInCurrentFile"); + field.setAccessible(true); + return field; + } + + private static void assumeEnabled(final String benchmark) { + Assume.assumeTrue( + "Enable with -D" + PREFIX + benchmark + "=true", Boolean.getBoolean(PREFIX + benchmark)); + } + + private static int positiveProperty(final String name, final int defaultValue) { + final int value = Integer.getInteger(PREFIX + name, defaultValue); + assertTrue(name + " must be positive", value > 0); + return value; + } + + private static void printWorkload(final int entryCount, final int warmups, final int iterations) { + System.out.printf( + Locale.ROOT, + "Subscription WAL benchmark: requests=%d, fragments=%d, warmups=%d, samples=%d, " + + "os=%s, java=%s%n", + entryCount, + FRAGMENTS, + warmups, + iterations, + System.getProperty("os.name"), + System.getProperty("java.version")); + } + + private static void printComparison( + final String name, final Sample[] baseline, final Sample[] optimized) { + final double baselineNanos = medianNanos(baseline); + final double optimizedNanos = medianNanos(optimized); + System.out.printf( + Locale.ROOT, + "%s: baselineMedian=%.6f ms, optimizedMedian=%.6f ms, speedup=%.2fx, " + + "baselineWork=%d, optimizedWork=%d%n", + name, + baselineNanos / 1_000_000.0, + optimizedNanos / 1_000_000.0, + baselineNanos / Math.max(1.0, optimizedNanos), + baseline[0].work, + optimized[0].work); + } + + private static double medianNanos(final Sample[] samples) { + final long[] values = new long[samples.length]; + for (int index = 0; index < samples.length; index++) { + values[index] = samples[index].elapsedNanos; + } + Arrays.sort(values); + final int middle = values.length / 2; + return values.length % 2 == 0 + ? values[middle - 1] + (values[middle] - values[middle - 1]) / 2.0 + : values[middle]; + } + + private static final class Sample { + private final long elapsedNanos; + private final long work; + + private Sample(final long elapsedNanos, final long work) { + this.elapsedNanos = elapsedNanos; + this.work = work; + } + } +} diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/metric/SubscriptionMetricsRestartTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/metric/SubscriptionMetricsRestartTest.java index a2142ad61fd7..68eee16ee2f4 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/metric/SubscriptionMetricsRestartTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/metric/SubscriptionMetricsRestartTest.java @@ -24,6 +24,8 @@ import org.apache.iotdb.commons.service.metric.enums.Metric; import org.apache.iotdb.db.subscription.broker.SubscriptionPrefetchingQueue; import org.apache.iotdb.db.subscription.broker.consensus.ConsensusPrefetchingQueue; +import org.apache.iotdb.db.subscription.resource.SubscriptionDataNodeResourceManager; +import org.apache.iotdb.db.subscription.resource.SubscriptionMemoryManager; import org.apache.iotdb.metrics.config.MetricConfig; import org.apache.iotdb.metrics.config.MetricConfigDescriptor; import org.apache.iotdb.metrics.metricsets.IMetricSet; @@ -38,12 +40,15 @@ import org.junit.Test; import org.mockito.Mockito; +import java.lang.reflect.Field; import java.util.ArrayList; import java.util.List; import java.util.Map; import java.util.stream.Collectors; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; import static org.mockito.Mockito.when; /** Subscription metrics must survive a metric service restart, which drops all metrics. */ @@ -96,6 +101,52 @@ public void testPrefetchingQueueMetrics() { assertEquals(0, count(Metric.SUBSCRIPTION_EVENT_TRANSFER)); } + @Test + public void testSubscriptionMemoryMetrics() throws Exception { + final SubscriptionMetrics metrics = SubscriptionMetrics.getInstance(); + // Subscription can be disabled in the default test configuration. Give the shared allocator + // an explicit test budget and restore its original block after removing the metrics. + final long budget = 10L; + final SubscriptionMemoryManager memoryManager = SubscriptionDataNodeResourceManager.memory(); + final long originalOversizedEntryCount = memoryManager.getOversizedEntryCount(); + final Field blockField = SubscriptionMemoryManager.class.getDeclaredField("memoryBlock"); + blockField.setAccessible(true); + final Object originalBlock = blockField.get(memoryManager); + try { + blockField.set(memoryManager, blockField.get(new SubscriptionMemoryManager(budget))); + bind(metrics); + assertTrue(memoryManager.tryAllocate(budget + 1L)); + try { + service.restartService(); + assertEquals( + budget, ((AutoGauge) get(Metric.SUBSCRIPTION_MEMORY_LIMIT_BYTES)).getValue(), 0); + assertEquals( + budget + 1L, ((AutoGauge) get(Metric.SUBSCRIPTION_MEMORY_USED_BYTES)).getValue(), 0); + assertEquals( + 1, ((AutoGauge) get(Metric.SUBSCRIPTION_MEMORY_OVERCOMMIT_BYTES)).getValue(), 0); + assertEquals( + originalOversizedEntryCount + 1L, + ((AutoGauge) get(Metric.SUBSCRIPTION_MEMORY_OVERSIZED_ENTRY_COUNT)).getValue(), + 0); + assertFalse(memoryManager.tryAllocate(1L)); + } finally { + memoryManager.release(budget + 1L); + } + assertEquals(0, ((AutoGauge) get(Metric.SUBSCRIPTION_MEMORY_USED_BYTES)).getValue(), 0); + assertEquals(0, ((AutoGauge) get(Metric.SUBSCRIPTION_MEMORY_OVERCOMMIT_BYTES)).getValue(), 0); + service.removeMetricSet(metrics); + boundMetricSets.remove(metrics); + assertEquals(0, count(Metric.SUBSCRIPTION_MEMORY_USED_BYTES)); + assertEquals(0, count(Metric.SUBSCRIPTION_MEMORY_LIMIT_BYTES)); + assertEquals(0, count(Metric.SUBSCRIPTION_MEMORY_OVERCOMMIT_BYTES)); + assertEquals(0, count(Metric.SUBSCRIPTION_MEMORY_OVERSIZED_ENTRY_COUNT)); + } finally { + service.removeMetricSet(metrics); + boundMetricSets.remove(metrics); + blockField.set(memoryManager, originalBlock); + } + } + @Test public void testConsensusPrefetchingQueueMetrics() { final ConsensusSubscriptionPrefetchingQueueMetrics metrics = @@ -105,16 +156,52 @@ public void testConsensusPrefetchingQueueMetrics() { when(queue.getPrefetchingQueueId()).thenReturn(QUEUE_ID); when(queue.getConsensusGroupId()).thenReturn(new DataRegionId(1)); when(queue.getLag()).thenReturn(7L); + when(queue.getLastDeliveryIntervalMs()).thenReturn(60_001L); + when(queue.getMaxDeliveryIntervalMs()).thenReturn(65_000L); + when(queue.getDeliveryIdleTimeMs()).thenReturn(60_000L); + when(queue.getPrefetchDurationMs()).thenReturn(65_000L); + when(queue.getMaxPrefetchDurationMs()).thenReturn(69_000L); + when(queue.getPrefetchIdleTimeMs()).thenReturn(67_000L); metrics.register(queue); try { service.restartService(); assertEquals(7, ((AutoGauge) get(Metric.SUBSCRIPTION_CONSENSUS_LAG)).getValue(), 0); + assertEquals( + 60_001, + ((AutoGauge) get(Metric.SUBSCRIPTION_CONSENSUS_LAST_DELIVERY_INTERVAL_MS)).getValue(), + 0); + assertEquals( + 65_000, + ((AutoGauge) get(Metric.SUBSCRIPTION_CONSENSUS_MAX_DELIVERY_INTERVAL_MS)).getValue(), + 0); + assertEquals( + 60_000, + ((AutoGauge) get(Metric.SUBSCRIPTION_CONSENSUS_DELIVERY_IDLE_TIME_MS)).getValue(), + 0); + assertEquals( + 65_000, + ((AutoGauge) get(Metric.SUBSCRIPTION_CONSENSUS_PREFETCH_DURATION_MS)).getValue(), + 0); + assertEquals( + 69_000, + ((AutoGauge) get(Metric.SUBSCRIPTION_CONSENSUS_MAX_PREFETCH_DURATION_MS)).getValue(), + 0); + assertEquals( + 67_000, + ((AutoGauge) get(Metric.SUBSCRIPTION_CONSENSUS_PREFETCH_IDLE_TIME_MS)).getValue(), + 0); metrics.mark(QUEUE_ID, new DataRegionId(1).toString(), 5); assertEquals(5, ((Rate) get(Metric.SUBSCRIPTION_EVENT_TRANSFER)).getCount()); } finally { metrics.deregister(queue); } assertEquals(0, count(Metric.SUBSCRIPTION_CONSENSUS_LAG)); + assertEquals(0, count(Metric.SUBSCRIPTION_CONSENSUS_LAST_DELIVERY_INTERVAL_MS)); + assertEquals(0, count(Metric.SUBSCRIPTION_CONSENSUS_MAX_DELIVERY_INTERVAL_MS)); + assertEquals(0, count(Metric.SUBSCRIPTION_CONSENSUS_DELIVERY_IDLE_TIME_MS)); + assertEquals(0, count(Metric.SUBSCRIPTION_CONSENSUS_PREFETCH_DURATION_MS)); + assertEquals(0, count(Metric.SUBSCRIPTION_CONSENSUS_MAX_PREFETCH_DURATION_MS)); + assertEquals(0, count(Metric.SUBSCRIPTION_CONSENSUS_PREFETCH_IDLE_TIME_MS)); assertEquals(0, count(Metric.SUBSCRIPTION_EVENT_TRANSFER)); } diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/resource/SubscriptionMemoryManagerTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/resource/SubscriptionMemoryManagerTest.java index beda24e623bf..31129d357bb3 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/resource/SubscriptionMemoryManagerTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/resource/SubscriptionMemoryManagerTest.java @@ -35,6 +35,8 @@ public void testAllocateAndReleaseWithinBudget() { assertFalse(memoryManager.tryAllocate(5L)); assertEquals(6L, memoryManager.getUsedMemorySizeInBytes()); assertEquals(4L, memoryManager.getFreeMemorySizeInBytes()); + assertEquals(0L, memoryManager.getOvercommitSizeInBytes()); + assertEquals(0L, memoryManager.getOversizedEntryCount()); memoryManager.release(6L); assertEquals(0L, memoryManager.getUsedMemorySizeInBytes()); @@ -48,10 +50,19 @@ public void testAllowOnlyOneOversizedEntryWhenBudgetIsEmpty() { assertTrue(memoryManager.tryAllocate(11L)); assertFalse(memoryManager.tryAllocate(1L)); assertEquals(11L, memoryManager.getUsedMemorySizeInBytes()); + assertEquals(1L, memoryManager.getOvercommitSizeInBytes()); + assertEquals(1L, memoryManager.getOversizedEntryCount()); memoryManager.release(11L); assertEquals(0L, memoryManager.getUsedMemorySizeInBytes()); + assertEquals(0L, memoryManager.getOvercommitSizeInBytes()); assertTrue(memoryManager.tryAllocate(10L)); + assertFalse(memoryManager.tryAllocate(11L)); + assertEquals(1L, memoryManager.getOversizedEntryCount()); + memoryManager.release(10L); + assertTrue(memoryManager.tryAllocate(12L)); + assertEquals(2L, memoryManager.getOvercommitSizeInBytes()); + assertEquals(2L, memoryManager.getOversizedEntryCount()); } @Test @@ -60,6 +71,7 @@ public void testZeroBudgetRejectsAllocation() { assertFalse(memoryManager.tryAllocate(1L)); assertEquals(0L, memoryManager.getUsedMemorySizeInBytes()); + assertEquals(0L, memoryManager.getOversizedEntryCount()); } @Test @@ -133,6 +145,8 @@ public void testOversizedQueueEntryIsExplicitlyRejectedEvenWhenNodeIsEmpty() { SubscriptionMemoryManager.AllocationRejectionReason.OVERSIZED_ENTRY, queue.tryAllocate(11L).getRejectionReason()); assertEquals(0L, memoryManager.getUsedMemorySizeInBytes()); + assertEquals(0L, memoryManager.getOvercommitSizeInBytes()); + assertEquals(0L, memoryManager.getOversizedEntryCount()); assertTrue(queue.tryAllocate(10L).isAccepted()); } assertEquals(0L, memoryManager.getUsedMemorySizeInBytes()); 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 e20919801e42..e6569beb7e6f 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 @@ -233,7 +233,20 @@ public enum Metric { SUBSCRIPTION_CONSENSUS_PENDING_EVENT_COUNT("subscription_consensus_pending_event_count"), SUBSCRIPTION_CONSENSUS_LAST_POLL_TIME("subscription_consensus_last_poll_time"), SUBSCRIPTION_CONSENSUS_LAST_PROGRESS_TIME("subscription_consensus_last_progress_time"), + SUBSCRIPTION_CONSENSUS_LAST_DELIVERY_INTERVAL_MS( + "subscription_consensus_last_delivery_interval_ms"), + SUBSCRIPTION_CONSENSUS_MAX_DELIVERY_INTERVAL_MS( + "subscription_consensus_max_delivery_interval_ms"), + SUBSCRIPTION_CONSENSUS_DELIVERY_IDLE_TIME_MS("subscription_consensus_delivery_idle_time_ms"), + SUBSCRIPTION_CONSENSUS_PREFETCH_DURATION_MS("subscription_consensus_prefetch_duration_ms"), + SUBSCRIPTION_CONSENSUS_MAX_PREFETCH_DURATION_MS( + "subscription_consensus_max_prefetch_duration_ms"), + SUBSCRIPTION_CONSENSUS_PREFETCH_IDLE_TIME_MS("subscription_consensus_prefetch_idle_time_ms"), SUBSCRIPTION_CONSENSUS_PROGRESS_STATUS("subscription_consensus_progress_status"), + SUBSCRIPTION_MEMORY_USED_BYTES("subscription_memory_used_bytes"), + SUBSCRIPTION_MEMORY_LIMIT_BYTES("subscription_memory_limit_bytes"), + SUBSCRIPTION_MEMORY_OVERCOMMIT_BYTES("subscription_memory_overcommit_bytes"), + SUBSCRIPTION_MEMORY_OVERSIZED_ENTRY_COUNT("subscription_memory_oversized_entry_count"), SUBSCRIPTION_CONSENSUS_MEMORY_USED("subscription_consensus_memory_used"), SUBSCRIPTION_CONSENSUS_MEMORY_QUOTA("subscription_consensus_memory_quota"), SUBSCRIPTION_CONSENSUS_MEMORY_MAXIMUM("subscription_consensus_memory_maximum"),