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..6ec06422a8a79 --- /dev/null +++ b/modules/benchmarks/src/main/java/org/apache/ignite/internal/benchmarks/jmh/cache/JmhPageEvictionBenchmark.java @@ -0,0 +1,224 @@ +/* + * 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.Map; +import java.util.TreeMap; +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.cache.CacheAtomicityMode; +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.Measurement; +import org.openjdk.jmh.annotations.Mode; +import org.openjdk.jmh.annotations.OutputTimeUnit; +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 overhead of page eviction on the write paths of an in-memory data region with enabled page eviction. + * + * The benchmark uses the public API only, so the same class can be run on another branch (e.g. {@code master}) with the + * same parameters to compare the results. + */ +@Fork(1) +@Threads(4) +@BenchmarkMode(Mode.Throughput) +@OutputTimeUnit(TimeUnit.MILLISECONDS) +@Warmup(iterations = 3, time = 5) +@Measurement(iterations = 5, time = 10) +public class JmhPageEvictionBenchmark { + /** Atomic cache name. */ + private static final String ATOMIC_CACHE = "atomic"; + + /** Transactional cache name. */ + private static final String TX_CACHE = "tx"; + + /** Value size (bytes): several rows per data page. */ + private static final int VALUE_SIZE = 1024; + + /** Data region size (bytes). */ + private static final long REGION_SIZE = 256 * 1024L * 1024L; + + /** Bounded key range of {@link #putBelowThreshold}: about 15% of the region, far below the eviction threshold. */ + private static final int BELOW_THRESHOLD_KEYS = 32_000; + + /** Pre-filled entries of the at-threshold scenarios: more than the region holds, so eviction has already started. */ + private static final int PRE_FILL_ENTRIES = 400_000; + + /** Batch size of {@link #putAllAtThreshold}. */ + private static final int BATCH_SIZE = 100; + + /** Value (reused to avoid allocation noise in the hot path). */ + private static final byte[] VAL = new byte[VALUE_SIZE]; + + /** Node with an empty data region. */ + @State(Scope.Benchmark) + public static class BelowThreshold { + /** Atomic cache. */ + private IgniteCache cache; + + /** Key source. */ + private final AtomicInteger keyGen = new AtomicInteger(); + + /** */ + @Setup + public void setup() { + cache = startNode().cache(ATOMIC_CACHE); + } + + /** */ + @TearDown + public void tearDown() { + Ignition.stopAll(true); + } + } + + /** Node with a data region pre-filled up to the eviction threshold. */ + @State(Scope.Benchmark) + public static class AtThreshold { + /** Atomic cache. */ + private IgniteCache atomicCache; + + /** Transactional cache. */ + private IgniteCache txCache; + + /** Key source: every key is new, so every put inserts a new row. */ + private final AtomicInteger keyGen = new AtomicInteger(PRE_FILL_ENTRIES); + + /** */ + @Setup + public void setup() { + Ignite ignite = startNode(); + + atomicCache = ignite.cache(ATOMIC_CACHE); + txCache = ignite.cache(TX_CACHE); + + try (IgniteDataStreamer ldr = ignite.dataStreamer(ATOMIC_CACHE)) { + for (int i = 0; i < PRE_FILL_ENTRIES; i++) + ldr.addData(i, VAL); + } + } + + /** */ + @TearDown + public void tearDown() { + Ignition.stopAll(true); + } + } + + /** + * Atomic put below the eviction threshold (updates of a bounded key range). + * + * @param state State. + */ + @Benchmark + public void putBelowThreshold(BelowThreshold state) { + state.cache.put(state.keyGen.incrementAndGet() % BELOW_THRESHOLD_KEYS, VAL); + } + + /** + * Atomic put of a new row into a region at the eviction threshold. + * + * @param state State. + */ + @Benchmark + public void putAtThreshold(AtThreshold state) { + state.atomicCache.put(state.keyGen.incrementAndGet(), VAL); + } + + /** + * Atomic putAll of new rows into a region at the eviction threshold. + * + * @param state State. + */ + @Benchmark + public void putAllAtThreshold(AtThreshold state) { + int base = state.keyGen.getAndAdd(BATCH_SIZE); + + Map batch = new TreeMap<>(); + + for (int i = 0; i < BATCH_SIZE; i++) + batch.put(base + i, VAL); + + state.atomicCache.putAll(batch); + } + + /** + * Transactional put (implicit transaction) of a new row into a region at the eviction threshold. + * + * @param state State. + */ + @Benchmark + public void txPutAtThreshold(AtThreshold state) { + state.txCache.put(state.keyGen.incrementAndGet(), VAL); + } + + /** @return Started node with an atomic and a transactional cache in an in-memory region with page eviction. */ + private static Ignite startNode() { + DataStorageConfiguration dsCfg = new DataStorageConfiguration() + .setDefaultDataRegionConfiguration(new DataRegionConfiguration() + .setPersistenceEnabled(false) + .setInitialSize(REGION_SIZE) + .setMaxSize(REGION_SIZE) + .setPageEvictionMode(DataPageEvictionMode.RANDOM_LRU)); + + IgniteConfiguration cfg = new IgniteConfiguration() + .setIgniteInstanceName("page-eviction-benchmark") + .setLocalHost("127.0.0.1") + .setDataStorageConfiguration(dsCfg) + .setCacheConfiguration( + new CacheConfiguration(ATOMIC_CACHE), + new CacheConfiguration(TX_CACHE).setAtomicityMode(CacheAtomicityMode.TRANSACTIONAL)); + + return Ignition.start(cfg); + } + + /** + * 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 a0bac1118da80..75d7733309ea1 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,8 @@ public BinaryObjectImpl(BinaryContext ctx, byte[] arr, int start) { /** {@inheritDoc} */ @Override public int valueBytesLength(CacheObjectValueContext ctx) throws IgniteCheckedException { + marshal(ctx); + return CacheObjectUtils.objectPutSize(valBytes.length); } diff --git a/modules/core/src/main/java/org/apache/ignite/IgniteSystemProperties.java b/modules/core/src/main/java/org/apache/ignite/IgniteSystemProperties.java index 611195665f10b..33bd6b8f8d8cb 100644 --- a/modules/core/src/main/java/org/apache/ignite/IgniteSystemProperties.java +++ b/modules/core/src/main/java/org/apache/ignite/IgniteSystemProperties.java @@ -88,6 +88,7 @@ import static org.apache.ignite.internal.processors.cache.persistence.diagnostic.pagelocktracker.PageLockTrackerFactory.DFLT_PAGE_LOCK_TRACKER_CAPACITY; import static org.apache.ignite.internal.processors.cache.persistence.diagnostic.pagelocktracker.PageLockTrackerFactory.HEAP_LOG; import static org.apache.ignite.internal.processors.cache.persistence.diagnostic.pagelocktracker.SharedPageLockTracker.DFLT_PAGE_LOCK_TRACKER_CHECK_INTERVAL; +import static org.apache.ignite.internal.processors.cache.persistence.freelist.AbstractFreeList.DFLT_EVICTION_NO_PROGRESS_TIMEOUT_MILLIS; import static org.apache.ignite.internal.processors.cache.persistence.pagemem.FullPageIdTable.DFLT_LONG_LONG_HASH_MAP_LOAD_FACTOR; import static org.apache.ignite.internal.processors.cache.persistence.pagemem.PageMemoryImpl.DFLT_DELAYED_REPLACED_PAGE_WRITE; import static org.apache.ignite.internal.processors.cache.persistence.pagemem.PageMemoryImpl.DFLT_LOADED_PAGES_BACKWARD_SHIFT_MAP; @@ -1638,6 +1639,24 @@ public final class IgniteSystemProperties extends IgniteCommonsSystemProperties type = Integer.class, defaults = "" + DFLT_PAGE_LOCK_TRACKER_CHECK_INTERVAL) public static final String IGNITE_PAGE_LOCK_TRACKER_CHECK_INTERVAL = "IGNITE_PAGE_LOCK_TRACKER_CHECK_INTERVAL"; + /** + * Maximum time in milliseconds the size-aware eviction guard is willing to wait without a deficit decrease. This is + * a liveness guard against a true stall (all evictable entries locked); a progressing eviction that is simply slow + * (e.g. a large putAll on a nearly-full region) is not interrupted by this timeout. When it expires, an insert that + * can be refused (e.g. a primary atomic update) fails with an out-of-memory, while an insert that can not be safely + * refused (a backup update, a transaction commit, rebalancing) proceeds without a page reservation. After that, + * until the eviction makes progress again, such inserts proceed without waiting for the timeout. + *

+ * {@code 0} or a negative value disables the grace period: the guard gives up on the first no-progress iteration. + *

+ * Default is {@code 5000} (5 seconds). + */ + @SystemProperty(value = "Maximum time in milliseconds the size-aware eviction guard waits without progress. " + + "Then an insert that can be refused fails with out-of-memory, other inserts proceed without a page " + + "reservation. 0 or a negative value disables the grace period", type = Long.class, + defaults = "" + DFLT_EVICTION_NO_PROGRESS_TIMEOUT_MILLIS) + public static final String IGNITE_PAGE_EVICTION_NO_PROGRESS_TIMEOUT_MILLIS = "IGNITE_PAGE_EVICTION_NO_PROGRESS_TIMEOUT_MILLIS"; + /** * Enables threads locks dumping on critical node failure. * 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..0c0cc6485f7a5 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; @@ -323,6 +322,9 @@ public DataRegionConfiguration setPageReplacementMode(PageReplacementMode replac /** * Gets a threshold for memory pages eviction initiation. For instance, if the threshold is 0.9 it means that the * page memory will start the eviction only after 90% of the data region is occupied. + * The threshold applies to every insert into the region (primary and backup updates, transactions, rebalancing, + * data streaming): the eviction keeps the empty pages pool plus the space of the rows being inserted free once + * the threshold is reached. A row that does not fit below the threshold may use the space above it. * * @return Memory pages eviction threshold. */ @@ -346,9 +348,10 @@ 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 row plus this pool exceeds the available + * free pages, it is no longer required to increase this parameter up to the size of the largest cache entry. + * In addition, size-aware eviction keeps this many empty pages free as a single region-wide margin on top of the + * pages reserved by in-flight inserts (the margin is not multiplied by the number of concurrent writers). * * @return Minimum number of empty pages in reuse list. */ @@ -360,9 +363,10 @@ 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 row plus this pool exceeds the available + * free pages, it is no longer required to increase this parameter up to the size of the largest cache entry. + * In addition, size-aware eviction keeps this many empty pages free as a single region-wide margin on top of the + * pages reserved by in-flight inserts (the margin is not multiplied by the number of 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/GridCacheMapEntry.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMapEntry.java index 816a66f250572..0067bbbccb5da 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.IgniteCacheDatabaseSharedManager; 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; @@ -885,8 +886,10 @@ private EntryGetResult entryGetResult(CacheObject val, GridCacheVersion ver, boo boolean touch = false; + long reserved; + try { - ensureFreeSpace(); + reserved = reserveEntrySpace(ret, false); lockEntry(); @@ -932,6 +935,8 @@ private EntryGetResult entryGetResult(CacheObject val, GridCacheVersion ver, boo } finally { unlockEntry(); + + releaseReserved(reserved); } touch = true; @@ -988,7 +993,8 @@ protected void recordNodeId(UUID nodeId, AffinityTopologyVersion topVer) { long updateCntr0; - ensureFreeSpace(); + // Best-effort: a refused commit is not safe, other entries of the transaction may already be committed. + long reserved = reserveEntrySpace(val, true); lockListenerReadLock(); lockEntry(); @@ -1141,6 +1147,8 @@ else if (interceptorVal != val0) finally { unlockEntry(); unlockListenerReadLock(); + + releaseReserved(reserved); } onUpdateFinished(updateCntr0); @@ -1445,8 +1453,11 @@ else if (log.isDebugEnabled()) AtomicCacheUpdateClosure c; - if (!primary && !isNear()) - ensureFreeSpace(); + long reserved = 0; + + if (!primary && !isNear() && op == GridCacheOperation.UPDATE) + // Best-effort: a refused backup update is not safe, the primary has already applied it. + reserved = reserveEntrySpace((CacheObject)writeObj, true); lockListenerReadLock(); lockEntry(); @@ -1706,6 +1717,8 @@ else if (updateMetrics && REMOVE_NO_VAL.equals(updateRes.outcome()) finally { unlockEntry(); unlockListenerReadLock(); + + releaseReserved(reserved); } onUpdateFinished(c.updateRes.updateCounter()); @@ -2439,7 +2452,8 @@ private boolean skipInterceptor(@Nullable GridCacheVersion explicitVer) { ) throws IgniteCheckedException, GridCacheEntryRemovedException { assert !primary || !(preload || fromStore); - ensureFreeSpace(); + // A pre-created row (rebalancing) is already inserted into the free list, so there is nothing to reserve. + long reserved = row == null ? reserveEntrySpace(val, false) : 0; boolean deferred = false; boolean obsolete = false; @@ -2581,6 +2595,8 @@ else if (deletedUnlocked()) unlockEntry(); unlockListenerReadLock(); + releaseReserved(reserved); + // It is necessary to execute these callbacks outside of lock to avoid deadlocks. if (obsolete) { @@ -2677,6 +2693,9 @@ protected long nextPartitionCounter(IgniteInternalTx tx, @Nullable Long updateCn @Nullable IgniteCacheExpiryPolicy loadExpiryPlc, @Nullable ReaderArguments readerArgs ) throws IgniteCheckedException, GridCacheEntryRemovedException { + // The atomic cache reloads values while holding the entry locks, a reservation is not possible then. + long reserved = lockedByCurrentThread() ? 0 : reserveEntrySpace(val, false); + lockEntry(); try { @@ -2731,6 +2750,8 @@ protected long nextPartitionCounter(IgniteInternalTx tx, @Nullable Long updateCn } finally { unlockEntry(); + + releaseReserved(reserved); } } @@ -3557,13 +3578,51 @@ 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 val Value of the row, or {@code null} for a removal: no row is inserted, so nothing is reserved. + * @param bestEffort {@code True} if the insert can not be safely refused (see {@code FreeList#reserve}). + * @return Number of reserved pages, or {@code 0} if nothing is reserved. + * @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(@Nullable CacheObject val, boolean bestEffort) throws IgniteCheckedException { + // Deadlock alert: evicting a data page locks all entries on it. assert !lock.isHeldByCurrentThread(); - cctx.shared().database().ensureFreeSpace(cctx.dataRegion()); + IgniteCacheDatabaseSharedManager db = cctx.shared().database(); + + // Near entries keep values on heap (see GridNearCacheEntry#storeValue), so they are not reserved for. + // The row size is computed only if the reservation is engaged: it may require marshalling the value. + if (val == null || isNear() || !db.isReserveEnabled(cctx.dataRegion())) + return 0; + + return db.reserveInRegion(cctx.dataRegion(), rowSize(val), bestEffort); + } + + /** + * Releases a reservation made by {@link #reserveEntrySpace(CacheObject, boolean)}. 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) { + cctx.shared().database().releaseInRegion(cctx.dataRegion(), reserved); + } + + /** + * Computes the data row size for the given value. Used to reserve pages before the entry lock is acquired, when the + * version of the row may be not known yet: {@link CacheVersionIO#SIZE_V2} is the maximum of both version protocols, + * so the result is an upper bound. + * + * @param val Value. + * @return Estimated row size in bytes. + * @throws IgniteCheckedException If failed. + */ + private int rowSize(CacheObject val) throws IgniteCheckedException { + return CacheDataRowAdapter.rowSize(key, val, CacheVersionIO.SIZE_V2, cctx.cacheObjectContext(), + cctx.group().storeCacheIdInDataPage()); } /** diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtCacheAdapter.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtCacheAdapter.java index ec2e09483c0d1..d4980f7863123 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtCacheAdapter.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtCacheAdapter.java @@ -886,14 +886,6 @@ else if (storeEnabled) while (true) { GridCacheEntryEx entry = null; - try { - ctx.shared().database().ensureFreeSpace(ctx.dataRegion()); - } - catch (IgniteCheckedException e) { - // Wrap errors (will be unwrapped). - throw new GridClosureException(e); - } - ctx.shared().database().checkpointReadLock(); try { 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..2e65f36022b8a 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,9 @@ 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.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; @@ -135,6 +138,7 @@ import static org.apache.ignite.internal.processors.cache.GridCacheOperation.TRANSFORM; import static org.apache.ignite.internal.processors.cache.GridCacheOperation.UPDATE; import static org.apache.ignite.internal.processors.cache.GridCacheUtils.isNearEnabled; +import static org.apache.ignite.internal.processors.cache.persistence.CacheDataRowAdapter.rowSize; import static org.apache.ignite.internal.processors.dr.GridDrType.DR_BACKUP; import static org.apache.ignite.internal.processors.dr.GridDrType.DR_NONE; import static org.apache.ignite.internal.processors.dr.GridDrType.DR_PRIMARY; @@ -1806,8 +1810,10 @@ 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. @@ -1949,6 +1955,8 @@ private void updateAllAsyncInternal0( return; } finally { + ctx.shared().database().releaseInRegion(ctx.dataRegion(), batchReserved); + ctx.shared().database().checkpointReadUnlock(); } @@ -2978,6 +2986,53 @@ 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} makes room for the data pages + * of all rows of a batch with known value sizes, so their inserts are not expected to hit an out-of-memory in + * {@code FreeList.insertDataRow} (pages outside the reservation, e.g. B+tree pages, are covered by the empty pages + * pool). + *

+ * The full row size is reserved for updates too: an update inserts the new row before removing the old one, unless + * it is done in place (same row size, no SQL indexes, unchanged expire time), which can't be known before the entry + * lock. For in-place updates near the eviction threshold this keeps extra empty pages free (up to the size of the + * rows being written concurrently) instead of using them for data; it is not a correctness issue. + *

+ * Nothing is reserved for a remove: it inserts no rows. For a transform the result is unknown before the entry + * processor runs under the entry lock, so only a row without a value (key, version and row header) is reserved. + * + * @param req Update request. + * @return Reserved pages to release via {@code releaseInRegion}, or {@code 0} if reservation is not engaged. + * @throws IgniteCheckedException If failed. + */ + private long reserveBatchSpace(GridNearAtomicAbstractUpdateRequest req) throws IgniteCheckedException { + DataRegion dataRegion = ctx.dataRegion(); + + // Skip computing the row sizes (it may require marshalling) if the reservation is not engaged for the region. + if (req.operation() == GridCacheOperation.DELETE || !ctx.shared().database().isReserveEnabled(dataRegion)) + return 0; + + CacheObjectValueContext coctx = ctx.cacheObjectContext(); + + boolean storeCacheId = ctx.group().storeCacheIdInDataPage(); + + long sumRowSize = 0; + + for (int i = 0; i < req.size(); i++) { + KeyCacheObject key = req.key(i); + + // The value of a transform is unknown before the entry processor runs, a row without a value is reserved. + CacheObject val = req.operation() == GridCacheOperation.UPDATE ? req.value(i) : null; + + // 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 += rowSize(key, val, CacheVersionIO.SIZE_V2, coctx, storeCacheId); + } + + return ctx.shared().database().reserveInRegion(dataRegion, sumRowSize, false); + } + /** * 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 66059d557c9cc..960d5b1820f8f 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 @@ -28,6 +28,7 @@ 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.IncompleteCacheObject; import org.apache.ignite.internal.processors.cache.IncompleteObject; @@ -957,11 +958,32 @@ public boolean isReady() { /** {@inheritDoc} */ @Override public int size() throws IgniteCheckedException { - int len = key().valueBytesLength(null); + return rowSize(key(), value(), CacheVersionIO.size(version(), false), null, storeCacheId()); + } + + /** + * Computes the size a row with the given key and value occupies in page memory. + * + * @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 storeCacheId {@code True} if 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, + boolean storeCacheId + ) throws IgniteCheckedException { + int len = key.valueBytesLength(coctx); - len += value().valueBytesLength(null) + CacheVersionIO.size(version(), false) + 8; + len += (val != null ? val.valueBytesLength(coctx) : 0) + verSize + 8; - return len + (storeCacheId() ? 4 : 0); + return len + (storeCacheId ? 4 : 0); } /** 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..89ec2f352fae1 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 @@ -1223,23 +1223,50 @@ public void ensureFreeSpaceForInsert(DataRegion region, int dataRowSize) throws } /** - * See {@code GridCacheMapEntry#ensureFreeSpace()} + * Checks whether the size-aware page reservation is engaged for the given data region: only in-memory regions with + * enabled page eviction. Callers use it to skip computing the row size when no reservation will be made. * - * @param memPlc data region. + * @param memPlc Data region. + * @return {@code True} if {@link #reserveInRegion(DataRegion, long, boolean)} may reserve pages in the region. */ - public void ensureFreeSpace(DataRegion memPlc) throws IgniteCheckedException { - if (memPlc == null) - return; + public boolean isReserveEnabled(@Nullable DataRegion memPlc) { + return memPlc != null && !memPlc.config().isPersistenceEnabled() && + memPlc.config().getPageEvictionMode() != DataPageEvictionMode.DISABLED; + } - while (memPlc.evictionTracker().evictionRequired()) { - if (memPlc.metrics().onPageEvictionsStarted()) { - U.warn(log, "Page-based evictions started." + - " Consider increasing 'maxSize' on Data Region configuration: " + memPlc.config().getName()); - } + /** + * 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. + * + * @param memPlc Data region. + * @param rowSize Size of the row, or the total size of all rows of a batch. + * @param bestEffort {@code True} if the insert can not be safely refused (see {@link FreeList#reserve(long, boolean)}). + * @return Number of reserved pages to be released via {@link #releaseInRegion(DataRegion, long)}, or {@code 0} if + * nothing is reserved: the reservation is not engaged for this region (see {@link #isReserveEnabled(DataRegion)}) + * or a best-effort insert is admitted without a reservation. + * @throws IgniteCheckedException If failed. + */ + public long reserveInRegion(@Nullable DataRegion memPlc, long rowSize, boolean bestEffort) throws IgniteCheckedException { + if (!isReserveEnabled(memPlc)) + return 0; - memPlc.evictionTracker().evictDataPage(); + FreeList freeList = freeList(memPlc.config().getName()); + + return freeList == null ? 0 : freeList.reserve(rowSize, bestEffort); + } + + /** + * Releases a reservation made by {@link #reserveInRegion(DataRegion, long, boolean)}. + * + * @param memPlc Data region the reservation was made in. + * @param pages Number of pages to release. + */ + public void releaseInRegion(@Nullable DataRegion memPlc, long pages) { + if (memPlc != null && pages != 0) { + FreeList freeList = freeList(memPlc.config().getName()); - memPlc.metrics().updateEvictionRate(); + if (freeList != null) + freeList.releaseReserved(pages); } } @@ -1321,7 +1348,7 @@ protected PageEvictionTracker createPageEvictionTracker(DataRegionConfiguration PageMemoryNoStoreImpl pageMem0 = (PageMemoryNoStoreImpl)pageMem; if (Boolean.getBoolean("override.fair.fifo.page.eviction.tracker")) - return new FairFifoPageEvictionTracker(pageMem0, plc, cctx); + return new FairFifoPageEvictionTracker(pageMem0, cctx); switch (plc.getPageEvictionMode()) { case RANDOM_LRU: 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..4332a1f0bbedb 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 @@ -20,7 +20,6 @@ import java.util.LinkedList; import org.apache.ignite.IgniteCheckedException; import org.apache.ignite.IgniteException; -import org.apache.ignite.configuration.DataRegionConfiguration; import org.apache.ignite.internal.pagemem.PageIdUtils; import org.apache.ignite.internal.pagemem.impl.PageMemoryNoStoreImpl; import org.apache.ignite.internal.processors.cache.GridCacheSharedContext; @@ -34,14 +33,10 @@ public class FairFifoPageEvictionTracker extends PageAbstractEvictionTracker { /** * @param pageMem Page memory. - * @param plcCfg Data region configuration. * @param sharedCtx Shared context. */ - public FairFifoPageEvictionTracker( - PageMemoryNoStoreImpl pageMem, - DataRegionConfiguration plcCfg, - GridCacheSharedContext sharedCtx) { - super(pageMem, plcCfg, sharedCtx); + public FairFifoPageEvictionTracker(PageMemoryNoStoreImpl pageMem, GridCacheSharedContext sharedCtx) { + super(pageMem, sharedCtx); } /** {@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..d27d9e1c9eb85 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 @@ -47,11 +47,6 @@ public class NoOpPageEvictionTracker implements PageEvictionTracker { // No-op. } - /** {@inheritDoc} */ - @Override public boolean evictionRequired() { - return false; - } - /** {@inheritDoc} */ @Override public void trackFragmentPage(long pageId, long prevPageId, boolean isHeadPage) { // No-op. diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/evict/PageAbstractEvictionTracker.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/evict/PageAbstractEvictionTracker.java index 2330c0942662d..082380c3ca597 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/evict/PageAbstractEvictionTracker.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/evict/PageAbstractEvictionTracker.java @@ -18,14 +18,12 @@ import java.util.List; import org.apache.ignite.IgniteCheckedException; -import org.apache.ignite.configuration.DataRegionConfiguration; import org.apache.ignite.internal.pagemem.PageIdUtils; import org.apache.ignite.internal.pagemem.impl.PageMemoryNoStoreImpl; import org.apache.ignite.internal.processors.cache.GridCacheContext; import org.apache.ignite.internal.processors.cache.GridCacheEntryEx; import org.apache.ignite.internal.processors.cache.GridCacheSharedContext; import org.apache.ignite.internal.processors.cache.persistence.CacheDataRowAdapter; -import org.apache.ignite.internal.processors.cache.persistence.freelist.AbstractFreeList; import org.apache.ignite.internal.processors.cache.persistence.tree.io.DataPageIO; import org.apache.ignite.internal.processors.cache.persistence.tree.io.PageIO; import org.apache.ignite.internal.processors.cache.version.GridCacheVersionManager; @@ -53,40 +51,21 @@ public abstract class PageAbstractEvictionTracker implements PageEvictionTracker /** Shared context. */ private final GridCacheSharedContext sharedCtx; - /** Data region configuration. */ - private final DataRegionConfiguration regCfg; - /** * @param pageMem Page memory. - * @param plcCfg Data region configuration. * @param sharedCtx Shared context. */ - PageAbstractEvictionTracker( - PageMemoryNoStoreImpl pageMem, - DataRegionConfiguration plcCfg, - GridCacheSharedContext sharedCtx - ) { + PageAbstractEvictionTracker(PageMemoryNoStoreImpl pageMem, GridCacheSharedContext sharedCtx) { this.pageMem = pageMem; this.sharedCtx = sharedCtx; - regCfg = plcCfg; - trackingSize = pageMem.totalPages(); baseCompactTs = (U.currentTimeMillis() - DAY) >> COMPACT_TS_SHIFT; // We subtract day to avoid fail in case of daylight shift or timezone change. } - /** {@inheritDoc} */ - @Override public boolean evictionRequired() { - AbstractFreeList freeList = (AbstractFreeList)sharedCtx.database().freeList(regCfg.getName()); - - double pagesThreshold = regCfg.getEvictionThreshold() * regCfg.getMaxSize() / pageMem.systemPageSize(); - - return pageMem.loadedPages() > pagesThreshold && freeList.emptyDataPages() < regCfg.getEmptyPagesPoolSize(); - } - /** * @param pageIdx Page index. * @return true if at least one data row has been evicted 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..cb3d7df45fe03 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 @@ -31,13 +31,6 @@ public interface PageEvictionTracker extends LifecycleAware { */ public void touchPage(long pageId); - /** - * Check if page eviction is required according to the configured policy. - * - * @return {@code True} if eviction required. - */ - public boolean evictionRequired(); - /** * Evicts one data page. * In most cases, all entries will be removed from the 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..0f337f01ec7c5 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 @@ -58,7 +58,7 @@ public Random2LruPageEvictionTracker( DataRegionConfiguration plcCfg, GridCacheSharedContext sharedCtx ) { - super(pageMem, plcCfg, sharedCtx); + super(pageMem, sharedCtx); DataStorageConfiguration memCfg = sharedCtx.kernalContext().config().getDataStorageConfiguration(); @@ -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; + } + + continue; + } } } 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..78b49278de5d0 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 @@ -60,7 +60,7 @@ public RandomLruPageEvictionTracker( DataRegionConfiguration plcCfg, GridCacheSharedContext sharedCtx ) { - super((PageMemoryNoStoreImpl)pageMem, plcCfg, sharedCtx); + super((PageMemoryNoStoreImpl)pageMem, sharedCtx); DataStorageConfiguration memCfg = sharedCtx.kernalContext().config().getDataStorageConfiguration(); @@ -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; + } + + continue; + } } } 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 48055a157a270..3212628bd86ae 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 @@ -21,9 +21,13 @@ import java.util.Collection; 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.IgniteSystemProperties; +import org.apache.ignite.configuration.DataRegionConfiguration; 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; @@ -50,10 +54,13 @@ import org.apache.ignite.internal.processors.cache.persistence.tree.util.PageHandler; import org.apache.ignite.internal.util.GridCursorIteratorWrapper; import org.apache.ignite.internal.util.lang.GridCursor; +import org.apache.ignite.internal.util.typedef.internal.LT; import org.apache.ignite.internal.util.typedef.internal.U; import org.jetbrains.annotations.Nullable; +import static org.apache.ignite.IgniteSystemProperties.IGNITE_PAGE_EVICTION_NO_PROGRESS_TIMEOUT_MILLIS; import static org.apache.ignite.internal.pagemem.PageIdAllocator.FLAG_DATA; +import static org.apache.ignite.internal.pagemem.impl.PageMemoryNoStoreImpl.SEG_CNT; /** */ @@ -76,6 +83,30 @@ 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; + + /** + * Default maximum time (milliseconds) the size-aware eviction guard is willing to wait without the region's free + * page count increasing (the deficit shrinking) before giving up (see {@link #reserve(long, boolean)}). Configurable via + * {@link IgniteSystemProperties#IGNITE_PAGE_EVICTION_NO_PROGRESS_TIMEOUT_MILLIS}. + */ + public static final long DFLT_EVICTION_NO_PROGRESS_TIMEOUT_MILLIS = 5_000L; + + /** + * Maximum time (milliseconds) the size-aware eviction guard is willing to wait without the region's free page + * count increasing (the deficit shrinking) before giving up. This is a liveness guard: a + * {@code evictDataPage()} call can free no page (for example when the sampled rows are not actually reclaimable), + * so progress is measured by a real deficit decrease. + * An eviction that is merely slow but still frees pages keeps shrinking the deficit and thus keeps resetting the + * timer; a stall that frees nothing for the whole window gives up instead of spinning. + */ + private static final long EVICTION_NO_PROGRESS_TIMEOUT_MILLIS = IgniteSystemProperties.getLong( + IGNITE_PAGE_EVICTION_NO_PROGRESS_TIMEOUT_MILLIS, DFLT_EVICTION_NO_PROGRESS_TIMEOUT_MILLIS); + /** * 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. @@ -100,6 +131,26 @@ 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; + + /** + * Pages reserved by in-flight inserts (see {@link #reserve(long, boolean)}): only the rows' own pages, the empty pages pool + * is a separate region-wide margin. A reservation is released as a whole once its row (or batch of rows) is written. + */ + private final AtomicLong reservedPages = new AtomicLong(); + + /** + * Whether the eviction is stalled: set by a best-effort reservation that gave up because the eviction made no + * progress within {@link #EVICTION_NO_PROGRESS_TIMEOUT_MILLIS}, cleared by any reservation that observes progress or + * reserves pages. While set, best-effort reservations (including the already waiting ones) give up after an + * eviction attempt without progress, so a series of inserts does not wait for the timeout one by one. + */ + private volatile boolean stalled; + /** Page list cache limit. */ private final AtomicLong pageListCacheLimit; @@ -367,7 +418,7 @@ private final class WriteRowsHandler extends PageHandler, Integer> AbstractDataPageIO io = (AbstractDataPageIO)iox; // Fill the page up to the end. - while (written != COMPLETE || (!evictionTracker.evictionRequired() && cur.next())) { + while (written != COMPLETE || cur.next()) { T row = cur.get(); if (written == COMPLETE) { @@ -493,6 +544,10 @@ public AbstractFreeList( rmvRow = new RemoveRowHandler(cacheGrpId == 0); this.evictionTracker = dataRegion.evictionTracker(); + + reserveEnabled = ctx.cache().context().database().isReserveEnabled(dataRegion); + + this.dataRegion = dataRegion; this.reuseList = reuseList == null ? this : reuseList; int pageSize = pageMem.pageSize(); @@ -649,6 +704,9 @@ private long allocateDataPage(int part) throws IgniteCheckedException { * Rows are sequentially added to the page as long as there is enough free space on it. If the row is large then * those fragments that occupy the whole memory page are written to other pages, and the remainder is added to the * current one. + *

+ * The space of the whole batch is reserved upfront, before any page lock is taken: the eviction is not allowed + * under a page write lock, and a batch (e.g. of rebalancing) is small, so a pessimistic reservation is cheap. * * @param rows Rows. * @param statHolder Statistics holder to track IO operations. @@ -657,39 +715,40 @@ private long allocateDataPage(int part) throws IgniteCheckedException { @Override public void insertDataRows(Collection rows, IoStatisticsHolder statHolder) throws IgniteCheckedException { try { - GridCursor cur = new GridCursorIteratorWrapper<>(rows.iterator()); - - int written = COMPLETE; + // Best-effort: rows of a batch (e.g. of rebalancing) can not be safely refused. + long reserved = reserveEnabled ? reserve(batchSize(rows), true) : 0; - while (written != COMPLETE || cur.next()) { - T row = cur.get(); + try { + GridCursor cur = new GridCursorIteratorWrapper<>(rows.iterator()); - // If eviction is required - free up memory before locking the next page. - while (evictionTracker.evictionRequired()) { - evictionTracker.evictDataPage(); + int written = COMPLETE; - memMetrics.updateEvictionRate(); - } + while (written != COMPLETE || cur.next()) { + T row = cur.get(); - if (written == COMPLETE) { - written = writeWholePages(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(); + } - written = write(pageId, writeRowsHnd, initIo, cur, written, FAIL_I, statHolder); + written = write(pageId, writeRowsHnd, initIo, cur, written, FAIL_I, statHolder); - assert written != FAIL_I; // We can't fail here. + assert written != FAIL_I; // We can't fail here. + } + } + finally { + releaseReserved(reserved); } } catch (RuntimeException e) { @@ -697,6 +756,23 @@ private long allocateDataPage(int part) throws IgniteCheckedException { } } + /** + * Computes a pessimistic size of a batch of rows: every row is rounded up to whole data pages, as if no rows + * shared a page. + * + * @param rows Rows. + * @return Size of the batch in bytes. + * @throws IgniteCheckedException If failed. + */ + private long batchSize(Collection rows) throws IgniteCheckedException { + long size = 0; + + for (T row : rows) + size += requiredPages(row.size()) * MIN_SIZE_FOR_DATA_PAGE; + + return size; + } + /** * 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. @@ -1134,8 +1210,184 @@ public int emptyDataPages() { } } + /** {@inheritDoc} */ + @Override public long reserve(long rowSize, boolean bestEffort) throws IgniteOutOfMemoryException, IgniteCheckedException { + // Nothing to insert (e.g. an empty batch): no eviction is needed. + if (!reserveEnabled || rowSize <= 0) + return 0; + + long claimPages = requiredPages(rowSize); + + if (!fitsRegion(claimPages)) { + if (bestEffort) + return 0; // Let the page memory decide. + + throw outOfMemory(); + } + + long limitPages = limitPages(claimPages); + + long backoffNanos = EVICTION_BACKOFF_START_NANOS; + + long lastProgressMillis = U.currentTimeMillis(); + + // Deficit before the last eviction attempt, 0 until the first one. + long prevDeficit = 0; + + while (true) { + long reserved = reservedPages.get(); + long deficit = deficit(claimPages, limitPages, reserved); + + if (deficit <= 0) { + if (reservedPages.compareAndSet(reserved, reserved + claimPages)) { + clearStalled(); + + return claimPages; + } + + continue; + } + + if (prevDeficit > 0) { + if (deficit < prevDeficit) { + clearStalled(); + + lastProgressMillis = U.currentTimeMillis(); + + backoffNanos = EVICTION_BACKOFF_START_NANOS; + } + else if (bestEffort && stalled) + return 0; + else if (U.currentTimeMillis() - lastProgressMillis > EVICTION_NO_PROGRESS_TIMEOUT_MILLIS) { + if (!bestEffort) + throw outOfMemory(); + + stalled = true; + + LT.warn(log, "Page eviction made no progress, the row is inserted without a page reservation " + + "[region=" + dataRegion.config().getName() + ", timeout=" + EVICTION_NO_PROGRESS_TIMEOUT_MILLIS + "ms]"); + + return 0; + } + else { + LockSupport.parkNanos(backoffNanos); + + backoffNanos = Math.min(backoffNanos << 1, EVICTION_BACKOFF_MAX_NANOS); + } + } + + prevDeficit = deficit; + + if (memMetrics.onPageEvictionsStarted()) { + U.warn(log, "Page-based evictions started." + + " Consider increasing 'maxSize' on Data Region configuration: " + dataRegion.config().getName()); + } + + evictionTracker.evictDataPage(); + + memMetrics.updateEvictionRate(); + } + } + + /** + * Clears the {@link #stalled} flag once a reservation observes progress or reserves pages. The flag is read first, + * so the common path does not write to the shared field. + */ + private void clearStalled() { + if (stalled) + stalled = false; + } + + /** {@inheritDoc} */ + @Override public void releaseReserved(long pages) { + assert pages >= 0; + + if (reserveEnabled && pages != 0) + reservedPages.addAndGet(-pages); + } + /** {@inheritDoc} */ @Override public String toString() { return "FreeList [name=" + name() + ']'; } + + /** + * Computes the number of pages to claim for a row. The empty pages pool is not a part of the claim: it is a single + * region-wide margin that every reservation must leave free (see {@link #deficit}), so concurrent writers do not + * multiply it. + * + * @param rowSize Row size in bytes. + * @return Number of pages required for the row. + */ + private long requiredPages(long rowSize) { + return (rowSize + MIN_SIZE_FOR_DATA_PAGE - 1) / MIN_SIZE_FOR_DATA_PAGE; + } + + /** + * @param requiredPages Pages required for a row. + * @return {@code True} if the row together with the empty pages pool fits into the whole data region. + */ + private boolean fitsRegion(long requiredPages) { + return requiredPages + dataRegion.config().getEmptyPagesPoolSize() <= totalPages(); + } + + /** @return Total number of pages in the data region. */ + private long totalPages() { + return dataRegion.config().getMaxSize() / pageMem.systemPageSize(); + } + + /** + * Computes the number of loaded pages up to which the free space of the region is counted for a row. Normally it + * is the eviction threshold ({@link DataRegionConfiguration#getEvictionThreshold()}), so every insert path starts + * evicting at the same fill level. A row that does not fit below the threshold together with the empty pages pool + * may use the space above it, up to the whole region. + * + * @param requiredPages Pages required for the row. + * @return Limit of loaded pages for the row. + */ + private long limitPages(long requiredPages) { + long totalPages = totalPages(); + + long hardPages = totalPages > SEG_CNT ? totalPages - SEG_CNT : 0; + + long thresholdPages = Math.min((long)(dataRegion.config().getEvictionThreshold() * totalPages), hardPages); + + return requiredPages + dataRegion.config().getEmptyPagesPoolSize() > thresholdPages ? hardPages : thresholdPages; + } + + /** + * Computes how many pages are missing to admit a claim: the claim, the empty pages pool and the reservations of + * other in-flight inserts must fit into the free pages of the region (empty pages in this free list plus the + * headroom up to the limit). + * + * @param claimPages Pages to claim. + * @param limitPages Limit of loaded pages (see {@link #limitPages(long)}). + * @param reserved Pages reserved by other in-flight inserts. + * @return Number of missing pages, a non-positive value if the claim can be admitted. + */ + private long deficit(long claimPages, long limitPages, long reserved) { + long freePages = emptyDataPages() + Math.max(0, limitPages - pageMem.loadedPages()); + + return claimPages + dataRegion.config().getEmptyPagesPoolSize() + reserved - freePages; + } + + /** + * Creates an out-of-memory exception for a refused reservation. It is thrown only for inserts that can be refused + * (see {@link #reserve(long, boolean)}) and is not reported to the failure handler: nothing has been changed yet. + * + * @return New {@link IgniteOutOfMemoryException} for the data region of this free list. + */ + private IgniteOutOfMemoryException outOfMemory() { + DataRegionConfiguration cfg = dataRegion.config(); + + return 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" + ); + } } 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 3db46ed6a300e..daa2c2abe8f0c 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,6 +20,7 @@ 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; @@ -63,4 +64,30 @@ public void updateDataRow( * @param log Logger. */ public void dumpStatistics(IgniteLogger log); + + /** + * Reserves pages for a row (or a batch of rows) of the given size before it is inserted. Runs blocking eviction + * when necessary (the caller must not hold any entry or page locks). + * + * @param rowSize Size of the row (or the total size of a batch of rows) to be inserted. + * @param bestEffort {@code True} if the insert can not be safely refused (e.g. a backup update or a transaction + * commit): if the row does not fit into the region or the eviction makes no progress, the row is admitted + * without a reservation ({@code 0} is returned) and the page memory remains the hard limit. {@code False} if + * the insert can be refused: an out-of-memory is thrown instead. + * @return Number of reserved pages to be released via {@link #releaseReserved(long)} once the row insert + * completes, or {@code 0} if nothing is reserved (persistent data region, disabled page eviction or a best-effort + * admission without a reservation). + * @throws IgniteOutOfMemoryException If the insert is not best-effort and the row cannot fit into the region or + * eviction makes no progress. The exception is not reported to the failure handler: nothing has been changed + * yet, so the caller decides how to handle the refusal. + * @throws IgniteCheckedException If failed to evict data pages. + */ + public long reserve(long rowSize, boolean bestEffort) throws IgniteOutOfMemoryException, IgniteCheckedException; + + /** + * Releases a reservation made by {@link #reserve(long, boolean)}. + * + * @param pages Number of pages to release (must not be negative). + */ + public void releaseReserved(long pages); } 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..2bd40e87b37d0 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,19 +16,31 @@ */ package org.apache.ignite.internal.processors.cache.eviction.paged; +import javax.cache.Cache; +import javax.cache.configuration.Factory; +import javax.cache.expiry.Duration; +import javax.cache.integration.CacheLoaderException; +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; 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.DataRegionConfiguration; 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; + /** * */ @@ -46,7 +58,7 @@ public class PageEvictionAbstractTest extends GridCommonAbstractTest { private static final int EMPTY_PAGES_POOL_SIZE = 100; /** Eviction threshold. */ - private static final double EVICTION_THRESHOLD = 0.9; + static final double EVICTION_THRESHOLD = 0.9; /** Default policy name. */ private static final String DEFAULT_POLICY_NAME = "dfltPlc"; @@ -130,4 +142,111 @@ protected CacheConfiguration cacheConfig( return cacheConfiguration; } + + /** + * @param ignite Node. + * @param cacheName Cache name. + * @return Atomic cache without expiry. + */ + protected static IgniteCache createCache(IgniteEx ignite, String cacheName) { + return createCache(ignite, cacheName, 0, false); + } + + /** + * @param ignite Node. + * @param cacheName Cache name. + * @param ttl TTL in milliseconds ({@code 0} for no expiry). + * @param transactional {@code True} for a transactional cache, {@code false} for an atomic one. + * @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 The number of loaded pages in the default data region. */ + protected static long loadedPages(IgniteEx ignite) throws IgniteCheckedException { + return defaultRegion(ignite).pageMemory().loadedPages(); + } + + /** @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 {@code totalPages * fraction} page-sized entries (one data page each), i.e. loads about + * {@code fraction} of 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; + } + + /** */ + protected static class TestStoreFactory implements Factory { + /** {@inheritDoc} */ + @Override public TestCacheStore create() { + return new TestCacheStore(); + } + } + + /** */ + protected 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/PageEvictionConcurrentWritesAbstractTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/eviction/paged/PageEvictionConcurrentWritesAbstractTest.java new file mode 100644 index 0000000000000..0f3fffecc2a56 --- /dev/null +++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/eviction/paged/PageEvictionConcurrentWritesAbstractTest.java @@ -0,0 +1,260 @@ +/* + * 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 { + super.afterTest(); + + stopAllGrids(); + } + + /** */ + @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); + + 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(); + + 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)); + + 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); + } + + /** */ + @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); + } + + /** */ + @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)); + + int probeKey = keyGen.incrementAndGet(); + byte[] probeVal = valueFor(probeKey, valSize); + + cache.put(probeKey, probeVal); + + byte[] read = cache.get(probeKey); + + assertNotNull(read); + assertTrue(Arrays.equals(probeVal, read)); + } + + /** */ + 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; + } + + /** @return Deterministic per-key value content. */ + private 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; + } +} diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/eviction/paged/PageEvictionDisabledEvictionSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/eviction/paged/PageEvictionDisabledEvictionSelfTest.java new file mode 100644 index 0000000000000..784f38dae8132 --- /dev/null +++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/eviction/paged/PageEvictionDisabledEvictionSelfTest.java @@ -0,0 +1,204 @@ +/* + * 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.HashSet; +import java.util.Set; +import java.util.concurrent.atomic.AtomicInteger; +import org.apache.ignite.Ignite; +import org.apache.ignite.IgniteCache; +import org.apache.ignite.cache.CacheAtomicityMode; +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.failure.FailureContext; +import org.apache.ignite.failure.FailureHandler; +import org.apache.ignite.internal.IgniteEx; +import org.apache.ignite.internal.mem.IgniteOutOfMemoryException; +import org.apache.ignite.internal.processors.cache.persistence.DataRegion; +import org.apache.ignite.internal.processors.cache.persistence.freelist.FreeList; +import org.apache.ignite.internal.util.typedef.X; +import org.apache.ignite.testframework.GridTestUtils; +import org.junit.Test; + +/** + * Checks that the size-aware page reservation does not change the behavior of an in-memory data region with disabled + * page eviction: inserts (including read-through loads) are still rejected by the out-of-memory check at the threshold, + * while removals still succeed and do not trigger the failure handler. + */ +public class PageEvictionDisabledEvictionSelfTest extends PageEvictionAbstractTest { + /** Number of out-of-memory failures reported to the failure handler. */ + private static final AtomicInteger OOM_FAILURES = new AtomicInteger(); + + /** {@inheritDoc} */ + @Override protected IgniteConfiguration getConfiguration(String gridName) throws Exception { + return setEvictionMode(DataPageEvictionMode.DISABLED, super.getConfiguration(gridName)); + } + + /** {@inheritDoc} */ + @Override protected FailureHandler getFailureHandler(String igniteInstanceName) { + return new OomCountingFailureHandler(); + } + + /** {@inheritDoc} */ + @Override protected void beforeTest() throws Exception { + super.beforeTest(); + + OOM_FAILURES.set(0); + } + + /** {@inheritDoc} */ + @Override protected void afterTest() throws Exception { + super.afterTest(); + + stopAllGrids(); + } + + /** */ + @Test(timeout = 180_000) + public void testRemoveAtOomThreshold() throws Exception { + IgniteEx ignite = startGrids(2); + + IgniteCache cache = ignite.createCache(new CacheConfiguration(DEFAULT_CACHE_NAME) + .setAffinity(new RendezvousAffinityFunction(false, 32)) + .setAtomicityMode(CacheAtomicityMode.ATOMIC) + .setBackups(1) + .setWriteSynchronizationMode(CacheWriteSynchronizationMode.FULL_SYNC)); + + awaitPartitionMapExchange(); + + int filled = fillUntilOom(cache, pageSize(ignite)); + + OOM_FAILURES.set(0); + + int half = filled / 2; + + // Primary single-key removes (and backup deletes on the other node). + for (int i = 0; i < half; i++) + cache.remove(i); + + // Primary batch removes. The key that failed with OOM is included: it may be stored on one of the nodes. + Set keys = new HashSet<>(); + + for (int i = half; i <= filled; i++) + keys.add(i); + + cache.removeAll(keys); + + assertEquals(0, cache.size()); + assertEquals("Removals must not report out-of-memory to the failure handler", 0, OOM_FAILURES.get()); + } + + /** + * The size-aware reservation is not engaged for a region with disabled page eviction: it neither reserves nor + * performs an early out-of-memory check, even for a row larger than the region. The out-of-memory check of an + * actual insert ({@code ensureFreeSpaceForInsert}) is kept as is. + */ + @Test + public void testReservationNotEngaged() throws Exception { + IgniteEx ignite = startGrid(0); + + createCache(ignite, DEFAULT_CACHE_NAME).put(0, 0); + + DataRegion region = defaultRegion(ignite); + + FreeList freeList = ignite.context().cache().context().database().freeList(region.config().getName()); + + int hugeRow = (int)Math.min(Integer.MAX_VALUE, regionMaxSize(ignite) * 2); + + assertEquals(0, ignite.context().cache().context().database().reserveInRegion(region, hugeRow, false)); + assertEquals(0, freeList.reserve(hugeRow, false)); + assertEquals(0, OOM_FAILURES.get()); + + GridTestUtils.assertThrows(log, () -> { + ignite.context().cache().context().database().ensureFreeSpaceForInsert(region, hugeRow); + + return null; + }, IgniteOutOfMemoryException.class, null); + } + + /** Read-through load into a region at the out-of-memory threshold is rejected by the insert check. */ + @Test(timeout = 120_000) + public void testReadThroughGetOomAtThreshold() throws Exception { + IgniteEx ignite = startGrid(0); + + IgniteCache cache = ignite.getOrCreateCache(readThroughCacheConfig()); + + int filled = fillUntilOom(cache, pageSize(ignite)); + + GridTestUtils.assertThrowsWithCause(() -> cache.get(filled + 1), IgniteOutOfMemoryException.class); + } + + /** Read-through load below the threshold succeeds. */ + @Test + public void testReadThroughGetBelowThresholdSucceeds() throws Exception { + byte[] val = startGrid(0).getOrCreateCache(readThroughCacheConfig()).get(1); + + assertNotNull("Read-through load must succeed below threshold", val); + assertEquals(PAGE_SIZE - 200, val.length); + } + + /** @return Atomic read-through cache configuration. */ + private static CacheConfiguration readThroughCacheConfig() { + return new CacheConfiguration(DEFAULT_CACHE_NAME) + .setAffinity(new RendezvousAffinityFunction(false, 32)) + .setAtomicityMode(CacheAtomicityMode.ATOMIC) + .setReadThrough(true) + .setCacheStoreFactory(new TestStoreFactory()); + } + + /** + * Puts page-sized entries until the insert is rejected with an out-of-memory. + * + * @param cache Cache. + * @param pageSize Page size. + * @return Number of entries written before the out-of-memory. + */ + private static int fillUntilOom(IgniteCache cache, int pageSize) { + byte[] val = new byte[pageSize - 200]; + + for (int i = 0; i < 100_000; i++) { + try { + cache.put(i, val); + } + catch (Exception e) { + assertTrue("Expected IgniteOutOfMemoryException while filling the region, got: " + e, + X.hasCause(e, IgniteOutOfMemoryException.class)); + + assertTrue(i > 0); + + return i; + } + } + + throw new AssertionError("Region was not filled up to the out-of-memory threshold"); + } + + /** Counts out-of-memory failures and ignores them (does not stop the node). */ + private static class OomCountingFailureHandler implements FailureHandler { + /** {@inheritDoc} */ + @Override public boolean onFailure(Ignite ignite, FailureContext failureCtx) { + if (X.hasCause(failureCtx.error(), IgniteOutOfMemoryException.class)) + OOM_FAILURES.incrementAndGet(); + + return false; + } + } +} diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/eviction/paged/PageEvictionReservedPagesSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/eviction/paged/PageEvictionReservedPagesSelfTest.java new file mode 100644 index 0000000000000..b8a6dccc46105 --- /dev/null +++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/eviction/paged/PageEvictionReservedPagesSelfTest.java @@ -0,0 +1,302 @@ +/* + * 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.HashMap; +import java.util.Map; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicLong; +import org.apache.ignite.Ignite; +import org.apache.ignite.IgniteCache; +import org.apache.ignite.IgniteCheckedException; +import org.apache.ignite.Ignition; +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.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.IgniteCacheOffheapManager.CacheDataStore; +import org.apache.ignite.internal.processors.cache.KeyCacheObject; +import org.apache.ignite.internal.processors.cache.persistence.CacheDataRow; +import org.apache.ignite.internal.processors.cache.persistence.freelist.FreeList; +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 the page reservation accounting of an in-memory free list with enabled page eviction: the reservation is fully + * released after batch inserts (whole-page fragments, fill-up-handler rows, whole-page-aligned rows, concurrent batch + * inserts), concurrent puts and putAll on a backup node, and read-through loads; the empty pages pool is shared by + * all concurrent reservations instead of being added to each of them. + */ +public class PageEvictionReservedPagesSelfTest extends PageEvictionAbstractTest { + /** Empty pages pool size: close to the allowed maximum of 10% of the region. */ + private static final int LARGE_EMPTY_PAGES_POOL_SIZE = 900; + + /** {@inheritDoc} */ + @Override protected IgniteConfiguration getConfiguration(String gridName) throws Exception { + IgniteConfiguration cfg = setEvictionMode(DataPageEvictionMode.RANDOM_LRU, super.getConfiguration(gridName)); + + cfg.getDataStorageConfiguration().getDefaultDataRegionConfiguration().setEmptyPagesPoolSize(LARGE_EMPTY_PAGES_POOL_SIZE); + + return cfg; + } + + /** {@inheritDoc} */ + @Override protected void afterTest() throws Exception { + super.afterTest(); + + stopAllGrids(); + } + + /** */ + @Test + public void testReservedPagesReleasedAfterBatchInserts() throws Exception { + IgniteEx ignite = startGrid(0); + + createCache(ignite, DEFAULT_CACHE_NAME).put(0, 0); + + FreeList freeList = freeList(ignite); + + // Phase 1: small rows that fit into a single page (fill-up path, no whole-page fragments). + insertBatchesAndCheckReservedPages(ignite, freeList, 10); + + // Phase 2: large rows that span multiple pages (whole-page fragments + fill-up remainder). + insertBatchesAndCheckReservedPages(ignite, freeList, 10_000); + + // Phase 3: rows sized to an exact multiple of the whole-page fragment size (whole-page fragments only). + int minPageSize = getFieldValue(freeList, "MIN_SIZE_FOR_DATA_PAGE"); + + int fixedSize = row(cacheContext(ignite), 0, 0).size(); + + insertBatchesAndCheckReservedPages(ignite, freeList, 2 * minPageSize - fixedSize); + } + + /** */ + @Test + public void testReservedPagesReleasedAfterConcurrentBatchInserts() throws Exception { + IgniteEx ignite = startGrid(0); + + createCache(ignite, DEFAULT_CACHE_NAME).put(0, 0); + + FreeList freeList = freeList(ignite); + + GridCacheContext cctx = cacheContext(ignite); + + int rowsPerThread = 200; + + 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"); + + assertAllReleased(10_000); + } + + /** */ + @Test + public void testReservedPagesReleasedAfterConcurrentPuts() throws Exception { + IgniteCache cache = startGridsWithBackupCache(); + + AtomicInteger keyGen = new AtomicInteger(); + + GridTestUtils.runMultiThreaded(() -> { + for (int i = 0; i < 300; i++) + cache.put(keyGen.incrementAndGet(), new byte[500]); + }, 8, "concurrent-put"); + + assertAllReleased(10_000); + } + + /** */ + @Test + public void testReservedPagesReleasedAfterConcurrentPutAll() throws Exception { + IgniteCache cache = startGridsWithBackupCache(); + + AtomicInteger keyGen = new AtomicInteger(); + + GridTestUtils.runMultiThreaded(() -> { + for (int b = 0; b < 10; b++) { + Map batch = new HashMap<>(); + + for (int i = 0; i < 100; i++) + batch.put(keyGen.incrementAndGet(), new byte[300]); + + cache.putAll(batch); + } + }, 4, "concurrent-putAll"); + + assertAllReleased(10_000); + } + + /** Read-through loads reserve pages before the entry lock and must release them, also on the backup. */ + @Test + public void testReservedPagesReleasedAfterReadThrough() throws Exception { + startGrids(2); + + IgniteCache cache = grid(0).getOrCreateCache(new CacheConfiguration(DEFAULT_CACHE_NAME) + .setAffinity(new RendezvousAffinityFunction(false, 32)) + .setWriteSynchronizationMode(CacheWriteSynchronizationMode.FULL_SYNC) + .setBackups(1) + .setReadThrough(true) + .setCacheStoreFactory(new TestStoreFactory())); + + awaitPartitionMapExchange(); + + for (int i = 1; i <= 15_000; i++) + cache.get(i); + + assertTrue("Eviction must have started during read-through loads", isEvictionsStarted(grid(0))); + + assertEquals(PAGE_SIZE - 200, cache.get(1).length); + + assertAllReleased(10_000); + } + + /** + * Holds reservations of single-page rows of many writers at once in an empty region: the sum of per-writer pools + * would exceed the region (and there is nothing to evict), but a single shared pool leaves room for all of them. + */ + @Test + public void testSimultaneousReservationsShareThePool() throws Exception { + IgniteEx ignite = startGrid(0); + + createCache(ignite, DEFAULT_CACHE_NAME).put(0, 0); + + int writers = 16; + + assertTrue(writers * LARGE_EMPTY_PAGES_POOL_SIZE > totalPages(ignite)); + + FreeList freeList = freeList(ignite); + + long[] reserved = new long[writers]; + + try { + for (int i = 0; i < writers; i++) { + reserved[i] = freeList.reserve(pageSize(ignite) / 2, false); + + assertEquals(1, reserved[i]); + } + } + finally { + for (long pages : reserved) + freeList.releaseReserved(pages); + } + + assertEquals(0, reservedPages(freeList)); + } + + /** @return Cache with a backup on two started nodes. */ + private IgniteCache startGridsWithBackupCache() 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(); + + return cache; + } + + /** + * 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 timeoutMs Wait timeout in milliseconds. + * @throws Exception If the reservation does not drain in time. + */ + private static void assertAllReleased(long timeoutMs) throws Exception { + for (Ignite ignite : Ignition.allGrids()) { + for (CacheDataStore store : cacheContext((IgniteEx)ignite).offheap().cacheDataStores()) { + FreeList freeList = (FreeList)store.rowStore().freeList(); + + assertTrue(GridTestUtils.waitForCondition(() -> reservedPages(freeList) == 0, timeoutMs)); + } + } + } + + /** */ + private static long reservedPages(FreeList freeList) { + return ((AtomicLong)getFieldValue(freeList, "reservedPages")).get(); + } + + /** */ + private static FreeList freeList(IgniteEx ignite) { + return (FreeList)cacheContext(ignite).offheap().cacheDataStores().iterator().next().rowStore().freeList(); + } + + /** */ + private static void insertBatchesAndCheckReservedPages( + IgniteEx ignite, + FreeList freeList, + int payloadSize + ) throws IgniteCheckedException { + int batchSize = 200; + + 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(); + } + + /** @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(), false); + } +} 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..03b7a6d315728 --- /dev/null +++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/eviction/paged/PageEvictionRowSizeBinaryObjectTest.java @@ -0,0 +1,100 @@ +/* + * 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; + +/** + * {@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 { + super.afterTest(); + + stopAllGrids(); + } + + /** */ + @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)); + + 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)); + } + + /** */ + @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..bb2e63f4493b6 --- /dev/null +++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/eviction/paged/PageEvictionSizeAwareAbstractTest.java @@ -0,0 +1,460 @@ +/* + * 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.Arrays; +import java.util.Collection; +import java.util.HashMap; +import java.util.Map; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicLong; +import java.util.concurrent.locks.Lock; +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.configuration.DataRegionConfiguration; +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.internal.processors.cache.persistence.DataRegion; +import org.apache.ignite.internal.processors.cache.persistence.IgniteCacheDatabaseSharedManager; +import org.apache.ignite.internal.processors.cache.persistence.freelist.AbstractFreeList; +import org.apache.ignite.internal.processors.cache.persistence.freelist.FreeList; +import org.apache.ignite.internal.util.typedef.internal.U; +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 { + /** Name of the rebalance supplier node, which has a twice larger data region. */ + private static final String SUPPLIER = "supplier"; + + /** {@inheritDoc} */ + @Override protected IgniteConfiguration getConfiguration(String gridName) throws Exception { + IgniteConfiguration cfg = super.getConfiguration(gridName); + + if (SUPPLIER.equals(gridName)) { + DataRegionConfiguration regCfg = cfg.getDataStorageConfiguration().getDefaultDataRegionConfiguration(); + + regCfg.setMaxSize(regCfg.getMaxSize() * 2); + } + + return cfg; + } + + /** {@inheritDoc} */ + @Override protected void afterTest() throws Exception { + super.afterTest(); + + stopAllGrids(); + } + + /** + * Rebalancing inserts the received rows in batches reserved as a whole: the demander, which receives more data than + * fits below its threshold, evicts at the threshold and releases the reservations. + */ + @Test(timeout = 300_000) + public void testRebalanceHonorsThreshold() throws Exception { + IgniteEx supplier = startGrid(SUPPLIER); + + IgniteCache cache = supplier.createCache(new CacheConfiguration(DEFAULT_CACHE_NAME) + .setAffinity(new RendezvousAffinityFunction(false, 32)) + .setBackups(1)); + + byte[] val = new byte[pageSize(supplier) - 200]; + + // One page per entry: 1.5 times the demander region. + int keys = (int)(totalPages(supplier) * 0.75); + + for (int base = 0; base < keys; base += 100) { + Map batch = new HashMap<>(); + + for (int i = base; i < Math.min(base + 100, keys); i++) + batch.put(i, val); + + cache.putAll(batch); + } + + IgniteEx demander = startGrid(1); + + awaitPartitionMapExchange(); + + assertHonorsThreshold(demander); + + assertEquals(0, ((AtomicLong)GridTestUtils.getFieldValue(defaultFreeList(demander), "reservedPages")).get()); + } + + /** */ + @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 + ); + } + + /** */ + @Test + public void testPutAllLargeRows() 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); + + 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)); + } + + /** */ + @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)); + } + + /** */ + @Test + public void testLargeRowBelowThresholdUsesHeadroomNoEviction() throws Exception { + IgniteEx ignite = startGrid(1); + + IgniteCache cache = createCache(ignite, DEFAULT_CACHE_NAME); + + 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)); + + 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)); + } + + /** */ + @Test + public void testLargeRowAboveThresholdEvictsAndSucceeds() throws Exception { + IgniteEx ignite = startGrid(1); + + IgniteCache cache = createCache(ignite, DEFAULT_CACHE_NAME); + + int smallEntries = fillRegion(ignite, cache, 0.85); + + assertTrue("Pre-fill must load >80% of pages", loadedPages(ignite) > totalPages(ignite) * 0.80); + + 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)); + } + + /** Atomic cache with a backup: primary batch and backup updates must evict at the same threshold. */ + @Test(timeout = 300_000) + public void testAtomicBackupHonorsThreshold() throws Exception { + IgniteEx ignite = startGrids(2); + + IgniteCache cache = ignite.createCache(new CacheConfiguration(DEFAULT_CACHE_NAME) + .setAffinity(new RendezvousAffinityFunction(false, 32)) + .setBackups(1) + .setWriteSynchronizationMode(CacheWriteSynchronizationMode.FULL_SYNC)); + + awaitPartitionMapExchange(); + + byte[] val = new byte[pageSize(ignite) - 200]; + + int keys = (int)(totalPages(ignite) * 3); + + for (int base = 0; base < keys; base += 100) { + Map batch = new HashMap<>(); + + for (int i = base; i < base + 100; i++) + batch.put(i, val); + + cache.putAll(batch); + } + + assertHonorsThreshold(grid(0)); + assertHonorsThreshold(grid(1)); + } + + /** Transactional cache: commits must evict at the threshold. */ + @Test(timeout = 300_000) + public void testTransactionalHonorsThreshold() throws Exception { + IgniteEx ignite = startGrid(0); + + IgniteCache cache = createCache(ignite, DEFAULT_CACHE_NAME, 0, true); + + byte[] val = new byte[pageSize(ignite) - 200]; + + int keys = (int)(totalPages(ignite) * 3); + + for (int i = 0; i < keys; i++) + cache.put(i, val); + + assertHonorsThreshold(ignite); + } + + /** A row that does not fit below the threshold together with the pool may use the space above it. */ + @Test + public void testRowLargerThanThresholdFits() throws Exception { + IgniteEx ignite = startGrid(0); + + IgniteCache cache = createCache(ignite, DEFAULT_CACHE_NAME); + + fillRegion(ignite, cache, 0.5); + + int largeValSize = (int)(regionMaxSize(ignite) * EVICTION_THRESHOLD); + + cache.put(-1, new byte[largeValSize]); + + assertEquals(largeValSize, ((byte[])cache.get(-1)).length); + } + + /** + * When no page can be evicted (all resident entries are locked), the eviction progress guard refuses an insert that + * can be refused (a primary atomic update) with an out-of-memory, while a best-effort reservation (backup update, + * transaction commit, rebalancing) gives up without an error. Once the stall is detected, the already waiting and + * the following best-effort reservations give up without waiting for the timeout, until the eviction makes progress + * again. + */ + @Test(timeout = 180_000) + public void testGuardWhenAllEntriesLocked() throws Exception { + IgniteEx ignite = startGrid(1); + + IgniteCache txCache = createCache(ignite, "tx-cache", 0, true); + IgniteCache atomicCache = createCache(ignite, DEFAULT_CACHE_NAME); + + int fillEntries = fillRegion(ignite, txCache, 0.85); + + Collection keys = new ArrayList<>(fillEntries); + + for (int i = 0; i < fillEntries; i++) + keys.add(i); + + CountDownLatch ready = new CountDownLatch(1); + CountDownLatch release = new CountDownLatch(1); + + IgniteInternalFuture lockerFut = GridTestUtils.runAsync(() -> { + Lock lock = txCache.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)); + + int largeValSize = 4 * 1024 * 1024; + + GridTestUtils.assertThrowsWithCause( + () -> atomicCache.put(0, new byte[largeValSize]), + IgniteOutOfMemoryException.class + ); + + IgniteCacheDatabaseSharedManager db = ignite.context().cache().context().database(); + + DataRegion region = defaultRegion(ignite); + + FreeList freeList = db.freeList(region.config().getName()); + + long timeout = AbstractFreeList.DFLT_EVICTION_NO_PROGRESS_TIMEOUT_MILLIS; + + assertFalse(GridTestUtils.getFieldValue(freeList, "stalled")); + + IgniteInternalFuture firstFut = GridTestUtils.runAsync(() -> db.reserveInRegion(region, largeValSize, true), + "size-aware-guard-first"); + + U.sleep(timeout / 2); + + // Starts waiting before the stall is detected and must be released once the first reservation detects it. + long start = System.currentTimeMillis(); + + assertEquals(0, db.reserveInRegion(region, largeValSize, true)); + + assertTrue("Waiting best-effort reservation must give up once the stall is detected", + System.currentTimeMillis() - start < timeout * 3 / 4); + + assertEquals(0, (long)firstFut.get(timeout, TimeUnit.MILLISECONDS)); + + assertTrue(GridTestUtils.getFieldValue(freeList, "stalled")); + + // The stall is not a time window: once the timeout passes, best-effort reservations still do not wait. + U.sleep(timeout + 500); + + start = System.currentTimeMillis(); + + assertEquals(0, db.reserveInRegion(region, largeValSize, true)); + + assertTrue("Best-effort reservation must not wait for the timeout while the eviction is stalled", + System.currentTimeMillis() - start < timeout / 2); + + release.countDown(); + + lockerFut.get(10, TimeUnit.SECONDS); + + // The unlocked entries can be evicted again: the reservation succeeds and clears the stall. While stalled, a + // best-effort reservation gives up after a single eviction attempt, which may free no page, so it is retried. + long reserved = 0; + + long deadline = System.currentTimeMillis() + timeout; + + while (reserved == 0 && System.currentTimeMillis() < deadline) + reserved = db.reserveInRegion(region, largeValSize, true); + + try { + assertTrue("Reservation must succeed once the entries are unlocked", reserved > 0); + + assertFalse(GridTestUtils.getFieldValue(freeList, "stalled")); + } + finally { + db.releaseInRegion(region, reserved); + } + } + finally { + release.countDown(); + + lockerFut.get(10, TimeUnit.SECONDS); + } + } + + /** + * @param ignite Node. + * @return Free list of the default data region. + */ + private static FreeList defaultFreeList(IgniteEx ignite) throws IgniteCheckedException { + return ignite.context().cache().context().database().freeList(defaultRegion(ignite).config().getName()); + } + + /** + * Asserts that the node evicted at the threshold: loaded pages reached it but do not exceed it by more than the + * empty pages pool and 1% of non-data pages (B+tree, free list metadata). + * + * @param ignite Node. + */ + private static void assertHonorsThreshold(IgniteEx ignite) throws IgniteCheckedException { + assertTrue("Evictions must have started", isEvictionsStarted(ignite)); + + long total = totalPages(ignite); + long loaded = loadedPages(ignite); + long limit = (long)(EVICTION_THRESHOLD * total) + defaultRegion(ignite).config().getEmptyPagesPoolSize() + total / 100; + + assertTrue("Loaded pages must reach the threshold [loaded=" + loaded + ", total=" + total + ']', + loaded >= total * 0.85); + assertTrue("Loaded pages exceed the threshold [node=" + ignite.name() + ", loaded=" + loaded + + ", limit=" + limit + ']', loaded <= limit); + } +} 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..f915b2ac3d57b --- /dev/null +++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/eviction/paged/PageEvictionWithExpiryPolicyAbstractTest.java @@ -0,0 +1,135 @@ +/* + * 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 { + super.afterTest(); + + stopAllGrids(); + } + + /** */ + @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)); + + 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)); + } + + /** */ + @Test + public void testTtlFreedSpaceAccountedForByEviction() throws Exception { + IgniteEx ignite = startGrid(1); + + IgniteCache plainCache = createCache(ignite, "plain-cache"); + IgniteCache ttlCache = createCache(ignite, "ttl-cache", TTL, false); + + // Pre-fill + two 5% TTL rows + the empty pages pool must stay below the 0.9 eviction threshold, which applies + // to every insert including the size of the row being inserted. + int smallEntries = fillRegion(ignite, plainCache, 0.75); + + 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/database/CacheFreeListSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/database/CacheFreeListSelfTest.java index 0b0d253edc462..8f344114f2941 100644 --- a/modules/core/src/test/java/org/apache/ignite/internal/processors/database/CacheFreeListSelfTest.java +++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/database/CacheFreeListSelfTest.java @@ -41,10 +41,13 @@ import org.apache.ignite.internal.pagemem.impl.PageMemoryNoStoreImpl; import org.apache.ignite.internal.processors.cache.CacheObject; import org.apache.ignite.internal.processors.cache.CacheObjectValueContext; +import org.apache.ignite.internal.processors.cache.GridCacheProcessor; +import org.apache.ignite.internal.processors.cache.GridCacheSharedContext; import org.apache.ignite.internal.processors.cache.KeyCacheObject; 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.DataRegionMetricsImpl; +import org.apache.ignite.internal.processors.cache.persistence.IgniteCacheDatabaseSharedManager; import org.apache.ignite.internal.processors.cache.persistence.diagnostic.pagelocktracker.PageLockTrackerManager; import org.apache.ignite.internal.processors.cache.persistence.evict.NoOpPageEvictionTracker; import org.apache.ignite.internal.processors.cache.persistence.freelist.CacheFreeList; @@ -528,6 +531,18 @@ private FreeList createFreeList(int pageSize) throws Exception { when(pageLockTrackerMgr.createPageLockTracker(anyString())).thenReturn(PageLockTrackerManager.NOOP_LSNR); + GridTestKernalContext freeListCtx = new GridTestKernalContext(log); + + GridCacheSharedContext sharedCtx = mock(GridCacheSharedContext.class); + + when(sharedCtx.database()).thenReturn(new IgniteCacheDatabaseSharedManager(freeListCtx)); + + GridCacheProcessor cacheProc = mock(GridCacheProcessor.class); + + when(cacheProc.context()).thenReturn(sharedCtx); + + freeListCtx.add(cacheProc, false); + return new CacheFreeList( 1, "freelist", @@ -536,7 +551,7 @@ private FreeList createFreeList(int pageSize) throws Exception { metaPageId, true, pageLockTrackerMgr, - new GridTestKernalContext(log), + freeListCtx, null, PageIdAllocator.FLAG_IDX ); 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..6873adc3ed5f8 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,19 +38,28 @@ 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.PageEvictionDisabledEvictionSelfTest; 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.PageEvictionReadThroughTest; +import org.apache.ignite.internal.processors.cache.eviction.paged.PageEvictionReservedPagesSelfTest; +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; @@ -100,12 +109,27 @@ 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, PageEvictionPagesRecyclingAndReusingTest.class, ignoredTests); GridTestUtils.addTestIfNeeded(suite, DhtAndNearEvictionTest.class, ignoredTests); + GridTestUtils.addTestIfNeeded(suite, PageEvictionRowSizeBinaryObjectTest.class, ignoredTests); + + GridTestUtils.addTestIfNeeded(suite, PageEvictionReservedPagesSelfTest.class, ignoredTests); + + GridTestUtils.addTestIfNeeded(suite, PageEvictionDisabledEvictionSelfTest.class, ignoredTests); + return suite; } }