IGNITE-29032 Improvement of the page eviction mechanism for in-memory

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.
+ * <ul>
+ *     <li>{@link #putBelowThreshold} — puts within a bounded key range that keeps the region below the eviction
+ *     threshold: the admission fast path, no eviction.</li>
+ *     <li>{@link #putAtThreshold}, {@link #putAllAtThreshold}, {@link #txPutAtThreshold} — new rows inserted into a
+ *     region pre-filled up to the eviction threshold: steady-state eviction on the atomic single put, atomic batch and
+ *     transactional commit paths.</li>
+ * </ul>
+ * 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<Integer, Object> 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<Integer, Object> atomicCache;
+
+        /** Transactional cache. */
+        private IgniteCache<Integer, Object> 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<Integer, Object> 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<Integer, Object> 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<Integer, Object>(ATOMIC_CACHE),
+                new CacheConfiguration<Integer, Object>(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();
+    }
+}

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);
     }
 

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,23 @@ 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.
+     * <p>
+     * {@code 0} or a negative value disables the grace period: the guard gives up on the first no-progress iteration.
+     * <p>
+     * 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.
      *

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.

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;