diff --git a/iotdb-core/consensus/src/main/i18n/en/org/apache/iotdb/consensus/i18n/IoTConsensusMessages.java b/iotdb-core/consensus/src/main/i18n/en/org/apache/iotdb/consensus/i18n/IoTConsensusMessages.java index 7f6efd07aebc5..6c1e4c1dc17c5 100644 --- a/iotdb-core/consensus/src/main/i18n/en/org/apache/iotdb/consensus/i18n/IoTConsensusMessages.java +++ b/iotdb-core/consensus/src/main/i18n/en/org/apache/iotdb/consensus/i18n/IoTConsensusMessages.java @@ -330,6 +330,10 @@ private IoTConsensusMessages() {} public static final String LOG_SUBSCRIPTION_QUEUE_FULL_DROPPED_ARG_ENTRY_S_LAST_ARG_MS_2AD8AB3D = "Subscription queue full, dropped {} entry(s) in the last {} ms, latest "; public static final String LOG_SEARCHINDEX_ARG_QUEUESIZE_ARG_QUEUEREMAINING_ARG_2EA619ED = "searchIndex={}, queueSize={}, queueRemaining={}"; public static final String LOG_SUBSCRIPTION_QUEUE_FULL_DROPPED_ENTRY_SEARCHINDEX_ARG_DROPPEDCOUNT_ARG_61F126B8 = "Subscription queue full, dropped entry searchIndex={}, droppedCount={}"; + public static final String LOG_SUBSCRIPTION_REALTIME_ADMISSION_REJECTED_ARG_ENTRY_S_IN_THE_LAST_ARG_MS_WAL_REPLAY_REQUIRED_GROUP_ARG_LATEST_SEARCHINDEX_ARG_REASONCODE_ARG_QUEUESIZE_ARG_QUEUEREMAINING_ARG_0FBF6226 = + "Subscription realtime admission rejected {} entry(s) in the last {} ms; WAL replay required, group={}, latest searchIndex={}, reasonCode={}, queueSize={}, queueRemaining={}"; + public static final String LOG_SUBSCRIPTION_REALTIME_ADMISSION_REJECTED_ENTRY_WAL_REPLAY_REQUIRED_GROUP_ARG_SEARCHINDEX_ARG_REASONCODE_ARG_REJECTEDCOUNT_ARG_7F76D6A9 = + "Subscription realtime admission rejected entry; WAL replay required, group={}, searchIndex={}, reasonCode={}, rejectedCount={}"; public static final String LOG_RESERVED_ARG_BYTES_BATCH_ARG_ARG_CURRENT_TOTAL_USAGE_ARG_308AE9C2 = "Reserved {} bytes for batch {}-{}, current total usage {}"; public static final String LOG_ARG_FAILED_SEND_IDLE_WRITER_SAFE_TIME_BARRIER_ARG_STATUS_AE047EAD = "{}: Failed to send idle writer safe-time barrier to {}. status={}"; public static final String LOG_ARG_WRITE_OPERATION_FAILED_SEARCHINDEX_ARG_CODE_ARG_SUBSCRIPTIONQUEUES_ARG_THIS_ARG_F4B17576 = "{}: write operation failed. searchIndex: {}. Code: {}, subscriptionQueues: {}, this: {}"; diff --git a/iotdb-core/consensus/src/main/i18n/zh/org/apache/iotdb/consensus/i18n/IoTConsensusMessages.java b/iotdb-core/consensus/src/main/i18n/zh/org/apache/iotdb/consensus/i18n/IoTConsensusMessages.java index 7cc8ac622d194..a7ef49bf46741 100644 --- a/iotdb-core/consensus/src/main/i18n/zh/org/apache/iotdb/consensus/i18n/IoTConsensusMessages.java +++ b/iotdb-core/consensus/src/main/i18n/zh/org/apache/iotdb/consensus/i18n/IoTConsensusMessages.java @@ -328,6 +328,10 @@ private IoTConsensusMessages() {} public static final String LOG_SUBSCRIPTION_QUEUE_FULL_DROPPED_ARG_ENTRY_S_LAST_ARG_MS_2AD8AB3D = "订阅队列已满，丢弃了 {} 个 entry，最近 {} ms，最新 "; public static final String LOG_SEARCHINDEX_ARG_QUEUESIZE_ARG_QUEUEREMAINING_ARG_2EA619ED = "searchIndex={}, queueSize={}, queueRemaining={}"; public static final String LOG_SUBSCRIPTION_QUEUE_FULL_DROPPED_ENTRY_SEARCHINDEX_ARG_DROPPEDCOUNT_ARG_61F126B8 = "订阅队列已满，丢弃 entry，searchIndex={}，droppedCount={}"; + public static final String LOG_SUBSCRIPTION_REALTIME_ADMISSION_REJECTED_ARG_ENTRY_S_IN_THE_LAST_ARG_MS_WAL_REPLAY_REQUIRED_GROUP_ARG_LATEST_SEARCHINDEX_ARG_REASONCODE_ARG_QUEUESIZE_ARG_QUEUEREMAINING_ARG_0FBF6226 = + "订阅实时准入拒绝了 {} 个 entry（最近 {} ms）；需通过 WAL 回放补齐，group={}，最新 searchIndex={}，reasonCode={}，queueSize={}，queueRemaining={}"; + public static final String LOG_SUBSCRIPTION_REALTIME_ADMISSION_REJECTED_ENTRY_WAL_REPLAY_REQUIRED_GROUP_ARG_SEARCHINDEX_ARG_REASONCODE_ARG_REJECTEDCOUNT_ARG_7F76D6A9 = + "订阅实时准入拒绝 entry；需通过 WAL 回放补齐，group={}，searchIndex={}，reasonCode={}，rejectedCount={}"; public static final String LOG_RESERVED_ARG_BYTES_BATCH_ARG_ARG_CURRENT_TOTAL_USAGE_ARG_308AE9C2 = "预留 {} 字节给批次 {}-{}，当前总使用量 {}"; public static final String LOG_ARG_FAILED_SEND_IDLE_WRITER_SAFE_TIME_BARRIER_ARG_STATUS_AE047EAD = "{}：无法向 {} 发送 idle writer safe-time barrier。状态={}"; public static final String LOG_ARG_WRITE_OPERATION_FAILED_SEARCHINDEX_ARG_CODE_ARG_SUBSCRIPTIONQUEUES_ARG_THIS_ARG_F4B17576 = "{}：写入操作失败。searchIndex: {}。Code: {}，订阅队列：{}，当前对象：{}"; diff --git a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/subscription/SubscriptionQueueAdmission.java b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/subscription/SubscriptionQueueAdmission.java new file mode 100644 index 0000000000000..18c43fd7e168a --- /dev/null +++ b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/subscription/SubscriptionQueueAdmission.java @@ -0,0 +1,25 @@ +/* + * 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.consensus.iot.subscription; + +/** Exposes the reason for the calling thread's most recent subscription queue offer rejection. */ +public interface SubscriptionQueueAdmission { + + SubscriptionQueueRejectionReason getLastRejectionReason(); +} diff --git a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/subscription/SubscriptionQueueRegistry.java b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/subscription/SubscriptionQueueRegistry.java index d63e42f12c98d..0cc06eeff5278 100644 --- a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/subscription/SubscriptionQueueRegistry.java +++ b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/subscription/SubscriptionQueueRegistry.java @@ -32,20 +32,22 @@ import java.util.concurrent.BlockingQueue; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.TimeUnit; -import java.util.concurrent.atomic.AtomicLong; import java.util.function.LongSupplier; public class SubscriptionQueueRegistry { private static final Logger LOGGER = LoggerFactory.getLogger(SubscriptionQueueRegistry.class); - private static final long QUEUE_FULL_LOG_INTERVAL_MS = TimeUnit.SECONDS.toMillis(10); + private static final long REJECTION_LOG_INTERVAL_MS = TimeUnit.SECONDS.toMillis(10); private final String consensusGroupId; private final Map, SubscriptionQueueRegistration> queues = new ConcurrentHashMap<>(); - private final AtomicLong droppedEntries = new AtomicLong(); - private final AtomicLong lastDropLogTimeMs = new AtomicLong(); + // offer is synchronized, so these counts remain scoped to one reason without extra atomics. + private final long[] rejectedEntriesByReason = + new long[SubscriptionQueueRejectionReason.values().length]; + private final long[] lastRejectionLogTimeMsByReason = + new long[SubscriptionQueueRejectionReason.values().length]; public SubscriptionQueueRegistry(final String consensusGroupId) { this.consensusGroupId = consensusGroupId; @@ -141,27 +143,34 @@ public synchronized boolean offer(final IndexedConsensusRequest indexedConsensus queue.remainingCapacity()); } if (!offered) { - final long droppedCount = droppedEntries.incrementAndGet(); + final SubscriptionQueueRejectionReason rejectionReason = + queue instanceof SubscriptionQueueAdmission + ? ((SubscriptionQueueAdmission) queue).getLastRejectionReason() + : SubscriptionQueueRejectionReason.QUEUE_CAPACITY; + final int reasonIndex = rejectionReason.ordinal(); + final long rejectedCount = ++rejectedEntriesByReason[reasonIndex]; final long now = System.currentTimeMillis(); - final long lastLogTime = lastDropLogTimeMs.get(); - if (now - lastLogTime >= QUEUE_FULL_LOG_INTERVAL_MS - && lastDropLogTimeMs.compareAndSet(lastLogTime, now)) { + if (now - lastRejectionLogTimeMsByReason[reasonIndex] >= REJECTION_LOG_INTERVAL_MS) { + lastRejectionLogTimeMsByReason[reasonIndex] = now; + rejectedEntriesByReason[reasonIndex] = 0L; LOGGER.warn( IoTConsensusMessages - .LOG_SUBSCRIPTION_QUEUE_FULL_DROPPED_ARG_ENTRY_S_LAST_ARG_MS_2AD8AB3D - + IoTConsensusMessages - .LOG_SEARCHINDEX_ARG_QUEUESIZE_ARG_QUEUEREMAINING_ARG_2EA619ED, - droppedEntries.getAndSet(0), - QUEUE_FULL_LOG_INTERVAL_MS, + .LOG_SUBSCRIPTION_REALTIME_ADMISSION_REJECTED_ARG_ENTRY_S_IN_THE_LAST_ARG_MS_WAL_REPLAY_REQUIRED_GROUP_ARG_LATEST_SEARCHINDEX_ARG_REASONCODE_ARG_QUEUESIZE_ARG_QUEUEREMAINING_ARG_0FBF6226, + rejectedCount, + REJECTION_LOG_INTERVAL_MS, + consensusGroupId, indexedConsensusRequest.getSearchIndex(), + rejectionReason.getCode(), queue.size(), queue.remainingCapacity()); } else if (LOGGER.isDebugEnabled()) { LOGGER.debug( IoTConsensusMessages - .LOG_SUBSCRIPTION_QUEUE_FULL_DROPPED_ENTRY_SEARCHINDEX_ARG_DROPPEDCOUNT_ARG_61F126B8, + .LOG_SUBSCRIPTION_REALTIME_ADMISSION_REJECTED_ENTRY_WAL_REPLAY_REQUIRED_GROUP_ARG_SEARCHINDEX_ARG_REASONCODE_ARG_REJECTEDCOUNT_ARG_7F76D6A9, + consensusGroupId, indexedConsensusRequest.getSearchIndex(), - droppedCount); + rejectionReason.getCode(), + rejectedCount); } } } diff --git a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/subscription/SubscriptionQueueRejectionReason.java b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/subscription/SubscriptionQueueRejectionReason.java new file mode 100644 index 0000000000000..852bc55e2f62f --- /dev/null +++ b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/subscription/SubscriptionQueueRejectionReason.java @@ -0,0 +1,42 @@ +/* + * 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.consensus.iot.subscription; + +/** Stable error codes used when a consensus subscription queue rejects realtime input. */ +public enum SubscriptionQueueRejectionReason { + NONE("NONE"), + QUEUE_CAPACITY("SUBSCRIPTION_QUEUE_CAPACITY"), + SUBSCRIPTION_MEMORY_QUOTA("SUBSCRIPTION_MEMORY_QUOTA"), + SUBSCRIPTION_MEMORY_LIMIT("SUBSCRIPTION_MEMORY_LIMIT"), + SUBSCRIPTION_OVERSIZED_ENTRY("SUBSCRIPTION_OVERSIZED_ENTRY"), + WRITER_BACKLOG("SUBSCRIPTION_WRITER_BACKLOG"), + SEEK_IN_PROGRESS("SUBSCRIPTION_SEEK_IN_PROGRESS"), + CONSENSUS_REQUEST_MEMORY_LIMIT("CONSENSUS_REQUEST_MEMORY_LIMIT"), + INACTIVE_OR_CLOSED("SUBSCRIPTION_QUEUE_INACTIVE_OR_CLOSED"); + + private final String code; + + SubscriptionQueueRejectionReason(final String code) { + this.code = code; + } + + public String getCode() { + return code; + } +} 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 1c0362b627b3a..0404be4ae16c6 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 @@ -2645,4 +2645,6 @@ private DataNodePipeMessages() {} "Interrupted while reading OPC UA server operation limits, use defaults: maxNodesPerWrite={}, maxNodesPerNodeManagement={}"; public static final String LOG_FAILED_TO_READ_OPC_UA_SERVER_OPERATION_LIMITS_USE_DEFAULTS_MAXNODESPERWRITE_ARG_MAXNODESPERNODEMANAGEMENT_ARG_65460871 = "Failed to read OPC UA server operation limits, use defaults: maxNodesPerWrite={}, maxNodesPerNodeManagement={}"; + public static final String MESSAGE_ARG_SUBSCRIPTION_ENTRY_REQUIRES_ARG_BYTES_ABOVE_THE_CURRENT_PER_QUEUE_MAXIMUM_ARG_BYTES_DATANODE_BUDGET_ARG_BYTES_REDUCE_THE_WRITE_BATCH_FIELD_SIZE_OR_INCREASE_SUBSCRIPTION_MATERIALIZATION_MEMORY_WAL_PROGRESS_HAS_NOT_ADVANCED_AFCBC7FC = + "[%s] Subscription entry requires %d bytes, above the current per-queue maximum %d bytes (DataNode budget %d bytes). Reduce the write batch/field size or increase subscription materialization memory. WAL progress has not advanced."; } 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 49096659ade1b..2dd0af59e9fa1 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 @@ -2469,4 +2469,6 @@ private DataNodePipeMessages() {} "读取 OPC UA 服务器操作限制时被中断，使用默认值：maxNodesPerWrite={}，maxNodesPerNodeManagement={}"; public static final String LOG_FAILED_TO_READ_OPC_UA_SERVER_OPERATION_LIMITS_USE_DEFAULTS_MAXNODESPERWRITE_ARG_MAXNODESPERNODEMANAGEMENT_ARG_65460871 = "读取 OPC UA 服务器操作限制失败，使用默认值：maxNodesPerWrite={}，maxNodesPerNodeManagement={}"; + public static final String MESSAGE_ARG_SUBSCRIPTION_ENTRY_REQUIRES_ARG_BYTES_ABOVE_THE_CURRENT_PER_QUEUE_MAXIMUM_ARG_BYTES_DATANODE_BUDGET_ARG_BYTES_REDUCE_THE_WRITE_BATCH_FIELD_SIZE_OR_INCREASE_SUBSCRIPTION_MATERIALIZATION_MEMORY_WAL_PROGRESS_HAS_NOT_ADVANCED_AFCBC7FC = + "[%s] 订阅 entry 需要 %d 字节，超过当前队列可用上限 %d 字节（DataNode 预算 %d 字节）。请减小写入 batch/字段大小或增加订阅物化内存。WAL 进度未推进。"; } 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 b692ec94ed046..921892f200fbd 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 @@ -30,6 +30,8 @@ import org.apache.iotdb.consensus.iot.SubscriptionWalRetentionPolicy; import org.apache.iotdb.consensus.iot.log.ConsensusReqReader; import org.apache.iotdb.consensus.iot.logdispatcher.IoTConsensusMemoryManager; +import org.apache.iotdb.consensus.iot.subscription.SubscriptionQueueAdmission; +import org.apache.iotdb.consensus.iot.subscription.SubscriptionQueueRejectionReason; import org.apache.iotdb.db.conf.IoTDBDescriptor; import org.apache.iotdb.db.i18n.DataNodeMiscMessages; import org.apache.iotdb.db.i18n.DataNodePipeMessages; @@ -92,10 +94,11 @@ import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicLong; +import java.util.concurrent.atomic.AtomicLongArray; import java.util.concurrent.atomic.AtomicReference; import java.util.concurrent.locks.ReentrantReadWriteLock; import java.util.function.BiConsumer; -import java.util.function.BooleanSupplier; +import java.util.function.Consumer; import java.util.function.Function; import java.util.function.Predicate; import java.util.function.Supplier; @@ -151,6 +154,27 @@ public class ConsensusPrefetchingQueue { private SubscriptionMemoryManager subscriptionMemoryManager; + /** Queue-local view of the DataNode subscription memory budget. */ + private SubscriptionMemoryManager.MemoryHandle subscriptionMemoryHandle; + + private final AtomicLong realtimeAdmissionRejectionCount = new AtomicLong(0L); + + private final AtomicLongArray realtimeAdmissionRejectionsByReason = + new AtomicLongArray(SubscriptionQueueRejectionReason.values().length); + + private final AtomicLong subscriptionMemoryRejectionCount = new AtomicLong(0L); + + private final AtomicLong oversizedEntryRejectionCount = new AtomicLong(0L); + + private final AtomicReference lastRejectionReason = + new AtomicReference<>(SubscriptionQueueRejectionReason.NONE); + + private volatile SubscriptionQueueRejectionReason realtimeAdmissionBlockReason = + SubscriptionQueueRejectionReason.NONE; + + private volatile SubscriptionQueueRejectionReason memoryBlockReason = + SubscriptionQueueRejectionReason.NONE; + /** * Incremented on each seek to distinguish batches/events created before and after the seek. The * value is copied into commit contexts so stale events cannot be committed after a later seek. @@ -415,30 +439,48 @@ protected String getDetail() { } private static final class WakeableIndexedConsensusQueue - extends LinkedBlockingDeque { + extends LinkedBlockingDeque implements SubscriptionQueueAdmission { private final Runnable wakeupHook; - private final BooleanSupplier admissionSupplier; + private final Supplier admissionSupplier; + private final Consumer rejectionListener; private final IoTConsensusMemoryManager requestMemoryManager = IoTConsensusMemoryManager.getInstance(); private final AtomicLong retainedRequestBytes = new AtomicLong(0L); + // The registry reads this immediately after offer on the same writer thread. A queue-wide + // value could be overwritten by another writer before the registry logs the rejection. + private final ThreadLocal lastOfferRejectionReason = + ThreadLocal.withInitial(() -> SubscriptionQueueRejectionReason.NONE); private WakeableIndexedConsensusQueue( - final int capacity, final Runnable wakeupHook, final BooleanSupplier admissionSupplier) { + final int capacity, + final Runnable wakeupHook, + final Supplier admissionSupplier, + final Consumer rejectionListener) { super(capacity); this.wakeupHook = wakeupHook; this.admissionSupplier = admissionSupplier; + this.rejectionListener = rejectionListener; + } + + private boolean reject(final SubscriptionQueueRejectionReason reason) { + final SubscriptionQueueRejectionReason effectiveReason = + Objects.isNull(reason) ? SubscriptionQueueRejectionReason.QUEUE_CAPACITY : reason; + lastOfferRejectionReason.set(effectiveReason); + rejectionListener.accept(effectiveReason); + return false; } @Override public boolean offer(final IndexedConsensusRequest request) { final boolean offered; synchronized (this) { - if (!admissionSupplier.getAsBoolean()) { - return false; + final SubscriptionQueueRejectionReason admissionReason = admissionSupplier.get(); + if (admissionReason != SubscriptionQueueRejectionReason.NONE) { + return reject(admissionReason); } if (!requestMemoryManager.reserve(request)) { - return false; + return reject(SubscriptionQueueRejectionReason.CONSENSUS_REQUEST_MEMORY_LIMIT); } try { offered = super.offer(request); @@ -450,6 +492,7 @@ public boolean offer(final IndexedConsensusRequest request) { retainedRequestBytes.addAndGet(request.getRetainedMemorySize()); } else { requestMemoryManager.free(request); + reject(SubscriptionQueueRejectionReason.QUEUE_CAPACITY); } } if (offered) { @@ -478,6 +521,11 @@ private void release(final List requests) { private long getRetainedRequestBytes() { return retainedRequestBytes.get(); } + + @Override + public SubscriptionQueueRejectionReason getLastRejectionReason() { + return lastOfferRejectionReason.get(); + } } private static final class PendingSeekRequest { @@ -559,6 +607,8 @@ public ConsensusPrefetchingQueue( this.tableModel = converter.isTableModel(); this.commitManager = commitManager; this.subscriptionMemoryManager = SubscriptionDataNodeResourceManager.memory(); + this.subscriptionMemoryHandle = subscriptionMemoryManager.registerQueue(); + this.subscriptionMemoryHandle.setActive(initialActive); this.fallbackCommittedRegionProgress = fallbackCommittedRegionProgress; this.fallbackTailSearchIndex = tailStartSearchIndex; this.runtimeVersion = initialRuntimeVersion; @@ -577,7 +627,8 @@ public ConsensusPrefetchingQueue( new WakeableIndexedConsensusQueue( PENDING_QUEUE_CAPACITY, this::requestPrefetchForRealtimeEntry, - this::canAcceptRealtimeEntry); + this::getRealtimeAdmissionRejectionReason, + this::recordRealtimeAdmissionRejection); serverImpl.registerSubscriptionQueue( pendingEntries, retentionPolicy, this::getRequiredRetainedMinVersionId); @@ -602,7 +653,12 @@ void setSubscriptionMemoryManager(final SubscriptionMemoryManager subscriptionMe if (retainedTabletBytes.get() != 0L) { throw new IllegalStateException(); } + if (Objects.nonNull(this.subscriptionMemoryHandle)) { + this.subscriptionMemoryHandle.close(); + } this.subscriptionMemoryManager = Objects.requireNonNull(subscriptionMemoryManager); + this.subscriptionMemoryHandle = this.subscriptionMemoryManager.registerQueue(); + this.subscriptionMemoryHandle.setActive(isActive); memoryBlockedEntryBytes = -1L; realtimeAdmissionBlocked.set(false); } @@ -641,23 +697,53 @@ private void requestPrefetchForRealtimeEntry() { } } - private boolean canAcceptRealtimeEntry() { - return isActive - && !closeRequested - && !isClosed - && !realtimeAdmissionBlocked.get() - && subscriptionMemoryManager.getFreeMemorySizeInBytes() > 0L; + private SubscriptionQueueRejectionReason getRealtimeAdmissionRejectionReason() { + if (!isActive || closeRequested || isClosed) { + return SubscriptionQueueRejectionReason.INACTIVE_OR_CLOSED; + } + if (realtimeAdmissionBlocked.get()) { + return realtimeAdmissionBlockReason; + } + // Allow an empty queue to probe an entry even with a zero quota so materialization can report + // an explicit oversized-entry error. Pending requests have separate consensus memory + // accounting. + if (subscriptionMemoryHandle.getUsedMemorySizeInBytes() > 0L + && subscriptionMemoryHandle.getFreeMemorySizeInBytes() <= 0L) { + return toQueueRejectionReason(subscriptionMemoryHandle.inspectRejection(1L)); + } + return SubscriptionQueueRejectionReason.NONE; } - private void blockRealtimeAdmission() { + private void blockRealtimeAdmission(final SubscriptionQueueRejectionReason reason) { + realtimeAdmissionBlockReason = reason; realtimeAdmissionBlocked.set(true); - pendingEntries.clear(); } private void unblockRealtimeAdmission() { realtimeAdmissionBlocked.set(false); } + private void recordRealtimeAdmissionRejection( + final SubscriptionQueueRejectionReason rejectionReason) { + if (Objects.isNull(rejectionReason) + || rejectionReason == SubscriptionQueueRejectionReason.NONE) { + return; + } + lastRejectionReason.set(rejectionReason); + realtimeAdmissionRejectionCount.incrementAndGet(); + realtimeAdmissionRejectionsByReason.incrementAndGet(rejectionReason.ordinal()); + recordMemoryRejection(rejectionReason); + } + + private void recordMemoryRejection(final SubscriptionQueueRejectionReason rejectionReason) { + if (rejectionReason == SubscriptionQueueRejectionReason.SUBSCRIPTION_MEMORY_QUOTA + || rejectionReason == SubscriptionQueueRejectionReason.SUBSCRIPTION_MEMORY_LIMIT) { + subscriptionMemoryRejectionCount.incrementAndGet(); + } else if (rejectionReason == SubscriptionQueueRejectionReason.SUBSCRIPTION_OVERSIZED_ENTRY) { + oversizedEntryRejectionCount.incrementAndGet(); + } + } + private ConsensusPrefetchSubtask ensurePrefetchSubtaskBound() { if (closeRequested || isClosed) { return null; @@ -777,6 +863,26 @@ public SubscriptionEvent poll(final String consumerId, final RegionProgress regi } else if (Objects.isNull(event) && shouldRecoverPrefetchBindingAfterEmptyPoll()) { requestPrefetch(); } + // Deliver reserved events first so ACK can drain memory. An entry that cannot fit even in an + // empty queue is an explicit configuration error, with no replay/commit progress advancement. + final long blockedEntryBytes = memoryBlockedEntryBytes; + if (Objects.isNull(event) + && blockedEntryBytes > 0L + && subscriptionMemoryHandle.inspectRejection(blockedEntryBytes) + == SubscriptionMemoryManager.AllocationRejectionReason.OVERSIZED_ENTRY) { + return new SubscriptionEvent( + SubscriptionPollResponseType.ERROR.getType(), + new ErrorPayload( + String.format( + DataNodePipeMessages + .MESSAGE_ARG_SUBSCRIPTION_ENTRY_REQUIRES_ARG_BYTES_ABOVE_THE_CURRENT_PER_QUEUE_MAXIMUM_ARG_BYTES_DATANODE_BUDGET_ARG_BYTES_REDUCE_THE_WRITE_BATCH_FIELD_SIZE_OR_INCREASE_SUBSCRIPTION_MATERIALIZATION_MEMORY_WAL_PROGRESS_HAS_NOT_ADVANCED_AFCBC7FC, + SubscriptionQueueRejectionReason.SUBSCRIPTION_OVERSIZED_ENTRY.getCode(), + blockedEntryBytes, + subscriptionMemoryHandle.getMaximumMemorySizeInBytes(), + subscriptionMemoryManager.getTotalMemorySizeInBytes()), + true), + createNonCommittableContext(IoTDBDescriptor.getInstance().getConfig().getDataNodeId())); + } return event; } finally { releaseReadLock(); @@ -1428,7 +1534,7 @@ public PrefetchRoundResult drivePrefetchOnce() { recycleInFlightEvents(); if (!isActive) { - blockRealtimeAdmission(); + blockRealtimeAdmission(SubscriptionQueueRejectionReason.INACTIVE_OR_CLOSED); return computeIdleRoundResult(); } if (prefetchingQueue.size() >= MAX_PREFETCHING_QUEUE_SIZE) { @@ -1451,12 +1557,12 @@ public PrefetchRoundResult drivePrefetchOnce() { if (prefetchingQueue.size() >= MAX_PREFETCHING_QUEUE_SIZE || !realtimeEntriesByWriter.isEmpty()) { if (!realtimeEntriesByWriter.isEmpty()) { - blockRealtimeAdmission(); + blockRealtimeAdmission(SubscriptionQueueRejectionReason.WRITER_BACKLOG); } return computeIdleRoundResult(); } if (shouldWaitForSubscriptionMemory()) { - blockRealtimeAdmission(); + blockRealtimeAdmission(memoryBlockReason); if (!lingerBatch.isEmpty() && !flushBatch(lingerBatch, observedSeekGeneration)) { resetRoundStateForSeek(seekGeneration.get()); return PrefetchRoundResult.rescheduleNow(); @@ -1501,7 +1607,7 @@ public PrefetchRoundResult drivePrefetchOnce() { if (batchResult == MaterializationResult.MEMORY_BLOCKED) { // Publish already reserved Tablets immediately so ACK can release their memory. The // blocked request and the drained suffix are recovered from WAL on a later round. - blockRealtimeAdmission(); + blockRealtimeAdmission(memoryBlockReason); if (!lingerBatch.isEmpty() && !flushBatch(lingerBatch, observedSeekGeneration)) { resetRoundStateForSeek(seekGeneration.get()); return PrefetchRoundResult.rescheduleNow(); @@ -1516,7 +1622,7 @@ public PrefetchRoundResult drivePrefetchOnce() { if (batch.isEmpty() && lingerBatch.isEmpty()) { final MaterializationResult walResult = tryCatchUpFromWAL(observedSeekGeneration); if (walResult == MaterializationResult.MEMORY_BLOCKED) { - blockRealtimeAdmission(); + blockRealtimeAdmission(memoryBlockReason); return PrefetchRoundResult.rescheduleAfter(MEMORY_RETRY_SLEEP_MS); } if (walResult == MaterializationResult.STALE) { @@ -2321,11 +2427,19 @@ private boolean tryReserveTabletMemory(final long bytes) { memoryBlockedEntryBytes = -1L; return true; } - if (!subscriptionMemoryManager.tryAllocate(bytes)) { + final SubscriptionMemoryManager.AllocationResult allocationResult = + subscriptionMemoryHandle.tryAllocate(bytes); + if (!allocationResult.isAccepted()) { memoryBlockedEntryBytes = bytes; + final SubscriptionQueueRejectionReason rejectionReason = + toQueueRejectionReason(allocationResult.getRejectionReason()); + memoryBlockReason = rejectionReason; + lastRejectionReason.set(rejectionReason); + recordMemoryRejection(rejectionReason); return false; } memoryBlockedEntryBytes = -1L; + memoryBlockReason = SubscriptionQueueRejectionReason.NONE; retainedTabletBytes.addAndGet(bytes); return true; } @@ -2333,22 +2447,21 @@ private boolean tryReserveTabletMemory(final long bytes) { private boolean shouldWaitForSubscriptionMemory() { final long blockedEntryBytes = memoryBlockedEntryBytes; if (blockedEntryBytes <= 0L) { - return subscriptionMemoryManager.getFreeMemorySizeInBytes() <= 0L; - } - - final long totalMemorySizeInBytes = subscriptionMemoryManager.getTotalMemorySizeInBytes(); - final long usedMemorySizeInBytes = subscriptionMemoryManager.getUsedMemorySizeInBytes(); - final boolean canAllocateWithinLimit = - blockedEntryBytes <= totalMemorySizeInBytes - && totalMemorySizeInBytes - usedMemorySizeInBytes >= blockedEntryBytes; - final boolean canAllocateSingleOversizedEntry = - totalMemorySizeInBytes > 0L - && usedMemorySizeInBytes == 0L - && blockedEntryBytes > totalMemorySizeInBytes; - if (canAllocateWithinLimit || canAllocateSingleOversizedEntry) { + if (subscriptionMemoryHandle.getUsedMemorySizeInBytes() > 0L + && subscriptionMemoryHandle.getFreeMemorySizeInBytes() <= 0L) { + memoryBlockReason = toQueueRejectionReason(subscriptionMemoryHandle.inspectRejection(1L)); + return memoryBlockReason != SubscriptionQueueRejectionReason.NONE; + } + return false; + } + final SubscriptionMemoryManager.AllocationRejectionReason rejectionReason = + subscriptionMemoryHandle.inspectRejection(blockedEntryBytes); + if (rejectionReason == SubscriptionMemoryManager.AllocationRejectionReason.NONE) { memoryBlockedEntryBytes = -1L; + memoryBlockReason = SubscriptionQueueRejectionReason.NONE; return false; } + memoryBlockReason = toQueueRejectionReason(rejectionReason); return true; } @@ -2366,7 +2479,22 @@ private void releaseTabletMemory(final long bytes) { releasedBytes = Math.min(bytes, currentRetainedBytes); } while (!retainedTabletBytes.compareAndSet( currentRetainedBytes, currentRetainedBytes - releasedBytes)); - subscriptionMemoryManager.release(releasedBytes); + subscriptionMemoryHandle.release(releasedBytes); + } + + private SubscriptionQueueRejectionReason toQueueRejectionReason( + final SubscriptionMemoryManager.AllocationRejectionReason rejectionReason) { + switch (rejectionReason) { + case NONE: + return SubscriptionQueueRejectionReason.NONE; + case MEMORY_QUOTA: + return SubscriptionQueueRejectionReason.SUBSCRIPTION_MEMORY_QUOTA; + case OVERSIZED_ENTRY: + return SubscriptionQueueRejectionReason.SUBSCRIPTION_OVERSIZED_ENTRY; + case MEMORY_LIMIT: + default: + return SubscriptionQueueRejectionReason.SUBSCRIPTION_MEMORY_LIMIT; + } } private void reconcileRetainedTabletMemoryAfterCleanup() { @@ -3032,7 +3160,7 @@ private void recycleInFlightEvents() { // ======================== Cleanup ======================== public void cleanUp() { - blockRealtimeAdmission(); + blockRealtimeAdmission(SubscriptionQueueRejectionReason.INACTIVE_OR_CLOSED); acquireWriteLock(); try { prefetchingQueue.forEach(event -> cleanUpEvent(event, true)); @@ -3048,7 +3176,8 @@ public void cleanUp() { materializedProgressByWriter.clear(); // A prefetch round that already held the read lock may have reopened admission while this // cleanup was waiting for the write lock. Fence and clear once more under the write lock. - blockRealtimeAdmission(); + blockRealtimeAdmission(SubscriptionQueueRejectionReason.INACTIVE_OR_CLOSED); + pendingEntries.clear(); discardBatch(lingerBatch); reconcileRetainedTabletMemoryAfterCleanup(); memoryBlockedEntryBytes = -1L; @@ -3059,6 +3188,8 @@ public void cleanUp() { pendingSubscriptionWalResetSearchIndex = Long.MIN_VALUE; pendingSubscriptionWalResetGeneration = Long.MIN_VALUE; closeSubscriptionWALIterator(); + subscriptionMemoryHandle.setActive(false); + subscriptionMemoryHandle.close(); } finally { releaseWriteLock(); @@ -3317,7 +3448,7 @@ private void failPendingSeekBeforeScheduling(final PendingSeekRequest request) { } private void applySeekResetUnderWriteLock(final PendingSeekRequest request) { - blockRealtimeAdmission(); + blockRealtimeAdmission(SubscriptionQueueRejectionReason.SEEK_IN_PROGRESS); // 1. Clean up all queued and in-flight events prefetchingQueue.forEach(event -> cleanUpEvent(event, true)); @@ -3327,6 +3458,7 @@ private void applySeekResetUnderWriteLock(final PendingSeekRequest request) { discardRetainedEventMemory(); // 2. Discard stale pending entries from in-memory queue + pendingEntries.clear(); memoryBlockedEntryBytes = -1L; unblockRealtimeAdmission(); @@ -4038,12 +4170,14 @@ private void setActiveUnderRuntimeLock(final boolean active) { if (this.isActive != active) { if (active) { this.isActive = true; + subscriptionMemoryHandle.setActive(true); unblockRealtimeAdmission(); } else { // Fence admission before waiting for the write lock so no new realtime reference can race // with the lifecycle cleanup below. this.isActive = false; - realtimeAdmissionBlocked.set(true); + subscriptionMemoryHandle.setActive(false); + blockRealtimeAdmission(SubscriptionQueueRejectionReason.INACTIVE_OR_CLOSED); acquireWriteLock(); try { seekRequestToFail = pendingSeekRequest; @@ -4394,6 +4528,39 @@ public long getSubscriptionMemoryLimitInBytes() { return subscriptionMemoryManager.getTotalMemorySizeInBytes(); } + public long getSubscriptionMemoryUsedInBytes() { + return subscriptionMemoryHandle.getUsedMemorySizeInBytes(); + } + + public long getSubscriptionMemoryQuotaInBytes() { + return subscriptionMemoryHandle.getMemoryQuotaInBytes(); + } + + public long getSubscriptionMemoryMaximumInBytes() { + return subscriptionMemoryHandle.getMaximumMemorySizeInBytes(); + } + + public long getRealtimeAdmissionRejectionCount() { + return realtimeAdmissionRejectionCount.get(); + } + + public long getRealtimeAdmissionRejectionCount( + final SubscriptionQueueRejectionReason rejectionReason) { + return realtimeAdmissionRejectionsByReason.get(rejectionReason.ordinal()); + } + + public long getSubscriptionMemoryRejectionCount() { + return subscriptionMemoryRejectionCount.get(); + } + + public long getOversizedEntryRejectionCount() { + return oversizedEntryRejectionCount.get(); + } + + public String getLastAdmissionRejectionCode() { + return lastRejectionReason.get().getCode(); + } + /** Exposes the current seek generation for runtime tests and metrics. */ public long getCurrentSeekGeneration() { return seekGeneration.get(); @@ -4471,6 +4638,17 @@ public Map coreReportMessage() { result.put( "subscriptionMemoryLimitInBytes", String.valueOf(subscriptionMemoryManager.getTotalMemorySizeInBytes())); + result.put("subscriptionMemoryUsedInBytes", String.valueOf(getSubscriptionMemoryUsedInBytes())); + result.put( + "subscriptionMemoryQuotaInBytes", String.valueOf(getSubscriptionMemoryQuotaInBytes())); + result.put( + "subscriptionMemoryMaximumInBytes", String.valueOf(getSubscriptionMemoryMaximumInBytes())); + result.put( + "realtimeAdmissionRejectionCount", String.valueOf(getRealtimeAdmissionRejectionCount())); + result.put( + "subscriptionMemoryRejectionCount", String.valueOf(getSubscriptionMemoryRejectionCount())); + result.put("oversizedEntryRejectionCount", String.valueOf(getOversizedEntryRejectionCount())); + result.put("lastAdmissionRejectionCode", getLastAdmissionRejectionCode()); result.put("pendingPathAcceptedEntries", String.valueOf(getPendingPathAcceptedEntries())); result.put("walPathAcceptedEntries", String.valueOf(getWalPathAcceptedEntries())); result.put("seekGeneration", String.valueOf(seekGeneration.get())); 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 19076e9abaeb2..0a549565b06e1 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 @@ -21,6 +21,7 @@ import org.apache.iotdb.commons.service.metric.enums.Metric; import org.apache.iotdb.commons.service.metric.enums.Tag; +import org.apache.iotdb.consensus.iot.subscription.SubscriptionQueueRejectionReason; import org.apache.iotdb.db.i18n.DataNodePipeMessages; import org.apache.iotdb.db.subscription.broker.consensus.ConsensusPrefetchingQueue; import org.apache.iotdb.metrics.AbstractMetricService; @@ -199,6 +200,53 @@ private void createAutoGauge(final QueueMetricsKey key) { queue, ConsensusPrefetchingQueue::getProgressStatus, key.getTags()); + metricService.createAutoGauge( + Metric.SUBSCRIPTION_CONSENSUS_MEMORY_USED.toString(), + MetricLevel.IMPORTANT, + queue, + ConsensusPrefetchingQueue::getSubscriptionMemoryUsedInBytes, + key.getTags()); + metricService.createAutoGauge( + Metric.SUBSCRIPTION_CONSENSUS_MEMORY_QUOTA.toString(), + MetricLevel.IMPORTANT, + queue, + ConsensusPrefetchingQueue::getSubscriptionMemoryQuotaInBytes, + key.getTags()); + metricService.createAutoGauge( + Metric.SUBSCRIPTION_CONSENSUS_MEMORY_MAXIMUM.toString(), + MetricLevel.IMPORTANT, + queue, + ConsensusPrefetchingQueue::getSubscriptionMemoryMaximumInBytes, + key.getTags()); + metricService.createAutoGauge( + Metric.SUBSCRIPTION_CONSENSUS_ADMISSION_REJECTIONS.toString(), + MetricLevel.IMPORTANT, + queue, + ConsensusPrefetchingQueue::getRealtimeAdmissionRejectionCount, + key.getTags()); + for (final SubscriptionQueueRejectionReason reason : + SubscriptionQueueRejectionReason.values()) { + if (reason != SubscriptionQueueRejectionReason.NONE) { + metricService.createAutoGauge( + Metric.SUBSCRIPTION_CONSENSUS_ADMISSION_REJECTIONS_BY_REASON.toString(), + MetricLevel.IMPORTANT, + queue, + q -> q.getRealtimeAdmissionRejectionCount(reason), + key.getTagsWithReason(reason)); + } + } + metricService.createAutoGauge( + Metric.SUBSCRIPTION_CONSENSUS_MEMORY_REJECTIONS.toString(), + MetricLevel.IMPORTANT, + queue, + ConsensusPrefetchingQueue::getSubscriptionMemoryRejectionCount, + key.getTags()); + metricService.createAutoGauge( + Metric.SUBSCRIPTION_CONSENSUS_OVERSIZED_REJECTIONS.toString(), + MetricLevel.IMPORTANT, + queue, + ConsensusPrefetchingQueue::getOversizedEntryRejectionCount, + key.getTags()); } private void createRate(final QueueMetricsKey key) { @@ -309,6 +357,37 @@ private void removeAutoGauge(final QueueMetricsKey key) { MetricType.AUTO_GAUGE, Metric.SUBSCRIPTION_CONSENSUS_PROGRESS_STATUS.toString(), key.getTags()); + metricService.remove( + MetricType.AUTO_GAUGE, Metric.SUBSCRIPTION_CONSENSUS_MEMORY_USED.toString(), key.getTags()); + metricService.remove( + MetricType.AUTO_GAUGE, + Metric.SUBSCRIPTION_CONSENSUS_MEMORY_QUOTA.toString(), + key.getTags()); + metricService.remove( + MetricType.AUTO_GAUGE, + Metric.SUBSCRIPTION_CONSENSUS_MEMORY_MAXIMUM.toString(), + key.getTags()); + metricService.remove( + MetricType.AUTO_GAUGE, + Metric.SUBSCRIPTION_CONSENSUS_ADMISSION_REJECTIONS.toString(), + key.getTags()); + for (final SubscriptionQueueRejectionReason reason : + SubscriptionQueueRejectionReason.values()) { + if (reason != SubscriptionQueueRejectionReason.NONE) { + metricService.remove( + MetricType.AUTO_GAUGE, + Metric.SUBSCRIPTION_CONSENSUS_ADMISSION_REJECTIONS_BY_REASON.toString(), + key.getTagsWithReason(reason)); + } + } + metricService.remove( + MetricType.AUTO_GAUGE, + Metric.SUBSCRIPTION_CONSENSUS_MEMORY_REJECTIONS.toString(), + key.getTags()); + metricService.remove( + MetricType.AUTO_GAUGE, + Metric.SUBSCRIPTION_CONSENSUS_OVERSIZED_REJECTIONS.toString(), + key.getTags()); } private void removeRate(final QueueMetricsKey key) { @@ -357,6 +436,17 @@ private String[] getTags() { }; } + private String[] getTagsWithReason(final SubscriptionQueueRejectionReason reason) { + return new String[] { + Tag.NAME.toString(), + queueId, + Tag.REGION.toString(), + regionId, + Tag.REASON.toString(), + reason.getCode(), + }; + } + @Override public boolean equals(final Object obj) { if (this == obj) { 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 bf71fad6c4b72..d250cd0ba167e 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 @@ -25,11 +25,29 @@ import org.apache.iotdb.commons.utils.TestOnly; import org.apache.iotdb.db.conf.IoTDBDescriptor; +import java.util.HashMap; +import java.util.Map; + +/** + * DataNode-wide memory manager for materialized subscription data. + * + *

Consensus subscription queues use a handle rather than the legacy manager-level allocation + * methods. Handles are accounted independently. Each active queue keeps half of its equal share + * protected; another queue may borrow only the remaining idle space. Borrowed allocations are + * drained rather than revoked when the active queue set changes. An entry larger than the maximum + * amount one queue can hold under this policy is rejected explicitly. + */ public class SubscriptionMemoryManager { private static final String MEMORY_BLOCK_NAME = "Subscription"; private final IMemoryBlock memoryBlock; + private static final int PROTECTED_SHARE_DIVISOR = 2; + private final Map handles = new HashMap<>(); + private long nextHandleId; + + // Kept for callers of the original manager-level API. It is deliberately not a fair-share owner. + private long legacyUsedMemoryInBytes; SubscriptionMemoryManager() { memoryBlock = @@ -46,44 +64,321 @@ public SubscriptionMemoryManager(final long totalMemorySizeInBytes) { MEMORY_BLOCK_NAME, null, totalMemorySizeInBytes, MemoryBlockType.DYNAMIC); } - /** - * Reserves memory for materialized subscription data. - * - *

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 - * consensus entry. - */ + /** Registers a queue for fair-share accounting. */ + public synchronized MemoryHandle registerQueue() { + final MemoryHandle handle = new MemoryHandle(++nextHandleId); + handles.put(handle.id, handle); + return handle; + } + + /** Reserves memory through the compatibility owner. New queues should use a handle. */ public synchronized boolean tryAllocate(final long sizeInBytes) { + return tryAllocateLegacy(sizeInBytes).isAccepted(); + } + + private AllocationResult tryAllocateLegacy(final long sizeInBytes) { if (sizeInBytes <= 0L) { - return true; + return AllocationResult.accepted(getTotalMemorySizeInBytes(), getFreeMemorySizeInBytes()); } - if (memoryBlock.allocate(sizeInBytes)) { - return true; + final long total = getTotalMemorySizeInBytes(); + final long used = getUsedMemorySizeInBytes(); + if (total > 0L && used <= total && sizeInBytes <= total - used) { + if (!memoryBlock.allocate(sizeInBytes)) { + return AllocationResult.rejected( + AllocationRejectionReason.MEMORY_LIMIT, total, getFreeMemorySizeInBytes()); + } + legacyUsedMemoryInBytes += sizeInBytes; + return AllocationResult.accepted(total, total - used - sizeInBytes); } - if (memoryBlock.getTotalMemorySizeInBytes() > 0L - && memoryBlock.getUsedMemoryInBytes() == 0L - && sizeInBytes > memoryBlock.getTotalMemorySizeInBytes()) { + if (total > 0L && used == 0L && sizeInBytes > total) { memoryBlock.forceAllocateWithoutLimitation(sizeInBytes); - return true; + legacyUsedMemoryInBytes += sizeInBytes; + return AllocationResult.accepted(total, total - sizeInBytes); } - return false; + return AllocationResult.rejected( + sizeInBytes > total + ? AllocationRejectionReason.OVERSIZED_ENTRY + : AllocationRejectionReason.MEMORY_LIMIT, + total, + total - used); } public synchronized void release(final long sizeInBytes) { - if (sizeInBytes > 0L) { - memoryBlock.release(sizeInBytes); + if (sizeInBytes <= 0L) { + return; + } + final long released = Math.min(sizeInBytes, legacyUsedMemoryInBytes); + if (released > 0L) { + legacyUsedMemoryInBytes -= released; + memoryBlock.release(released); } } - public long getTotalMemorySizeInBytes() { + public synchronized long getTotalMemorySizeInBytes() { return memoryBlock.getTotalMemorySizeInBytes(); } - public long getUsedMemorySizeInBytes() { + public synchronized long getUsedMemorySizeInBytes() { return memoryBlock.getUsedMemoryInBytes(); } - public long getFreeMemorySizeInBytes() { + public synchronized long getFreeMemorySizeInBytes() { return memoryBlock.getFreeMemoryInBytes(); } + + private synchronized AllocationResult tryAllocate( + final MemoryHandle handle, final long sizeInBytes) { + final AllocationResult decision = inspectAllocationUnderLock(handle, sizeInBytes); + if (!decision.isAccepted() || sizeInBytes <= 0L) { + return decision; + } + + if (!memoryBlock.allocate(sizeInBytes)) { + return AllocationResult.rejected( + AllocationRejectionReason.MEMORY_LIMIT, + handle.getMemoryQuotaInBytesUnderLock(), + getFreeMemorySizeInBytes()); + } + handle.usedMemoryInBytes += sizeInBytes; + return AllocationResult.accepted( + handle.getMemoryQuotaInBytesUnderLock(), handle.getFreeMemorySizeInBytes()); + } + + private AllocationResult inspectAllocationUnderLock( + final MemoryHandle handle, final long sizeInBytes) { + if (sizeInBytes <= 0L) { + return AllocationResult.accepted( + handle.getMemoryQuotaInBytes(), handle.getFreeMemorySizeInBytes()); + } + if (!handle.active || handle.closed || !handles.containsKey(handle.id)) { + return AllocationResult.rejected( + AllocationRejectionReason.MEMORY_LIMIT, getTotalMemorySizeInBytes(), 0L); + } + + final long total = getTotalMemorySizeInBytes(); + final long used = getUsedMemorySizeInBytes(); + final long quota = handle.getMemoryQuotaInBytesUnderLock(); + final long protectedShare = quota / PROTECTED_SHARE_DIVISOR; + final long maximum = handle.getMaximumMemorySizeInBytesUnderLock(); + long reservedByOthers = 0L; + for (final MemoryHandle other : handles.values()) { + if (other != handle && !other.closed && (other.active || other.usedMemoryInBytes > 0L)) { + reservedByOthers += Math.max(protectedShare, other.usedMemoryInBytes); + } + } + final long availableToOwner = Math.max(0L, total - legacyUsedMemoryInBytes - reservedByOthers); + final long ownerUsed = handle.usedMemoryInBytes; + final boolean fitsInNode = total > 0L && used <= total && sizeInBytes <= total - used; + final boolean fitsInShare = + ownerUsed <= availableToOwner && sizeInBytes <= availableToOwner - ownerUsed; + + if (fitsInNode && fitsInShare) { + return AllocationResult.accepted(quota, availableToOwner - ownerUsed); + } + + final AllocationRejectionReason reason; + if (sizeInBytes > maximum) { + reason = AllocationRejectionReason.OVERSIZED_ENTRY; + } else if (ownerUsed > maximum || sizeInBytes > maximum - ownerUsed) { + reason = AllocationRejectionReason.MEMORY_QUOTA; + } else { + reason = AllocationRejectionReason.MEMORY_LIMIT; + } + return AllocationResult.rejected( + reason, quota, Math.min(availableToOwner - ownerUsed, total - used)); + } + + private synchronized void release(final MemoryHandle handle, final long sizeInBytes) { + if (sizeInBytes <= 0L || handle.closed) { + return; + } + final long released = Math.min(sizeInBytes, handle.usedMemoryInBytes); + if (released > 0L) { + handle.usedMemoryInBytes -= released; + memoryBlock.release(released); + } + } + + private synchronized void unregister(final MemoryHandle handle) { + if (handle.closed) { + return; + } + final long remaining = handle.usedMemoryInBytes; + if (remaining > 0L) { + memoryBlock.release(remaining); + handle.usedMemoryInBytes = 0L; + } + handles.remove(handle.id); + handle.closed = true; + } + + private synchronized int activeHandleCountUnderLock() { + int count = 0; + for (final MemoryHandle handle : handles.values()) { + // An inactive queue keeps its share until lifecycle cleanup releases its allocations. + if (!handle.closed && (handle.active || handle.usedMemoryInBytes > 0L)) { + count++; + } + } + return Math.max(1, count); + } + + public enum AllocationRejectionReason { + NONE("NONE"), + MEMORY_QUOTA("SUBSCRIPTION_MEMORY_QUOTA"), + MEMORY_LIMIT("SUBSCRIPTION_MEMORY_LIMIT"), + OVERSIZED_ENTRY("SUBSCRIPTION_OVERSIZED_ENTRY"); + + private final String code; + + AllocationRejectionReason(final String code) { + this.code = code; + } + + public String getCode() { + return code; + } + } + + public static final class AllocationResult { + private final boolean accepted; + private final AllocationRejectionReason rejectionReason; + private final long quotaInBytes; + private final long freeMemoryInBytes; + + private AllocationResult( + final boolean accepted, + final AllocationRejectionReason rejectionReason, + final long quotaInBytes, + final long freeMemoryInBytes) { + this.accepted = accepted; + this.rejectionReason = rejectionReason; + this.quotaInBytes = quotaInBytes; + this.freeMemoryInBytes = freeMemoryInBytes; + } + + private static AllocationResult accepted( + final long quotaInBytes, final long freeMemoryInBytes) { + return new AllocationResult( + true, AllocationRejectionReason.NONE, quotaInBytes, freeMemoryInBytes); + } + + private static AllocationResult rejected( + final AllocationRejectionReason reason, + final long quotaInBytes, + final long freeMemoryInBytes) { + return new AllocationResult(false, reason, quotaInBytes, freeMemoryInBytes); + } + + public boolean isAccepted() { + return accepted; + } + + public AllocationRejectionReason getRejectionReason() { + return rejectionReason; + } + + public long getQuotaInBytes() { + return quotaInBytes; + } + + public long getFreeMemoryInBytes() { + return freeMemoryInBytes; + } + } + + /** Per-queue view of the shared subscription memory budget. */ + public final class MemoryHandle implements AutoCloseable { + private final long id; + private long usedMemoryInBytes; + private boolean active = true; + private boolean closed; + + private MemoryHandle(final long id) { + this.id = id; + } + + public AllocationResult tryAllocate(final long sizeInBytes) { + return SubscriptionMemoryManager.this.tryAllocate(this, sizeInBytes); + } + + public void release(final long sizeInBytes) { + SubscriptionMemoryManager.this.release(this, sizeInBytes); + } + + public long getUsedMemorySizeInBytes() { + synchronized (SubscriptionMemoryManager.this) { + return usedMemoryInBytes; + } + } + + public long getMemoryQuotaInBytes() { + synchronized (SubscriptionMemoryManager.this) { + return getMemoryQuotaInBytesUnderLock(); + } + } + + private long getMemoryQuotaInBytesUnderLock() { + return Math.max(0L, getTotalMemorySizeInBytes() - legacyUsedMemoryInBytes) + / activeHandleCountUnderLock(); + } + + public long getMaximumMemorySizeInBytes() { + synchronized (SubscriptionMemoryManager.this) { + return getMaximumMemorySizeInBytesUnderLock(); + } + } + + private long getMaximumMemorySizeInBytesUnderLock() { + final long budget = Math.max(0L, getTotalMemorySizeInBytes() - legacyUsedMemoryInBytes); + final long protectedShare = getMemoryQuotaInBytesUnderLock() / PROTECTED_SHARE_DIVISOR; + return Math.max(0L, budget - protectedShare * (activeHandleCountUnderLock() - 1L)); + } + + public long getFreeMemorySizeInBytes() { + synchronized (SubscriptionMemoryManager.this) { + // Probe the size of one byte to account for protected shares and live borrowed allocations. + return Math.max( + 0L, + SubscriptionMemoryManager.this + .inspectAllocationUnderLock(this, 1L) + .getFreeMemoryInBytes()); + } + } + + public boolean isActive() { + synchronized (SubscriptionMemoryManager.this) { + return active && !closed; + } + } + + public void setActive(final boolean active) { + synchronized (SubscriptionMemoryManager.this) { + if (!closed) { + this.active = active; + } + } + } + + public boolean canAllocate(final long sizeInBytes) { + synchronized (SubscriptionMemoryManager.this) { + return SubscriptionMemoryManager.this + .inspectAllocationUnderLock(this, sizeInBytes) + .isAccepted(); + } + } + + public AllocationRejectionReason inspectRejection(final long sizeInBytes) { + synchronized (SubscriptionMemoryManager.this) { + return SubscriptionMemoryManager.this + .inspectAllocationUnderLock(this, sizeInBytes) + .getRejectionReason(); + } + } + + @Override + public void close() { + SubscriptionMemoryManager.this.unregister(this); + } + } } 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 60f30834cad0a..c7ebf06ab055d 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 @@ -26,12 +26,15 @@ import org.apache.iotdb.consensus.iot.SubscriptionWalRetentionPolicy; import org.apache.iotdb.consensus.iot.WriterSafeFrontierTracker; import org.apache.iotdb.consensus.iot.log.ConsensusReqReader; +import org.apache.iotdb.consensus.iot.subscription.SubscriptionQueueAdmission; +import org.apache.iotdb.consensus.iot.subscription.SubscriptionQueueRejectionReason; import org.apache.iotdb.db.conf.IoTDBDescriptor; import org.apache.iotdb.db.pipe.resource.memory.PipeMemoryWeightUtil; import org.apache.iotdb.db.queryengine.plan.statement.StatementTestUtils; import org.apache.iotdb.db.subscription.event.SubscriptionEvent; import org.apache.iotdb.db.subscription.resource.SubscriptionMemoryManager; 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.tsfile.enums.ColumnCategory; @@ -65,6 +68,192 @@ public class ConsensusPrefetchingQueueDataNodeMemoryTest { @Rule public final TemporaryFolder temporaryFolder = new TemporaryFolder(); + @Test + public void testLargeQueueQuotaDoesNotBlockSmallQueueAndPreservesPendingEntries() + throws Exception { + final String originalSystemDir = IoTDBDescriptor.getInstance().getConfig().getSystemDir(); + final int originalBatchMaxDelay = + CommonDescriptor.getInstance().getConfig().getSubscriptionConsensusBatchMaxDelayInMs(); + final int originalMaxWalEntries = + CommonDescriptor.getInstance().getConfig().getSubscriptionConsensusBatchMaxWalEntries(); + ConsensusPrefetchingQueue largeQueue = null; + ConsensusPrefetchingQueue smallQueue = null; + try { + CommonDescriptor.getInstance().getConfig().setSubscriptionConsensusBatchMaxDelayInMs(0); + CommonDescriptor.getInstance().getConfig().setSubscriptionConsensusBatchMaxWalEntries(1); + final Tablet largeTablet = createLargeTablet(); + final long largeBytes = PipeMemoryWeightUtil.calculateTabletSizeInBytes(largeTablet); + final long smallBytes = PipeMemoryWeightUtil.calculateTabletSizeInBytes(createTablet()); + final SubscriptionMemoryManager memoryManager = + // With two queues, a queue may use 3/4 of the budget; make that exactly one large entry. + new SubscriptionMemoryManager(largeBytes + largeBytes / 3L); + final ConsensusSubscriptionCommitManager commitManager = + newCommitManager(temporaryFolder.newFolder("large-small-fair-share")); + final FakeConsensusReqReader largeReader = new FakeConsensusReqReader(); + final FakeConsensusReqReader smallReader = new FakeConsensusReqReader(); + final AtomicInteger largeConversions = new AtomicInteger(); + final AtomicInteger smallConversions = new AtomicInteger(); + largeQueue = + newQueue( + "largeGroup", + new DataRegionId(1), + largeReader, + newConverter(largeConversions, largeTablet), + commitManager, + TopicConstant.ORDER_MODE_LEADER_ONLY_VALUE); + smallQueue = + newQueue( + "smallGroup", + new DataRegionId(2), + smallReader, + newConverter(smallConversions), + commitManager, + TopicConstant.ORDER_MODE_LEADER_ONLY_VALUE); + largeQueue.setSubscriptionMemoryManager(memoryManager); + smallQueue.setSubscriptionMemoryManager(memoryManager); + assertNull(largeQueue.poll("largeConsumer")); + assertNull(smallQueue.poll("smallConsumer")); + + largeReader.currentSearchIndex = 2L; + assertTrue(pendingEntries(largeQueue).offer(createRequest(1L))); + assertTrue(pendingEntries(largeQueue).offer(createRequest(2L))); + largeQueue.drivePrefetchOnce(); + assertEquals(largeBytes, largeQueue.getRetainedTabletBytes()); + final long retainedPendingBytes = largeQueue.getRetainedRequestBytes(); + assertTrue(retainedPendingBytes > 0L); + largeQueue.drivePrefetchOnce(); + assertEquals(1, largeConversions.get()); + assertEquals(1, pendingEntries(largeQueue).size()); + assertEquals(retainedPendingBytes, largeQueue.getRetainedRequestBytes()); + assertFalse(pendingEntries(largeQueue).offer(createRequest(3L))); + assertEquals( + SubscriptionQueueRejectionReason.SUBSCRIPTION_MEMORY_QUOTA, + ((SubscriptionQueueAdmission) pendingEntries(largeQueue)).getLastRejectionReason()); + assertEquals(1L, largeQueue.getRealtimeAdmissionRejectionCount()); + assertEquals( + 1L, + largeQueue.getRealtimeAdmissionRejectionCount( + SubscriptionQueueRejectionReason.SUBSCRIPTION_MEMORY_QUOTA)); + + smallReader.currentSearchIndex = 1L; + assertTrue(pendingEntries(smallQueue).offer(createRequest(1L))); + smallQueue.drivePrefetchOnce(); + assertEquals(1, smallConversions.get()); + assertEquals(largeBytes + smallBytes, memoryManager.getUsedMemorySizeInBytes()); + final SubscriptionEvent smallEvent = smallQueue.poll("smallConsumer"); + assertNotNull(smallEvent); + assertTrue(smallQueue.ack("smallConsumer", smallEvent.getCommitContext())); + assertEquals(largeBytes, memoryManager.getUsedMemorySizeInBytes()); + + final SubscriptionEvent largeEvent = largeQueue.poll("largeConsumer"); + assertNotNull(largeEvent); + assertTrue(largeQueue.ack("largeConsumer", largeEvent.getCommitContext())); + largeQueue.drivePrefetchOnce(); + assertEquals(2, largeConversions.get()); + assertEquals(0, pendingEntries(largeQueue).size()); + assertEquals(0L, largeQueue.getRetainedRequestBytes()); + final SubscriptionEvent secondLargeEvent = largeQueue.poll("largeConsumer"); + assertNotNull(secondLargeEvent); + assertTrue(largeQueue.ack("largeConsumer", secondLargeEvent.getCommitContext())); + assertEquals(0L, memoryManager.getUsedMemorySizeInBytes()); + } finally { + if (largeQueue != null) { + largeQueue.close(); + } + if (smallQueue != null) { + smallQueue.close(); + } + CommonDescriptor.getInstance() + .getConfig() + .setSubscriptionConsensusBatchMaxDelayInMs(originalBatchMaxDelay); + CommonDescriptor.getInstance() + .getConfig() + .setSubscriptionConsensusBatchMaxWalEntries(originalMaxWalEntries); + IoTDBDescriptor.getInstance().getConfig().setSystemDir(originalSystemDir); + } + } + + @Test + public void testOversizedEntryReportsErrorWhileOtherQueueHoldsMemory() throws Exception { + final String originalSystemDir = IoTDBDescriptor.getInstance().getConfig().getSystemDir(); + final int originalBatchMaxDelay = + CommonDescriptor.getInstance().getConfig().getSubscriptionConsensusBatchMaxDelayInMs(); + ConsensusPrefetchingQueue largeQueue = null; + ConsensusPrefetchingQueue smallQueue = null; + try { + CommonDescriptor.getInstance().getConfig().setSubscriptionConsensusBatchMaxDelayInMs(0); + final Tablet largeTablet = createLargeTablet(); + final long largeBytes = PipeMemoryWeightUtil.calculateTabletSizeInBytes(largeTablet); + // This entry exceeds the entire node budget, which is also partly held by another queue. + final SubscriptionMemoryManager memoryManager = + new SubscriptionMemoryManager(largeBytes / 2L); + final ConsensusSubscriptionCommitManager commitManager = + newCommitManager(temporaryFolder.newFolder("oversized-queue-error")); + final FakeConsensusReqReader largeReader = new FakeConsensusReqReader(); + final FakeConsensusReqReader smallReader = new FakeConsensusReqReader(); + final AtomicInteger largeConversions = new AtomicInteger(); + largeQueue = + newQueue( + "largeGroup", + new DataRegionId(1), + largeReader, + newConverter(largeConversions, largeTablet), + commitManager, + TopicConstant.ORDER_MODE_LEADER_ONLY_VALUE); + smallQueue = + newQueue( + "smallGroup", + new DataRegionId(2), + smallReader, + newConverter(new AtomicInteger()), + commitManager, + TopicConstant.ORDER_MODE_LEADER_ONLY_VALUE); + largeQueue.setSubscriptionMemoryManager(memoryManager); + smallQueue.setSubscriptionMemoryManager(memoryManager); + assertNull(largeQueue.poll("largeConsumer")); + assertNull(smallQueue.poll("smallConsumer")); + smallReader.currentSearchIndex = 1L; + assertTrue(pendingEntries(smallQueue).offer(createRequest(1L))); + smallQueue.drivePrefetchOnce(); + final long smallBytes = memoryManager.getUsedMemorySizeInBytes(); + assertTrue(smallBytes > 0L); + + largeReader.currentSearchIndex = 1L; + assertTrue(pendingEntries(largeQueue).offer(createRequest(1L))); + largeQueue.drivePrefetchOnce(); + assertEquals(1, largeConversions.get()); + assertEquals(1L, largeQueue.getCurrentReadSearchIndex()); + assertEquals(1L, largeQueue.getOversizedEntryRejectionCount()); + assertEquals(0L, largeQueue.getRealtimeAdmissionRejectionCount()); + assertEquals(smallBytes, memoryManager.getUsedMemorySizeInBytes()); + final SubscriptionEvent error = largeQueue.poll("largeConsumer"); + assertNotNull(error); + final ErrorPayload payload = (ErrorPayload) error.getCurrentResponse().getPayload(); + assertTrue(payload.isCritical()); + assertTrue(payload.getErrorMessage().contains("SUBSCRIPTION_OVERSIZED_ENTRY")); + for (int i = 0; i < 10; i++) { + largeQueue.drivePrefetchOnce(); + } + assertEquals(1, largeConversions.get()); + assertEquals(1L, largeQueue.getCurrentReadSearchIndex()); + final SubscriptionEvent smallEvent = smallQueue.poll("smallConsumer"); + assertNotNull(smallEvent); + assertTrue(smallQueue.ack("smallConsumer", smallEvent.getCommitContext())); + assertEquals(0L, memoryManager.getUsedMemorySizeInBytes()); + } finally { + if (largeQueue != null) { + largeQueue.close(); + } + if (smallQueue != null) { + smallQueue.close(); + } + CommonDescriptor.getInstance() + .getConfig() + .setSubscriptionConsensusBatchMaxDelayInMs(originalBatchMaxDelay); + IoTDBDescriptor.getInstance().getConfig().setSystemDir(originalSystemDir); + } + } + @Test public void testQueuesShareDataNodeMemoryBudget() throws Exception { final String originalSystemDir = IoTDBDescriptor.getInstance().getConfig().getSystemDir(); @@ -77,7 +266,8 @@ public void testQueuesShareDataNodeMemoryBudget() throws Exception { CommonDescriptor.getInstance().getConfig().setSubscriptionConsensusBatchMaxDelayInMs(0); final Tablet tablet = createTablet(); final long oneTabletBytes = PipeMemoryWeightUtil.calculateTabletSizeInBytes(tablet); - final SubscriptionMemoryManager memoryManager = new SubscriptionMemoryManager(oneTabletBytes); + final SubscriptionMemoryManager memoryManager = + new SubscriptionMemoryManager(oneTabletBytes * 2L); final ConsensusSubscriptionCommitManager commitManager = newCommitManager(systemDir); final FakeConsensusReqReader readerA = new FakeConsensusReqReader(); @@ -121,34 +311,28 @@ public void testQueuesShareDataNodeMemoryBudget() throws Exception { <= memoryManager.getTotalMemorySizeInBytes()); readerB.currentSearchIndex = 1L; - assertFalse(pendingEntries(queueB).offer(createRequest(1L))); + assertTrue(pendingEntries(queueB).offer(createRequest(1L))); queueB.drivePrefetchOnce(); - assertEquals(0, conversionCountB.get()); - assertEquals("true", queueB.coreReportMessage().get("realtimeAdmissionBlocked")); - assertEquals(oneTabletBytes, memoryManager.getUsedMemorySizeInBytes()); + assertEquals(1, conversionCountB.get()); + assertEquals("false", queueB.coreReportMessage().get("realtimeAdmissionBlocked")); + assertEquals(oneTabletBytes * 2L, memoryManager.getUsedMemorySizeInBytes()); + assertEquals(oneTabletBytes, queueB.getRetainedTabletBytes()); final SubscriptionEvent eventA = queueA.poll("consumerA"); assertNotNull(eventA); assertTrue(queueA.ack("consumerA", eventA.getCommitContext())); assertEquals(0L, queueA.getRetainedTabletBytes()); - assertEquals(0L, memoryManager.getUsedMemorySizeInBytes()); + assertEquals(oneTabletBytes, memoryManager.getUsedMemorySizeInBytes()); queueB.drivePrefetchOnce(); assertEquals("false", queueB.coreReportMessage().get("realtimeAdmissionBlocked")); - assertTrue(pendingEntries(queueB).offer(createRequest(1L))); - queueB.drivePrefetchOnce(); - - assertEquals(1, conversionCountB.get()); - assertEquals(oneTabletBytes, queueB.getRetainedTabletBytes()); - assertEquals(oneTabletBytes, memoryManager.getUsedMemorySizeInBytes()); - assertTrue( - queueA.getRetainedTabletBytes() + queueB.getRetainedTabletBytes() - <= memoryManager.getTotalMemorySizeInBytes()); final SubscriptionEvent eventB = queueB.poll("consumerB"); assertNotNull(eventB); assertTrue(queueB.ack("consumerB", eventB.getCommitContext())); assertEquals(0L, memoryManager.getUsedMemorySizeInBytes()); + queueB.drivePrefetchOnce(); + assertEquals("false", queueB.coreReportMessage().get("realtimeAdmissionBlocked")); } finally { if (queueA != null) { queueA.close(); @@ -203,23 +387,20 @@ public void testRealtimeBacklogIsDrainedBeforeMorePendingEntriesAreConverted() t assertEquals(1, conversionCount.get()); assertEquals(2L, queue.getCurrentReadSearchIndex()); assertEquals("1", queue.coreReportMessage().get("bufferedRealtimeEntryCount")); - assertEquals("0", queue.coreReportMessage().get("pendingEntriesSize")); + assertEquals("1", queue.coreReportMessage().get("pendingEntriesSize")); assertEquals("true", queue.coreReportMessage().get("realtimeAdmissionBlocked")); + assertFalse(pendingEntries(queue).offer(createRequest(3L))); + assertEquals("SUBSCRIPTION_WRITER_BACKLOG", queue.getLastAdmissionRejectionCode()); + assertEquals(0L, queue.getSubscriptionMemoryRejectionCount()); queue.setActiveWriterNodeIds(Collections.singleton(7)); queue.drivePrefetchOnce(); - assertEquals(1, conversionCount.get()); - assertEquals("0", queue.coreReportMessage().get("bufferedRealtimeEntryCount")); - assertEquals("false", queue.coreReportMessage().get("realtimeAdmissionBlocked")); - - assertTrue(pendingEntries(queue).offer(createRequest(2L))); - queue.drivePrefetchOnce(); - assertEquals(2, conversionCount.get()); assertEquals(3L, queue.getCurrentReadSearchIndex()); assertEquals("0", queue.coreReportMessage().get("bufferedRealtimeEntryCount")); assertEquals("0", queue.coreReportMessage().get("pendingEntriesSize")); + assertEquals("false", queue.coreReportMessage().get("realtimeAdmissionBlocked")); } finally { if (queue != null) { queue.close(); @@ -260,12 +441,17 @@ private static ConsensusPrefetchingQueue newQueue( } private static ConsensusLogToTabletConverter newConverter(final AtomicInteger conversionCount) { + return newConverter(conversionCount, createTablet()); + } + + private static ConsensusLogToTabletConverter newConverter( + final AtomicInteger conversionCount, final Tablet tablet) { final ConsensusLogToTabletConverter converter = mock(ConsensusLogToTabletConverter.class); when(converter.convert(any())) .thenAnswer( ignored -> { conversionCount.incrementAndGet(); - return Collections.singletonList(createTablet()); + return Collections.singletonList(tablet); }); when(converter.getDatabaseName()).thenReturn("db"); return converter; @@ -291,6 +477,21 @@ private static Tablet createTablet() { return tablet; } + private static Tablet createLargeTablet() { + final Tablet tablet = + new Tablet( + "sensors", + Arrays.asList("device", "payload"), + Arrays.asList(TSDataType.STRING, TSDataType.STRING), + Arrays.asList(ColumnCategory.TAG, ColumnCategory.FIELD), + 1); + tablet.addTimestamp(0, 1L); + tablet.addValue(0, 0, "d1"); + tablet.addValue(0, 1, "v".repeat(20_000)); + tablet.setRowSize(1); + return tablet; + } + private static IndexedConsensusRequest createRequest(final long searchIndex) { return new IndexedConsensusRequest( searchIndex, 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 0f321341401ed..68df431cf63d9 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 @@ -30,6 +30,8 @@ import org.apache.iotdb.consensus.iot.WriterSafeFrontierTracker; import org.apache.iotdb.consensus.iot.log.ConsensusReqReader; import org.apache.iotdb.consensus.iot.logdispatcher.IoTConsensusMemoryManager; +import org.apache.iotdb.consensus.iot.subscription.SubscriptionQueueAdmission; +import org.apache.iotdb.consensus.iot.subscription.SubscriptionQueueRejectionReason; import org.apache.iotdb.db.conf.IoTDBDescriptor; import org.apache.iotdb.db.queryengine.plan.statement.StatementTestUtils; import org.apache.iotdb.db.storageengine.dataregion.wal.io.ProgressWALReader; @@ -81,8 +83,9 @@ import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicReference; import java.util.concurrent.locks.ReentrantReadWriteLock; -import java.util.function.BooleanSupplier; +import java.util.function.Consumer; import java.util.function.LongSupplier; +import java.util.function.Supplier; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; @@ -635,7 +638,8 @@ public void testAdmissionClearCannotLeaveEntryEnqueuedAfterFence() throws Except final Class queueClass = Class.forName(ConsensusPrefetchingQueue.class.getName() + "$WakeableIndexedConsensusQueue"); final Constructor constructor = - queueClass.getDeclaredConstructor(int.class, Runnable.class, BooleanSupplier.class); + queueClass.getDeclaredConstructor( + int.class, Runnable.class, Supplier.class, Consumer.class); constructor.setAccessible(true); final AtomicBoolean admissionEnabled = new AtomicBoolean(true); @@ -645,7 +649,7 @@ public void testAdmissionClearCannotLeaveEntryEnqueuedAfterFence() throws Except final CountDownLatch clearCompleted = new CountDownLatch(1); final AtomicReference offered = new AtomicReference<>(); final AtomicReference asyncFailure = new AtomicReference<>(); - final BooleanSupplier admissionSupplier = + final Supplier admissionSupplier = () -> { final boolean admitted = admissionEnabled.get(); admissionChecked.countDown(); @@ -657,11 +661,17 @@ public void testAdmissionClearCannotLeaveEntryEnqueuedAfterFence() throws Except Thread.currentThread().interrupt(); throw new AssertionError(e); } - return admitted; + return admitted + ? SubscriptionQueueRejectionReason.NONE + : SubscriptionQueueRejectionReason.INACTIVE_OR_CLOSED; }; final BlockingQueue queue = (BlockingQueue) - constructor.newInstance(8, (Runnable) () -> {}, admissionSupplier); + constructor.newInstance( + 8, + (Runnable) () -> {}, + admissionSupplier, + (Consumer) reason -> {}); final Thread offerThread = new Thread( @@ -714,6 +724,77 @@ public void testAdmissionClearCannotLeaveEntryEnqueuedAfterFence() throws Except } assertTrue(Boolean.TRUE.equals(offered.get())); assertTrue(queue.isEmpty()); + assertFalse(queue.offer(createRequest(2L))); + assertEquals( + SubscriptionQueueRejectionReason.INACTIVE_OR_CLOSED, + ((SubscriptionQueueAdmission) queue).getLastRejectionReason()); + } + + @Test + @SuppressWarnings("unchecked") + public void testConcurrentOffersKeepTheirOwnRejectionReasons() throws Exception { + final Class queueClass = + Class.forName(ConsensusPrefetchingQueue.class.getName() + "$WakeableIndexedConsensusQueue"); + final Constructor constructor = + queueClass.getDeclaredConstructor( + int.class, Runnable.class, Supplier.class, Consumer.class); + constructor.setAccessible(true); + final BlockingQueue queue = + (BlockingQueue) + constructor.newInstance( + 8, + (Runnable) () -> {}, + (Supplier) + () -> + Thread.currentThread().getName().equals("first-offer") + ? SubscriptionQueueRejectionReason.SUBSCRIPTION_MEMORY_QUOTA + : SubscriptionQueueRejectionReason.WRITER_BACKLOG, + (Consumer) reason -> {}); + final CountDownLatch firstOffered = new CountDownLatch(1); + final CountDownLatch secondOffered = new CountDownLatch(1); + final AtomicReference firstReason = new AtomicReference<>(); + final AtomicReference secondReason = new AtomicReference<>(); + final AtomicReference asyncFailure = new AtomicReference<>(); + final Thread first = + new Thread( + () -> { + try { + assertFalse(queue.offer(createRequest(1L))); + firstOffered.countDown(); + assertTrue(secondOffered.await(5, TimeUnit.SECONDS)); + firstReason.set(((SubscriptionQueueAdmission) queue).getLastRejectionReason()); + } catch (final Throwable t) { + asyncFailure.compareAndSet(null, t); + } + }, + "first-offer"); + final Thread second = + new Thread( + () -> { + try { + assertTrue(firstOffered.await(5, TimeUnit.SECONDS)); + assertFalse(queue.offer(createRequest(2L))); + secondReason.set(((SubscriptionQueueAdmission) queue).getLastRejectionReason()); + } catch (final Throwable t) { + asyncFailure.compareAndSet(null, t); + } finally { + secondOffered.countDown(); + } + }, + "second-offer"); + first.setDaemon(true); + second.setDaemon(true); + first.start(); + second.start(); + first.join(TimeUnit.SECONDS.toMillis(5)); + second.join(TimeUnit.SECONDS.toMillis(5)); + assertFalse(first.isAlive()); + assertFalse(second.isAlive()); + if (asyncFailure.get() != null) { + throw new AssertionError(asyncFailure.get()); + } + assertEquals(SubscriptionQueueRejectionReason.SUBSCRIPTION_MEMORY_QUOTA, firstReason.get()); + assertEquals(SubscriptionQueueRejectionReason.WRITER_BACKLOG, secondReason.get()); } @Test @@ -722,7 +803,8 @@ public void testPendingQueueUsesSharedConsensusMemoryBudgetAndClearReleasesIt() final Class queueClass = Class.forName(ConsensusPrefetchingQueue.class.getName() + "$WakeableIndexedConsensusQueue"); final Constructor constructor = - queueClass.getDeclaredConstructor(int.class, Runnable.class, BooleanSupplier.class); + queueClass.getDeclaredConstructor( + int.class, Runnable.class, Supplier.class, Consumer.class); constructor.setAccessible(true); final Method retainedBytesMethod = queueClass.getDeclaredMethod("getRetainedRequestBytes"); retainedBytesMethod.setAccessible(true); @@ -734,10 +816,20 @@ public void testPendingQueueUsesSharedConsensusMemoryBudgetAndClearReleasesIt() new AtomicLongMemoryBlock("SubscriptionPendingTest", null, 100L); final BlockingQueue firstQueue = (BlockingQueue) - constructor.newInstance(8, (Runnable) () -> {}, (BooleanSupplier) () -> true); + constructor.newInstance( + 8, + (Runnable) () -> {}, + (Supplier) + () -> SubscriptionQueueRejectionReason.NONE, + (Consumer) reason -> {}); final BlockingQueue secondQueue = (BlockingQueue) - constructor.newInstance(8, (Runnable) () -> {}, (BooleanSupplier) () -> true); + constructor.newInstance( + 8, + (Runnable) () -> {}, + (Supplier) + () -> SubscriptionQueueRejectionReason.NONE, + (Consumer) reason -> {}); memoryManager.init(testMemoryBlock, 0.6); try { @@ -749,6 +841,9 @@ public void testPendingQueueUsesSharedConsensusMemoryBudgetAndClearReleasesIt() assertEquals(40L, retainedBytesMethod.invoke(secondQueue)); assertFalse(firstQueue.offer(createSizedRequest(2L, 10L, 30))); + assertEquals( + SubscriptionQueueRejectionReason.CONSENSUS_REQUEST_MEMORY_LIMIT, + ((SubscriptionQueueAdmission) firstQueue).getLastRejectionReason()); assertEquals(40L, testMemoryBlock.getUsedMemoryInBytes()); firstQueue.clear(); @@ -988,6 +1083,7 @@ public void testFilteredEmptyEntryDoesNotAdvanceAcrossUncommittedData() throws E 1L, 1L, true); + queue.setSubscriptionMemoryManager(new SubscriptionMemoryManager(1024L * 1024L)); final IndexedConsensusRequest dataRequest = new IndexedConsensusRequest( @@ -1760,6 +1856,7 @@ public void testTabletMemoryReleasedAfterAck() throws Exception { 1L, 1L, true); + queue.setSubscriptionMemoryManager(new SubscriptionMemoryManager(1024L * 1024L)); final IndexedConsensusRequest dataRequest = new IndexedConsensusRequest( @@ -1828,6 +1925,7 @@ public void testLateAckDoesNotWaitForPrefetchReadLock() throws Exception { 1L, 1L, true); + queue.setSubscriptionMemoryManager(new SubscriptionMemoryManager(1024L * 1024L)); final IndexedConsensusRequest dataRequest = new IndexedConsensusRequest( @@ -2636,6 +2734,7 @@ public void testRequeueDoesNotIncrementNackCount() throws Exception { 1L, 1L, true); + queue.setSubscriptionMemoryManager(new SubscriptionMemoryManager(1024L * 1024L)); reader.currentSearchIndex = 2L; assertTrue(pendingEntries(queue).offer(createRequest(1L))); 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 b3783beab43dd..e25c77803cffa 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 @@ -154,7 +154,7 @@ public void testAckRecoversDrainedSuffixFromWalWithoutReoffer() throws Exception assertEquals(Arrays.asList(1L, 2L), conversionAttempts); assertEquals(2L, queue.getCurrentReadSearchIndex()); assertEquals(1, queue.getPrefetchedEventCount()); - assertTrue(pendingEntries(queue).isEmpty()); + assertEquals(0, pendingEntries(queue).size()); assertEquals("true", queue.coreReportMessage().get("realtimeAdmissionBlocked")); assertEquals(oneTabletBytes, memoryManager.getUsedMemorySizeInBytes()); assertMemoryBounded(queue, memoryManager); @@ -307,22 +307,11 @@ private TrackingSubscriptionMemoryManager(final long totalMemorySizeInBytes) { } @Override - public synchronized boolean tryAllocate(final long sizeInBytes) { - final boolean allocated = super.tryAllocate(sizeInBytes); - observeUsage(); - return allocated; - } - - @Override - public synchronized void release(final long sizeInBytes) { - super.release(sizeInBytes); - observeUsage(); - } - - private void observeUsage() { - final long usedMemorySizeInBytes = getUsedMemorySizeInBytes(); + public synchronized long getUsedMemorySizeInBytes() { + final long usedMemorySizeInBytes = super.getUsedMemorySizeInBytes(); maximumUsedMemorySizeInBytes = Math.max(maximumUsedMemorySizeInBytes, usedMemorySizeInBytes); exceededBudget |= usedMemorySizeInBytes > getTotalMemorySizeInBytes(); + return usedMemorySizeInBytes; } private long getMaximumUsedMemorySizeInBytes() { diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/metric/ConsensusSubscriptionPrefetchingQueueMetricsTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/metric/ConsensusSubscriptionPrefetchingQueueMetricsTest.java index 08880f4fcb325..782cebe30cdac 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/metric/ConsensusSubscriptionPrefetchingQueueMetricsTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/metric/ConsensusSubscriptionPrefetchingQueueMetricsTest.java @@ -22,6 +22,7 @@ import org.apache.iotdb.commons.consensus.DataRegionId; import org.apache.iotdb.commons.service.metric.enums.Metric; import org.apache.iotdb.commons.service.metric.enums.Tag; +import org.apache.iotdb.consensus.iot.subscription.SubscriptionQueueRejectionReason; import org.apache.iotdb.db.subscription.broker.consensus.ConsensusPrefetchingQueue; import org.apache.iotdb.metrics.AbstractMetricService; import org.apache.iotdb.metrics.type.Rate; @@ -132,6 +133,18 @@ public void testMetricsAreIsolatedByRegion() throws Exception { eq(queueId), eq(Tag.REGION.toString()), eq(firstRegionId.toString())); + verify(metricService) + .createAutoGauge( + eq(Metric.SUBSCRIPTION_CONSENSUS_ADMISSION_REJECTIONS_BY_REASON.toString()), + eq(MetricLevel.IMPORTANT), + eq(firstQueue), + any(), + eq(Tag.NAME.toString()), + eq(queueId), + eq(Tag.REGION.toString()), + eq(firstRegionId.toString()), + eq(Tag.REASON.toString()), + eq(SubscriptionQueueRejectionReason.SUBSCRIPTION_MEMORY_QUOTA.getCode())); verify(metricService) .createAutoGauge( eq(Metric.SUBSCRIPTION_CONSENSUS_LAG.toString()), @@ -164,6 +177,16 @@ public void testMetricsAreIsolatedByRegion() throws Exception { queueId, Tag.REGION.toString(), firstRegionId.toString()); + verify(metricService) + .remove( + MetricType.AUTO_GAUGE, + Metric.SUBSCRIPTION_CONSENSUS_ADMISSION_REJECTIONS_BY_REASON.toString(), + Tag.NAME.toString(), + queueId, + Tag.REGION.toString(), + firstRegionId.toString(), + Tag.REASON.toString(), + SubscriptionQueueRejectionReason.SUBSCRIPTION_MEMORY_QUOTA.getCode()); verify(metricService, never()) .remove( MetricType.AUTO_GAUGE, 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 1635a6e0ecf83..beda24e623bf6 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 @@ -61,4 +61,112 @@ public void testZeroBudgetRejectsAllocation() { assertFalse(memoryManager.tryAllocate(1L)); assertEquals(0L, memoryManager.getUsedMemorySizeInBytes()); } + + @Test + public void testQueueCanBorrowIdleMemoryWithoutConsumingAnotherQueuesProtectedShare() { + final SubscriptionMemoryManager memoryManager = new SubscriptionMemoryManager(10L); + final SubscriptionMemoryManager.MemoryHandle queueA = memoryManager.registerQueue(); + final SubscriptionMemoryManager.MemoryHandle queueB = memoryManager.registerQueue(); + + assertEquals(5L, queueA.getMemoryQuotaInBytes()); + assertEquals( + SubscriptionMemoryManager.AllocationRejectionReason.OVERSIZED_ENTRY, + queueA.inspectRejection(9L)); + assertEquals(0L, memoryManager.getUsedMemorySizeInBytes()); + + assertTrue(queueA.tryAllocate(6L).isAccepted()); + assertTrue(queueA.tryAllocate(2L).isAccepted()); + assertEquals( + SubscriptionMemoryManager.AllocationRejectionReason.MEMORY_QUOTA, + queueA.inspectRejection(1L)); + assertTrue(queueB.tryAllocate(2L).isAccepted()); + assertFalse(queueB.tryAllocate(1L).isAccepted()); + assertEquals(10L, memoryManager.getUsedMemorySizeInBytes()); + + queueA.close(); + assertEquals(10L, queueB.getMemoryQuotaInBytes()); + assertTrue(queueB.tryAllocate(8L).isAccepted()); + assertEquals(10L, memoryManager.getUsedMemorySizeInBytes()); + + queueB.close(); + assertEquals(0L, memoryManager.getUsedMemorySizeInBytes()); + } + + @Test + public void testBorrowedMemoryWaitsForAckWhenAnotherQueueNeedsMoreThanItsReserve() { + final SubscriptionMemoryManager memoryManager = new SubscriptionMemoryManager(100L); + try (SubscriptionMemoryManager.MemoryHandle largeQueue = memoryManager.registerQueue(); + SubscriptionMemoryManager.MemoryHandle smallQueue = memoryManager.registerQueue()) { + assertTrue(largeQueue.tryAllocate(75L).isAccepted()); + assertEquals(25L, smallQueue.getFreeMemorySizeInBytes()); + assertEquals( + SubscriptionMemoryManager.AllocationRejectionReason.MEMORY_LIMIT, + smallQueue.inspectRejection(30L)); + assertTrue(smallQueue.tryAllocate(25L).isAccepted()); + assertEquals(0L, smallQueue.getFreeMemorySizeInBytes()); + largeQueue.release(50L); + assertTrue(smallQueue.tryAllocate(30L).isAccepted()); + assertEquals(80L, memoryManager.getUsedMemorySizeInBytes()); + } + } + + @Test + public void testNewQueueWaitsForExistingBorrowedAllocationsToDrain() { + final SubscriptionMemoryManager memoryManager = new SubscriptionMemoryManager(100L); + try (SubscriptionMemoryManager.MemoryHandle first = memoryManager.registerQueue()) { + assertTrue(first.tryAllocate(100L).isAccepted()); + try (SubscriptionMemoryManager.MemoryHandle second = memoryManager.registerQueue()) { + assertEquals( + SubscriptionMemoryManager.AllocationRejectionReason.MEMORY_LIMIT, + second.inspectRejection(1L)); + first.release(25L); + assertTrue(second.tryAllocate(25L).isAccepted()); + } + } + } + + @Test + public void testOversizedQueueEntryIsExplicitlyRejectedEvenWhenNodeIsEmpty() { + final SubscriptionMemoryManager memoryManager = new SubscriptionMemoryManager(10L); + try (SubscriptionMemoryManager.MemoryHandle queue = memoryManager.registerQueue()) { + assertEquals( + SubscriptionMemoryManager.AllocationRejectionReason.OVERSIZED_ENTRY, + queue.tryAllocate(11L).getRejectionReason()); + assertEquals(0L, memoryManager.getUsedMemorySizeInBytes()); + assertTrue(queue.tryAllocate(10L).isAccepted()); + } + assertEquals(0L, memoryManager.getUsedMemorySizeInBytes()); + } + + @Test + public void testInactiveQueueKeepsShareUntilRetainedDataIsReleased() { + final SubscriptionMemoryManager memoryManager = new SubscriptionMemoryManager(10L); + try (SubscriptionMemoryManager.MemoryHandle queueA = memoryManager.registerQueue(); + SubscriptionMemoryManager.MemoryHandle queueB = memoryManager.registerQueue()) { + assertTrue(queueA.tryAllocate(5L).isAccepted()); + queueA.setActive(false); + assertEquals(5L, queueB.getMemoryQuotaInBytes()); + assertTrue(queueB.tryAllocate(5L).isAccepted()); + queueA.release(5L); + assertEquals(10L, queueB.getMemoryQuotaInBytes()); + assertTrue(queueB.tryAllocate(5L).isAccepted()); + } + assertEquals(0L, memoryManager.getUsedMemorySizeInBytes()); + } + + @Test + public void testAllocationProbeDoesNotReserveMemory() { + final SubscriptionMemoryManager memoryManager = new SubscriptionMemoryManager(10L); + try (SubscriptionMemoryManager.MemoryHandle queue = memoryManager.registerQueue()) { + for (int i = 0; i < 100; i++) { + assertTrue(queue.canAllocate(10L)); + assertEquals( + SubscriptionMemoryManager.AllocationRejectionReason.NONE, queue.inspectRejection(10L)); + } + assertEquals(0L, memoryManager.getUsedMemorySizeInBytes()); + assertTrue(queue.tryAllocate(10L).isAccepted()); + queue.release(100L); + assertEquals(0L, memoryManager.getUsedMemorySizeInBytes()); + } + } } diff --git a/iotdb-core/node-commons/src/assembly/resources/conf/iotdb-system.properties.template b/iotdb-core/node-commons/src/assembly/resources/conf/iotdb-system.properties.template index 158a079be6b19..b229f36d1fed5 100644 --- a/iotdb-core/node-commons/src/assembly/resources/conf/iotdb-system.properties.template +++ b/iotdb-core/node-commons/src/assembly/resources/conf/iotdb-system.properties.template @@ -1206,6 +1206,13 @@ may_cache_nonexist_series=true # The parameter form is a:b:c:d:e:f:g:h:i, where a, b, c, d, e, f, g, h and i are integers. for example: 1:1:1:1:1:1:1:1:1 , 1:100:200:50:200:200:200:50:250 # The legacy eight-part form is still accepted and defaults Subscription to 20% of query memory. # When subscription is disabled, its memory proportion is 0. +# The ninth (Subscription) share controls consensus subscription materialization, independently of +# subscription_cache_memory_usage_percentage. Each active queue protects half of its equal share; +# the remaining idle capacity can be borrowed by another queue, but not revoked until ACK. Size +# the budget for the largest materialized WAL entry * active queue count * at least 1.2; use a +# larger multiplier for multiple buffered entries per queue. An entry exceeding the maximum a +# queue can hold under these protected shares reports SUBSCRIPTION_OVERSIZED_ENTRY and does not +# advance WAL progress. New queues may wait for existing borrowed allocations to drain via ACK. # effectiveMode: restart chunk_timeseriesmeta_free_memory_proportion=1:100:200:50:200:200:200:50:250 @@ -2153,6 +2160,28 @@ pipe_all_sinks_rate_limit_bytes_per_second=-1 ### Subscription Consensus Configuration #################### +# Maximum fraction of pipe non-floating memory used to cache serialized subscription poll responses. +# Recommended starting value: 0.2, including large-field and large-batch workloads. This does not control +# consensus subscription tablet materialization; size that budget using the ninth +# chunk_timeseriesmeta_free_memory_proportion share instead. The cache may receive less than this +# maximum under pipe memory pressure. +# effectiveMode: restart +# Datatype: double, range: [0, 1] +subscription_cache_memory_usage_percentage=0.2 + +# Maximum number of consensus-subscription prefetch workers. For large-field workloads, compare +# 2 and 4 workers under load, checking CPU, heap and ACK latency; queues share the worker pool. +# effectiveMode: restart +# Datatype: int +subscription_consensus_prefetch_executor_max_thread_num=2 + +# Maximum materialized consensus-subscription batch size. This is a batching target, not a hard +# limit on one WAL entry/tablet. For large-field workloads, choose a value at least as large as a +# typical single tablet if batching is desired, and provide enough materialization memory per queue. +# effectiveMode: restart +# Datatype: long, Unit: byte +subscription_consensus_batch_max_size_in_bytes=8388608 + # The interval at which an idle consensus-subscription queue advances its writer safe-time barrier by watermark. # effectiveMode: restart # Datatype: long(duration) 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 5bcafffaa142b..d412d74d0ccb1 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 @@ -232,6 +232,14 @@ public enum Metric { SUBSCRIPTION_CONSENSUS_LAST_POLL_TIME("subscription_consensus_last_poll_time"), SUBSCRIPTION_CONSENSUS_LAST_PROGRESS_TIME("subscription_consensus_last_progress_time"), SUBSCRIPTION_CONSENSUS_PROGRESS_STATUS("subscription_consensus_progress_status"), + SUBSCRIPTION_CONSENSUS_MEMORY_USED("subscription_consensus_memory_used"), + SUBSCRIPTION_CONSENSUS_MEMORY_QUOTA("subscription_consensus_memory_quota"), + SUBSCRIPTION_CONSENSUS_MEMORY_MAXIMUM("subscription_consensus_memory_maximum"), + SUBSCRIPTION_CONSENSUS_ADMISSION_REJECTIONS("subscription_consensus_admission_rejections"), + SUBSCRIPTION_CONSENSUS_ADMISSION_REJECTIONS_BY_REASON( + "subscription_consensus_admission_rejections_by_reason"), + SUBSCRIPTION_CONSENSUS_MEMORY_REJECTIONS("subscription_consensus_memory_rejections"), + SUBSCRIPTION_CONSENSUS_OVERSIZED_REJECTIONS("subscription_consensus_oversized_rejections"), // load related ACTIVE_LOADING_FILES_NUMBER("active_loading_files_number"), ACTIVE_LOADING_FILES_SIZE("active_loading_files_size"), diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/service/metric/enums/Tag.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/service/metric/enums/Tag.java index 91b4307e02d61..0f0aa0dadc65d 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/service/metric/enums/Tag.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/service/metric/enums/Tag.java @@ -23,6 +23,7 @@ public enum Tag { TYPE("type"), NAME("name"), REGION("region"), + REASON("reason"), STATUS("status"), DATABASE("database"), FROM("from"),