>From Michael Blow <[email protected]>: Michael Blow has uploaded this change for review. ( https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/21627?usp=email )
Change subject: [NO ISSUE][STO] Reclaim idle memory components when the vbc is full ...................................................................... [NO ISSUE][STO] Reclaim idle memory components when the vbc is full Above a few thousand concurrently ingesting datasets the global virtual buffer cache fills with memory components which hold no data, and ingestion stalls permanently with nothing logged by any logger. allocateMemoryComponents() allocates every memory component of an index on the first write to it, and each pins a metadata page and a root page as soon as its BTree is created. Only the component which is written is flushed, and only a flush cleans a component up and returns its pages; the siblings stay allocated and empty until the dataset is closed, which nothing does under memory pressure. A dataset written once therefore leaves pages behind for data which never arrived, and enough of them fill the cache with components selectFlushIndex() will never pick, because it only picks components which are not empty. Writers then fail threadEnter()'s !vbc.isFull() check and retry every 100ms forever, with nothing left which can make it false. The deadlock watchdog sees no cycle- the writer is in a timed wait, not blocked on a monitor. When a scheduleFlush() finds the cache full, nothing flushable and no flush in flight- so no completion is coming to free pages or to wake the flush thread again- reclaim the pages held by components which are allocated but INACTIVE, and warn if even that finds nothing. Reclaiming costs the next writer to that index a re-allocation, which is the same one it already pays after every flush. The components are not reset: they are INACTIVE, so they have already been reset or were never used, and resetting again would decrement the index's scheduled flush count a second time. The pass takes each index's operation tracker, which is what a writer holds while it activates a component, and copies primaryIndexes under the cache monitor first so that the cache-then-op-tracker order selectFlushIndex() establishes is preserved rather than extended. This bounds the damage rather than removing the residue; allocating the sibling components lazily, when the index switches to them, is the fix for that and is not done here. Co-Authored-By: Claude Opus 5 <[email protected]> Change-Id: I4cccd9857d277b1a32017f0afcbd644e33388f2f Ext-ref: MB-73615 --- M asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/GlobalVirtualBufferCacheTest.java M asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/GlobalVirtualBufferCache.java 2 files changed, 178 insertions(+), 0 deletions(-) git pull ssh://asterix-gerrit.ics.uci.edu:29418/asterixdb refs/changes/27/21627/1 diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/GlobalVirtualBufferCacheTest.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/GlobalVirtualBufferCacheTest.java index 0a1fba2..3ebce64 100644 --- a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/GlobalVirtualBufferCacheTest.java +++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/GlobalVirtualBufferCacheTest.java @@ -24,6 +24,7 @@ import java.util.ArrayList; import java.util.Collections; import java.util.List; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicReference; import org.apache.asterix.app.bootstrap.TestNodeController; @@ -64,6 +65,7 @@ import org.apache.hyracks.storage.am.lsm.common.api.ILSMDiskComponent; import org.apache.hyracks.storage.am.lsm.common.api.ILSMIndex; import org.apache.hyracks.storage.am.lsm.common.impls.NoMergePolicyFactory; +import org.apache.hyracks.util.annotations.AiProvenance; import org.apache.logging.log4j.LogManager; import org.apache.logging.log4j.Logger; import org.junit.After; @@ -93,6 +95,9 @@ private static RecordTupleGenerator tupleGenerator; private static final int NUM_PARTITIONS = 2; + // bounds the fill loop so that a cache which cannot be filled fails the test rather than hanging it + private static final int MAX_IDLE_INDEXES = 256; + private static final int RECLAIM_TIMEOUT_SECONDS = 60; private static final long FILTERED_MEMORY_COMPONENT_SIZE = 16 * 1024l; @BeforeClass @@ -275,6 +280,78 @@ } } + /** + * A memory component which is allocated but holds no data still holds the pages its BTree pinned when it was + * created, and only a flush returns them. Since every memory component of an index is allocated on the first + * write, and only the one that is written is ever flushed, enough idle indexes can fill the cache with components + * that the flush selection will never pick, because it only picks components which are not empty. Writers then + * block on a full cache with nothing able to make it non-full, which is a permanent, silent stall. + * <p> + * Fills the cache with idle memory components and asserts that it recovers on its own. + */ + @AiProvenance(agent = AiProvenance.Agent.CLAUDE_OPUS_5, tool = AiProvenance.Tool.CLAUDE_CODE_CLI, contributionKind = AiProvenance.ContributionKind.TEST_GENERATED) + @Test + public void testReclaimsIdleMemoryComponentsWhenNothingIsFlushable() throws Exception { + final DataverseName dvName = DataverseName.createSinglePartName(StorageTestUtils.DATAVERSE_NAME); + final GlobalVirtualBufferCache globalVBC = (GlobalVirtualBufferCache) ncAppCtx.getVirtualBufferCache(); + final List<IIndexDataflowHelper> helpers = new ArrayList<>(); + int peakUsage = 0; + try { + // allocate idle indexes until the cache is full. Self-sizing rather than a fixed count, so that the test + // does not have to know how much of the budget the metadata datasets already hold. The loop reads the + // usage directly rather than calling isFull(), which would notify the flush thread and let the + // reclamation under test run while we are still filling + for (int i = 0; i < MAX_IDLE_INDEXES && peakUsage < globalVBC.getPageBudget(); i++) { + Dataset idleDataset = new TestDataset(dvName, "idle_ds" + i, dvName, StorageTestUtils.DATA_TYPE_NAME, + StorageTestUtils.NODE_GROUP_NAME, NoMergePolicyFactory.NAME, null, + new InternalDatasetDetails(null, PartitioningStrategy.HASH, StorageTestUtils.PARTITIONING_KEYS, + null, null, null, false, null, null), + null, DatasetType.INTERNAL, StorageTestUtils.DATASET_ID + 100 + i, 0); + PrimaryIndexInfo info = StorageTestUtils.createPrimaryIndex(nc, idleDataset, 0); + IIndexDataflowHelper helper = + new IndexDataflowHelperFactory(nc.getStorageManager(), info.getFileSplitProvider()) + .create(testCtxs[0].getJobletContext().getServiceContext(), 0); + helper.open(); + helpers.add(helper); + // allocates every memory component of the index and registers it with the cache, which is what the + // first write to a dataset does. None of them is written to, so all of them stay INACTIVE and empty + ((ILSMIndex) helper.getIndexInstance()).allocateMemoryComponents(); + peakUsage = Math.max(peakUsage, globalVBC.getUsage()); + } + Assert.assertTrue("could not fill the cache with " + helpers.size() + " idle index(es); peak usage was " + + peakUsage + " of " + globalVBC.getPageBudget() + " page(s)", + peakUsage >= globalVBC.getPageBudget()); + + // isFull() is what a blocked writer calls on every retry, and it is what notifies the flush thread. None + // of these components can be flushed, so without the reclamation this loop would never exit + final long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(RECLAIM_TIMEOUT_SECONDS); + while (globalVBC.isFull() && System.nanoTime() < deadline) { + Thread.sleep(50); + } + + final int usageAfter = globalVBC.getUsage(); + Assert.assertFalse( + "cache is still full at " + usageAfter + " of " + globalVBC.getPageBudget() + " page(s) after " + + RECLAIM_TIMEOUT_SECONDS + "s; idle memory components were not reclaimed", + globalVBC.isFull()); + Assert.assertTrue("no pages were returned to the cache: usage went from " + peakUsage + " to " + usageAfter + + " page(s)", usageAfter < peakUsage); + + // reclaiming is transparent: the next writer re-allocates, exactly as it does after a flush + for (IIndexDataflowHelper helper : helpers) { + ((ILSMIndex) helper.getIndexInstance()).allocateMemoryComponents(); + } + } finally { + for (IIndexDataflowHelper helper : helpers) { + try { + helper.destroy(); + } catch (Throwable t) { + LOGGER.warn("failed to destroy an idle index", t); + } + } + } + } + private void initializeNc() throws Exception { List<Pair<IOption, Object>> opts = new ArrayList<>(); opts.add(Pair.of(Option.STORAGE_MEMORYCOMPONENT_GLOBALBUDGET, 128 * 1024L)); diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/GlobalVirtualBufferCache.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/GlobalVirtualBufferCache.java index 43f04e4..5d95797 100644 --- a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/GlobalVirtualBufferCache.java +++ b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/GlobalVirtualBufferCache.java @@ -52,6 +52,7 @@ import org.apache.hyracks.storage.common.file.BufferedFileHandle; import org.apache.hyracks.storage.common.file.IFileMapManager; import org.apache.hyracks.util.ExitUtil; +import org.apache.hyracks.util.annotations.AiProvenance; import org.apache.logging.log4j.LogManager; import org.apache.logging.log4j.Logger; @@ -81,6 +82,9 @@ private final VirtualBufferCache vbc; private final AtomicBoolean isOpen = new AtomicBoolean(false); private final FlushThread flushThread = new FlushThread(); + // edge-triggers the starvation warning below: a blocked writer retries every 100ms, and each retry that + // finds the cache full notifies the flush thread, so a level-triggered warning would flood the log + private final AtomicBoolean starvationReported = new AtomicBoolean(); public GlobalVirtualBufferCache(ICacheMemoryAllocator allocator, StorageProperties storageProperties, int maxConcurrentFlushes) { @@ -485,12 +489,109 @@ private void scheduleFlush() throws HyracksDataException { ILSMIndex selectedIndex = null; + boolean starved; synchronized (GlobalVirtualBufferCache.this) { while (flushingIndexes.size() < maxConcurrentFlushes && ((selectedIndex = selectFlushIndex()) != null)) { LOGGER.debug("Waiting for flushing primary index {} to complete...", selectedIndex); flushingIndexes.add(selectedIndex); } + // nothing was selected and nothing is in flight: no flush completion is coming to free pages or to + // wake us again, so if the cache is full at this point it will stay that way on its own + starved = selectedIndex == null && flushingIndexes.isEmpty() && vbc.isFull(); + } + if (starved) { + reclaimIdleMemoryComponents(); + } else { + starvationReported.set(false); + } + } + + /** + * Frees the pages held by memory components which are allocated but hold no data, as a last resort when the + * cache is full and nothing can be flushed. + * <p> + * Every memory component of an index is allocated on the first write to it, and each one pins a metadata page + * and a root page as soon as it is created. Only the component that is written is flushed, and only a flush + * cleans a component up and returns its pages; the siblings stay allocated and empty until the dataset is + * closed, which nothing does under memory pressure. An index that is written once therefore leaves pages + * behind for data that never arrived, and enough such indexes fill the cache with components that + * {@link #selectFlushIndex()} will never select, because it only selects components which are not empty. + * Writers then block in {@code threadEnter} on {@code !vbc.isFull()} with nothing left to make it false. + * <p> + * Reclaiming costs the next writer to that index a re-allocation, which is the same one it already pays after + * every flush. We do not reset the components we clean up: they are INACTIVE, so they have already been reset + * (or were never used), and resetting again would decrement the index's scheduled flush count a second time. + */ + @AiProvenance(agent = AiProvenance.Agent.CLAUDE_OPUS_5, tool = AiProvenance.Tool.CLAUDE_CODE_CLI, contributionKind = AiProvenance.ContributionKind.GENERATED) + private void reclaimIdleMemoryComponents() throws HyracksDataException { + final List<ILSMIndex> candidates; + synchronized (GlobalVirtualBufferCache.this) { + // copied so that the op trackers below are taken without holding the cache monitor, preserving the + // cache-then-op-tracker order that selectFlushIndex() establishes + candidates = new ArrayList<>(primaryIndexes); + } + final int usageBefore = vbc.getUsage(); + int reclaimedFrom = 0; + for (ILSMIndex index : candidates) { + if (!vbc.isFull()) { + // enough has come back for the blocked writers to proceed; leave the rest allocated so that + // indexes which are still being written do not have to re-create their components + break; + } + if (isMetadataIndex(index)) { + // metadata indexes support full ACID transactions, and there are few of them; leave them alone + // for the same reason selectFlushIndex() will not wait on them + continue; + } + if (reclaimIdleMemoryComponents(index)) { + reclaimedFrom++; + } + } + final int reclaimedPages = usageBefore - vbc.getUsage(); + if (reclaimedPages > 0) { + LOGGER.info( + "reclaimed {} page(s) from the idle memory components of {} index(es); usage is now " + + "{} of {} page(s)", + reclaimedPages, reclaimedFrom, vbc.getUsage(), vbc.getPageBudget()); + starvationReported.set(false); + } else if (starvationReported.compareAndSet(false, true)) { + LOGGER.warn( + "global virtual buffer cache is full ({} of {} page(s)) with no memory component to " + + "flush and none idle to reclaim across {} primary index(es); writers are blocked until a " + + "dataset is closed. Raising storage.memorycomponent.globalbudget, or lowering " + + "storage.memorycomponent.pagesize or storage.memorycomponent.numcomponents, will admit " + + "more concurrently ingesting datasets", + vbc.getUsage(), vbc.getPageBudget(), candidates.size()); + } + } + + /** + * Frees the pages held by any memory component of this index which is allocated but holds no data. Taken + * under the index's operation tracker because that is what a writer holds while it activates a component, + * and activating one we are in the middle of destroying would race. + * + * @return whether any page was returned to the cache + */ + @AiProvenance(agent = AiProvenance.Agent.CLAUDE_OPUS_5, tool = AiProvenance.Tool.CLAUDE_CODE_CLI, contributionKind = AiProvenance.ContributionKind.GENERATED) + private boolean reclaimIdleMemoryComponents(ILSMIndex index) throws HyracksDataException { + final ILSMOperationTracker opTracker = index.getOperationTracker(); + final int usageBefore = vbc.getUsage(); + synchronized (opTracker) { + for (ILSMMemoryComponent memoryComponent : index.getMemoryComponents()) { + // INACTIVE is the whole test: a component holding data is never in that state, and cleanup() is + // a no-op on one which is not allocated + if (memoryComponent.getState() == ComponentState.INACTIVE && !memoryComponent.isModified()) { + memoryComponent.cleanup(); + } + } + final boolean reclaimed = vbc.getUsage() < usageBefore; + if (reclaimed) { + // writers blocked on a full cache wait on the op tracker; let them retry now rather than when + // their wait next times out + opTracker.notifyAll(); + } + return reclaimed; } } -- To view, visit https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/21627?usp=email To unsubscribe, or for help writing mail filters, visit https://asterix-gerrit.ics.uci.edu/settings?usp=email Gerrit-MessageType: newchange Gerrit-Project: asterixdb Gerrit-Branch: totoro Gerrit-Change-Id: I4cccd9857d277b1a32017f0afcbd644e33388f2f Gerrit-Change-Number: 21627 Gerrit-PatchSet: 1 Gerrit-Owner: Michael Blow <[email protected]>
