You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by ag...@apache.org on 2018/08/13 09:07:00 UTC
ignite git commit: IGNITE-9244 Partition eviction should not take all
threads in system pool
Repository: ignite
Updated Branches:
refs/heads/master af03d8413 -> 2d63040fd
IGNITE-9244 Partition eviction should not take all threads in system pool
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/2d63040f
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/2d63040f
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/2d63040f
Branch: refs/heads/master
Commit: 2d63040fdf77bbc4515790d796fa86a8f4f6a66c
Parents: af03d84
Author: Dmitriy Govorukhin <dm...@gmail.com>
Authored: Mon Aug 13 11:59:27 2018 +0300
Committer: Alexey Goncharuk <al...@gmail.com>
Committed: Mon Aug 13 11:59:27 2018 +0300
----------------------------------------------------------------------
.../apache/ignite/IgniteSystemProperties.java | 5 +
.../processors/cache/CacheGroupContext.java | 17 +-
.../processors/cache/GridCacheProcessor.java | 3 +
.../cache/GridCacheSharedContext.java | 48 +-
.../distributed/dht/GridDhtLocalPartition.java | 3 +-
.../dht/GridDhtPartitionsEvictor.java | 310 ----------
.../distributed/dht/PartitionsEvictManager.java | 566 +++++++++++++++++++
.../wal/reader/IgniteWalIteratorFactory.java | 2 +-
.../wal/IgniteWalIteratorSwitchSegmentTest.java | 2 +
.../pagemem/BPlusTreePageMemoryImplTest.java | 1 +
.../BPlusTreeReuseListPageMemoryImplTest.java | 1 +
.../pagemem/IndexStoragePageMemoryImplTest.java | 1 +
.../pagemem/PageMemoryImplNoLoadTest.java | 1 +
.../persistence/pagemem/PageMemoryImplTest.java | 1 +
.../loadtests/hashmap/GridCacheTestContext.java | 2 +
15 files changed, 630 insertions(+), 333 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/2d63040f/modules/core/src/main/java/org/apache/ignite/IgniteSystemProperties.java
----------------------------------------------------------------------
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 3b4aedb..21f6c5f 100644
--- a/modules/core/src/main/java/org/apache/ignite/IgniteSystemProperties.java
+++ b/modules/core/src/main/java/org/apache/ignite/IgniteSystemProperties.java
@@ -942,6 +942,11 @@ public final class IgniteSystemProperties {
public static final String IGNITE_THROTTLE_LOG_THRESHOLD = "IGNITE_THROTTLE_LOG_THRESHOLD";
/**
+ * Number of concurrent operation for evict partitions.
+ */
+ public static final String IGNITE_EVICTION_PERMITS = "IGNITE_EVICTION_PERMITS";
+
+ /**
* Enforces singleton.
*/
private IgniteSystemProperties() {
http://git-wip-us.apache.org/repos/asf/ignite/blob/2d63040f/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheGroupContext.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheGroupContext.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheGroupContext.java
index 7009575..c8fe283 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheGroupContext.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheGroupContext.java
@@ -41,7 +41,7 @@ import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtAffini
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtAffinityAssignmentResponse;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtPartitionTopology;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtPartitionTopologyImpl;
-import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtPartitionsEvictor;
+import org.apache.ignite.internal.processors.cache.distributed.dht.PartitionsEvictManager;
import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPreloader;
import org.apache.ignite.internal.processors.cache.persistence.DataRegion;
import org.apache.ignite.internal.processors.cache.persistence.GridCacheOffheapManager;
@@ -126,10 +126,6 @@ public class CacheGroupContext {
/** */
private GridCachePreloader preldr;
-
- /** Partition evictor. */
- private GridDhtPartitionsEvictor evictor;
-
/** */
private final DataRegion dataRegion;
@@ -260,13 +256,6 @@ public class CacheGroupContext {
}
/**
- * @return Partitions evictor.
- */
- public GridDhtPartitionsEvictor evictor() {
- return evictor;
- }
-
- /**
* @return IO policy for the given cache group.
*/
public byte ioPolicy() {
@@ -733,7 +722,7 @@ public class CacheGroupContext {
IgniteCheckedException err =
new IgniteCheckedException("Failed to wait for topology update, cache (or node) is stopping.");
- evictor.stop();
+ ctx.evict().onCacheGroupStopped(this);
aff.cancelFutures(err);
@@ -909,8 +898,6 @@ public class CacheGroupContext {
else
preldr = new GridCachePreloaderAdapter(this);
- evictor = new GridDhtPartitionsEvictor(this);
-
if (persistenceEnabled()) {
try {
offheapMgr = new GridCacheOffheapManager();
http://git-wip-us.apache.org/repos/asf/ignite/blob/2d63040f/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProcessor.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProcessor.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProcessor.java
index 09e2860..ed97982 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProcessor.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProcessor.java
@@ -81,6 +81,7 @@ import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtCache;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtCacheAdapter;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtLocalPartition;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtPartitionTopology;
+import org.apache.ignite.internal.processors.cache.distributed.dht.PartitionsEvictManager;
import org.apache.ignite.internal.processors.cache.distributed.dht.atomic.GridDhtAtomicCache;
import org.apache.ignite.internal.processors.cache.distributed.dht.colocated.GridDhtColocatedCache;
import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.StopCachesOnClientReconnectExchangeTask;
@@ -2463,6 +2464,7 @@ public class GridCacheProcessor extends GridProcessorAdapter {
GridCacheIoManager ioMgr = new GridCacheIoManager();
CacheAffinitySharedManager topMgr = new CacheAffinitySharedManager();
GridCacheSharedTtlCleanupManager ttl = new GridCacheSharedTtlCleanupManager();
+ PartitionsEvictManager evict = new PartitionsEvictManager();
CacheJtaManagerAdapter jta = JTA.createOptional();
@@ -2481,6 +2483,7 @@ public class GridCacheProcessor extends GridProcessorAdapter {
topMgr,
ioMgr,
ttl,
+ evict,
jta,
storeSesLsnrs
);
http://git-wip-us.apache.org/repos/asf/ignite/blob/2d63040f/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheSharedContext.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheSharedContext.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheSharedContext.java
index 381d5bf..4b98060 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheSharedContext.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheSharedContext.java
@@ -45,6 +45,7 @@ import org.apache.ignite.internal.pagemem.wal.IgniteWriteAheadLogManager;
import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtPartitionTopology;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtTopologyFuture;
+import org.apache.ignite.internal.processors.cache.distributed.dht.PartitionsEvictManager;
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearTxLocal;
import org.apache.ignite.internal.processors.cache.jta.CacheJtaManagerAdapter;
import org.apache.ignite.internal.processors.cache.persistence.IgniteCacheDatabaseSharedManager;
@@ -126,6 +127,9 @@ public class GridCacheSharedContext<K, V> {
/** Ttl cleanup manager. */
private GridCacheSharedTtlCleanupManager ttlMgr;
+ /** */
+ private PartitionsEvictManager evictMgr;
+
/** Cache contexts map. */
private ConcurrentHashMap<Integer, GridCacheContext<K, V>> ctxMap;
@@ -199,13 +203,30 @@ public class GridCacheSharedContext<K, V> {
CacheAffinitySharedManager<K, V> affMgr,
GridCacheIoManager ioMgr,
GridCacheSharedTtlCleanupManager ttlMgr,
+ PartitionsEvictManager evictMgr,
CacheJtaManagerAdapter jtaMgr,
Collection<CacheStoreSessionListener> storeSesLsnrs
) {
this.kernalCtx = kernalCtx;
- setManagers(mgrs, txMgr, jtaMgr, verMgr, mvccMgr, pageStoreMgr, walMgr, walStateMgr, dbMgr, snpMgr, depMgr,
- exchMgr, affMgr, ioMgr, ttlMgr);
+ setManagers(
+ mgrs,
+ txMgr,
+ jtaMgr,
+ verMgr,
+ mvccMgr,
+ pageStoreMgr,
+ walMgr,
+ walStateMgr,
+ dbMgr,
+ snpMgr,
+ depMgr,
+ exchMgr,
+ affMgr,
+ ioMgr,
+ ttlMgr,
+ evictMgr
+ );
this.storeSesLsnrs = storeSesLsnrs;
@@ -349,7 +370,9 @@ public class GridCacheSharedContext<K, V> {
void onReconnected(boolean active) throws IgniteCheckedException {
List<GridCacheSharedManager<K, V>> mgrs = new LinkedList<>();
- setManagers(mgrs, txMgr,
+ setManagers(
+ mgrs,
+ txMgr,
jtaMgr,
verMgr,
mvccMgr,
@@ -362,7 +385,9 @@ public class GridCacheSharedContext<K, V> {
new GridCachePartitionExchangeManager<K, V>(),
affMgr,
ioMgr,
- ttlMgr);
+ ttlMgr,
+ evictMgr
+ );
this.mgrs = mgrs;
@@ -404,7 +429,8 @@ public class GridCacheSharedContext<K, V> {
* @param ttlMgr Ttl cleanup manager.
*/
@SuppressWarnings("unchecked")
- private void setManagers(List<GridCacheSharedManager<K, V>> mgrs,
+ private void setManagers(
+ List<GridCacheSharedManager<K, V>> mgrs,
IgniteTxManager txMgr,
CacheJtaManagerAdapter jtaMgr,
GridCacheVersionManager verMgr,
@@ -418,7 +444,9 @@ public class GridCacheSharedContext<K, V> {
GridCachePartitionExchangeManager<K, V> exchMgr,
CacheAffinitySharedManager affMgr,
GridCacheIoManager ioMgr,
- GridCacheSharedTtlCleanupManager ttlMgr) {
+ GridCacheSharedTtlCleanupManager ttlMgr,
+ PartitionsEvictManager evictMgr
+ ) {
this.mvccMgr = add(mgrs, mvccMgr);
this.verMgr = add(mgrs, verMgr);
this.txMgr = add(mgrs, txMgr);
@@ -433,6 +461,7 @@ public class GridCacheSharedContext<K, V> {
this.affMgr = add(mgrs, affMgr);
this.ioMgr = add(mgrs, ioMgr);
this.ttlMgr = add(mgrs, ttlMgr);
+ this.evictMgr = add(mgrs, evictMgr);
}
/**
@@ -765,6 +794,13 @@ public class GridCacheSharedContext<K, V> {
}
/**
+ * @return Partition evict manager.
+ */
+ public PartitionsEvictManager evict() {
+ return evictMgr;
+ }
+
+ /**
* @return Node ID.
*/
public UUID localNodeId() {
http://git-wip-us.apache.org/repos/asf/ignite/blob/2d63040f/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtLocalPartition.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtLocalPartition.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtLocalPartition.java
index e13c952..e1e9f8e 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtLocalPartition.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtLocalPartition.java
@@ -684,7 +684,7 @@ public class GridDhtLocalPartition extends GridCacheConcurrentMapImpl implements
}
}
- grp.evictor().evictPartitionAsync(this);
+ ctx.evict().evictPartitionAsync(grp,this);
}
/**
@@ -696,6 +696,7 @@ public class GridDhtLocalPartition extends GridCacheConcurrentMapImpl implements
return;
clear = true;
+
clearAsync0(false);
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/2d63040f/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtPartitionsEvictor.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtPartitionsEvictor.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtPartitionsEvictor.java
deleted file mode 100644
index 7206397..0000000
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtPartitionsEvictor.java
+++ /dev/null
@@ -1,310 +0,0 @@
-/*
- * 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.distributed.dht;
-
-import java.util.Queue;
-import java.util.Set;
-import java.util.concurrent.Callable;
-import java.util.concurrent.LinkedBlockingQueue;
-import java.util.function.Function;
-import org.apache.ignite.IgniteCheckedException;
-import org.apache.ignite.IgniteLogger;
-import org.apache.ignite.IgniteSystemProperties;
-import org.apache.ignite.internal.IgniteInternalFuture;
-import org.apache.ignite.internal.processors.cache.CacheGroupContext;
-import org.apache.ignite.internal.processors.cache.GridCacheSharedContext;
-import org.apache.ignite.internal.util.GridConcurrentHashSet;
-import org.apache.ignite.internal.util.future.GridFutureAdapter;
-import org.apache.ignite.internal.util.typedef.internal.LT;
-import org.apache.ignite.internal.util.typedef.internal.U;
-
-/**
- * Class that serves asynchronous part eviction process.
- * Only one partition from group can be evicted at the moment.
- */
-public class GridDhtPartitionsEvictor {
- /** Default eviction progress show frequency. */
- private static final int DEFAULT_SHOW_EVICTION_PROGRESS_FREQ_MS = 2 * 60 * 1000; // 2 Minutes.
-
- /** Eviction progress frequency property name. */
- private static final String SHOW_EVICTION_PROGRESS_FREQ = "SHOW_EVICTION_PROGRESS_FREQ";
-
- /** */
- private final GridCacheSharedContext<?, ?> ctx;
-
- /** */
- private final CacheGroupContext grp;
-
- /** */
- private final IgniteLogger log;
-
- /** Lock object. */
- private final Object mux = new Object();
-
- /** Queue contains partitions scheduled for eviction. */
- private final DeduplicationQueue<Integer, GridDhtLocalPartition> evictionQueue = new DeduplicationQueue<>(GridDhtLocalPartition::id);
-
- /**
- * Flag indicates that eviction process is running at the moment.
- * This is needed to schedule partition eviction if there are no currently running self-scheduling eviction tasks.
- * Guarded by {@link #mux}.
- */
- private boolean evictionRunning;
-
- /** Flag indicates that eviction process has stopped. */
- private volatile boolean stop;
-
- /** Future for currently running partition eviction task. */
- private volatile GridFutureAdapter<Boolean> evictionFut;
-
- /** Eviction progress frequency in ms. */
- private final long evictionProgressFreqMs = IgniteSystemProperties.getLong(SHOW_EVICTION_PROGRESS_FREQ,
- DEFAULT_SHOW_EVICTION_PROGRESS_FREQ_MS);
-
- /** Next time of show eviction progress. */
- private long nextShowProgressTime;
-
- /**
- * Constructor.
- *
- * @param grp Cache group context.
- */
- public GridDhtPartitionsEvictor(CacheGroupContext grp) {
- assert grp != null;
-
- this.grp = grp;
- this.ctx = grp.shared();
-
- this.log = ctx.logger(getClass());
- }
-
- /**
- * Adds partition to eviction queue and starts eviction process.
- *
- * @param part Partition to evict.
- */
- public void evictPartitionAsync(GridDhtLocalPartition part) {
- if (stop)
- return;
-
- boolean added = evictionQueue.offer(part);
-
- if (!added)
- return;
-
- synchronized (mux) {
- if (!evictionRunning) {
- nextShowProgressTime = U.currentTimeMillis() + evictionProgressFreqMs;
-
- scheduleNextPartitionEviction();
- }
- }
- }
-
- /**
- * Stops eviction process.
- * Method awaits last offered partition eviction.
- */
- public void stop() {
- stop = true;
-
- synchronized (mux) {
- // Wait for last offered partition eviction completion.
- IgniteInternalFuture<Boolean> evictionFut0 = evictionFut;
-
- if (evictionFut0 != null) {
- try {
- evictionFut0.get();
- }
- catch (IgniteCheckedException e) {
- if (log.isDebugEnabled())
- log.warning("Failed to await partition eviction during stopping", e);
- }
- }
- }
- }
-
- /**
- * Gets next partition from the queue and schedules it for eviction.
- */
- private void scheduleNextPartitionEviction() {
- if (stop)
- return;
-
- synchronized (mux) {
- GridDhtLocalPartition next = evictionQueue.poll();
-
- if (next != null) {
- showProgress();
-
- evictionFut = new GridFutureAdapter<>();
-
- ctx.kernalContext().closure().callLocalSafe(new PartitionEvictionTask(next, () -> stop), true);
- }
- else
- evictionRunning = false;
- }
- }
-
- /**
- * Shows progress of eviction.
- */
- private void showProgress() {
- if (U.currentTimeMillis() >= nextShowProgressTime) {
- int size = evictionQueue.size() + 1; // Queue size plus current partition.
-
- if (log.isInfoEnabled())
- log.info("Eviction in progress [grp=" + grp.cacheOrGroupName()
- + ", remainingPartsCnt=" + size + "]");
-
- nextShowProgressTime = U.currentTimeMillis() + evictionProgressFreqMs;
- }
- }
-
- /**
- * Task for self-scheduled partition eviction / clearing.
- */
- private class PartitionEvictionTask implements Callable<Boolean> {
- /** Partition to evict. */
- private final GridDhtLocalPartition part;
-
- /** Eviction context. */
- private final EvictionContext evictionCtx;
-
- /**
- * @param part Partition.
- * @param evictionCtx Eviction context.
- */
- public PartitionEvictionTask(GridDhtLocalPartition part, EvictionContext evictionCtx) {
- this.part = part;
- this.evictionCtx = evictionCtx;
- }
-
- /** {@inheritDoc} */
- @Override public Boolean call() throws Exception {
- if (stop) {
- evictionFut.onDone();
-
- return false;
- }
-
- try {
- boolean success = part.tryClear(evictionCtx);
-
- if (success) {
- if (part.state() == GridDhtPartitionState.EVICTED && part.markForDestroy())
- part.destroy();
- }
- else // Re-offer partition if clear was unsuccessful due to partition reservation.
- evictionQueue.offer(part);
-
- // Complete eviction future before schedule new to prevent deadlock with
- // simultaneous eviction stopping and scheduling new eviction.
- evictionFut.onDone();
-
- scheduleNextPartitionEviction();
-
- return true;
- }
- catch (Throwable ex) {
- evictionFut.onDone(ex);
-
- if (ctx.kernalContext().isStopping()) {
- LT.warn(log, ex, "Partition eviction failed (current node is stopping).",
- false,
- true);
- }
- else
- LT.error(log, ex, "Partition eviction failed, this can cause grid hang.");
- }
-
- return false;
- }
- }
-
- /**
- * Thread-safe blocking queue with items deduplication.
- *
- * @param <K> Key type of item used for deduplication.
- * @param <V> Queue item type.
- */
- private static class DeduplicationQueue<K, V> {
- /** Queue. */
- private final Queue<V> queue;
-
- /** Unique items set. */
- private final Set<K> uniqueItems;
-
- /** Key mapping function. */
- private final Function<V, K> keyMappingFunction;
-
- /**
- * Constructor.
- *
- * @param keyExtractor Function to extract a key from a queue item.
- * This key is used for deduplication if some item has offered twice.
- */
- public DeduplicationQueue(Function<V, K> keyExtractor) {
- keyMappingFunction = keyExtractor;
- queue = new LinkedBlockingQueue<>();
- uniqueItems = new GridConcurrentHashSet<>();
- }
-
- /**
- * Offers item to the queue.
- *
- * @param item Item.
- * @return {@code true} if item has been successfully offered to the queue,
- * {@code false} if item was rejected because already exists in the queue.
- */
- public boolean offer(V item) {
- K key = keyMappingFunction.apply(item);
-
- if (uniqueItems.add(key)) {
- queue.offer(item);
-
- return true;
- }
-
- return false;
- }
-
- /**
- * Polls next item from queue.
- *
- * @return Next item or {@code null} if queue is empty.
- */
- public V poll() {
- V item = queue.poll();
-
- if (item != null) {
- K key = keyMappingFunction.apply(item);
-
- uniqueItems.remove(key);
- }
-
- return item;
- }
-
- /**
- * @return Size of queue.
- */
- public int size() {
- return queue.size();
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/ignite/blob/2d63040f/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/PartitionsEvictManager.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/PartitionsEvictManager.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/PartitionsEvictManager.java
new file mode 100644
index 0000000..f76310d
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/PartitionsEvictManager.java
@@ -0,0 +1,566 @@
+/*
+ * 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.distributed.dht;
+
+import java.util.Collection;
+import java.util.Comparator;
+import java.util.Map;
+import java.util.Queue;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.PriorityBlockingQueue;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.internal.IgniteInternalFuture;
+import org.apache.ignite.internal.managers.communication.GridIoPolicy;
+import org.apache.ignite.internal.processors.cache.CacheGroupContext;
+import org.apache.ignite.internal.processors.cache.GridCacheSharedManagerAdapter;
+import org.apache.ignite.internal.util.GridConcurrentHashSet;
+import org.apache.ignite.internal.util.future.GridFutureAdapter;
+import org.apache.ignite.internal.util.typedef.internal.LT;
+import org.apache.ignite.internal.util.typedef.internal.U;
+
+import static org.apache.ignite.IgniteSystemProperties.IGNITE_EVICTION_PERMITS;
+import static org.apache.ignite.IgniteSystemProperties.getInteger;
+import static org.apache.ignite.IgniteSystemProperties.getLong;
+
+/**
+ * Class that serves asynchronous part eviction process.
+ * Multiple partition from group can be evicted at the same time.
+ */
+public class PartitionsEvictManager extends GridCacheSharedManagerAdapter {
+
+ /** Default eviction progress show frequency. */
+ private static final int DEFAULT_SHOW_EVICTION_PROGRESS_FREQ_MS = 2 * 60 * 1000; // 2 Minutes.
+
+ /** Eviction progress frequency property name. */
+ private static final String SHOW_EVICTION_PROGRESS_FREQ = "SHOW_EVICTION_PROGRESS_FREQ";
+
+ /** Eviction thread pool policy. */
+ private static final byte EVICT_POOL_PLC = GridIoPolicy.SYSTEM_POOL;
+
+ /** Eviction progress frequency in ms. */
+ private final long evictionProgressFreqMs = getLong(SHOW_EVICTION_PROGRESS_FREQ, DEFAULT_SHOW_EVICTION_PROGRESS_FREQ_MS);
+
+ /** */
+ private final int confPermits = getInteger(IGNITE_EVICTION_PERMITS, -1);
+
+ /** Next time of show eviction progress. */
+ private long nextShowProgressTime;
+
+ private final Map<Integer, GroupEvictionContext> evictionGroupsMap = new ConcurrentHashMap<>();
+
+ /** Flag indicates that eviction process has stopped. */
+ private volatile boolean stop;
+
+ /** Check stop eviction context. */
+ private final EvictionContext sharedEvictionContext = () -> stop;
+
+ /** Number of maximum concurrent operations. */
+ private volatile int threads;
+
+ /** How many eviction task may execute concurrent. */
+ private volatile int permits;
+
+ /** Bucket queue for load balance partitions to the threads via count of partition size.
+ * Is not thread-safe.
+ * All method should be called under mux synchronization.
+ */
+ private volatile BucketQueue evictionQueue;
+
+ /** Lock object. */
+ private final Object mux = new Object();
+
+ /**
+ * Stops eviction process for group.
+ *
+ * Method awaits last offered partition eviction.
+ *
+ * @param grp Group context.
+ */
+ public void onCacheGroupStopped(CacheGroupContext grp){
+ GroupEvictionContext groupEvictionContext = evictionGroupsMap.remove(grp.groupId());
+
+ if (groupEvictionContext != null){
+ groupEvictionContext.stop();
+
+ groupEvictionContext.awaitFinishAll();
+ }
+ }
+
+ /**
+ * Adds partition to eviction queue and starts eviction process if permit available.
+ *
+ * @param grp Group context.
+ * @param part Partition to evict.
+ */
+ public void evictPartitionAsync(CacheGroupContext grp, GridDhtLocalPartition part) {
+ // Check node stop.
+ if (sharedEvictionContext.shouldStop())
+ return;
+
+ GroupEvictionContext groupEvictionContext = evictionGroupsMap.computeIfAbsent(
+ grp.groupId(), (k) -> new GroupEvictionContext(grp));
+
+ PartitionEvictionTask evictionTask = groupEvictionContext.createEvictPartitionTask(part);
+
+ if (evictionTask == null)
+ return;
+
+ int bucket;
+
+ synchronized (mux) {
+ bucket = evictionQueue.offer(evictionTask);
+ }
+
+ scheduleNextPartitionEviction(bucket);
+ }
+
+ /**
+ * Gets next partition from the queue and schedules it for eviction.
+ *
+ * @param bucket Bucket.
+ */
+ private void scheduleNextPartitionEviction(int bucket) {
+ // Check node stop.
+ if (sharedEvictionContext.shouldStop())
+ return;
+
+ synchronized (mux) {
+ // Check that we have permits for next operation.
+ if (permits > 0) {
+ // If queue is empty not need to do.
+ if (evictionQueue.isEmpty())
+ return;
+
+ // Get task until we have permits.
+ while (permits >= 0) {
+ // Get task from bucket.
+ PartitionEvictionTask evictionTask = evictionQueue.poll(bucket);
+
+ // If bucket empty try get from another.
+ if (evictionTask == null) {
+ // Until queue have tasks.
+ while (!evictionQueue.isEmpty()) {
+ // Get task from any other bucket.
+ evictionTask = evictionQueue.pollAny();
+
+ // Stop iteration if we found task.
+ if (evictionTask != null)
+ break;
+ }
+
+ // If task not found no need to do some.
+ if (evictionTask == null)
+ return;
+ }
+
+ // Print current eviction progress.
+ showProgress();
+
+ GroupEvictionContext groupEvictionContext = evictionTask.groupEvictionContext;
+
+ // Check that group or node stopping.
+ if (groupEvictionContext.shouldStop())
+ continue;
+
+ // Get permit for this task.
+ permits--;
+
+ // Register task future, may need if group or node will be stopped.
+ groupEvictionContext.taskScheduled(evictionTask);
+
+ evictionTask.finishFut.listen(f -> {
+ synchronized (mux) {
+ // Return permit after task completed.
+ permits++;
+ }
+
+ // Re-schedule new one task form same bucket.
+ scheduleNextPartitionEviction(bucket);
+ });
+
+ // Submit task to executor.
+ cctx.kernalContext()
+ .closure()
+ .runLocalSafe(evictionTask, EVICT_POOL_PLC);
+ }
+ }
+ }
+ }
+
+ /**
+ * Shows progress of eviction.
+ */
+ private void showProgress() {
+ if (U.currentTimeMillis() >= nextShowProgressTime) {
+ int size = evictionQueue.size() + 1; // Queue size plus current partition.
+
+ if (log.isInfoEnabled())
+ log.info("Eviction in progress [permits=" + permits+
+ ", threads=" + threads +
+ ", groups=" + evictionGroupsMap.keySet().size() +
+ ", remainingPartsToEvict=" + size + "]");
+
+ evictionGroupsMap.values().forEach(GroupEvictionContext::showProgress);
+
+ nextShowProgressTime = U.currentTimeMillis() + evictionProgressFreqMs;
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void start0() throws IgniteCheckedException {
+ super.start0();
+
+ // If property is not setup, calculate permits as parts of sys pool.
+ if (confPermits == -1) {
+ int sysPoolSize = cctx.kernalContext().config().getSystemThreadPoolSize();
+
+ threads = permits = sysPoolSize / 4;
+ }
+ else
+ threads = permits = confPermits;
+
+ // Avoid 0 permits if sys pool size less that 4.
+ if (threads == 0)
+ threads = permits = 1;
+
+ log.info("Evict partition permits=" + permits);
+
+ evictionQueue = new BucketQueue(threads);
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void stop0(boolean cancel) {
+ super.stop0(cancel);
+
+ stop = true;
+
+ Collection<GroupEvictionContext> evictionGrps = evictionGroupsMap.values();
+
+ evictionGrps.forEach(GroupEvictionContext::stop);
+
+ evictionGrps.forEach(GroupEvictionContext::awaitFinishAll);
+ }
+
+ /**
+ *
+ */
+ private class GroupEvictionContext implements EvictionContext {
+ /** */
+ private final CacheGroupContext grp;
+
+ /** Deduplicate set partition ids. */
+ private final Set<Integer> partIds = new GridConcurrentHashSet<>();
+
+ /** Future for currently running partition eviction task. */
+ private final Map<Integer, IgniteInternalFuture<?>> partsEvictFutures = new ConcurrentHashMap<>();
+
+ /** Flag indicates that eviction process has stopped for this group. */
+ private volatile boolean stop;
+
+ /** Total partition to evict. */
+ private AtomicInteger totalTasks = new AtomicInteger();
+
+ /** Total partition evict in progress. */
+ private int taskInProgress;
+
+ /**
+ * @param grp Group context.
+ */
+ private GroupEvictionContext(CacheGroupContext grp) {
+ this.grp = grp;
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean shouldStop() {
+ return stop || sharedEvictionContext.shouldStop();
+ }
+
+ /**
+ *
+ * @param part Grid local partition.
+ */
+ private PartitionEvictionTask createEvictPartitionTask(GridDhtLocalPartition part){
+ if (shouldStop() || !partIds.add(part.id()))
+ return null;
+
+ totalTasks.incrementAndGet();
+
+ return new PartitionEvictionTask(part, this);
+ }
+
+ /**
+ *
+ * @param task Partition eviction task.
+ */
+ private synchronized void taskScheduled(PartitionEvictionTask task) {
+ if (shouldStop())
+ return;
+
+ taskInProgress++;
+
+ GridFutureAdapter<?> fut = task.finishFut;
+
+ int partId = task.part.id();
+
+ partsEvictFutures.put(partId, fut);
+
+ fut.listen(f -> {
+ synchronized (this) {
+ taskInProgress--;
+
+ partsEvictFutures.remove(partId, f);
+
+ if (totalTasks.decrementAndGet() == 0)
+ evictionGroupsMap.remove(grp.groupId());
+ }
+ });
+ }
+
+ /**
+ * Stop eviction for group.
+ */
+ private void stop() {
+ stop = true;
+ }
+
+ /**
+ * Await evict finish.
+ */
+ private void awaitFinishAll(){
+ partsEvictFutures.forEach(this::awaitFinish);
+
+ evictionGroupsMap.remove(grp.groupId());
+ }
+
+ /**
+ * Await evict finish partition.
+ */
+ private void awaitFinish(Integer part, IgniteInternalFuture<?> fut) {
+ // Wait for last offered partition eviction completion
+ try {
+ log.info("Await partition evict, grpName=" + grp.cacheOrGroupName() +
+ ", grpId=" + grp.groupId() + ", partId=" + part);
+
+ fut.get();
+ }
+ catch (IgniteCheckedException e) {
+ if (log.isDebugEnabled())
+ log.warning("Failed to await partition eviction during stopping.", e);
+ }
+ }
+
+ /**
+ * Shows progress group of eviction.
+ */
+ private void showProgress() {
+ if (log.isInfoEnabled())
+ log.info("Group eviction in progress [grpName=" + grp.cacheOrGroupName()+
+ ", grpId=" + grp.groupId() +
+ ", remainingPartsToEvict=" + (totalTasks.get() - taskInProgress) +
+ ", partsEvictInProgress=" + taskInProgress +
+ ", totalParts= " + grp.topology().localPartitions().size() + "]");
+ }
+ }
+
+ /**
+ * Task for self-scheduled partition eviction / clearing.
+ */
+ private class PartitionEvictionTask implements Runnable {
+ /** Partition to evict. */
+ private final GridDhtLocalPartition part;
+
+ private final long size;
+
+ /** Eviction context. */
+ private final GroupEvictionContext groupEvictionContext;
+
+ /** */
+ private final GridFutureAdapter<?> finishFut = new GridFutureAdapter<>();
+
+ /**
+ * @param part Partition.
+ */
+ private PartitionEvictionTask(
+ GridDhtLocalPartition part,
+ GroupEvictionContext groupEvictionContext
+ ) {
+ this.part = part;
+ this.groupEvictionContext = groupEvictionContext;
+
+ size = part.fullSize();
+ }
+
+ /** {@inheritDoc} */
+ @Override public void run() {
+ if (groupEvictionContext.shouldStop())
+ return;
+
+ try {
+ boolean success = part.tryClear(groupEvictionContext);
+
+ if (success) {
+ if (part.state() == GridDhtPartitionState.EVICTED && part.markForDestroy())
+ part.destroy();
+ }
+ else // Re-offer partition if clear was unsuccessful due to partition reservation.
+ evictionQueue.offer(this);
+
+ // Complete eviction future before schedule new to prevent deadlock with
+ // simultaneous eviction stopping and scheduling new eviction.
+ finishFut.onDone();
+ }
+ catch (Throwable ex) {
+ finishFut.onDone(ex);
+
+ if (cctx.kernalContext().isStopping()) {
+ LT.warn(log, ex, "Partition eviction failed (current node is stopping).",
+ false,
+ true);
+ }
+ else{
+ LT.error(log, ex, "Partition eviction failed, this can cause grid hang.");
+ }
+ }
+ }
+ }
+
+ /**
+ *
+ */
+ private class BucketQueue {
+ /** Queues contains partitions scheduled for eviction. */
+ private final Queue<PartitionEvictionTask>[] buckets;
+
+ /** */
+ private final long[] bucketSizes;
+
+ /**
+ * @param buckets Number of buckets.
+ */
+ BucketQueue(int buckets) {
+ this.buckets = new Queue[buckets];
+
+ for (int i = 0; i < buckets; i++)
+ this.buckets[i] = createEvictPartitionQueue();
+
+ bucketSizes = new long[buckets];
+ }
+
+ /**
+ * Poll eviction task from queue for specific bucket.
+ *
+ * @param bucket Bucket index.
+ * @return Partition evict task, or {@code null} if bucket queue is empty.
+ */
+ PartitionEvictionTask poll(int bucket) {
+ PartitionEvictionTask task = buckets[bucket].poll();
+
+ if (task != null)
+ bucketSizes[bucket] -= task.size;
+
+ return task;
+ }
+
+ /**
+ * Poll eviction task from queue (bucket is not specific).
+ *
+ * @return Partition evict task.
+ */
+ PartitionEvictionTask pollAny() {
+ for (int bucket = 0; bucket < bucketSizes.length; bucket++){
+ if (!buckets[bucket].isEmpty())
+ return poll(bucket);
+ }
+
+ return null;
+ }
+
+ /**
+ * Offer task to queue.
+ *
+ * @return Bucket index.
+ */
+ int offer(PartitionEvictionTask task) {
+ int bucket = calculateBucket();
+
+ buckets[bucket].offer(task);
+
+ bucketSizes[bucket] += task.size;
+
+ return bucket;
+ }
+
+
+ /**
+ * @return {@code True} if queue is empty, {@code} False if not empty.
+ */
+ boolean isEmpty(){
+ return size() == 0;
+ }
+
+ /**
+ * @return Queue size.
+ */
+ int size(){
+ int size = 0;
+
+ for (Queue<PartitionEvictionTask> queue : buckets) {
+ size += queue.size();
+ }
+
+ return size;
+ }
+
+ /***
+ * @return Bucket index.
+ */
+ private int calculateBucket() {
+ int min = 0;
+
+ for (int bucket = min; bucket < bucketSizes.length; bucket++) {
+ if (bucketSizes[min] > bucketSizes[bucket])
+ min = bucket;
+ }
+
+ return min;
+ }
+
+ /**
+ * 0 - PRIORITY QUEUE (compare by partition size).
+ * default (any other values) - FIFO.
+ */
+ private static final byte QUEUE_TYPE = 1;
+
+ /**
+ *
+ * @return Queue for evict partitions.
+ */
+ private Queue<PartitionEvictionTask> createEvictPartitionQueue() {
+ switch (QUEUE_TYPE) {
+ case 1:
+ return new PriorityBlockingQueue<>(
+ 1000, Comparator.comparingLong(p -> p.part.fullSize()));
+ default:
+ return new LinkedBlockingQueue<>();
+ }
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/2d63040f/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/wal/reader/IgniteWalIteratorFactory.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/wal/reader/IgniteWalIteratorFactory.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/wal/reader/IgniteWalIteratorFactory.java
index eeb52b3..aae9775 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/wal/reader/IgniteWalIteratorFactory.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/wal/reader/IgniteWalIteratorFactory.java
@@ -322,7 +322,7 @@ public class IgniteWalIteratorFactory {
kernalCtx, null, null, null,
null, null, null, dbMgr, null,
null, null, null, null,
- null, null, null
+ null, null,null, null
);
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/2d63040f/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/db/wal/IgniteWalIteratorSwitchSegmentTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/db/wal/IgniteWalIteratorSwitchSegmentTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/db/wal/IgniteWalIteratorSwitchSegmentTest.java
index 00ed6f1..9dbef5d 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/db/wal/IgniteWalIteratorSwitchSegmentTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/db/wal/IgniteWalIteratorSwitchSegmentTest.java
@@ -142,6 +142,7 @@ public class IgniteWalIteratorSwitchSegmentTest extends GridCommonAbstractTest {
null,
null,
null,
+ null,
null)
).createSerializer(serVer);
@@ -370,6 +371,7 @@ public class IgniteWalIteratorSwitchSegmentTest extends GridCommonAbstractTest {
new GridCacheIoManager(),
null,
null,
+ null,
null
);
http://git-wip-us.apache.org/repos/asf/ignite/blob/2d63040f/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/BPlusTreePageMemoryImplTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/BPlusTreePageMemoryImplTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/BPlusTreePageMemoryImplTest.java
index 3737204..7719b43 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/BPlusTreePageMemoryImplTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/BPlusTreePageMemoryImplTest.java
@@ -60,6 +60,7 @@ public class BPlusTreePageMemoryImplTest extends BPlusTreeSelfTest {
null,
null,
null,
+ null,
null
);
http://git-wip-us.apache.org/repos/asf/ignite/blob/2d63040f/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/BPlusTreeReuseListPageMemoryImplTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/BPlusTreeReuseListPageMemoryImplTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/BPlusTreeReuseListPageMemoryImplTest.java
index c36ecce..71eb129 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/BPlusTreeReuseListPageMemoryImplTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/BPlusTreeReuseListPageMemoryImplTest.java
@@ -62,6 +62,7 @@ public class BPlusTreeReuseListPageMemoryImplTest extends BPlusTreeReuseSelfTest
null,
null,
null,
+ null,
null
);
http://git-wip-us.apache.org/repos/asf/ignite/blob/2d63040f/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/IndexStoragePageMemoryImplTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/IndexStoragePageMemoryImplTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/IndexStoragePageMemoryImplTest.java
index 9087b1c..43fbb6e 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/IndexStoragePageMemoryImplTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/IndexStoragePageMemoryImplTest.java
@@ -75,6 +75,7 @@ public class IndexStoragePageMemoryImplTest extends IndexStorageSelfTest {
null,
null,
null,
+ null,
null
);
http://git-wip-us.apache.org/repos/asf/ignite/blob/2d63040f/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/PageMemoryImplNoLoadTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/PageMemoryImplNoLoadTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/PageMemoryImplNoLoadTest.java
index 34fd93b..52aff0c 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/PageMemoryImplNoLoadTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/PageMemoryImplNoLoadTest.java
@@ -65,6 +65,7 @@ public class PageMemoryImplNoLoadTest extends PageMemoryNoLoadSelfTest {
null,
null,
null,
+ null,
null
);
http://git-wip-us.apache.org/repos/asf/ignite/blob/2d63040f/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/PageMemoryImplTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/PageMemoryImplTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/PageMemoryImplTest.java
index 3697c4c..000131a 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/PageMemoryImplTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/PageMemoryImplTest.java
@@ -297,6 +297,7 @@ public class PageMemoryImplTest extends GridCommonAbstractTest {
null,
null,
null,
+ null,
null
);
http://git-wip-us.apache.org/repos/asf/ignite/blob/2d63040f/modules/core/src/test/java/org/apache/ignite/loadtests/hashmap/GridCacheTestContext.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/loadtests/hashmap/GridCacheTestContext.java b/modules/core/src/test/java/org/apache/ignite/loadtests/hashmap/GridCacheTestContext.java
index 3933953..64d29e6 100644
--- a/modules/core/src/test/java/org/apache/ignite/loadtests/hashmap/GridCacheTestContext.java
+++ b/modules/core/src/test/java/org/apache/ignite/loadtests/hashmap/GridCacheTestContext.java
@@ -37,6 +37,7 @@ import org.apache.ignite.internal.processors.cache.GridCacheSharedTtlCleanupMana
import org.apache.ignite.internal.processors.cache.GridCacheTtlManager;
import org.apache.ignite.internal.processors.cache.WalStateManager;
import org.apache.ignite.internal.processors.cache.datastructures.CacheDataStructuresManager;
+import org.apache.ignite.internal.processors.cache.distributed.dht.PartitionsEvictManager;
import org.apache.ignite.internal.processors.cache.dr.GridOsCacheDrManager;
import org.apache.ignite.internal.processors.cache.jta.CacheNoopJtaManager;
import org.apache.ignite.internal.processors.cache.persistence.IgniteCacheDatabaseSharedManager;
@@ -78,6 +79,7 @@ public class GridCacheTestContext<K, V> extends GridCacheContext<K, V> {
new CacheAffinitySharedManager<K, V>(),
new GridCacheIoManager(),
new GridCacheSharedTtlCleanupManager(),
+ new PartitionsEvictManager(),
new CacheNoopJtaManager(),
null
),