diff --git a/modules/benchmarks/src/main/java/org/apache/ignite/internal/benchmarks/jmh/cache/JmhPageEvictionBenchmark.java b/modules/benchmarks/src/main/java/org/apache/ignite/internal/benchmarks/jmh/cache/JmhPageEvictionBenchmark.java new file mode 100644 index 0000000000000..8124aafa96839 --- /dev/null +++ b/modules/benchmarks/src/main/java/org/apache/ignite/internal/benchmarks/jmh/cache/JmhPageEvictionBenchmark.java @@ -0,0 +1,158 @@ +/* + * 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.ignite.internal.benchmarks.jmh.cache; + +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import org.apache.ignite.Ignite; +import org.apache.ignite.IgniteCache; +import org.apache.ignite.IgniteDataStreamer; +import org.apache.ignite.Ignition; +import org.apache.ignite.configuration.CacheConfiguration; +import org.apache.ignite.configuration.DataPageEvictionMode; +import org.apache.ignite.configuration.DataRegionConfiguration; +import org.apache.ignite.configuration.DataStorageConfiguration; +import org.apache.ignite.configuration.IgniteConfiguration; +import org.apache.ignite.internal.benchmarks.jmh.runner.JmhIdeBenchmarkRunner; +import org.openjdk.jmh.annotations.Benchmark; +import org.openjdk.jmh.annotations.BenchmarkMode; +import org.openjdk.jmh.annotations.Fork; +import org.openjdk.jmh.annotations.Level; +import org.openjdk.jmh.annotations.Measurement; +import org.openjdk.jmh.annotations.Mode; +import org.openjdk.jmh.annotations.OutputTimeUnit; +import org.openjdk.jmh.annotations.Param; +import org.openjdk.jmh.annotations.Scope; +import org.openjdk.jmh.annotations.Setup; +import org.openjdk.jmh.annotations.State; +import org.openjdk.jmh.annotations.TearDown; +import org.openjdk.jmh.annotations.Threads; +import org.openjdk.jmh.annotations.Warmup; + +/** + * Measures the impact of size-aware page eviction on an in-memory data region. + *

+ * {@link #putSmall()} — small values within a bounded key range that keeps the region below the eviction threshold + * (size-aware reserve fast path). + *

+ * {@link #putLarge()} — large values against a pre-filled region near capacity (size-aware eviction loop). + */ +@State(Scope.Benchmark) +@Fork(1) +@Threads(4) +@BenchmarkMode(Mode.Throughput) +@OutputTimeUnit(TimeUnit.MILLISECONDS) +@Warmup(iterations = 3, time = 5) +@Measurement(iterations = 5, time = 10) +public class JmhPageEvictionBenchmark { + /** Default cache name. */ + private static final String CACHE_NAME = "default"; + + /** Small value size (bytes): a single data page, far below the empty-pages pool. */ + private static final int SMALL_VALUE_SIZE = 1024; + + /** Large value size (bytes): larger than the empty-pages pool in page terms. */ + private static final int LARGE_VALUE_SIZE = 2 * 1024 * 1024; + + /** Pre-fill entries for the LARGE scenario; exceeds region capacity to pin it at the eviction threshold. */ + private static final int PRE_FILL_ENTRIES = 400_000; + + /** Bounded key range for {@link #putSmall()} to keep the region below the eviction threshold. */ + private static final int SMALL_KEY_RANGE = 32_000; + + /** Benchmark scenario: selects the value size and the pre-fill strategy. */ + @Param({"SMALL", "LARGE"}) + private String scenario; + + /** Ignite cache. */ + private IgniteCache cache; + + /** Pre-allocated small value (reused to avoid allocation noise in the hot path). */ + private final byte[] smallVal = new byte[SMALL_VALUE_SIZE]; + + /** Pre-allocated large value (reused to avoid allocation noise in the hot path). */ + private final byte[] largeVal = new byte[LARGE_VALUE_SIZE]; + + /** Monotonic key source. */ + private final AtomicInteger keyGen = new AtomicInteger(); + + /** Small value put within a bounded key range (fast path, no eviction). */ + @Benchmark + public void putSmall() { + int key = keyGen.incrementAndGet() % SMALL_KEY_RANGE; + + cache.put(key, smallVal); + } + + /** Large value put against a nearly-full region (size-aware eviction loop). Single-threaded to avoid OOM. */ + @Benchmark + @Threads(1) + public void putLarge() { + int key = keyGen.incrementAndGet(); + + cache.put(key, largeVal); + } + + /** */ + @Setup(Level.Trial) + public void setup() { + DataStorageConfiguration dsCfg = new DataStorageConfiguration() + .setDefaultDataRegionConfiguration(new DataRegionConfiguration() + .setPersistenceEnabled(false) + .setMaxSize(256 * 1024L * 1024L) + .setPageEvictionMode(DataPageEvictionMode.RANDOM_LRU)); + + IgniteConfiguration cfg = new IgniteConfiguration() + .setIgniteInstanceName("test") + .setLocalHost("127.0.0.1") + .setDataStorageConfiguration(dsCfg); + + Ignite ignite = Ignition.start(cfg); + + cache = ignite.getOrCreateCache(new CacheConfiguration(CACHE_NAME).setBackups(0)); + + if ("LARGE".equalsIgnoreCase(scenario)) { + try (IgniteDataStreamer ldr = ignite.dataStreamer(CACHE_NAME)) { + ldr.perNodeBufferSize(1024); + + for (int i = 0; i < PRE_FILL_ENTRIES; i++) + ldr.addData(i, smallVal); + } + + keyGen.set(PRE_FILL_ENTRIES); + } + } + + /** */ + @TearDown + public void tearDown() { + Ignition.stopAll(true); + } + + /** + * Runs benchmark. + * + * @param args Ignored. + * @throws Exception If failed. + */ + public static void main(String[] args) throws Exception { + JmhIdeBenchmarkRunner.create() + .benchmarks(JmhPageEvictionBenchmark.class.getSimpleName()) + .run(); + } +} diff --git a/modules/binary/impl/src/main/java/org/apache/ignite/internal/binary/BinaryObjectImpl.java b/modules/binary/impl/src/main/java/org/apache/ignite/internal/binary/BinaryObjectImpl.java index 3359c73459de8..2c99d107627fa 100644 --- a/modules/binary/impl/src/main/java/org/apache/ignite/internal/binary/BinaryObjectImpl.java +++ b/modules/binary/impl/src/main/java/org/apache/ignite/internal/binary/BinaryObjectImpl.java @@ -218,6 +218,12 @@ public BinaryObjectImpl(BinaryContext ctx, byte[] arr, int start) { /** {@inheritDoc} */ @Override public int valueBytesLength(CacheObjectValueContext ctx) throws IgniteCheckedException { + if (valBytes == null) { + assert arr != null; + + valBytes = valueBytesFromArray(ctx); + } + return CacheObjectUtils.objectPutSize(valBytes.length); } diff --git a/modules/core/src/main/java/org/apache/ignite/configuration/DataRegionConfiguration.java b/modules/core/src/main/java/org/apache/ignite/configuration/DataRegionConfiguration.java index 1293ed7989486..e8ddb992aa34e 100644 --- a/modules/core/src/main/java/org/apache/ignite/configuration/DataRegionConfiguration.java +++ b/modules/core/src/main/java/org/apache/ignite/configuration/DataRegionConfiguration.java @@ -18,7 +18,6 @@ import java.io.Serializable; import org.apache.ignite.DataRegionMetrics; -import org.apache.ignite.internal.mem.IgniteOutOfMemoryException; import org.apache.ignite.internal.util.typedef.internal.S; import org.apache.ignite.mem.MemoryAllocator; import org.apache.ignite.mxbean.MetricsMxBean; @@ -346,9 +345,11 @@ public DataRegionConfiguration setEvictionThreshold(double evictionThreshold) { * Specifies the minimal number of empty pages to be present in reuse lists for this data region. * This parameter ensures that Ignite will be able to successfully evict old data entries when the size of * (key, value) pair is slightly larger than page size / 2. - * Increase this parameter if cache can contain very big entries (total size of pages in this pool should be enough - * to contain largest cache entry). - * Increase this parameter if {@link IgniteOutOfMemoryException} occurred with enabled page eviction. + * Since size-aware eviction automatically frees additional pages when the inserted row is larger than this pool, + * it is no longer required to increase this parameter up to the size of the largest cache entry. + * It also acts as the concurrency-safety margin for size-aware eviction: the reserve keeps this many empty pages + * on top of each row's own requirement, so the insert path is safe for up to {@code emptyPagesPoolSize} concurrent + * writers. * * @return Minimum number of empty pages in reuse list. */ @@ -360,9 +361,11 @@ public int getEmptyPagesPoolSize() { * Specifies the minimal number of empty pages to be present in reuse lists for this data region. * This parameter ensures that Ignite will be able to successfully evict old data entries when the size of * (key, value) pair is slightly larger than page size / 2. - * Increase this parameter if cache can contain very big entries (total size of pages in this pool should be enough - * to contain largest cache entry). - * Increase this parameter if {@link IgniteOutOfMemoryException} occurred with enabled page eviction. + * Since size-aware eviction automatically frees additional pages when the inserted row is larger than this pool, + * it is no longer required to increase this parameter up to the size of the largest cache entry. + * It also acts as the concurrency-safety margin for size-aware eviction: the reserve keeps this many empty pages + * on top of each row's own requirement, so the insert path is safe for up to {@code emptyPagesPoolSize} concurrent + * writers. * * @param emptyPagesPoolSize Empty pages pool size. * @return {@code this} for chaining. diff --git a/modules/core/src/main/java/org/apache/ignite/internal/pagemem/impl/PageMemoryNoStoreImpl.java b/modules/core/src/main/java/org/apache/ignite/internal/pagemem/impl/PageMemoryNoStoreImpl.java index bd5b48f8d8d8a..d9bd2b8bacaf0 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/pagemem/impl/PageMemoryNoStoreImpl.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/pagemem/impl/PageMemoryNoStoreImpl.java @@ -109,7 +109,7 @@ public class PageMemoryNoStoreImpl implements PageMemory { private static final int SEG_BITS = 4; /** Number of bits required to store segment index. */ - private static final int SEG_CNT = (1 << SEG_BITS); + public static final int SEG_CNT = (1 << SEG_BITS); /** Number of bits left to store page index. */ private static final int IDX_BITS = PageIdUtils.PAGE_IDX_SIZE - SEG_BITS; diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheEntryEx.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheEntryEx.java index bda9d3fbdc198..9755e56228058 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheEntryEx.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheEntryEx.java @@ -665,6 +665,10 @@ public GridCacheVersionedEntryEx versionedEntry(final boolean keepB * Sets new value if passed in version matches the current version * (used for read-through only). * + *

This overload reserves page space internally (before acquiring the entry lock). Use + * {@link #versionedValue(CacheObject, GridCacheVersion, GridCacheVersion, IgniteCacheExpiryPolicy, + * ReaderArguments, long)} when the caller has already reserved (or the entry is already locked). + * * @param val New value. * @param curVer Version to match or {@code null} if match is not required. * @param newVer Version to set. @@ -681,6 +685,29 @@ public EntryGetResult versionedValue(CacheObject val, @Nullable ReaderArguments readerArgs) throws IgniteCheckedException, GridCacheEntryRemovedException; + /** + * Sets new value if passed in version matches the current version (used for read-through only). + * + * @param val New value. + * @param curVer Version to match or {@code null} if match is not required. + * @param newVer Version to set. + * @param loadExpiryPlc Expiry policy if entry is loaded from store. + * @param readerArgs Reader will be added if not null. + * @param reserved Number of pages reserved by a prior call to {@code reserveEntrySpace} (before acquiring the + * entry lock), or {@code 0} if no reservation was made. + * @return Current version and value. + * @throws IgniteCheckedException If index could not be updated. + * @throws GridCacheEntryRemovedException If entry was removed. + */ + public EntryGetResult versionedValue( + CacheObject val, + @Nullable GridCacheVersion curVer, + @Nullable GridCacheVersion newVer, + @Nullable IgniteCacheExpiryPolicy loadExpiryPlc, + @Nullable ReaderArguments readerArgs, + long reserved + ) throws IgniteCheckedException, GridCacheEntryRemovedException; + /** * Checks if the candidate is either owner or pending. * diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMapEntry.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMapEntry.java index c0c5bbaf68c6e..ca4ae6327df42 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMapEntry.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMapEntry.java @@ -55,9 +55,10 @@ import org.apache.ignite.internal.processors.cache.extras.GridCacheTtlEntryExtras; import org.apache.ignite.internal.processors.cache.persistence.CacheDataRow; import org.apache.ignite.internal.processors.cache.persistence.CacheDataRowAdapter; -import org.apache.ignite.internal.processors.cache.persistence.DataRegion; +import org.apache.ignite.internal.processors.cache.persistence.RowStore; import org.apache.ignite.internal.processors.cache.persistence.StorageException; import org.apache.ignite.internal.processors.cache.persistence.snapshot.dump.DumpEntryChangeListener; +import org.apache.ignite.internal.processors.cache.persistence.tree.io.CacheVersionIO; import org.apache.ignite.internal.processors.cache.persistence.wal.WALPointer; import org.apache.ignite.internal.processors.cache.query.continuous.CacheContinuousQueryListener; import org.apache.ignite.internal.processors.cache.transactions.IgniteInternalTx; @@ -903,8 +904,10 @@ private EntryGetResult entryGetResult(CacheObject val, GridCacheVersion ver, boo boolean touch = false; + long reserved; + try { - ensureFreeSpace(); + reserved = reserveEntrySpace(rowSize(ret, ver)); lockEntry(); @@ -950,6 +953,8 @@ private EntryGetResult entryGetResult(CacheObject val, GridCacheVersion ver, boo } finally { unlockEntry(); + + releaseReserved(reserved); } touch = true; @@ -1006,7 +1011,7 @@ protected void recordNodeId(UUID nodeId, AffinityTopologyVersion topVer) { long updateCntr0; - ensureFreeSpace(); + long reserved = reserveEntrySpace(rowSize(val, ver)); lockListenerReadLock(); lockEntry(); @@ -1159,6 +1164,8 @@ else if (interceptorVal != val0) finally { unlockEntry(); unlockListenerReadLock(); + + releaseReserved(reserved); } onUpdateFinished(updateCntr0); @@ -1463,8 +1470,12 @@ else if (log.isDebugEnabled()) AtomicCacheUpdateClosure c; + long reserved = 0; + if (!primary && !isNear()) - ensureFreeSpace(); + reserved = reserveEntrySpace(rowSize( + op == GridCacheOperation.UPDATE && writeObj instanceof CacheObject ? (CacheObject)writeObj : null, + newVer)); lockListenerReadLock(); lockEntry(); @@ -1724,6 +1735,8 @@ else if (updateMetrics && REMOVE_NO_VAL.equals(updateRes.outcome()) finally { unlockEntry(); unlockListenerReadLock(); + + releaseReserved(reserved); } onUpdateFinished(c.updateRes.updateCounter()); @@ -2473,7 +2486,7 @@ private boolean skipInterceptor(@Nullable GridCacheVersion explicitVer) { ) throws IgniteCheckedException, GridCacheEntryRemovedException { assert !primary || !(preload || fromStore); - ensureFreeSpace(); + long reserved = reserveEntrySpace(rowSize(val, ver)); boolean deferred = false; boolean obsolete = false; @@ -2615,6 +2628,8 @@ else if (deletedUnlocked()) unlockEntry(); unlockListenerReadLock(); + releaseReserved(reserved); + // It is necessary to execute these callbacks outside of lock to avoid deadlocks. if (obsolete) { @@ -2705,11 +2720,26 @@ protected long nextPartitionCounter(IgniteInternalTx tx, @Nullable Long updateCn } /** {@inheritDoc} */ - @Override public EntryGetResult versionedValue(CacheObject val, + @Override public EntryGetResult versionedValue( + CacheObject val, GridCacheVersion curVer, GridCacheVersion newVer, @Nullable IgniteCacheExpiryPolicy loadExpiryPlc, @Nullable ReaderArguments readerArgs + ) throws IgniteCheckedException, GridCacheEntryRemovedException { + long reserved = reserveEntrySpace(rowSize(val, newVer != null ? newVer : ver)); + + return versionedValue(val, curVer, newVer, loadExpiryPlc, readerArgs, reserved); + } + + /** {@inheritDoc} */ + @Override public EntryGetResult versionedValue( + CacheObject val, + GridCacheVersion curVer, + GridCacheVersion newVer, + @Nullable IgniteCacheExpiryPolicy loadExpiryPlc, + @Nullable ReaderArguments readerArgs, + long reserved ) throws IgniteCheckedException, GridCacheEntryRemovedException { lockEntry(); @@ -2765,6 +2795,8 @@ protected long nextPartitionCounter(IgniteInternalTx tx, @Nullable Long updateCn } finally { unlockEntry(); + + releaseReserved(reserved); } } @@ -3619,13 +3651,49 @@ protected void removeValue() throws IgniteCheckedException { } /** - * Evicts necessary number of data pages if per-page eviction is configured in current {@link DataRegion}. + * Reserves pages for the row about to be inserted. Must be called before {@link #lockEntry()} — eviction may + * block on other entries' locks. + * + * @param rowSize Size of the row to be inserted. + * @return Number of reserved pages, or {@code 0} if reservation is not engaged. + * @throws IgniteCheckedException If failed. */ - private void ensureFreeSpace() throws IgniteCheckedException { - // Deadlock alert: evicting data page causes removing (and locking) all entries on the page one by one. + private long reserveEntrySpace(int rowSize) throws IgniteCheckedException { + // Deadlock alert: evicting a data page locks all entries on it. assert !lock.isHeldByCurrentThread(); - cctx.shared().database().ensureFreeSpace(cctx.dataRegion()); + // Near entries keep values on heap and never write rows to the free list (storeValue is a no-op for them). + if (isNear()) + return 0; + + return cctx.shared().database().reserveInRegion(cctx.dataRegion(), rowSize); + } + + /** + * Releases a reservation made by {@link #reserveEntrySpace(int)}. The whole claim is released as a unit: + * the insert does not consume the reservation page by page, so the release is exact on any outcome + * (including an exception in the middle of the row insert). + * + * @param reserved Number of reserved pages. + */ + private void releaseReserved(long reserved) { + if (reserved > 0) + cctx.shared().database().releaseInRegion(cctx.dataRegion(), reserved); + } + + /** + * Computes the data row size for the given value and version (see {@link RowStore#rowSize}). + * Used to reserve pages before the entry lock is acquired. + * + * @param val Value (maybe {@code null} if unknown — under-reserves safely). + * @param ver Version. + * @return Estimated row size in bytes. + * @throws IgniteCheckedException If failed. + */ + private int rowSize(@Nullable CacheObject val, GridCacheVersion ver) throws IgniteCheckedException { + int cacheId = cctx.group().storeCacheIdInDataPage() ? cctx.cacheId() : CU.UNDEFINED_CACHE_ID; + + return RowStore.rowSize(key, val, CacheVersionIO.size(ver, false), cctx.cacheObjectContext(), cacheId); } /** diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java index 496f579a33fcf..7a125d2bf597b 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java @@ -55,6 +55,7 @@ import org.apache.ignite.internal.processors.cache.CacheInvokeResult; import org.apache.ignite.internal.processors.cache.CacheLazyEntry; import org.apache.ignite.internal.processors.cache.CacheObject; +import org.apache.ignite.internal.processors.cache.CacheObjectValueContext; import org.apache.ignite.internal.processors.cache.CacheOperationContext; import org.apache.ignite.internal.processors.cache.CacheStoppedException; import org.apache.ignite.internal.processors.cache.CacheStorePartialUpdateException; @@ -90,7 +91,10 @@ import org.apache.ignite.internal.processors.cache.dr.GridCacheDrExpirationInfo; import org.apache.ignite.internal.processors.cache.dr.GridCacheDrInfo; import org.apache.ignite.internal.processors.cache.persistence.CacheDataRow; +import org.apache.ignite.internal.processors.cache.persistence.DataRegion; +import org.apache.ignite.internal.processors.cache.persistence.RowStore; import org.apache.ignite.internal.processors.cache.persistence.StorageException; +import org.apache.ignite.internal.processors.cache.persistence.tree.io.CacheVersionIO; import org.apache.ignite.internal.processors.cache.transactions.IgniteTxLocalEx; import org.apache.ignite.internal.processors.cache.version.GridCacheVersion; import org.apache.ignite.internal.processors.cache.version.GridCacheVersionConflictContext; @@ -162,6 +166,13 @@ public class GridDhtAtomicCache extends GridDhtCacheAdapter { private static final int DEFERRED_UPDATE_RESPONSE_TIMEOUT = Integer.getInteger(IGNITE_ATOMIC_DEFERRED_ACK_TIMEOUT, DFLT_ATOMIC_DEFERRED_ACK_TIMEOUT); + /** + * Upper bound of a single batch reservation in bytes: caps the sum of row sizes passed to + * {@code FreeList.reserve} to keep its page arithmetic within {@code int}. Batches above the cap are silently + * under-reserved (they are pathological — default per-request size limits are far below it). + */ + private static final int MAX_RESERVE_BATCH_BYTES = 1_000_000_000; + /** */ private final ThreadLocal> defRes = new ThreadLocal>() { @@ -1806,9 +1817,13 @@ private void updateAllAsyncInternal0( ctx.shared().database().checkpointReadLock(); + long batchReserved = 0; + try { ctx.shared().database().ensureFreeSpace(ctx.dataRegion()); + batchReserved = reserveBatchSpace(req); + // If batch store update is enabled, we need to lock all entries. // First, need to acquire locks on cache entries, then check filter. List locked = lockEntries(req, req.topologyVersion()); @@ -1949,6 +1964,8 @@ private void updateAllAsyncInternal0( return; } finally { + releaseBatchSpace(batchReserved); + ctx.shared().database().checkpointReadUnlock(); } @@ -2489,7 +2506,7 @@ private void reloadIfNeeded(final List entries) throws Ignite try { GridCacheVersion ver = entry.version(); - entry.versionedValue(ctx.toCacheObject(v), null, ver, null, null); + entry.versionedValue(ctx.toCacheObject(v), null, ver, null, null, 0L); } catch (GridCacheEntryRemovedException e) { assert false : "Entry should not get obsolete while holding lock [entry=" + entry + @@ -2978,6 +2995,62 @@ else if (GridDhtCacheEntry.ReaderId.contains(readers, nearNode.id())) { } } + /** + * Reserves free-list pages for the total size of all rows in the batch: size-aware admission control for + * the primary atomic path, which acquires entry locks via {@link #lockEntries} before {@code innerUpdate} and so + * cannot reserve (or evict) per entry there. Reserving before {@link #lockEntries} guarantees every row of a + * batch with known value sizes can be inserted without an out-of-memory in {@code FreeList.insertDataRow}. + * + *

Best-effort only: {@code TRANSFORM} rows reserve the key sizes (processor results are unknown), and batches + * above {@link #MAX_RESERVE_BATCH_BYTES} are reserved for the capped size. + * + * @param req Update request. + * @return Reserved pages to release via {@link #releaseBatchSpace}, or {@code 0} if reservation is not engaged. + * @throws IgniteCheckedException If failed. + */ + private long reserveBatchSpace(GridNearAtomicAbstractUpdateRequest req) throws IgniteCheckedException { + DataRegion dataRegion = ctx.dataRegion(); + + if (dataRegion == null || dataRegion.config().isPersistenceEnabled()) + return 0; + + CacheObjectValueContext coctx = ctx.cacheObjectContext(); + + int cacheId = ctx.group().storeCacheIdInDataPage() ? ctx.cacheId() : CU.UNDEFINED_CACHE_ID; + + long sumRowSize = 0; + + for (int i = 0; i < req.size(); i++) { + KeyCacheObject key = req.key(i); + + CacheObject val = req.writeValue(i); + + if (val == null && req.operation() == GridCacheOperation.UPDATE) + val = req.value(i); + + // CacheVersionIO.SIZE_V2 is the maximum of both version protocols (V1 is smaller), so this is an upper + // bound: the actual rows can not be larger than estimated. + sumRowSize += RowStore.rowSize(key, val, CacheVersionIO.SIZE_V2, coctx, cacheId); + } + + if (sumRowSize == 0) + return 0; + + int reserveSize = sumRowSize > MAX_RESERVE_BATCH_BYTES ? MAX_RESERVE_BATCH_BYTES : (int)sumRowSize; + + return ctx.shared().database().reserveInRegion(dataRegion, reserveSize); + } + + /** + * Releases the batch-level reservation made by {@link #reserveBatchSpace}. + * + * @param reserved Reserved pages. + */ + private void releaseBatchSpace(long reserved) { + if (reserved != 0) + ctx.shared().database().releaseInRegion(ctx.dataRegion(), reserved); + } + /** * Acquires java-level locks on cache entries. Returns collection of locked entries. * diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/CacheDataRowAdapter.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/CacheDataRowAdapter.java index 810423838a3a8..9625bcc91fc3c 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/CacheDataRowAdapter.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/CacheDataRowAdapter.java @@ -890,11 +890,7 @@ public boolean isReady() { /** {@inheritDoc} */ @Override public int size() throws IgniteCheckedException { - int len = key().valueBytesLength(null); - - len += value().valueBytesLength(null) + CacheVersionIO.size(version(), false) + 8; - - return len + (cacheId() != 0 ? 4 : 0); + return RowStore.rowSize(key(), value(), CacheVersionIO.size(version(), false), null, cacheId()); } /** diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/IgniteCacheDatabaseSharedManager.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/IgniteCacheDatabaseSharedManager.java index e39607e65f29b..77046c62df0f9 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/IgniteCacheDatabaseSharedManager.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/IgniteCacheDatabaseSharedManager.java @@ -39,15 +39,12 @@ import org.apache.ignite.configuration.DataRegionConfiguration; import org.apache.ignite.configuration.DataStorageConfiguration; import org.apache.ignite.configuration.WarmUpConfiguration; -import org.apache.ignite.failure.FailureContext; -import org.apache.ignite.failure.FailureType; import org.apache.ignite.internal.GridKernalContext; import org.apache.ignite.internal.IgniteInternalFuture; import org.apache.ignite.internal.IgniteKernal; import org.apache.ignite.internal.managers.discovery.GridDiscoveryManager; import org.apache.ignite.internal.mem.DirectMemoryProvider; import org.apache.ignite.internal.mem.DirectMemoryRegion; -import org.apache.ignite.internal.mem.IgniteOutOfMemoryException; import org.apache.ignite.internal.mem.file.MappedFileMemoryProvider; import org.apache.ignite.internal.mem.unsafe.UnsafeMemoryProvider; import org.apache.ignite.internal.pagemem.PageIdAllocator; @@ -68,7 +65,6 @@ import org.apache.ignite.internal.processors.cache.persistence.evict.RandomLruPageEvictionTracker; import org.apache.ignite.internal.processors.cache.persistence.filename.NodeFileTree; import org.apache.ignite.internal.processors.cache.persistence.filename.PdsFolderSettings; -import org.apache.ignite.internal.processors.cache.persistence.freelist.AbstractFreeList; import org.apache.ignite.internal.processors.cache.persistence.freelist.CacheFreeList; import org.apache.ignite.internal.processors.cache.persistence.freelist.FreeList; import org.apache.ignite.internal.processors.cache.persistence.metastorage.MetaStorage; @@ -1160,70 +1156,10 @@ public WALPointer latestWalPointerReservedForPreloading() { } /** - * Checks that the given {@code region} has enough space for putting a new entry. - * - * This method makes sense then and only then - * the data region is not persisted {@link DataRegionConfiguration#isPersistenceEnabled()} - * and page eviction is disabled {@link DataPageEvictionMode#DISABLED}. - * - * The non-persistent region should reserve a number of pages to support a free list {@link AbstractFreeList}. - * For example, removing a row from underlying store may require allocating a new data page - * in order to move a tracked page from one bucket to another one which does not have a free space for a new stripe. - * See {@link AbstractFreeList#removeDataRowByLink}. - * Therefore, inserting a new entry should be prevented in case of some threshold is exceeded. - * - * @param region Data region to be checked. - * @param dataRowSize Size of data row to be inserted. - * @throws IgniteOutOfMemoryException In case of the given data region does not have enough free space - * for putting a new entry. - */ - public void ensureFreeSpaceForInsert(DataRegion region, int dataRowSize) throws IgniteOutOfMemoryException { - if (region == null) - return; - - DataRegionConfiguration regCfg = region.config(); - - if (regCfg.getPageEvictionMode() != DataPageEvictionMode.DISABLED || regCfg.isPersistenceEnabled()) - return; - - long memorySize = regCfg.getMaxSize(); - - PageMemory pageMem = region.pageMemory(); - - CacheFreeList freeList = freeListMap.get(regCfg.getName()); - - long nonEmptyPages = (pageMem.loadedPages() - freeList.emptyDataPages()); - - // The maximum number of pages that can be allocated (memorySize / systemPageSize) - // should be greater or equal to pages required for inserting a new entry plus - // the current number of non-empty pages plus the number of pages that may be required in order to move - // all pages to a reuse bucket, that is equal to nonEmptyPages * 8 / pageSize, where 8 is the size of a link. - // Note that not the whole page can be used to storing links, - // see PagesListNodeIO and PagesListMetaIO#getCapacity(), so we pessimistically multiply the result on 1.5, - // in any way, the number of required pages is less than 1 percent. - boolean oomThreshold = (memorySize / pageMem.systemPageSize()) < - ((double)dataRowSize / pageMem.pageSize() + nonEmptyPages * (8.0 * 1.5 / pageMem.pageSize() + 1) + 256 /*one page per bucket*/); - - if (oomThreshold) { - IgniteOutOfMemoryException oom = new IgniteOutOfMemoryException("Out of memory in data region [" + - "name=" + regCfg.getName() + - ", initSize=" + U.readableSize(regCfg.getInitialSize(), false) + - ", maxSize=" + U.readableSize(regCfg.getMaxSize(), false) + - ", persistenceEnabled=" + regCfg.isPersistenceEnabled() + "] Try the following:" + U.nl() + - " ^-- Increase maximum off-heap memory size (DataRegionConfiguration.maxSize)" + U.nl() + - " ^-- Enable Ignite persistence (DataRegionConfiguration.persistenceEnabled)" + U.nl() + - " ^-- Enable eviction or expiration policies" - ); - - if (cctx.kernalContext() != null) - cctx.kernalContext().failure().process(new FailureContext(FailureType.CRITICAL_ERROR, oom)); - - throw oom; - } - } - - /** - * See {@code GridCacheMapEntry#ensureFreeSpace()} + * Evicts data pages from the given region until the configured eviction threshold is no longer exceeded or the + * empty-pages pool is refilled. This is the coarse, size-agnostic admission gate called at the batch level (before + * per-entry locks are acquired); size-aware per-row admission is handled separately by + * {@link org.apache.ignite.internal.processors.cache.persistence.freelist.AbstractFreeList#reserve(int)}. * * @param memPlc data region. */ @@ -1243,6 +1179,35 @@ public void ensureFreeSpace(DataRegion memPlc) throws IgniteCheckedException { } } + /** + * Reserves free-list pages in the given data region for a row (or a batch of rows totaling {@code rowSize} bytes) + * about to be inserted. See {@link FreeList#reserve(int)} for the caller contract (no entry/page locks held). + * + * @param memPlc Data region ({@code null} is tolerated). + * @param rowSize Size of the row, or the total size of all rows of a batch. + * @return Number of reserved pages to be passed to {@link FreeList#insertDataRow} and released via + * {@link #releaseInRegion(DataRegion, long)}, or {@code 0} if the reservation is not engaged for this region + * (persistent data region or disabled page eviction). + * @throws IgniteCheckedException If failed. + */ + public long reserveInRegion(DataRegion memPlc, int rowSize) throws IgniteCheckedException { + if (memPlc == null || memPlc.config().isPersistenceEnabled()) + return 0; + + return freeList(memPlc.config().getName()).reserve(rowSize); + } + + /** + * Releases the unconsumed portion of a reservation made by {@link #reserveInRegion(DataRegion, int)}. + * + * @param memPlc Data region the reservation was made in ({@code null} is tolerated). + * @param pages Number of pages to release. + */ + public void releaseInRegion(DataRegion memPlc, long pages) { + if (memPlc != null && pages != 0) + freeList(memPlc.config().getName()).releaseReserved(pages); + } + /** * @param memCfg memory configuration with common parameters. * @param plcCfg data region with PageMemory specific parameters. diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/RowStore.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/RowStore.java index cffcf9b1e5be0..a9af7a2c80f82 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/RowStore.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/RowStore.java @@ -24,11 +24,17 @@ import org.apache.ignite.internal.pagemem.PageIdUtils; import org.apache.ignite.internal.pagemem.PageMemory; import org.apache.ignite.internal.processors.cache.CacheGroupContext; +import org.apache.ignite.internal.processors.cache.CacheObject; import org.apache.ignite.internal.processors.cache.CacheObjectContext; +import org.apache.ignite.internal.processors.cache.CacheObjectValueContext; import org.apache.ignite.internal.processors.cache.GridCacheSharedContext; +import org.apache.ignite.internal.processors.cache.KeyCacheObject; import org.apache.ignite.internal.processors.cache.persistence.freelist.FreeList; +import org.apache.ignite.internal.processors.cache.persistence.tree.io.CacheVersionIO; +import org.apache.ignite.internal.processors.cache.version.GridCacheVersion; import org.apache.ignite.internal.processors.query.GridQueryRowCacheCleaner; import org.apache.ignite.internal.util.typedef.internal.U; +import org.jetbrains.annotations.Nullable; /** * Data store for H2 rows. @@ -101,11 +107,12 @@ public void removeRow(long link, IoStatisticsHolder statHolder) throws IgniteChe /** * @param row Row. + * @param statHolder Statistics holder to track IO operations. * @throws IgniteCheckedException If failed. */ public void addRow(CacheDataRow row, IoStatisticsHolder statHolder) throws IgniteCheckedException { if (!persistenceEnabled) { - ctx.database().ensureFreeSpaceForInsert(grp.dataRegion(), row.size()); + freeList.checkFitsForInsert(row.size()); freeList.insertDataRow(row, statHolder); } @@ -132,8 +139,7 @@ public void addRow(CacheDataRow row, IoStatisticsHolder statHolder) throws Ignit * @param statHolder Statistics holder to track IO operations. * @throws IgniteCheckedException If failed. */ - public void addRows(Collection rows, - IoStatisticsHolder statHolder) throws IgniteCheckedException { + public void addRows(Collection rows, IoStatisticsHolder statHolder) throws IgniteCheckedException { assert ctx.database().checkpointLockIsHeldByThread(); freeList.insertDataRows(rows, statHolder); @@ -173,4 +179,31 @@ public void setRowCacheCleaner(Supplier rowCacheCleane this.rowCacheCleaner = rowCacheCleaner; } + + /** + * Computes the size a row with the given key and value occupies in page memory. Single source of truth for the + * row size: {@link CacheDataRowAdapter#size()} delegates here, and callers use it to reserve free-list pages + * before the actual row is created. + * + * @param key Row key. + * @param val Row value ({@code null} if unknown — under-reserves safely). + * @param verSize Serialized version size (see {@link CacheVersionIO#size(GridCacheVersion, boolean)}). + * @param coctx Cache object context to compute key/value bytes lengths. + * @param cacheId Cache ID stored in the data page. + * @return Row size in bytes. + * @throws IgniteCheckedException If failed to compute key or value bytes length. + */ + public static int rowSize( + KeyCacheObject key, + @Nullable CacheObject val, + int verSize, + CacheObjectValueContext coctx, + int cacheId + ) throws IgniteCheckedException { + int len = key.valueBytesLength(coctx); + + len += (val != null ? val.valueBytesLength(coctx) : 0) + verSize + 8; + + return len + (cacheId != 0 ? 4 : 0); + } } diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/evict/FairFifoPageEvictionTracker.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/evict/FairFifoPageEvictionTracker.java index 04454d05acbac..53b0609b670a7 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/evict/FairFifoPageEvictionTracker.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/evict/FairFifoPageEvictionTracker.java @@ -65,8 +65,10 @@ public FairFifoPageEvictionTracker( } /** {@inheritDoc} */ - @Override public synchronized void evictDataPage() throws IgniteCheckedException { - evictDataPage(pageUsageList.pollFirst()); + @Override public synchronized boolean evictDataPage() throws IgniteCheckedException { + Integer pageIdx = pageUsageList.pollFirst(); + + return pageIdx != null && evictDataPage(pageIdx); } /** {@inheritDoc} */ diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/evict/NoOpPageEvictionTracker.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/evict/NoOpPageEvictionTracker.java index 4409e35e7446b..7f11ffd4df9ef 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/evict/NoOpPageEvictionTracker.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/evict/NoOpPageEvictionTracker.java @@ -38,8 +38,8 @@ public class NoOpPageEvictionTracker implements PageEvictionTracker { } /** {@inheritDoc} */ - @Override public void evictDataPage() { - // No-op. + @Override public boolean evictDataPage() { + return false; } /** {@inheritDoc} */ diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/evict/PageEvictionTracker.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/evict/PageEvictionTracker.java index 82970fb52521c..ff5cb78966d5c 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/evict/PageEvictionTracker.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/evict/PageEvictionTracker.java @@ -44,9 +44,10 @@ public interface PageEvictionTracker extends LifecycleAware { * Method guarantees removing at least one entry from "evicted" data page. Removing all entries may be * not possible, as some of them can be used by active transactions. * + * @return {@code true} if at least one data row has been evicted. * @throws IgniteCheckedException In case of page memory error. */ - public void evictDataPage() throws IgniteCheckedException; + public boolean evictDataPage() throws IgniteCheckedException; /** * Call this method when last entry is removed from data page. diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/evict/Random2LruPageEvictionTracker.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/evict/Random2LruPageEvictionTracker.java index 01300ffb73fa0..e075ec1f573c1 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/evict/Random2LruPageEvictionTracker.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/evict/Random2LruPageEvictionTracker.java @@ -123,7 +123,7 @@ public Random2LruPageEvictionTracker( } /** {@inheritDoc} */ - @Override public void evictDataPage() throws IgniteCheckedException { + @Override public boolean evictDataPage() throws IgniteCheckedException { ThreadLocalRandom rnd = ThreadLocalRandom.current(); int evictAttemptsCnt = 0; @@ -161,7 +161,19 @@ public Random2LruPageEvictionTracker( firstTs = first(trackingData); - assert firstTs >= 0 : "[firstTs=" + firstTs + ", secondTs=" + second(trackingData) + "]"; + // The link read above may be stale: a concurrently deleted row frees the head page, which can be + // re-used as an intermediate fragment holding a link again. Legal state: skip and resample. + if (firstTs < 0) { + sampleSpinCnt++; + + if (sampleSpinCnt > SAMPLE_SPIN_LIMIT) { + LT.warn(log, "Too many attempts to choose data page: " + SAMPLE_SPIN_LIMIT); + + return false; + } + + continue; + } } } @@ -187,17 +199,19 @@ public Random2LruPageEvictionTracker( if (sampleSpinCnt > SAMPLE_SPIN_LIMIT) { LT.warn(log, "Too many attempts to choose data page: " + SAMPLE_SPIN_LIMIT); - return; + return false; } } if (evictDataPage(pageIdx(lruTrackingIdx))) - return; + return true; evictAttemptsCnt++; } LT.warn(log, "Too many failed attempts to evict page: " + EVICT_ATTEMPTS_LIMIT); + + return false; } /** {@inheritDoc} */ diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/evict/RandomLruPageEvictionTracker.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/evict/RandomLruPageEvictionTracker.java index b940681403c15..217abc826d6d6 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/evict/RandomLruPageEvictionTracker.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/evict/RandomLruPageEvictionTracker.java @@ -109,7 +109,7 @@ public RandomLruPageEvictionTracker( } /** {@inheritDoc} */ - @Override public void evictDataPage() throws IgniteCheckedException { + @Override public boolean evictDataPage() throws IgniteCheckedException { ThreadLocalRandom rnd = ThreadLocalRandom.current(); int evictAttemptsCnt = 0; @@ -141,7 +141,19 @@ public RandomLruPageEvictionTracker( compactTs = GridUnsafe.getIntVolatile(null, trackingArrPtr + sampleTrackingIdx * 4L); - assert compactTs >= 0 : "[compactTs=" + compactTs + "]"; + // The link read above may be stale: a concurrently deleted row frees the head page, which can be + // re-used as an intermediate fragment holding a link again. Legal state: skip and resample. + if (compactTs < 0) { + sampleSpinCnt++; + + if (sampleSpinCnt > SAMPLE_SPIN_LIMIT) { + LT.warn(log, "Too many attempts to choose data page: " + SAMPLE_SPIN_LIMIT); + + return false; + } + + continue; + } } } @@ -161,17 +173,19 @@ public RandomLruPageEvictionTracker( if (sampleSpinCnt > SAMPLE_SPIN_LIMIT) { LT.warn(log, "Too many attempts to choose data page: " + SAMPLE_SPIN_LIMIT); - return; + return false; } } if (evictDataPage(pageIdx(lruTrackingIdx))) - return; + return true; evictAttemptsCnt++; } LT.warn(log, "Too many failed attempts to evict page: " + EVICT_ATTEMPTS_LIMIT); + + return false; } /** {@inheritDoc} */ diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/freelist/AbstractFreeList.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/freelist/AbstractFreeList.java index 83e05562a4d34..028fdd2989bf4 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/freelist/AbstractFreeList.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/freelist/AbstractFreeList.java @@ -19,11 +19,18 @@ import java.util.Arrays; import java.util.Collection; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicLong; import java.util.concurrent.atomic.AtomicReferenceArray; +import java.util.concurrent.locks.LockSupport; import org.apache.ignite.IgniteCheckedException; import org.apache.ignite.IgniteLogger; +import org.apache.ignite.configuration.DataPageEvictionMode; +import org.apache.ignite.configuration.DataRegionConfiguration; +import org.apache.ignite.failure.FailureContext; +import org.apache.ignite.failure.FailureType; import org.apache.ignite.internal.GridKernalContext; +import org.apache.ignite.internal.mem.IgniteOutOfMemoryException; import org.apache.ignite.internal.metric.IoStatisticsHolder; import org.apache.ignite.internal.metric.IoStatisticsHolderNoOp; import org.apache.ignite.internal.pagemem.PageIdAllocator; @@ -52,6 +59,7 @@ import org.jetbrains.annotations.Nullable; import static org.apache.ignite.internal.pagemem.PageIdAllocator.FLAG_DATA; +import static org.apache.ignite.internal.pagemem.impl.PageMemoryNoStoreImpl.SEG_CNT; /** */ @@ -74,6 +82,31 @@ public abstract class AbstractFreeList extends PagesList imp /** */ private static final int MIN_PAGE_FREE_SPACE = 8; + /** Initial backoff (nanoseconds) between consecutive no-progress eviction attempts. */ + private static final long EVICTION_BACKOFF_START_NANOS = 50_000L; + + /** Upper bound (nanoseconds) for the backoff between no-progress eviction attempts. */ + private static final long EVICTION_BACKOFF_MAX_NANOS = 1_000_000L; + + /** + * Maximum time (milliseconds) the size-aware eviction guard is willing to wait without either an eviction or a + * new highest empty-pages count before failing with an out-of-memory. + */ + private static final long EVICTION_NO_PROGRESS_TIMEOUT_MILLIS = 1_000L; + + /** + * Hard upper bound (milliseconds) on the total duration of one size-aware eviction cycle, regardless of + * per-iteration progress. + */ + private static final long EVICTION_MAX_CYCLE_TIME_MILLIS = 5 * EVICTION_NO_PROGRESS_TIMEOUT_MILLIS; + + /** + * Number of compare-and-set attempts in {@link #tryReserve(int)}. A deficit (the space is genuinely unavailable) + * defers the row immediately; only a lost CAS race (concurrent reservers) is retried, up to this many attempts, + * before the row is deferred to the blocking context as well. + */ + private static final int TRY_RESERVE_CAS_ATTEMPTS = 4; + /** * Step between buckets in free list, measured in powers of two. * For example, for page size 4096 and 256 buckets, shift is 4 and step is 16 bytes. @@ -98,6 +131,18 @@ public abstract class AbstractFreeList extends PagesList imp /** */ private final PageEvictionTracker evictionTracker; + /** Whether the page reservation is engaged for this free list: only for in-memory regions with enabled page eviction. */ + private final boolean reserveEnabled; + + /** Data region this free list belongs to. */ + private final DataRegion dataRegion; + + /** Kernal context. */ + private final GridKernalContext ctx; + + /** Pages reserved by in-flight inserts (see {@link #reserve(int)}). */ + private final AtomicLong reservedPages = new AtomicLong(); + /** Page list cache limit. */ private final AtomicLong pageListCacheLimit; @@ -319,8 +364,29 @@ protected void putPage(int freeSpace, long pageId, long page, long pageAddr, IoS } } + /** + * Per-call write context of {@link #insertDataRows}: the rows cursor plus the reservation accounting of the row + * currently being written, shared between {@link WriteRowsHandler} and the outer loop of {@link #insertDataRows}. + */ + private static final class RowsWriteContext { + /** Rows cursor. */ + private GridCursor cur; + + /** Reservation made by {@link #reserve(int)} for the current row. */ + private long reserved; + + /** Number of the current row's pages that have already consumed the reservation. */ + private int consumed; + + /** + * Set when {@link WriteRowsHandler} could not reserve space for a row without eviction (forbidden under the + * page write lock); the cursor is positioned on that row, and the outer loop must continue without advancing. + */ + private boolean pending; + } + /** */ - private final class WriteRowsHandler extends PageHandler, Integer> { + private final class WriteRowsHandler extends PageHandler, Integer> { /** {@inheritDoc} */ @Override public Integer run( int cacheId, @@ -329,19 +395,31 @@ private final class WriteRowsHandler extends PageHandler, Integer> long pageAddr, PageIO iox, Boolean walPlc, - GridCursor cur, + RowsWriteContext ctx, int written, IoStatisticsHolder statHolder ) throws IgniteCheckedException { AbstractDataPageIO io = (AbstractDataPageIO)iox; // Fill the page up to the end. - while (written != COMPLETE || (!evictionTracker.evictionRequired() && cur.next())) { - T row = cur.get(); + while (written != COMPLETE || (!evictionTracker.evictionRequired() && ctx.cur.next())) { + T row = ctx.cur.get(); if (written == COMPLETE) { + long tryClaim = tryReserve(row.size()); + + if (tryClaim < 0) { + ctx.pending = true; + + break; + } + + claim(ctx, tryClaim); + + written = writeWholePages(ctx, row, statHolder); + // If the data row was completely written without remainder, proceed to the next. - if ((written = writeWholePages(row, statHolder)) == COMPLETE) + if (written == COMPLETE) continue; if (io.getFreeSpace(pageAddr) < row.size() - written) @@ -461,7 +539,13 @@ public AbstractFreeList( rmvRow = new RemoveRowHandler(cacheGrpId == 0); - this.evictionTracker = dataRegion.evictionTracker(); + evictionTracker = dataRegion.evictionTracker(); + + reserveEnabled = !dataRegion.config().isPersistenceEnabled() && + dataRegion.config().getPageEvictionMode() != DataPageEvictionMode.DISABLED; + + this.dataRegion = dataRegion; + this.ctx = ctx; this.reuseList = reuseList == null ? this : reuseList; int pageSize = pageMem.pageSize(); @@ -482,7 +566,7 @@ public AbstractFreeList( this.shift = shift; - this.memMetrics = dataRegion.metrics(); + memMetrics = dataRegion.metrics(); this.pageListCacheLimit = pageListCacheLimit; @@ -580,17 +664,6 @@ private int bucket(int freeSpace, boolean allowReuse) { return freeSpace > MIN_PAGE_FREE_SPACE ? bucket(freeSpace, false) : -1; } - /** - * @param part Partition. - * @return Page ID. - * @throws IgniteCheckedException If failed. - */ - private long allocateDataPage(int part) throws IgniteCheckedException { - assert part <= PageIdAllocator.MAX_PARTITION_ID; - - return pageMem.allocatePage(grpId, part, FLAG_DATA); - } - /** {@inheritDoc} */ @Override public void insertDataRow(T row, IoStatisticsHolder statHolder) throws IgniteCheckedException { int written = 0; @@ -604,7 +677,7 @@ private long allocateDataPage(int part) throws IgniteCheckedException { } while (written != COMPLETE); } - catch (IgniteCheckedException | Error e) { + catch (IgniteCheckedException | Error | IgniteOutOfMemoryException e) { throw e; } catch (Throwable t) { @@ -623,60 +696,101 @@ private long allocateDataPage(int part) throws IgniteCheckedException { * @param statHolder Statistics holder to track IO operations. * @throws IgniteCheckedException If failed. */ - @Override public void insertDataRows(Collection rows, - IoStatisticsHolder statHolder) throws IgniteCheckedException { + @Override public void insertDataRows(Collection rows, IoStatisticsHolder statHolder) throws IgniteCheckedException { try { - GridCursor cur = new GridCursorIteratorWrapper<>(rows.iterator()); + RowsWriteContext ctx = new RowsWriteContext<>(); + + ctx.cur = new GridCursorIteratorWrapper<>(rows.iterator()); int written = COMPLETE; - while (written != COMPLETE || cur.next()) { - T row = cur.get(); + try { + while (written != COMPLETE || ctx.pending || ctx.cur.next()) { + ctx.pending = false; - // If eviction is required - free up memory before locking the next page. - while (evictionTracker.evictionRequired()) { - evictionTracker.evictDataPage(); + T row = ctx.cur.get(); - memMetrics.updateEvictionRate(); - } + if (written == COMPLETE) { + written = reserveAndWriteWholePages(ctx, row, statHolder); - if (written == COMPLETE) { - written = writeWholePages(row, statHolder); + continue; + } - continue; - } + AbstractDataPageIO initIo = null; - AbstractDataPageIO initIo = null; + long pageId = takePage(row.size() - written, row, statHolder); - long pageId = takePage(row.size() - written, row, statHolder); + if (pageId == 0L) { + pageId = allocateDataPage(row.partition()); - if (pageId == 0L) { - pageId = allocateDataPage(row.partition()); + initIo = row.ioVersions().latest(); + } - initIo = row.ioVersions().latest(); - } + if (reserveEnabled) { + reservedPages.decrementAndGet(); + + ctx.consumed++; + } - written = write(pageId, writeRowsHnd, initIo, cur, written, FAIL_I, statHolder); + written = write(pageId, writeRowsHnd, initIo, ctx, written, FAIL_I, statHolder); - assert written != FAIL_I; // We can't fail here. + assert written != FAIL_I; // We can't fail here. + } + } + finally { + releaseReserved(ctx.reserved - ctx.consumed); } } + catch (IgniteOutOfMemoryException e) { + throw e; + } catch (RuntimeException e) { throw new CorruptedFreeListException("Failed to insert data rows", e, grpId); } } + /** + * Reserves the space for the next row (releasing the leftover reservation of the previous one) and writes its + * whole-page fragments, each consuming the reservation. + * + * @param ctx Write context holding the reservation accounting of the current row. + * @param row Row to process. + * @param statHolder Statistics holder to track IO operations. + * @return Number of bytes written, {@link #COMPLETE} if the row was fully written by whole-page fragments. + * @throws IgniteCheckedException If failed. + */ + private int reserveAndWriteWholePages(RowsWriteContext ctx, T row, IoStatisticsHolder statHolder) throws IgniteCheckedException { + claim(ctx, reserve(row.size())); + + return writeWholePages(ctx, row, statHolder); + } + + /** + * Releases the unconsumed remainder of the previous row's reservation and stores the new claim for the current row. + * + * @param ctx Write context holding the reservation accounting of the current row. + * @param claimPages Pages claimed for the current row. + */ + private void claim(RowsWriteContext ctx, long claimPages) { + releaseReserved(ctx.reserved - ctx.consumed); + + ctx.reserved = claimPages; + ctx.consumed = 0; + } + /** * Write fragments of the row, which occupy the whole memory page. A data row is ignored if it is less than the max - * payload of an empty data page. + * payload of an empty data page. Every written fragment consumes one reserved page (accounted in {@code ctx} + * right next to the {@code reservedPages} decrement), so the accounting is exact on any outcome. * + * @param ctx Write context holding the reservation accounting of the current row. * @param row Row to process. * @param statHolder Statistics holder to track IO operations. * @return Number of bytes written, {@link #COMPLETE} if the row was fully written, {@code 0} if data row was * ignored because it is less than the max payload of an empty data page. * @throws IgniteCheckedException If failed. */ - private int writeWholePages(T row, IoStatisticsHolder statHolder) throws IgniteCheckedException { + private int writeWholePages(RowsWriteContext ctx, T row, IoStatisticsHolder statHolder) throws IgniteCheckedException { assert row.link() == 0 : row.link(); int written = 0; @@ -685,6 +799,12 @@ private int writeWholePages(T row, IoStatisticsHolder statHolder) throws IgniteC while (rowSize - written >= MIN_SIZE_FOR_DATA_PAGE) { written = writeSinglePage(row, written, statHolder); + if (reserveEnabled) { + reservedPages.decrementAndGet(); + + ctx.consumed++; + } + memMetrics.incrementLargeEntriesPages(); } @@ -692,7 +812,9 @@ private int writeWholePages(T row, IoStatisticsHolder statHolder) throws IgniteC } /** - * Take a page and write row on it. + * Take a page and write row on it. The reservation made by {@link #reserve(int)} (when engaged for the row's + * insert) is released as a whole by the caller once the insert completes, so pages taken here are not + * individually accounted against it. * * @param row Row to write. * @param written Written size. @@ -923,7 +1045,203 @@ public int emptyDataPages() { } /** {@inheritDoc} */ - @Override public String toString() { - return "FreeList [name=" + name() + ']'; + @Override public long reserve(int rowSize) throws IgniteOutOfMemoryException, IgniteCheckedException { + if (!reserveEnabled) { + checkDisabledEvictionOom(rowSize); + + return 0; + } + + long claimPages = pagesToReserve(rowSize); + + long backoffNanos = EVICTION_BACKOFF_START_NANOS; + + long lastProgressNanos = System.nanoTime(); + + long cycleDeadlineNanos = lastProgressNanos + TimeUnit.MILLISECONDS.toNanos(EVICTION_MAX_CYCLE_TIME_MILLIS); + + long totalPages = totalPages(); + + while (true) { + long deficit = claimPages - (freePages(totalPages) - reservedPages.get()); + + if (deficit <= 0) { + long reserved = reservedPages.get(); + + if (reservedPages.compareAndSet(reserved, reserved + claimPages)) + return claimPages; + + continue; + } + + if (memMetrics.onPageEvictionsStarted()) { + U.warn(log, "Page-based evictions started." + + " Consider increasing 'maxSize' on Data Region configuration: " + dataRegion.config().getName()); + } + + boolean evicted = evictionTracker.evictDataPage(); + + memMetrics.updateEvictionRate(); + + long curDeficit = claimPages - (freePages(totalPages) - reservedPages.get()); + + if (evicted || curDeficit < deficit) { + lastProgressNanos = System.nanoTime(); + + backoffNanos = EVICTION_BACKOFF_START_NANOS; + } + else { + LockSupport.parkNanos(backoffNanos); + + backoffNanos = Math.min(backoffNanos << 1, EVICTION_BACKOFF_MAX_NANOS); + } + + long nowNanos = System.nanoTime(); + + if (nowNanos - lastProgressNanos > TimeUnit.MILLISECONDS.toNanos(EVICTION_NO_PROGRESS_TIMEOUT_MILLIS) || + nowNanos > cycleDeadlineNanos) + throw outOfMemory(dataRegion.config()); + } + } + + /** {@inheritDoc} */ + @Override public void releaseReserved(long pages) { + assert pages >= 0; + + if (reserveEnabled && pages != 0) + reservedPages.addAndGet(-pages); + } + + /** {@inheritDoc} */ + @Override public void checkFitsForInsert(int rowSize) throws IgniteOutOfMemoryException { + checkDisabledEvictionOom(rowSize); + } + + /** + * Attempts to reserve space for a row without blocking on eviction. Unlike {@link #reserve(int)}, this method + * never calls {@code evictionTracker.evictDataPage()} (which would take an entry lock), so it is safe to call + * while holding a page write lock. + * + * @param rowSize Row size in bytes. + * @return The number of claimed pages (>= 0) if the reservation succeeded without eviction, or {@code -1} if + * the claim would require eviction (the caller should defer the row to a context where blocking is safe). + * @throws IgniteOutOfMemoryException If the row is larger than the region or the region is exhausted (disabled + * eviction in-memory OOM check). + */ + private long tryReserve(int rowSize) throws IgniteOutOfMemoryException { + if (!reserveEnabled) { + checkDisabledEvictionOom(rowSize); + + return 0; + } + + long claimPages = pagesToReserve(rowSize); + long totalPages = totalPages(); + + for (int i = 0; i < TRY_RESERVE_CAS_ATTEMPTS; i++) { + long reserved = reservedPages.get(); + + if (claimPages - (freePages(totalPages) - reserved) > 0) + return -1; + + if (reservedPages.compareAndSet(reserved, reserved + claimPages)) + return claimPages; + } + + return -1; + } + + /** + * Computes the number of pages to reserve for a row, validating that the row can fit into the data region. + * + * @param rowSize Row size in bytes. + * @return Number of pages to claim: the pages required for the row plus the empty pages pool, capped at the region size. + * @throws IgniteOutOfMemoryException If the row is larger than the whole data region. + */ + private long pagesToReserve(int rowSize) throws IgniteOutOfMemoryException { + long totalPages = totalPages(); + + long requiredPages = (rowSize + MIN_SIZE_FOR_DATA_PAGE - 1) / MIN_SIZE_FOR_DATA_PAGE; + + if (requiredPages > totalPages) + throw outOfMemory(dataRegion.config()); + + return Math.min(requiredPages + dataRegion.config().getEmptyPagesPoolSize(), totalPages); + } + + /** @return Total number of pages in the data region (maxSize / systemPageSize). */ + private long totalPages() { + return dataRegion.config().getMaxSize() / pageMem.systemPageSize(); + } + + /** + * OOM threshold check for the disabled-eviction in-memory case (previously in {@code ensureFreeSpace}). + * + * @param rowSize Row size in bytes. + * @throws IgniteOutOfMemoryException If the region is near exhaustion. + */ + private void checkDisabledEvictionOom(int rowSize) throws IgniteOutOfMemoryException { + if (!dataRegion.config().isPersistenceEnabled() && dataRegion.config().getPageEvictionMode() == DataPageEvictionMode.DISABLED) { + long nonEmptyPages = pageMem.loadedPages() - emptyDataPages(); + + boolean oomThreshold = + totalPages() < ((double)rowSize / pageMem.pageSize() + nonEmptyPages * (8.0 * 1.5 / pageMem.pageSize() + 1) + 256); + + if (oomThreshold) + throw outOfMemory(dataRegion.config()); + } + } + + /** + * @param part Partition. + * @return Page ID. + * @throws IgniteCheckedException If failed. + */ + private long allocateDataPage(int part) throws IgniteCheckedException { + assert part <= PageIdAllocator.MAX_PARTITION_ID; + + return pageMem.allocatePage(grpId, part, FLAG_DATA); + } + + /** + * @param totalPages Total pages in the region (maxSize / systemPageSize). + * @return Free pages of the region: empty pages in this free list plus unallocated headroom. Effective headroom + * accounts for per-segment allocation overhead (up to one page per segment), so it does not trust headroom + * that cannot actually be allocated. + */ + private long freePages(long totalPages) { + return emptyDataPages() + effectiveHeadroom(totalPages, pageMem.loadedPages()); + } + + /** + * Computes the effective headroom (unallocated pages that can actually be allocated), accounting for per-segment + * allocation overhead (up to one page per segment). + * + * @param totalPages Total pages in the region (maxSize / systemPageSize). + * @param loadedPages Currently loaded pages. + * @return Effective headroom, never negative. + */ + private static long effectiveHeadroom(long totalPages, long loadedPages) { + return Math.max(0, (totalPages > SEG_CNT ? totalPages - SEG_CNT : 0) - loadedPages); + } + + /** + * @param cfg Data region configuration. + * @return New {@link IgniteOutOfMemoryException} (also reported as a critical failure) for the given region. + */ + private IgniteOutOfMemoryException outOfMemory(DataRegionConfiguration cfg) { + IgniteOutOfMemoryException oom = new IgniteOutOfMemoryException("Out of memory in data region [" + + "name=" + cfg.getName() + + ", initSize=" + U.readableSize(cfg.getInitialSize(), false) + + ", maxSize=" + U.readableSize(cfg.getMaxSize(), false) + + ", persistenceEnabled=" + cfg.isPersistenceEnabled() + "] Try the following:" + U.nl() + + " ^-- Increase maximum off-heap memory size (DataRegionConfiguration.maxSize)" + U.nl() + + " ^-- Enable Ignite persistence (DataRegionConfiguration.persistenceEnabled)" + U.nl() + + " ^-- Enable eviction or expiration policies" + ); + + ctx.failure().process(new FailureContext(FailureType.CRITICAL_ERROR, oom)); + + return oom; } } diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/freelist/FreeList.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/freelist/FreeList.java index 7d492fd46f8a3..38902e6cf321f 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/freelist/FreeList.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/freelist/FreeList.java @@ -20,13 +20,19 @@ import java.util.Collection; import org.apache.ignite.IgniteCheckedException; import org.apache.ignite.IgniteLogger; +import org.apache.ignite.internal.mem.IgniteOutOfMemoryException; import org.apache.ignite.internal.metric.IoStatisticsHolder; import org.apache.ignite.internal.processors.cache.persistence.Storable; -/** - */ +/** */ public interface FreeList { /** + * Inserts a data page by page. When the caller reserved space via {@link #reserve(int)} before acquiring the + * entry lock, the whole reservation is released by the caller via {@link #releaseReserved(long)} once the + * insert completes (normally or not): pages taken here are not individually accounted against it, so an + * exception in the middle of the insert cannot drift the reservation counter. Does not run eviction or acquire + * space. + * * @param row Row. * @param statHolder Statistics holder to track IO operations. * @throws IgniteCheckedException If failed. @@ -60,4 +66,34 @@ public interface FreeList { * @param log Logger. */ public void dumpStatistics(IgniteLogger log); + + /** + * Reserves pages for a row of the given size before the entry lock is acquired. Runs blocking eviction when + * necessary (the caller must not hold any entry or page locks). + * + * @param rowSize Size of the row to be inserted. + * @return Number of reserved pages to be released via {@link #releaseReserved(long)} once the row insert + * completes, or {@code 0} if the reservation is not engaged for this free list (persistent + * data region or disabled page eviction). + * @throws IgniteOutOfMemoryException If the row cannot fit into the region, or eviction makes no progress. + * @throws IgniteCheckedException If failed to evict data pages. + */ + public long reserve(int rowSize) throws IgniteOutOfMemoryException, IgniteCheckedException; + + /** + * Releases the unconsumed portion of a reservation made by {@link #reserve(int)}. + * + * @param pages Number of pages to release (must not be negative). + */ + public void releaseReserved(long pages); + + /** + * Early out-of-memory check for an in-memory data region with disabled page eviction: verifies that the region + * still has room for a row of the given size with a safety margin for free-list bookkeeping pages. No-op for + * persistent regions or regions with eviction enabled. + * + * @param rowSize Row size in bytes. + * @throws IgniteOutOfMemoryException If the region is near exhaustion. + */ + public void checkFitsForInsert(int rowSize) throws IgniteOutOfMemoryException; } diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/tree/io/CacheVersionIO.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/tree/io/CacheVersionIO.java index fd7f2f5db8691..c2ba4e0c998ed 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/tree/io/CacheVersionIO.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/tree/io/CacheVersionIO.java @@ -40,7 +40,7 @@ public class CacheVersionIO { private static final int SIZE_V1 = 17; /** Serialized size in bytes. */ - private static final int SIZE_V2 = 33; + public static final int SIZE_V2 = 33; /** * @param ver Version. diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/eviction/paged/PageEvictionAbstractTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/eviction/paged/PageEvictionAbstractTest.java index 18622516a2030..f7f278bd7ca15 100644 --- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/eviction/paged/PageEvictionAbstractTest.java +++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/eviction/paged/PageEvictionAbstractTest.java @@ -16,6 +16,9 @@ */ package org.apache.ignite.internal.processors.cache.eviction.paged; +import javax.cache.expiry.Duration; +import org.apache.ignite.IgniteCache; +import org.apache.ignite.IgniteCheckedException; import org.apache.ignite.cache.CacheAtomicityMode; import org.apache.ignite.cache.CacheMode; import org.apache.ignite.cache.CacheWriteSynchronizationMode; @@ -26,9 +29,14 @@ import org.apache.ignite.configuration.DataStorageConfiguration; import org.apache.ignite.configuration.IgniteConfiguration; import org.apache.ignite.configuration.NearCacheConfiguration; +import org.apache.ignite.internal.IgniteEx; +import org.apache.ignite.internal.processors.cache.persistence.DataRegion; import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest; import org.jetbrains.annotations.NotNull; +import static java.util.concurrent.TimeUnit.MILLISECONDS; +import static javax.cache.expiry.CreatedExpiryPolicy.factoryOf; + /** * */ @@ -51,6 +59,16 @@ public class PageEvictionAbstractTest extends GridCommonAbstractTest { /** Default policy name. */ private static final String DEFAULT_POLICY_NAME = "dfltPlc"; + /** @return Deterministic per-key value content. */ + public static byte[] valueFor(int key, int size) { + byte[] val = new byte[size]; + + for (int i = 0; i < size; i++) + val[i] = (byte)(key * 31 + i); + + return val; + } + /** * @param mode Eviction mode. * @param configuration Configuration. @@ -130,4 +148,73 @@ protected CacheConfiguration cacheConfig( return cacheConfiguration; } + + /** @param cacheName Cache name. */ + protected static IgniteCache createCache(IgniteEx ignite, String cacheName) { + return createCache(ignite, cacheName, 0, false); + } + + /** + * @param ttl TTL in milliseconds ({@code 0} for no expiry). + * @return Cache with eager TTL expiry if {@code ttl > 0}. + */ + protected static IgniteCache createCache(IgniteEx ignite, String cacheName, long ttl, boolean transactional) { + CacheConfiguration ccfg = new CacheConfiguration(cacheName) + .setAffinity(new RendezvousAffinityFunction(false, 32)); + + if (transactional) + ccfg.setAtomicityMode(CacheAtomicityMode.TRANSACTIONAL); + + if (ttl > 0) + ccfg.setExpiryPolicyFactory(factoryOf(new Duration(MILLISECONDS, ttl))).setEagerTtl(true); + + return ignite.createCache(ccfg); + } + + /** @return Default data region. */ + protected static DataRegion defaultRegion(IgniteEx ignite) throws IgniteCheckedException { + return ignite.context().cache().context().database().dataRegion(null); + } + + /** @return The maximum default data region size in bytes. */ + protected static long regionMaxSize(IgniteEx ignite) throws IgniteCheckedException { + return defaultRegion(ignite).config().getMaxSize(); + } + + /** @return The page size in bytes for the default data region. */ + protected static int pageSize(IgniteEx ignite) throws IgniteCheckedException { + return defaultRegion(ignite).pageMemory().pageSize(); + } + + /** @return The total number of pages in the default data region. */ + protected static long totalPages(IgniteEx ignite) throws IgniteCheckedException { + return regionMaxSize(ignite) / defaultRegion(ignite).pageMemory().systemPageSize(); + } + + /** @return {@code true} if page eviction has started in the default data region. */ + protected static boolean isEvictionsStarted(IgniteEx ignite) throws IgniteCheckedException { + return defaultRegion(ignite).metrics().isEvictionsStarted(); + } + + /** + * Pre-fills the cache with page-sized entries until the loaded page count reaches {@code fraction} of the total + * pages in the default data region. Entries are written with zero-filled {@code pageSize - 200} byte values at + * consecutive keys starting from {@code 0}. + * + * @param ignite Node. + * @param cache Cache. + * @param fraction Target fraction of total pages to load (e.g. {@code 0.85}). + * @return Number of entries written. + * @throws Exception If failed. + */ + protected static int fillRegion(IgniteEx ignite, IgniteCache cache, double fraction) throws Exception { + int entries = (int)(totalPages(ignite) * fraction); + + byte[] small = new byte[pageSize(ignite) - 200]; + + for (int i = 0; i < entries; i++) + cache.put(i, small); + + return entries; + } } diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/eviction/paged/PageEvictionConcurrentWritesAbstractTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/eviction/paged/PageEvictionConcurrentWritesAbstractTest.java new file mode 100644 index 0000000000000..8c42fd7e93c30 --- /dev/null +++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/eviction/paged/PageEvictionConcurrentWritesAbstractTest.java @@ -0,0 +1,319 @@ +/* + * 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.ignite.internal.processors.cache.eviction.paged; + +import java.util.Arrays; +import java.util.HashMap; +import java.util.Map; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import org.apache.ignite.IgniteCache; +import org.apache.ignite.cache.CacheWriteSynchronizationMode; +import org.apache.ignite.cache.affinity.rendezvous.RendezvousAffinityFunction; +import org.apache.ignite.configuration.CacheConfiguration; +import org.apache.ignite.internal.IgniteEx; +import org.apache.ignite.internal.IgniteInternalFuture; +import org.apache.ignite.internal.util.typedef.internal.U; +import org.apache.ignite.testframework.GridTestUtils; +import org.junit.Test; + +/** Concurrent deadlock and data-integrity tests for size-aware page eviction. */ +public abstract class PageEvictionConcurrentWritesAbstractTest extends PageEvictionAbstractTest { + /** {@inheritDoc} */ + @Override protected void afterTest() throws Exception { + stopAllGrids(); + } + + /** + * Concurrent large inserts into a region pre-filled with small entries must complete without deadlock and without + * corrupting the free list. + * + * @throws Exception If failed. + */ + @Test + public void testConcurrentLargeWritesNoDeadlock() throws Exception { + int smallEntries = 1_000; + int largeRowsPerThread = 20; + int threads = 10; + + IgniteEx ignite = startGrid(1); + + IgniteCache cache = createCache(ignite, DEFAULT_CACHE_NAME); + + long regionMax = regionMaxSize(ignite); + + // Small values fill ~80% of the region, below the 0.9 eviction threshold. + int smallValSize = (int)(regionMax * 0.8 / smallEntries); + + for (int i = 0; i < smallEntries; i++) + cache.put(i, valueFor(i, smallValSize)); + + assertFalse("Eviction must not have started during pre-fill", isEvictionsStarted(ignite)); + + CountDownLatch startLatch = new CountDownLatch(1); + AtomicInteger threadIdx = new AtomicInteger(); + + // Large values (5% of the region) trigger eviction during concurrent writes. + int largeValSize = (int)(regionMax / 20); + + IgniteInternalFuture fut = GridTestUtils.runMultiThreadedAsync(() -> { + U.awaitQuiet(startLatch); + + int idx = threadIdx.getAndIncrement(); + + for (int k = 0; k < largeRowsPerThread; k++) { + int key = smallEntries + idx * largeRowsPerThread + k; + + cache.put(key, valueFor(key, largeValSize)); + } + }, threads, "paged-writer"); + + startLatch.countDown(); + + fut.get(TimeUnit.MINUTES.toMillis(3)); + + assertTrue("Eviction must have started during concurrent large writes", isEvictionsStarted(ignite)); + + // Probe put+get: the free list must not be corrupted. + int probeKey = smallEntries + threads * largeRowsPerThread + 1; + + byte[] probeVal = valueFor(probeKey, pageSize(ignite) - 200); + + cache.put(probeKey, probeVal); + + byte[] read = (byte[])cache.get(probeKey); + + assertNotNull(read); + assertTrue(Arrays.equals(probeVal, read)); + + verifyContent(cache, 0, smallEntries, smallValSize); + + int survivedLarge = verifyContent(cache, smallEntries, threads * largeRowsPerThread, largeValSize); + + assertTrue("At least one large entry must survive", survivedLarge > 0); + } + + /** + * Batch insert path test. + * + * @throws Exception If failed. + */ + @Test + public void testRebalanceBatchInsertsWithEviction() throws Exception { + IgniteEx ignite1 = startGrid(1); + + int largeValSize = (int)(regionMaxSize(ignite1) / 20); + int largeRows = 25; + + IgniteCache cache = ignite1.createCache(new CacheConfiguration(DEFAULT_CACHE_NAME) + .setAffinity(new RendezvousAffinityFunction(false, 32)).setBackups(1)); + + for (int k = 0; k < largeRows; k++) + cache.put(k, valueFor(k, largeValSize)); + + byte[] smallVal = new byte[pageSize(ignite1) - 200]; + + int smallKeyBase = largeRows + 1; + + for (int i = 0; i < totalPages(ignite1); i++) { + cache.put(smallKeyBase + i, smallVal); + + if (isEvictionsStarted(ignite1)) + break; + } + + assertTrue("Eviction must start during overflow fill", isEvictionsStarted(ignite1)); + + IgniteEx ignite2 = startGrid(2); + + // Writers keep adding large rows on the coordinator while the joining node stores rebalance batches. + CountDownLatch startLatch = new CountDownLatch(1); + AtomicInteger threadIdx = new AtomicInteger(); + + int writers = 4; + int rowsPerWriter = 5; + + int writerKeyBase = smallKeyBase + (int)totalPages(ignite1); + + IgniteInternalFuture fut = GridTestUtils.runMultiThreadedAsync(() -> { + U.awaitQuiet(startLatch); + + int idx = threadIdx.getAndIncrement(); + + for (int k = 0; k < rowsPerWriter; k++) { + int key = writerKeyBase + idx * rowsPerWriter + k; + + cache.put(key, valueFor(key, largeValSize)); + } + }, writers, "paged-writer"); + + startLatch.countDown(); + + awaitPartitionMapExchange(); + + cache.rebalance().get(TimeUnit.MINUTES.toMillis(2)); + + fut.get(TimeUnit.MINUTES.toMillis(2)); + + assertTrue("Eviction must have started on the rebalanced node", isEvictionsStarted(ignite2)); + + int survived = verifyContent(cache, 0, largeRows, largeValSize) + + verifyContent(cache, writerKeyBase, writers * rowsPerWriter, largeValSize); + + assertTrue("At least one large entry must survive", survived > 0); + } + + /** + * Concurrent {@code putAll} batches on a near-full region with enabled eviction must complete without deadlock. + * + * @throws Exception If failed. + */ + @Test + public void testConcurrentPutAllNoDeadlock() throws Exception { + IgniteEx ignite = startGrid(0); + + IgniteCache cache = ignite.createCache(new CacheConfiguration(DEFAULT_CACHE_NAME) + .setAffinity(new RendezvousAffinityFunction(false, 32)) + .setBackups(0) + .setWriteSynchronizationMode(CacheWriteSynchronizationMode.PRIMARY_SYNC)); + + int pSize = pageSize(ignite); + + // Pre-fill to ~70% of the region. + int smallValSize = pSize / 4 - 50; + int smallEntries = (int)(regionMaxSize(ignite) * 0.7 / smallValSize); + + for (int i = 0; i < smallEntries; i++) + cache.put(i, valueFor(i, smallValSize)); + + int valSize = pSize / 4 - 50; + + CountDownLatch startLatch = new CountDownLatch(1); + AtomicInteger keyGen = new AtomicInteger(smallEntries); + + IgniteInternalFuture fut = GridTestUtils.runMultiThreadedAsync(() -> { + U.awaitQuiet(startLatch); + + for (int b = 0; b < 10; b++) { + Map batch = new HashMap<>(); + + for (int i = 0; i < 50; i++) { + int key = keyGen.incrementAndGet(); + + batch.put(key, valueFor(key, valSize)); + } + + cache.putAll(batch); + } + }, 4, "batch-writer"); + + startLatch.countDown(); + + fut.get(TimeUnit.MINUTES.toMillis(3)); + + assertTrue("Eviction must have started during concurrent batch writes", isEvictionsStarted(ignite)); + + int probeKey = keyGen.incrementAndGet(); + byte[] probeVal = valueFor(probeKey, smallValSize); + + cache.put(probeKey, probeVal); + + byte[] read = cache.get(probeKey); + + assertNotNull(read); + assertTrue(Arrays.equals(probeVal, read)); + } + + /** + * Concurrent single-row puts interleaved with {@code putAll} batches on a near-full region. + * + * @throws Exception If failed. + */ + @Test + public void testConcurrentPutsAndPutAllNoDeadlock() throws Exception { + IgniteEx ignite = startGrid(0); + + IgniteCache cache = ignite.createCache(new CacheConfiguration(DEFAULT_CACHE_NAME) + .setAffinity(new RendezvousAffinityFunction(false, 32)) + .setBackups(0) + .setWriteSynchronizationMode(CacheWriteSynchronizationMode.PRIMARY_SYNC)); + + int valSize = pageSize(ignite) / 4 - 50; + int smallEntries = (int)(regionMaxSize(ignite) * 0.7 / valSize); + + for (int i = 0; i < smallEntries; i++) + cache.put(i, valueFor(i, valSize)); + + AtomicInteger keyGen = new AtomicInteger(smallEntries); + + CountDownLatch startLatch = new CountDownLatch(1); + + AtomicInteger threadIdx = new AtomicInteger(); + + IgniteInternalFuture fut = GridTestUtils.runMultiThreadedAsync(() -> { + U.awaitQuiet(startLatch); + + // Even threads do putAll batches, odd threads do single puts. + boolean isBatch = (threadIdx.getAndIncrement() & 1) == 0; + + for (int op = 0; op < 50; op++) { + if (isBatch) { + Map batch = new HashMap<>(); + + for (int i = 0; i < 20; i++) { + int key = keyGen.incrementAndGet(); + + batch.put(key, valueFor(key, valSize)); + } + + cache.putAll(batch); + } + else { + int key = keyGen.incrementAndGet(); + + cache.put(key, valueFor(key, valSize)); + } + } + }, 6, "mixed-writer"); + + startLatch.countDown(); + + fut.get(TimeUnit.MINUTES.toMillis(3)); + + assertTrue("Eviction must have started", isEvictionsStarted(ignite)); + } + + /** Asserts every non-null value in {@code [base, base+cnt)} is byte-identical to {@code valueFor(k, valSize)}. */ + private static int verifyContent(IgniteCache cache, int base, int cnt, int valSize) { + int survived = 0; + + for (int k = base; k < base + cnt; k++) { + byte[] read = (byte[])cache.get(k); + + if (read == null) + continue; // Evicted. + + assertTrue("Content mismatch for key " + k, Arrays.equals(valueFor(k, valSize), read)); + + survived++; + } + + return survived; + } +} diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/eviction/paged/PageEvictionDisabledEvictionReadThroughOomSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/eviction/paged/PageEvictionDisabledEvictionReadThroughOomSelfTest.java new file mode 100644 index 0000000000000..f1967cd8f5fa6 --- /dev/null +++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/eviction/paged/PageEvictionDisabledEvictionReadThroughOomSelfTest.java @@ -0,0 +1,158 @@ +/* + * 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.ignite.internal.processors.cache.eviction.paged; + +import javax.cache.Cache; +import javax.cache.configuration.Factory; +import javax.cache.integration.CacheLoaderException; +import org.apache.ignite.IgniteCache; +import org.apache.ignite.cache.CacheAtomicityMode; +import org.apache.ignite.cache.CacheMode; +import org.apache.ignite.cache.CacheWriteSynchronizationMode; +import org.apache.ignite.cache.affinity.rendezvous.RendezvousAffinityFunction; +import org.apache.ignite.cache.store.CacheStoreAdapter; +import org.apache.ignite.configuration.CacheConfiguration; +import org.apache.ignite.configuration.DataPageEvictionMode; +import org.apache.ignite.configuration.IgniteConfiguration; +import org.apache.ignite.internal.IgniteEx; +import org.apache.ignite.internal.mem.IgniteOutOfMemoryException; +import org.apache.ignite.internal.util.typedef.X; +import org.apache.ignite.testframework.GridTestUtils; +import org.junit.Test; + +/** + * Regression test for the disabled-eviction in-memory OOM guard on insert paths that bypass {@code FreeList.reserve}: + * read-through inside {@code get()} runs under the entry lock and cannot reserve, so the per-insert guard previously + * living in {@code RowStore.addRow} (now {@code FreeList.checkFitsForInsert}) must still fire and reject the insert + * with {@link IgniteOutOfMemoryException} when the region is at the threshold — rather than silently consuming the + * free-list bookkeeping margin. + */ +public class PageEvictionDisabledEvictionReadThroughOomSelfTest extends PageEvictionAbstractTest { + /** {@inheritDoc} */ + @Override protected IgniteConfiguration getConfiguration(String gridName) throws Exception { + return setEvictionMode(DataPageEvictionMode.DISABLED, super.getConfiguration(gridName)); + } + + /** {@inheritDoc} */ + @Override protected void afterTest() throws Exception { + super.afterTest(); + + stopAllGrids(); + } + + /** + * Fills the region until a {@code put} hits the disabled-eviction OOM guard, then verifies that a read-through + * {@code get} on a missing key also throws {@link IgniteOutOfMemoryException} from the same guard (the bypass + * path must not silently consume the remaining free-list margin). + * + * @throws Exception If failed. + */ + @Test(timeout = 120_000) + public void testReadThroughGetOomAtThreshold() throws Exception { + IgniteEx ignite = startGrid(0); + + CacheConfiguration cfg = new CacheConfiguration(DEFAULT_CACHE_NAME) + .setAffinity(new RendezvousAffinityFunction(false, 32)) + .setCacheMode(CacheMode.PARTITIONED) + .setAtomicityMode(CacheAtomicityMode.ATOMIC) + .setWriteSynchronizationMode(CacheWriteSynchronizationMode.PRIMARY_SYNC) + .setReadThrough(true) + .setCacheStoreFactory(new TestStoreFactory()); + + IgniteCache cache = grid(0).getOrCreateCache(cfg); + + byte[] small = new byte[pageSize(ignite) - 200]; + + // Fill the region until the disabled-eviction OOM guard fires on a put. + int filled = 0; + + for (int i = 0; i < 100_000; i++) { + try { + cache.put(i, small); + + filled++; + } + catch (Exception e) { + assertTrue("Expected IgniteOutOfMemoryException while filling the region, got: " + e, + X.hasCause(e, IgniteOutOfMemoryException.class)); + + break; + } + } + + assertTrue("Region must be filled before OOM (filled=" + filled + ")", filled > 0); + + // Read-through on a missing key: the store returns a page-sized value, which must be rejected by the + // disabled-eviction OOM guard at the RowStore level (the bypass path cannot reserve). + int missingKey = filled + 1; + + GridTestUtils.assertThrowsWithCause(() -> cache.get(missingKey), IgniteOutOfMemoryException.class); + } + + /** + * Below the threshold, read-through {@code get} must succeed — the restored guard must not spuriously reject. + * + * @throws Exception If failed. + */ + @Test + public void testReadThroughGetBelowThresholdSucceeds() throws Exception { + startGrid(0); + + CacheConfiguration cfg = new CacheConfiguration(DEFAULT_CACHE_NAME) + .setAffinity(new RendezvousAffinityFunction(false, 32)) + .setCacheMode(CacheMode.PARTITIONED) + .setAtomicityMode(CacheAtomicityMode.ATOMIC) + .setWriteSynchronizationMode(CacheWriteSynchronizationMode.PRIMARY_SYNC) + .setReadThrough(true) + .setCacheStoreFactory(new TestStoreFactory()); + + IgniteCache cache = grid(0).getOrCreateCache(cfg); + + // A single read-through load well below capacity must succeed. + byte[] val = cache.get(1); + + assertNotNull("Read-through load must succeed below threshold", val); + assertEquals(PAGE_SIZE - 200, val.length); + } + + /** */ + private static class TestStoreFactory implements Factory { + /** {@inheritDoc} */ + @Override public TestCacheStore create() { + return new TestCacheStore(); + } + } + + /** */ + private static class TestCacheStore extends CacheStoreAdapter { + /** {@inheritDoc} */ + @Override public byte[] load(Integer key) throws CacheLoaderException { + return new byte[PAGE_SIZE - 200]; + } + + /** {@inheritDoc} */ + @Override public void write(Cache.Entry entry) { + // No-op. + } + + /** {@inheritDoc} */ + @Override public void delete(Object key) { + // No-op. + } + } +} diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/eviction/paged/PageEvictionEnsureFreeSpaceThresholdTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/eviction/paged/PageEvictionEnsureFreeSpaceThresholdTest.java new file mode 100644 index 0000000000000..240a38118cba9 --- /dev/null +++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/eviction/paged/PageEvictionEnsureFreeSpaceThresholdTest.java @@ -0,0 +1,85 @@ +/* + * 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.ignite.internal.processors.cache.eviction.paged; + +import org.apache.ignite.IgniteCache; +import org.apache.ignite.configuration.DataPageEvictionMode; +import org.apache.ignite.configuration.IgniteConfiguration; +import org.apache.ignite.internal.IgniteEx; +import org.junit.Test; + +/** + * Regression test for C2: the batch-level {@code ensureFreeSpace} must evict at the configured + * {@code evictionThreshold} (default 0.9), not silently skip eviction when the headroom exceeds the empty-pages pool. + */ +public class PageEvictionEnsureFreeSpaceThresholdTest extends PageEvictionAbstractTest { + /** {@inheritDoc} */ + @Override protected IgniteConfiguration getConfiguration(String gridName) throws Exception { + return setEvictionMode(DataPageEvictionMode.RANDOM_2_LRU, super.getConfiguration(gridName)); + } + + /** {@inheritDoc} */ + @Override protected void afterTest() throws Exception { + super.afterTest(); + + stopAllGrids(); + } + + /** */ + @Test + public void testSequentialLargePutsEvictAtThreshold() throws Exception { + IgniteEx ignite = startGrid(1); + + IgniteCache cache = createCache(ignite, DEFAULT_CACHE_NAME); + + long regionMax = regionMaxSize(ignite); + + int largeValSize = (int)(regionMax / 20); + + int largeRows = 25; + + for (int k = 0; k < largeRows; k++) { + byte[] val = new byte[largeValSize]; + + for (int i = 0; i < val.length; i++) + val[i] = (byte)(k * 31 + i); + + cache.put(k, val); + } + + assertTrue("Eviction must have started", isEvictionsStarted(ignite)); + + int survived = 0; + + for (int k = 0; k < largeRows; k++) { + byte[] read = (byte[])cache.get(k); + + if (read == null) + continue; + + survived++; + + assertEquals(largeValSize, read.length); + + for (int i = 0; i < read.length; i++) + assertEquals((byte)(k * 31 + i), read[i]); + } + + assertTrue("At least one large entry must survive", survived > 0); + } +} diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/eviction/paged/PageEvictionGuardOomTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/eviction/paged/PageEvictionGuardOomTest.java new file mode 100644 index 0000000000000..1de971a5ce2df --- /dev/null +++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/eviction/paged/PageEvictionGuardOomTest.java @@ -0,0 +1,106 @@ +/* + * 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.ignite.internal.processors.cache.eviction.paged; + +import java.util.ArrayList; +import java.util.Collection; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.locks.Lock; +import org.apache.ignite.IgniteCache; +import org.apache.ignite.configuration.DataPageEvictionMode; +import org.apache.ignite.configuration.IgniteConfiguration; +import org.apache.ignite.internal.IgniteEx; +import org.apache.ignite.internal.IgniteInternalFuture; +import org.apache.ignite.internal.mem.IgniteOutOfMemoryException; +import org.apache.ignite.testframework.GridTestUtils; +import org.junit.Test; + +/** Negative test for the size-aware eviction progress guard. */ +public class PageEvictionGuardOomTest extends PageEvictionAbstractTest { + /** {@inheritDoc} */ + @Override protected IgniteConfiguration getConfiguration(String gridName) throws Exception { + return setEvictionMode(DataPageEvictionMode.RANDOM_LRU, super.getConfiguration(gridName)); + } + + /** {@inheritDoc} */ + @Override protected void afterTest() throws Exception { + stopAllGrids(); + } + + /** + * With every candidate entry locked, eviction cannot free any page, so a row needing more free pages than remain + * must fail with OOM (bounded time), not hang. + * + * @throws Exception If failed. + */ + @Test(timeout = 180_000) + public void testGuardOomWhenAllEntriesLocked() throws Exception { + IgniteEx ignite = startGrid(1); + + IgniteCache cache = createCache(ignite, DEFAULT_CACHE_NAME, 0, true); + + // Fill to ~85% of total pages (below the 0.9 threshold), ~1 page per entry. + int fillEntries = (int)(totalPages(ignite) * 0.85); + + byte[] fillVal = new byte[pageSize(ignite) - 200]; + + for (int i = 1; i <= fillEntries; i++) + cache.put(i, fillVal); + + Collection keys = new ArrayList<>(fillEntries); + + for (int i = 1; i <= fillEntries; i++) + keys.add(i); + + CountDownLatch ready = new CountDownLatch(1); + + CountDownLatch release = new CountDownLatch(1); + + // Hold entry locks on every resident key from a background thread so eviction has no evictable page. + IgniteInternalFuture lockerFut = GridTestUtils.runAsync(() -> { + Lock lock = cache.lockAll(keys); + + lock.lock(); + + try { + ready.countDown(); + + release.await(); + } + finally { + lock.unlock(); + } + }, "size-aware-guard-locker"); + + try { + assertTrue("Timed out waiting for entries to be locked", ready.await(60, TimeUnit.SECONDS)); + + // All pages locked → eviction cannot free anything → OOM. + GridTestUtils.assertThrowsWithCause( + () -> cache.put(fillEntries + 1, new byte[4 * 1024 * 1024]), + IgniteOutOfMemoryException.class + ); + } + finally { + release.countDown(); + + lockerFut.get(10, TimeUnit.SECONDS); + } + } +} diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/eviction/paged/PageEvictionMetricTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/eviction/paged/PageEvictionMetricTest.java index 9f40cf4958431..fa774690ea372 100644 --- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/eviction/paged/PageEvictionMetricTest.java +++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/eviction/paged/PageEvictionMetricTest.java @@ -49,14 +49,41 @@ public void testPageEvictionMetric() throws Exception { checkPageEvictionMetric(CacheAtomicityMode.ATOMIC); } + /** + * Small records below the eviction threshold must not trigger page eviction. + * + * @throws Exception If failed. + */ + @Test + public void testNoEvictionBelowThreshold() throws Exception { + IgniteEx ignite = startGrid(0); + + DataRegionMetricsImpl metrics = defaultRegion(ignite).metrics(); + + metrics.enableMetrics(); + + CacheConfiguration cfg = cacheConfig("no-evict-below-threshold", null, + CacheMode.PARTITIONED, CacheAtomicityMode.ATOMIC, CacheWriteSynchronizationMode.PRIMARY_SYNC); + + IgniteCache cache = ignite.getOrCreateCache(cfg); + + for (int i = 1; i <= 500; i++) + cache.put(i, new TestObject(PAGE_SIZE / 6)); + + assertFalse("Page eviction must not start while the region is below the eviction threshold", + metrics.isEvictionsStarted()); + + assertEquals("Eviction rate must be zero while the region is below the eviction threshold", + 0f, metrics.getEvictionRate(), 0f); + } + /** * @throws Exception If failed. */ private void checkPageEvictionMetric(CacheAtomicityMode atomicityMode) throws Exception { IgniteEx ignite = startGrid(0); - DataRegionMetricsImpl metrics = - ignite.context().cache().context().database().dataRegion(null).metrics(); + DataRegionMetricsImpl metrics = defaultRegion(ignite).metrics(); metrics.enableMetrics(); diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/eviction/paged/PageEvictionMultinodeAbstractTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/eviction/paged/PageEvictionMultinodeAbstractTest.java index 2bf21fbd32fb9..898cb287fc98f 100644 --- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/eviction/paged/PageEvictionMultinodeAbstractTest.java +++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/eviction/paged/PageEvictionMultinodeAbstractTest.java @@ -30,6 +30,9 @@ * */ public abstract class PageEvictionMultinodeAbstractTest extends PageEvictionAbstractTest { + /** Number of entries, sized to fill the data region past the eviction threshold. */ + private static final int MULTINODE_ENTRIES = 16_000; + /** Cache modes. */ private static final CacheMode[] CACHE_MODES = {CacheMode.PARTITIONED, CacheMode.REPLICATED}; @@ -86,7 +89,7 @@ public void testPageEviction() throws Exception { protected void createCacheAndTestEviction(CacheConfiguration cfg) throws Exception { IgniteCache cache = clientGrid().getOrCreateCache(cfg); - for (int i = 1; i <= ENTRIES; i++) { + for (int i = 1; i <= MULTINODE_ENTRIES; i++) { ThreadLocalRandom r = ThreadLocalRandom.current(); if (r.nextInt() % 5 == 0) @@ -101,7 +104,7 @@ else if (r.nextInt() % 11 == 0) else if (r.nextInt() % 13 == 0) cache.put(r.nextInt(i), new TestObject(r.nextInt(PAGE_SIZE / 2))); // Update. - if (i % (ENTRIES / 10) == 0) + if (i % (MULTINODE_ENTRIES / 10) == 0) System.out.println(">>> Entries put: " + i); } @@ -110,7 +113,7 @@ else if (r.nextInt() % 13 == 0) System.out.println(">>> Resulting size: " + resultingSize); // Eviction started, no OutOfMemory occurred, success. - assertTrue(resultingSize < ENTRIES * 71 / 77); + assertTrue(resultingSize < MULTINODE_ENTRIES * 71 / 77); clientGrid().destroyCache(cfg.getName()); } diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/eviction/paged/PageEvictionPutLargeObjectsAbstractTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/eviction/paged/PageEvictionPutLargeObjectsAbstractTest.java index b141fa82bb357..c2d3a7a378909 100644 --- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/eviction/paged/PageEvictionPutLargeObjectsAbstractTest.java +++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/eviction/paged/PageEvictionPutLargeObjectsAbstractTest.java @@ -70,6 +70,6 @@ public void testPutLargeObjects() throws Exception { for (Integer key : primaryKeys(grid(1).cache(DEFAULT_CACHE_NAME), ENTRIES)) cache.put(key, val); - assertTrue(cache.size() < ENTRIES); + assertTrue(cache.size() > 0 && cache.size() < ENTRIES); } } diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/eviction/paged/PageEvictionReadThroughReservedPagesSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/eviction/paged/PageEvictionReadThroughReservedPagesSelfTest.java new file mode 100644 index 0000000000000..e94ce5198fe0e --- /dev/null +++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/eviction/paged/PageEvictionReadThroughReservedPagesSelfTest.java @@ -0,0 +1,140 @@ +/* + * 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.ignite.internal.processors.cache.eviction.paged; + +import javax.cache.Cache; +import javax.cache.configuration.Factory; +import javax.cache.integration.CacheLoaderException; +import org.apache.ignite.IgniteCache; +import org.apache.ignite.cache.CacheAtomicityMode; +import org.apache.ignite.cache.CacheMode; +import org.apache.ignite.cache.CacheWriteSynchronizationMode; +import org.apache.ignite.cache.affinity.rendezvous.RendezvousAffinityFunction; +import org.apache.ignite.cache.store.CacheStoreAdapter; +import org.apache.ignite.configuration.CacheConfiguration; +import org.apache.ignite.configuration.DataPageEvictionMode; +import org.apache.ignite.configuration.IgniteConfiguration; +import org.apache.ignite.internal.processors.cache.persistence.freelist.FreeListReservedPagesAbstractTest; +import org.junit.Test; + +/** + * Read-through loads go through {@code versionedValue}, which must reserve pages before the entry lock and release the + * remainder after the insert. After loads complete and the cluster quiesces, {@code reservedPages} must be 0. + */ +public class PageEvictionReadThroughReservedPagesSelfTest extends PageEvictionAbstractTest { + /** {@inheritDoc} */ + @Override protected IgniteConfiguration getConfiguration(String gridName) throws Exception { + return setEvictionMode(DataPageEvictionMode.RANDOM_LRU, super.getConfiguration(gridName)); + } + + /** {@inheritDoc} */ + @Override protected void afterTest() throws Exception { + super.afterTest(); + + stopAllGrids(); + } + + /** + * Read-through loads trigger {@code versionedValue}; {@code reservedPages} must be released. + * + * @throws Exception If failed. + */ + @Test + public void testReservedPagesReleasedAfterReadThrough() throws Exception { + startGrid(0); + + CacheConfiguration cfg = new CacheConfiguration(DEFAULT_CACHE_NAME) + .setAffinity(new RendezvousAffinityFunction(false, 32)) + .setCacheMode(CacheMode.PARTITIONED) + .setAtomicityMode(CacheAtomicityMode.ATOMIC) + .setWriteSynchronizationMode(CacheWriteSynchronizationMode.PRIMARY_SYNC) + .setReadThrough(true) + .setCacheStoreFactory(new TestStoreFactory()); + + IgniteCache cache = grid(0).getOrCreateCache(cfg); + + for (int i = 1; i <= 20_000; i++) + cache.get(i); + + assertTrue("Eviction must have started during read-through loads", isEvictionsStarted(grid(0))); + + byte[] val = cache.get(1); + + assertNotNull("Value for key 1 must not be null", val); + assertEquals(PAGE_SIZE - 200, val.length); + + FreeListReservedPagesAbstractTest.assertAllReleased(DEFAULT_CACHE_NAME, 10_000); + } + + /** + * Read-through on a 2-node cluster with backups; {@code reservedPages} must be released on every node. + * + * @throws Exception If failed. + */ + @Test + public void testReservedPagesReleasedAfterReadThroughOnBackup() throws Exception { + startGrids(2); + + CacheConfiguration cfg = new CacheConfiguration(DEFAULT_CACHE_NAME) + .setAffinity(new RendezvousAffinityFunction(false, 32)) + .setCacheMode(CacheMode.PARTITIONED) + .setAtomicityMode(CacheAtomicityMode.ATOMIC) + .setWriteSynchronizationMode(CacheWriteSynchronizationMode.FULL_SYNC) + .setBackups(1) + .setReadThrough(true) + .setCacheStoreFactory(new TestStoreFactory()); + + IgniteCache cache = grid(0).getOrCreateCache(cfg); + + awaitPartitionMapExchange(); + + for (int i = 1; i <= 15_000; i++) + cache.get(i); + + assertTrue("Eviction must have started", isEvictionsStarted(grid(0))); + + FreeListReservedPagesAbstractTest.assertAllReleased(DEFAULT_CACHE_NAME, 10_000); + } + + /** */ + private static class TestStoreFactory implements Factory { + /** {@inheritDoc} */ + @Override public TestCacheStore create() { + return new TestCacheStore(); + } + } + + /** */ + private static class TestCacheStore extends CacheStoreAdapter { + /** {@inheritDoc} */ + @Override public byte[] load(Integer key) throws CacheLoaderException { + // Fills most of a page so ~20k entries exceed the 20 MB region and trigger eviction. + return new byte[PAGE_SIZE - 200]; + } + + /** {@inheritDoc} */ + @Override public void write(Cache.Entry entry) { + // No-op. + } + + /** {@inheritDoc} */ + @Override public void delete(Object key) { + // No-op. + } + } +} diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/eviction/paged/PageEvictionRowSizeBinaryObjectTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/eviction/paged/PageEvictionRowSizeBinaryObjectTest.java new file mode 100644 index 0000000000000..8878e868d39ba --- /dev/null +++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/eviction/paged/PageEvictionRowSizeBinaryObjectTest.java @@ -0,0 +1,108 @@ +/* + * 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.ignite.internal.processors.cache.eviction.paged; + +import org.apache.ignite.IgniteCache; +import org.apache.ignite.IgniteDataStreamer; +import org.apache.ignite.cache.CacheAtomicityMode; +import org.apache.ignite.cache.CacheMode; +import org.apache.ignite.cache.CacheWriteSynchronizationMode; +import org.apache.ignite.cache.affinity.rendezvous.RendezvousAffinityFunction; +import org.apache.ignite.configuration.CacheConfiguration; +import org.apache.ignite.configuration.DataPageEvictionMode; +import org.apache.ignite.configuration.IgniteConfiguration; +import org.apache.ignite.internal.IgniteEx; +import org.junit.Test; + +/** + * Regression test for C1: {@code GridCacheMapEntry#rowSize()} must not throw NPE when the value is an unmarshalled + * {@code BinaryObjectImpl} or a POJO whose key has not yet been marshalled on the reservation path. + */ +public class PageEvictionRowSizeBinaryObjectTest extends PageEvictionAbstractTest { + /** {@inheritDoc} */ + @Override protected IgniteConfiguration getConfiguration(String gridName) throws Exception { + return setEvictionMode(DataPageEvictionMode.RANDOM_2_LRU, super.getConfiguration(gridName)); + } + + /** {@inheritDoc} */ + @Override protected void afterTest() throws Exception { + stopAllGrids(); + } + + /** + * POJO puts on the transactional commit path must not throw NPE inside {@code rowSize}. + * + * @throws Exception If failed. + */ + @Test + public void testPojoPutTxDoesNotNpeInRowSize() throws Exception { + IgniteEx ignite = startGrid(0); + + IgniteCache cache = ignite.createCache( + new CacheConfiguration(DEFAULT_CACHE_NAME) + .setAffinity(new RendezvousAffinityFunction(false, 32)) + .setAtomicityMode(CacheAtomicityMode.TRANSACTIONAL) + .setCacheMode(CacheMode.REPLICATED) + .setWriteSynchronizationMode(CacheWriteSynchronizationMode.PRIMARY_SYNC)); + + // ~1 row per page; fill past the eviction threshold. + int entries = 15_000; + + for (int i = 1; i <= entries; i++) + cache.put(i, new TestObject(PAGE_SIZE / 6)); + + assertTrue("Eviction must have started", isEvictionsStarted(ignite)); + + int probeKey = entries + 1; + + cache.put(probeKey, new TestObject(PAGE_SIZE / 6)); + + assertNotNull(cache.get(probeKey)); + } + + /** + * POJO puts via a {@code DataStreamer} must not throw NPE inside {@code rowSize}. + * + * @throws Exception If failed. + */ + @Test + public void testPojoDataStreamDoesNotNpeInRowSize() throws Exception { + IgniteEx ignite = startGrid(0); + + IgniteCache cache = ignite.createCache( + new CacheConfiguration(DEFAULT_CACHE_NAME) + .setAffinity(new RendezvousAffinityFunction(false, 32))); + + int entries = 15_000; + + try (IgniteDataStreamer streamer = ignite.dataStreamer(DEFAULT_CACHE_NAME)) { + streamer.allowOverwrite(true); + + for (int i = 1; i <= entries; i++) + streamer.addData(i, new TestObject(PAGE_SIZE / 6)); + } + + assertTrue("Eviction must have started", isEvictionsStarted(ignite)); + + int probeKey = entries + 1; + + cache.put(probeKey, new TestObject(PAGE_SIZE / 6)); + + assertNotNull(cache.get(probeKey)); + } +} diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/eviction/paged/PageEvictionSizeAwareAbstractTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/eviction/paged/PageEvictionSizeAwareAbstractTest.java new file mode 100644 index 0000000000000..87a3d32018778 --- /dev/null +++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/eviction/paged/PageEvictionSizeAwareAbstractTest.java @@ -0,0 +1,275 @@ +/* + * 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.ignite.internal.processors.cache.eviction.paged; + +import java.util.Arrays; +import java.util.HashMap; +import java.util.Map; +import org.apache.ignite.IgniteCache; +import org.apache.ignite.IgniteCheckedException; +import org.apache.ignite.internal.IgniteEx; +import org.apache.ignite.internal.mem.IgniteOutOfMemoryException; +import org.apache.ignite.testframework.GridTestUtils; +import org.junit.Test; + +/** Tests size-aware page eviction on in-memory (non-persistent) data regions. */ +public abstract class PageEvictionSizeAwareAbstractTest extends PageEvictionAbstractTest { + /** {@inheritDoc} */ + @Override protected void afterTest() throws Exception { + stopAllGrids(); + } + + /** + * A record larger than the whole region, and a {@code putAll} whose total size equals the region capacity, must + * both fail with OOM (structural pages leave no room for the data). + * + * @throws Exception If failed. + */ + @Test + public void testRecordLargerThanRegionOom() throws Exception { + IgniteEx ignite = startGrid(1); + + IgniteCache cache = createCache(ignite, DEFAULT_CACHE_NAME); + + long regionSize = regionMaxSize(ignite); + + GridTestUtils.assertThrowsWithCause( + () -> cache.put(1, new byte[(int)(regionSize * 2)]), + IgniteOutOfMemoryException.class + ); + + GridTestUtils.assertThrowsWithCause( + () -> { + Map batch = new HashMap<>(); + + Object val = new byte[(int)(regionSize / 4)]; + + for (int i = 0; i < 4; i++) + batch.put(i, val); + + cache.putAll(batch); + }, + IgniteOutOfMemoryException.class + ); + } + + /** + * A {@code putAll} of large records (each larger than the empty-pages pool) must succeed with eviction enabled. + * + * @throws Exception If failed. + */ + @Test + public void testPutAllLargeRows() throws Exception { + IgniteEx ignite = startGrid(1); + + IgniteCache cache = createCache(ignite, DEFAULT_CACHE_NAME); + + // Pre-fill to ~85% of capacity so less than one large row remains available. + int smallEntries = (int)(totalPages(ignite) * 0.85); + + byte[] small = new byte[pageSize(ignite) - 200]; + + for (int i = 0; i < smallEntries; i++) + cache.put(smallEntries + i, small); + + assertTrue("Pre-fill must load >80% of pages", loadedPages(ignite) > totalPages(ignite) * 0.80); + assertFalse("Eviction must not start during pre-fill", isEvictionsStarted(ignite)); + + int putAllLargeRows = 3; + + Map large = new HashMap<>(); + + Object val = new byte[(int)(regionMaxSize(ignite) / 4)]; + + for (int i = 0; i < putAllLargeRows; i++) + large.put(i, val); + + cache.putAll(large); + + assertTrue("Eviction must start after putAll of large rows near capacity", isEvictionsStarted(ignite)); + + for (int i = 0; i < putAllLargeRows; i++) + assertNotNull("Large row " + i + " must be readable after putAll", cache.get(i)); + } + + /** + * Updating a small record to a large value near capacity must succeed via the size-aware reserve. + * + * @throws Exception If failed. + */ + @Test + public void testUpdateRowGrows() throws Exception { + IgniteEx ignite = startGrid(1); + + IgniteCache cache = createCache(ignite, DEFAULT_CACHE_NAME); + + int smallEntries = (int)(totalPages(ignite) * 0.85); + + byte[] small = new byte[pageSize(ignite) - 200]; + + for (int i = 0; i < smallEntries; i++) + cache.put(smallEntries + i, small); + + cache.put(1, new byte[1024]); + + byte[] big = new byte[(int)(regionMaxSize(ignite) / 4)]; + + Arrays.fill(big, (byte)7); + + cache.put(1, big); + + assertTrue("Eviction must have started after growing a row near capacity", isEvictionsStarted(ignite)); + + byte[] read = (byte[])cache.get(1); + + assertNotNull("Updated large value must be readable", read); + assertTrue("Updated value must equal the stored value", Arrays.equals(big, read)); + } + + /** + * Fast path: below the eviction threshold, a large row that fits in the remaining headroom is written without + * triggering eviction. + * + * @throws Exception If failed. + */ + @Test + public void testLargeRowBelowThresholdUsesHeadroomNoEviction() throws Exception { + IgniteEx ignite = startGrid(1); + + IgniteCache cache = createCache(ignite, DEFAULT_CACHE_NAME); + + // Pre-fill to ~40% of total pages, leaving ~50% headroom. + int preFill = fillRegion(ignite, cache, 0.4); + + long loadedPages = loadedPages(ignite); + long totalPages = totalPages(ignite); + + assertTrue("Pre-fill must load ~40% of pages", loadedPages > totalPages * 0.30 && loadedPages < totalPages * 0.50); + assertFalse("Eviction must not have started during pre-fill below threshold", isEvictionsStarted(ignite)); + + // ~30% of the region fits in the remaining headroom (total ~70% < 0.9 threshold). + byte[] val = new byte[(int)(regionMaxSize(ignite) * 0.3)]; + + Arrays.fill(val, (byte)1); + + cache.put(preFill, val); + + assertFalse("Eviction must not start when the region stays below the threshold", isEvictionsStarted(ignite)); + + for (int i = 0; i < preFill; i++) + assertNotNull("Pre-filled entry " + i + " must not be evicted below threshold", cache.get(i)); + + byte[] read = (byte[])cache.get(preFill); + + assertNotNull("Large row must be readable", read); + assertTrue("Value read back must equal the stored value", Arrays.equals(val, read)); + } + + /** + * Above the eviction threshold, a large row exceeding the headroom + empty-pages pool is written by evicting + * pre-filled entries. + * + * @throws Exception If failed. + */ + @Test + public void testLargeRowAboveThresholdEvictsAndSucceeds() throws Exception { + IgniteEx ignite = startGrid(1); + + IgniteCache cache = createCache(ignite, DEFAULT_CACHE_NAME); + + // Pre-fill to ~85% of capacity (above the 0.9 threshold with page overhead). + int smallEntries = fillRegion(ignite, cache, 0.85); + + assertTrue("Pre-fill must load >80% of pages", loadedPages(ignite) > totalPages(ignite) * 0.80); + + // Large row exceeding headroom + empty-pages pool → reserve must evict to make room. + byte[] val = new byte[(int)(regionMaxSize(ignite) / 4)]; + + Arrays.fill(val, (byte)1); + + int largeKey = smallEntries + 1; + + cache.put(largeKey, val); + + assertTrue("Eviction must have started after a large put near capacity", isEvictionsStarted(ignite)); + + byte[] read = (byte[])cache.get(largeKey); + + assertNotNull("Large row must be readable after eviction", read); + assertTrue("Value read back must equal the stored value", Arrays.equals(val, read)); + } + + /** + * Threshold-gate semantics: once loaded pages cross the 0.9 threshold, the gate evicts on every put, so the region + * caps near the threshold. Exercises both single-put and {@code putAll} paths. + * + * @throws Exception If failed. + */ + @Test + public void testRegionEvictsAtThreshold() throws Exception { + IgniteEx ignite = startGrid(1); + + IgniteCache cache = createCache(ignite, DEFAULT_CACHE_NAME); + + long total = totalPages(ignite); + + int cap = (int)total * 2; + + byte[] small = new byte[pageSize(ignite) - 200]; + + // Fill until eviction starts. + for (int i = 0; i < cap; i++) { + cache.put(i, small); + + if (isEvictionsStarted(ignite)) + break; + } + + assertTrue("Eviction must start once the region crosses the 0.9 threshold", isEvictionsStarted(ignite)); + assertTrue("Region must reach at least ~85% before eviction engages", loadedPages(ignite) >= total * 0.85); + + stopGrid(1); + ignite = startGrid(1); + + assertEquals(0, loadedPages(ignite)); + + cache = createCache(ignite, DEFAULT_CACHE_NAME); + + int batchSize = 100; + + // Fill via putAll batches until eviction starts. + for (int base = 0; base < cap && !isEvictionsStarted(ignite); base += batchSize) { + Map batch = new HashMap<>(); + + for (int i = 0; i < batchSize; i++) + batch.put(base + i, small); + + cache.putAll(batch); + } + + assertTrue("Eviction must start via putAll once the region crosses the 0.9 threshold", + isEvictionsStarted(ignite)); + assertTrue("Region must reach at least ~85% before eviction engages via putAll", + loadedPages(ignite) >= total * 0.85); + } + + /** @return Loaded pages. */ + private long loadedPages(IgniteEx ignite) throws IgniteCheckedException { + return defaultRegion(ignite).pageMemory().loadedPages(); + } +} diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/eviction/paged/PageEvictionWithExpiryPolicyAbstractTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/eviction/paged/PageEvictionWithExpiryPolicyAbstractTest.java new file mode 100644 index 0000000000000..37ab38b0afb54 --- /dev/null +++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/eviction/paged/PageEvictionWithExpiryPolicyAbstractTest.java @@ -0,0 +1,141 @@ +/* + * 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.ignite.internal.processors.cache.eviction.paged; + +import java.util.Arrays; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import org.apache.ignite.IgniteCache; +import org.apache.ignite.internal.IgniteEx; +import org.apache.ignite.internal.IgniteInternalFuture; +import org.apache.ignite.internal.util.typedef.internal.U; +import org.apache.ignite.testframework.GridTestUtils; +import org.junit.Test; + +import static org.apache.ignite.testframework.GridTestUtils.waitForCondition; + +/** Tests the synergy between ExpiryPolicy (TTL cleanup) and size-aware page eviction on an in-memory data region. */ +public abstract class PageEvictionWithExpiryPolicyAbstractTest extends PageEvictionAbstractTest { + /** Short TTL applied to some entries. */ + private static final long TTL = 2000; + + /** {@inheritDoc} */ + @Override protected void afterTest() throws Exception { + stopAllGrids(); + } + + /** + * Concurrent page eviction and TTL cleanup on the same data region must not deadlock: large inserts trigger + * eviction while page-sized TTL entries expire and free real pages concurrently. + * + * @throws Exception If failed. + */ + @Test + public void testLargePutWithExpiryNoDeadlock() throws Exception { + IgniteEx ignite = startGrid(1); + + IgniteCache plainCache = createCache(ignite, "plain-cache"); + IgniteCache ttlCache = createCache(ignite, "ttl-cache", TTL, false); + + int smallEntries = 1_000; + + for (int i = 0; i < smallEntries; i++) + plainCache.put(i, new byte[16 * 1024]); + + assertFalse("Eviction must not have started during pre-fill", isEvictionsStarted(ignite)); + + // Page-sized TTL entries: their cleanup frees real pages concurrently with eviction. + int ttlEntries = 100; + byte[] ttlVal = new byte[pageSize(ignite) - 200]; + + for (int i = 0; i < ttlEntries; i++) + ttlCache.put(i, ttlVal); + + int largeRowsPerThread = 40; + byte[] largeVal = new byte[1024 * 1024]; + + CountDownLatch startLatch = new CountDownLatch(1); + AtomicInteger threadIdx = new AtomicInteger(); + + IgniteInternalFuture fut = GridTestUtils.runMultiThreadedAsync(() -> { + U.awaitQuiet(startLatch); + + int idx = threadIdx.getAndIncrement(); + + for (int k = 0; k < largeRowsPerThread; k++) + plainCache.put(smallEntries + idx * largeRowsPerThread + k, largeVal); + }, 10, "paged-writer"); + + startLatch.countDown(); + + assertTrue("TTL entries must expire after TTL duration", + waitForCondition(() -> { + for (int i = 0; i < ttlEntries; i++) { + if (ttlCache.get(i) != null) + return false; + } + + return true; + }, TTL + 10_000)); + + fut.get(TimeUnit.MINUTES.toMillis(3)); + + assertTrue("Eviction must have started during concurrent large writes", isEvictionsStarted(ignite)); + } + + /** + * After short-TTL entries expire and free their pages, putting the same data to a non-expiring cache must succeed. + * + * @throws Exception If failed. + */ + @Test + public void testTtlFreedSpaceAccountedForByEviction() throws Exception { + IgniteEx ignite = startGrid(1); + + IgniteCache plainCache = createCache(ignite, "plain-cache"); + IgniteCache ttlCache = createCache(ignite, "ttl-cache", TTL, false); + + int smallEntries = fillRegion(ignite, plainCache, 0.80); + + assertFalse("Eviction must not start during pre-fill", isEvictionsStarted(ignite)); + + int ttlEntries = 2; + byte[] val = new byte[(int)(regionMaxSize(ignite) * 0.05)]; + + for (int i = 0; i < ttlEntries; i++) + ttlCache.put(i, val); + + assertNotNull("TTL entry must be present before expiry", ttlCache.get(0)); + + assertTrue("TTL entries must expire", waitForCondition(() -> ttlCache.size() == 0, 10_000)); + + for (int i = 0; i < ttlEntries; i++) + plainCache.put(smallEntries + i, val); + + for (int i = 0; i < ttlEntries; i++) { + byte[] read = (byte[])plainCache.get(smallEntries + i); + + assertNotNull("Fresh large record must be present", read); + assertTrue("Fresh large record must match", Arrays.equals(val, read)); + } + + for (int i = 0; i < smallEntries; i++) + assertNotNull("Pre-filled entry " + i + " must not be evicted", plainCache.get(i)); + } +} diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/eviction/paged/Random2LruPageEvictionConcurrentWritesTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/eviction/paged/Random2LruPageEvictionConcurrentWritesTest.java new file mode 100644 index 0000000000000..22ebaad0f453e --- /dev/null +++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/eviction/paged/Random2LruPageEvictionConcurrentWritesTest.java @@ -0,0 +1,29 @@ +/* + * 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.ignite.internal.processors.cache.eviction.paged; + +import org.apache.ignite.configuration.DataPageEvictionMode; +import org.apache.ignite.configuration.IgniteConfiguration; + +/** Concurrent eviction/insertion test for {@link DataPageEvictionMode#RANDOM_2_LRU}. */ +public class Random2LruPageEvictionConcurrentWritesTest extends PageEvictionConcurrentWritesAbstractTest { + /** {@inheritDoc} */ + @Override protected IgniteConfiguration getConfiguration(String gridName) throws Exception { + return setEvictionMode(DataPageEvictionMode.RANDOM_2_LRU, super.getConfiguration(gridName)); + } +} diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/eviction/paged/Random2LruPageEvictionSizeAwareTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/eviction/paged/Random2LruPageEvictionSizeAwareTest.java new file mode 100644 index 0000000000000..58f5cab37585f --- /dev/null +++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/eviction/paged/Random2LruPageEvictionSizeAwareTest.java @@ -0,0 +1,29 @@ +/* + * 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.ignite.internal.processors.cache.eviction.paged; + +import org.apache.ignite.configuration.DataPageEvictionMode; +import org.apache.ignite.configuration.IgniteConfiguration; + +/** Size-aware page eviction test for {@link DataPageEvictionMode#RANDOM_2_LRU}. */ +public class Random2LruPageEvictionSizeAwareTest extends PageEvictionSizeAwareAbstractTest { + /** {@inheritDoc} */ + @Override protected IgniteConfiguration getConfiguration(String gridName) throws Exception { + return setEvictionMode(DataPageEvictionMode.RANDOM_2_LRU, super.getConfiguration(gridName)); + } +} diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/eviction/paged/Random2LruPageEvictionWithExpiryPolicyTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/eviction/paged/Random2LruPageEvictionWithExpiryPolicyTest.java new file mode 100644 index 0000000000000..5563620507507 --- /dev/null +++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/eviction/paged/Random2LruPageEvictionWithExpiryPolicyTest.java @@ -0,0 +1,29 @@ +/* + * 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.ignite.internal.processors.cache.eviction.paged; + +import org.apache.ignite.configuration.DataPageEvictionMode; +import org.apache.ignite.configuration.IgniteConfiguration; + +/** ExpiryPolicy + page eviction synergy test for {@link DataPageEvictionMode#RANDOM_2_LRU}. */ +public class Random2LruPageEvictionWithExpiryPolicyTest extends PageEvictionWithExpiryPolicyAbstractTest { + /** {@inheritDoc} */ + @Override protected IgniteConfiguration getConfiguration(String gridName) throws Exception { + return setEvictionMode(DataPageEvictionMode.RANDOM_2_LRU, super.getConfiguration(gridName)); + } +} diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/eviction/paged/RandomLruPageEvictionConcurrentWritesTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/eviction/paged/RandomLruPageEvictionConcurrentWritesTest.java new file mode 100644 index 0000000000000..be2dbcea3c91d --- /dev/null +++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/eviction/paged/RandomLruPageEvictionConcurrentWritesTest.java @@ -0,0 +1,29 @@ +/* + * 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.ignite.internal.processors.cache.eviction.paged; + +import org.apache.ignite.configuration.DataPageEvictionMode; +import org.apache.ignite.configuration.IgniteConfiguration; + +/** Concurrent eviction/insertion test for {@link DataPageEvictionMode#RANDOM_LRU}. */ +public class RandomLruPageEvictionConcurrentWritesTest extends PageEvictionConcurrentWritesAbstractTest { + /** {@inheritDoc} */ + @Override protected IgniteConfiguration getConfiguration(String gridName) throws Exception { + return setEvictionMode(DataPageEvictionMode.RANDOM_LRU, super.getConfiguration(gridName)); + } +} diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/eviction/paged/RandomLruPageEvictionSizeAwareTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/eviction/paged/RandomLruPageEvictionSizeAwareTest.java new file mode 100644 index 0000000000000..cb70cf540f825 --- /dev/null +++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/eviction/paged/RandomLruPageEvictionSizeAwareTest.java @@ -0,0 +1,29 @@ +/* + * 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.ignite.internal.processors.cache.eviction.paged; + +import org.apache.ignite.configuration.DataPageEvictionMode; +import org.apache.ignite.configuration.IgniteConfiguration; + +/** Size-aware page eviction test for {@link DataPageEvictionMode#RANDOM_LRU}. */ +public class RandomLruPageEvictionSizeAwareTest extends PageEvictionSizeAwareAbstractTest { + /** {@inheritDoc} */ + @Override protected IgniteConfiguration getConfiguration(String gridName) throws Exception { + return setEvictionMode(DataPageEvictionMode.RANDOM_LRU, super.getConfiguration(gridName)); + } +} diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/eviction/paged/RandomLruPageEvictionWithExpiryPolicyTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/eviction/paged/RandomLruPageEvictionWithExpiryPolicyTest.java new file mode 100644 index 0000000000000..98f5e7bf42a36 --- /dev/null +++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/eviction/paged/RandomLruPageEvictionWithExpiryPolicyTest.java @@ -0,0 +1,29 @@ +/* + * 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.ignite.internal.processors.cache.eviction.paged; + +import org.apache.ignite.configuration.DataPageEvictionMode; +import org.apache.ignite.configuration.IgniteConfiguration; + +/** ExpiryPolicy + page eviction synergy test for {@link DataPageEvictionMode#RANDOM_LRU}. */ +public class RandomLruPageEvictionWithExpiryPolicyTest extends PageEvictionWithExpiryPolicyAbstractTest { + /** {@inheritDoc} */ + @Override protected IgniteConfiguration getConfiguration(String gridName) throws Exception { + return setEvictionMode(DataPageEvictionMode.RANDOM_LRU, super.getConfiguration(gridName)); + } +} diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/freelist/FreeListReservedPagesAbstractTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/freelist/FreeListReservedPagesAbstractTest.java new file mode 100644 index 0000000000000..0e9d83b3b1056 --- /dev/null +++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/freelist/FreeListReservedPagesAbstractTest.java @@ -0,0 +1,97 @@ +/* + * 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.ignite.internal.processors.cache.persistence.freelist; + +import java.util.concurrent.atomic.AtomicLong; +import org.apache.ignite.Ignite; +import org.apache.ignite.Ignition; +import org.apache.ignite.configuration.DataPageEvictionMode; +import org.apache.ignite.configuration.DataRegionConfiguration; +import org.apache.ignite.configuration.DataStorageConfiguration; +import org.apache.ignite.configuration.IgniteConfiguration; +import org.apache.ignite.internal.IgniteEx; +import org.apache.ignite.internal.processors.cache.IgniteCacheOffheapManager.CacheDataStore; +import org.apache.ignite.internal.processors.cache.persistence.CacheDataRow; +import org.apache.ignite.testframework.GridTestUtils; +import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest; + +import static org.apache.ignite.testframework.GridTestUtils.getFieldValue; + +/** + * Base for tests verifying that the atomic page reservation of an in-memory free list with enabled page eviction is + * fully released after inserts. + */ +public abstract class FreeListReservedPagesAbstractTest extends GridCommonAbstractTest { + /** {@inheritDoc} */ + @Override protected IgniteConfiguration getConfiguration(String igniteInstanceName) throws Exception { + IgniteConfiguration cfg = super.getConfiguration(igniteInstanceName); + + cfg.setDataStorageConfiguration(new DataStorageConfiguration() + .setDefaultDataRegionConfiguration(new DataRegionConfiguration() + .setInitialSize(10 * 1024 * 1024L) + .setMaxSize(20 * 1024 * 1024L) + .setPageEvictionMode(DataPageEvictionMode.RANDOM_LRU))); + + return cfg; + } + + /** {@inheritDoc} */ + @Override protected void afterTest() throws Exception { + super.afterTest(); + + stopAllGrids(); + } + + /** + * @return Pages currently reserved by in-flight inserts on the given free list. + */ + protected static long reservedPages(FreeList freeList) { + return ((AtomicLong)getFieldValue(freeList, "reservedPages")).get(); + } + + /** + * @param ignite Node. + * @param cacheName Cache name. + * @return The free list of the first partition data store of the named cache on the given node. + */ + protected static FreeList freeList(IgniteEx ignite, String cacheName) { + return (FreeList)ignite.context().cache().cache(cacheName).context().offheap() + .cacheDataStores().iterator().next().rowStore().freeList(); + } + + /** + * Asserts that {@code reservedPages} is {@code 0} on every data store of every started node, waiting up to + * {@code timeoutMs} for the reservation to drain. + * + * @param cacheName Cache name. + * @param timeoutMs Wait timeout in milliseconds. + * @throws Exception If the reservation does not drain in time. + */ + public static void assertAllReleased(String cacheName, long timeoutMs) throws Exception { + for (Ignite ignite : Ignition.allGrids()) { + for (CacheDataStore store : ((IgniteEx)ignite).context().cache().cache(cacheName).context().offheap().cacheDataStores()) { + FreeList freeList = (FreeList)store.rowStore().freeList(); + + if (reservedPages(freeList) == 0) + continue; + + assertTrue(GridTestUtils.waitForCondition(() -> reservedPages(freeList) == 0, timeoutMs)); + } + } + } +} diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/freelist/FreeListReservedPagesSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/freelist/FreeListReservedPagesSelfTest.java new file mode 100644 index 0000000000000..08845e27713b1 --- /dev/null +++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/freelist/FreeListReservedPagesSelfTest.java @@ -0,0 +1,253 @@ +/* + * 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.ignite.internal.processors.cache.persistence.freelist; + +import java.util.ArrayList; +import java.util.Collection; +import java.util.HashMap; +import java.util.Map; +import java.util.concurrent.atomic.AtomicInteger; +import org.apache.ignite.IgniteCache; +import org.apache.ignite.IgniteCheckedException; +import org.apache.ignite.cache.CacheWriteSynchronizationMode; +import org.apache.ignite.cache.affinity.rendezvous.RendezvousAffinityFunction; +import org.apache.ignite.configuration.CacheConfiguration; +import org.apache.ignite.internal.IgniteEx; +import org.apache.ignite.internal.metric.IoStatisticsHolderNoOp; +import org.apache.ignite.internal.processors.cache.CacheObject; +import org.apache.ignite.internal.processors.cache.GridCacheContext; +import org.apache.ignite.internal.processors.cache.KeyCacheObject; +import org.apache.ignite.internal.processors.cache.persistence.CacheDataRow; +import org.apache.ignite.internal.processors.cache.tree.DataRow; +import org.apache.ignite.internal.processors.cache.version.GridCacheVersion; +import org.apache.ignite.testframework.GridTestUtils; +import org.junit.Test; + +import static org.apache.ignite.testframework.GridTestUtils.getFieldValue; + +/** + * Tests that the atomic page reservation of an in-memory free list with enabled page eviction is fully released after + * batch inserts: whole-page fragments, fill-up-handler rows, whole-page-aligned rows, concurrent batch inserts, and + * the end-to-end backup entry path. + */ +public class FreeListReservedPagesSelfTest extends FreeListReservedPagesAbstractTest { + /** */ + @Test + public void testReservedPagesReleasedAfterSmallRowsBatch() throws Exception { + assertReservedPagesReleasedAfterBatch(10); + } + + /** */ + @Test + public void testReservedPagesReleasedAfterLargeRowsBatch() throws Exception { + assertReservedPagesReleasedAfterBatch(10_000); + } + + /** */ + @Test + public void testReservedPagesReleasedAfterWholePageAlignedRowsBatch() throws Exception { + IgniteEx ignite = startGrid(0); + + createCache(ignite); + + FreeList freeList = freeList(ignite, DEFAULT_CACHE_NAME); + + int minPageSize = getFieldValue(freeList, "MIN_SIZE_FOR_DATA_PAGE"); + + // Size the value so the whole row is an exact multiple of the whole-page fragment size. + int fixedSize = row(cacheContext(ignite), 0, 0).size(); + + insertBatches(ignite, freeList, 2 * minPageSize - fixedSize); + + assertEquals(0, reservedPages(freeList)); + } + + /** */ + @Test + public void testReservedPagesReleasedAfterConcurrentBatchInserts() throws Exception { + IgniteEx ignite = startGrid(0); + + createCache(ignite); + + final FreeList freeList = freeList(ignite, DEFAULT_CACHE_NAME); + + final GridCacheContext cctx = cacheContext(ignite); + + int rowsPerThread = 200; + + final AtomicInteger keyGen = new AtomicInteger(); + + GridTestUtils.runMultiThreaded(() -> { + try { + Collection rows = new ArrayList<>(rowsPerThread); + + for (int i = 0; i < rowsPerThread; i++) + rows.add(row(cctx, keyGen.incrementAndGet(), 100)); + + freeList.insertDataRows(rows, IoStatisticsHolderNoOp.INSTANCE); + } + catch (IgniteCheckedException e) { + throw new RuntimeException(e); + } + }, 8, "concurrent-batch-insert"); + + assertTrue(GridTestUtils.waitForCondition(() -> reservedPages(freeList) == 0, 10_000)); + } + + /** + * End-to-end entry path: a 2-node cluster with {@code backups=1} so the backup node reserves pages before the + * entry lock. After puts and quiescence, {@code reservedPages} must be 0 on both nodes. + * + * @throws Exception If failed. + */ + @Test + public void testReservedPagesReleasedAfterEntryPathPutsOnBackup() throws Exception { + startGrids(2); + + IgniteCache cache = grid(0).createCache(new CacheConfiguration(DEFAULT_CACHE_NAME) + .setAffinity(new RendezvousAffinityFunction(false, 32)) + .setBackups(1) + .setWriteSynchronizationMode(CacheWriteSynchronizationMode.FULL_SYNC)); + + awaitPartitionMapExchange(); + + int keyCnt = 500; + int payloadSize = 200; + + for (int i = 0; i < keyCnt; i++) + cache.put(i, new byte[payloadSize]); + + assertAllReleased(DEFAULT_CACHE_NAME, 10_000); + } + + /** + * Concurrent single-row puts on a 2-node cluster with backups; {@code reservedPages} must be 0 after quiescence. + * + * @throws Exception If failed. + */ + @Test + public void testReservedPagesReleasedAfterConcurrentPuts() throws Exception { + startGrids(2); + + IgniteCache cache = grid(0).createCache(new CacheConfiguration(DEFAULT_CACHE_NAME) + .setAffinity(new RendezvousAffinityFunction(false, 32)) + .setBackups(1) + .setWriteSynchronizationMode(CacheWriteSynchronizationMode.FULL_SYNC)); + + awaitPartitionMapExchange(); + + AtomicInteger keyGen = new AtomicInteger(); + + GridTestUtils.runMultiThreaded(() -> { + for (int i = 0; i < 300; i++) { + int key = keyGen.incrementAndGet(); + + cache.put(key, new byte[500]); + } + }, 8, "concurrent-put"); + + assertAllReleased(DEFAULT_CACHE_NAME, 10_000); + } + + /** + * Concurrent {@code putAll} batches on a 2-node cluster with backups; {@code reservedPages} must be 0 after + * quiescence. + * + * @throws Exception If failed. + */ + @Test + public void testReservedPagesReleasedAfterConcurrentPutAll() throws Exception { + startGrids(2); + + IgniteCache cache = grid(0).createCache(new CacheConfiguration(DEFAULT_CACHE_NAME) + .setAffinity(new RendezvousAffinityFunction(false, 32)) + .setBackups(1) + .setWriteSynchronizationMode(CacheWriteSynchronizationMode.FULL_SYNC)); + + awaitPartitionMapExchange(); + + AtomicInteger keyGen = new AtomicInteger(); + + GridTestUtils.runMultiThreaded(() -> { + for (int b = 0; b < 10; b++) { + Map batch = new HashMap<>(); + + for (int i = 0; i < 100; i++) { + int key = keyGen.incrementAndGet(); + + batch.put(key, new byte[300]); + } + + cache.putAll(batch); + } + }, 4, "concurrent-putAll"); + + assertAllReleased(DEFAULT_CACHE_NAME, 10_000); + } + + /** */ + private void assertReservedPagesReleasedAfterBatch(int payloadSize) throws Exception { + IgniteEx ignite = startGrid(0); + + createCache(ignite); + + insertBatches(ignite, freeList(ignite, DEFAULT_CACHE_NAME), payloadSize); + } + + /** */ + private static void insertBatches(IgniteEx ignite, FreeList freeList, int payloadSize) throws IgniteCheckedException { + int batchSize = 500; + + for (int batch = 0; batch < 3; batch++) { + GridCacheContext cctx = cacheContext(ignite); + + Collection rows = new ArrayList<>(batchSize); + + for (int i = 0; i < batchSize; i++) + rows.add(row(cctx, batch * batchSize + i, payloadSize)); + + freeList.insertDataRows(rows, IoStatisticsHolderNoOp.INSTANCE); + + assertEquals(0, reservedPages(freeList)); + } + } + + /** @return Context of the default cache. */ + private static GridCacheContext cacheContext(IgniteEx ignite) { + return ignite.context().cache().cache(DEFAULT_CACHE_NAME).context(); + } + + /** */ + private static void createCache(IgniteEx ignite) { + IgniteCache cache = ignite.createCache(new CacheConfiguration(DEFAULT_CACHE_NAME) + .setAffinity(new RendezvousAffinityFunction(false, 32)).setBackups(0)); + + cache.put(0, 0); + } + + /** @return Data row with the given key and a byte array value of the given size. */ + private static CacheDataRow row(GridCacheContext cctx, int key, int payloadSize) throws IgniteCheckedException { + KeyCacheObject keyObj = cctx.toCacheKeyObject(key); + CacheObject valObj = cctx.toCacheObject(new byte[payloadSize]); + + keyObj.valueBytes(cctx.cacheObjectContext()); + valObj.valueBytes(cctx.cacheObjectContext()); + + return new DataRow(keyObj, valObj, new GridCacheVersion(), 0, 0L, cctx.cacheId()); + } +} diff --git a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheEvictionSelfTestSuite.java b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheEvictionSelfTestSuite.java index 03d95e5fabbcb..dc325d3e9865b 100644 --- a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheEvictionSelfTestSuite.java +++ b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheEvictionSelfTestSuite.java @@ -38,22 +38,34 @@ import org.apache.ignite.internal.processors.cache.eviction.lru.LruEvictionPolicySelfTest; import org.apache.ignite.internal.processors.cache.eviction.lru.LruNearEvictionPolicySelfTest; import org.apache.ignite.internal.processors.cache.eviction.lru.LruNearOnlyNearEvictionPolicySelfTest; +import org.apache.ignite.internal.processors.cache.eviction.paged.PageEvictionDisabledEvictionReadThroughOomSelfTest; +import org.apache.ignite.internal.processors.cache.eviction.paged.PageEvictionEnsureFreeSpaceThresholdTest; +import org.apache.ignite.internal.processors.cache.eviction.paged.PageEvictionGuardOomTest; import org.apache.ignite.internal.processors.cache.eviction.paged.PageEvictionMetricTest; import org.apache.ignite.internal.processors.cache.eviction.paged.PageEvictionPagesRecyclingAndReusingTest; +import org.apache.ignite.internal.processors.cache.eviction.paged.PageEvictionReadThroughReservedPagesSelfTest; import org.apache.ignite.internal.processors.cache.eviction.paged.PageEvictionReadThroughTest; +import org.apache.ignite.internal.processors.cache.eviction.paged.PageEvictionRowSizeBinaryObjectTest; import org.apache.ignite.internal.processors.cache.eviction.paged.PageEvictionTouchOrderTest; import org.apache.ignite.internal.processors.cache.eviction.paged.Random2LruNearEnabledPageEvictionMultinodeTest; +import org.apache.ignite.internal.processors.cache.eviction.paged.Random2LruPageEvictionConcurrentWritesTest; import org.apache.ignite.internal.processors.cache.eviction.paged.Random2LruPageEvictionDataStreamerTest; import org.apache.ignite.internal.processors.cache.eviction.paged.Random2LruPageEvictionMultinodeTest; import org.apache.ignite.internal.processors.cache.eviction.paged.Random2LruPageEvictionPutLargeObjectsTest; +import org.apache.ignite.internal.processors.cache.eviction.paged.Random2LruPageEvictionSizeAwareTest; +import org.apache.ignite.internal.processors.cache.eviction.paged.Random2LruPageEvictionWithExpiryPolicyTest; import org.apache.ignite.internal.processors.cache.eviction.paged.Random2LruPageEvictionWithRebalanceTest; import org.apache.ignite.internal.processors.cache.eviction.paged.RandomLruNearEnabledPageEvictionMultinodeTest; +import org.apache.ignite.internal.processors.cache.eviction.paged.RandomLruPageEvictionConcurrentWritesTest; import org.apache.ignite.internal.processors.cache.eviction.paged.RandomLruPageEvictionDataStreamerTest; import org.apache.ignite.internal.processors.cache.eviction.paged.RandomLruPageEvictionMultinodeTest; import org.apache.ignite.internal.processors.cache.eviction.paged.RandomLruPageEvictionPutLargeObjectsTest; +import org.apache.ignite.internal.processors.cache.eviction.paged.RandomLruPageEvictionSizeAwareTest; +import org.apache.ignite.internal.processors.cache.eviction.paged.RandomLruPageEvictionWithExpiryPolicyTest; import org.apache.ignite.internal.processors.cache.eviction.paged.RandomLruPageEvictionWithRebalanceTest; import org.apache.ignite.internal.processors.cache.eviction.sorted.SortedEvictionPolicyFactorySelfTest; import org.apache.ignite.internal.processors.cache.eviction.sorted.SortedEvictionPolicySelfTest; +import org.apache.ignite.internal.processors.cache.persistence.freelist.FreeListReservedPagesSelfTest; import org.apache.ignite.testframework.GridTestUtils; /** @@ -100,12 +112,33 @@ public static List> suite(Collection ignoredTests) { GridTestUtils.addTestIfNeeded(suite, RandomLruPageEvictionPutLargeObjectsTest.class, ignoredTests); GridTestUtils.addTestIfNeeded(suite, Random2LruPageEvictionPutLargeObjectsTest.class, ignoredTests); + GridTestUtils.addTestIfNeeded(suite, RandomLruPageEvictionSizeAwareTest.class, ignoredTests); + GridTestUtils.addTestIfNeeded(suite, Random2LruPageEvictionSizeAwareTest.class, ignoredTests); + + GridTestUtils.addTestIfNeeded(suite, RandomLruPageEvictionWithExpiryPolicyTest.class, ignoredTests); + GridTestUtils.addTestIfNeeded(suite, Random2LruPageEvictionWithExpiryPolicyTest.class, ignoredTests); + + GridTestUtils.addTestIfNeeded(suite, RandomLruPageEvictionConcurrentWritesTest.class, ignoredTests); + GridTestUtils.addTestIfNeeded(suite, Random2LruPageEvictionConcurrentWritesTest.class, ignoredTests); + GridTestUtils.addTestIfNeeded(suite, PageEvictionMetricTest.class, ignoredTests); + GridTestUtils.addTestIfNeeded(suite, PageEvictionGuardOomTest.class, ignoredTests); + GridTestUtils.addTestIfNeeded(suite, PageEvictionPagesRecyclingAndReusingTest.class, ignoredTests); GridTestUtils.addTestIfNeeded(suite, DhtAndNearEvictionTest.class, ignoredTests); + GridTestUtils.addTestIfNeeded(suite, FreeListReservedPagesSelfTest.class, ignoredTests); + + GridTestUtils.addTestIfNeeded(suite, PageEvictionReadThroughReservedPagesSelfTest.class, ignoredTests); + + GridTestUtils.addTestIfNeeded(suite, PageEvictionRowSizeBinaryObjectTest.class, ignoredTests); + + GridTestUtils.addTestIfNeeded(suite, PageEvictionEnsureFreeSpaceThresholdTest.class, ignoredTests); + + GridTestUtils.addTestIfNeeded(suite, PageEvictionDisabledEvictionReadThroughOomSelfTest.class, ignoredTests); + return suite; } }