ignite-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From sboi...@apache.org
Subject [12/53] [abbrv] incubator-ignite git commit: IGNITE-49 merged from sprint-1 into ignite-49
Date Tue, 27 Jan 2015 16:40:32 GMT
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/c7a9a1db/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheEvictionManager.java
----------------------------------------------------------------------
diff --cc modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheEvictionManager.java
index 0000000,ad6a1a2..ff06f9c7
mode 000000,100644..100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheEvictionManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheEvictionManager.java
@@@ -1,0 -1,2037 +1,2040 @@@
+ /*
+  * 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;
+ 
+ import org.apache.ignite.*;
+ import org.apache.ignite.cache.*;
+ import org.apache.ignite.cache.eviction.*;
+ import org.apache.ignite.cluster.*;
+ import org.apache.ignite.events.*;
+ import org.apache.ignite.internal.*;
+ import org.apache.ignite.internal.util.*;
+ import org.apache.ignite.lang.*;
+ import org.apache.ignite.thread.*;
+ import org.apache.ignite.internal.managers.eventstorage.*;
+ import org.apache.ignite.internal.processors.cache.distributed.dht.*;
+ import org.apache.ignite.internal.processors.cache.transactions.*;
+ import org.apache.ignite.internal.processors.timeout.*;
+ import org.apache.ignite.internal.util.future.*;
+ import org.apache.ignite.internal.util.lang.*;
+ import org.apache.ignite.internal.util.tostring.*;
+ import org.apache.ignite.internal.util.typedef.*;
+ import org.apache.ignite.internal.util.typedef.internal.*;
+ import org.apache.ignite.internal.util.worker.*;
+ import org.jdk8.backport.*;
+ import org.jetbrains.annotations.*;
+ import sun.misc.*;
+ 
+ import java.util.*;
+ import java.util.concurrent.*;
+ import java.util.concurrent.atomic.*;
+ import java.util.concurrent.locks.*;
+ import java.util.concurrent.locks.Lock;
+ 
+ import static java.util.concurrent.TimeUnit.*;
+ import static org.apache.ignite.events.IgniteEventType.*;
+ import static org.apache.ignite.cache.CacheMemoryMode.*;
+ import static org.apache.ignite.cache.CacheMode.*;
+ import static org.apache.ignite.internal.processors.cache.GridCacheUtils.*;
+ import static org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtPartitionState.*;
+ import static org.jdk8.backport.ConcurrentLinkedDeque8.*;
+ 
+ /**
+  * Cache eviction manager.
+  */
+ public class GridCacheEvictionManager<K, V> extends GridCacheManagerAdapter<K, V> {
+     /** Unsafe instance. */
+     private static final Unsafe unsafe = GridUnsafe.unsafe();
+ 
+     /** Eviction policy. */
+     private CacheEvictionPolicy<K, V> plc;
+ 
+     /** Eviction filter. */
+     private CacheEvictionFilter<K, V> filter;
+ 
+     /** Eviction buffer. */
+     private final ConcurrentLinkedDeque8<EvictionInfo> bufEvictQ = new ConcurrentLinkedDeque8<>();
+ 
+     /** Attribute name used to queue node in entry metadata. */
+     private final String meta = UUID.randomUUID().toString();
+ 
+     /** Active eviction futures. */
+     private final Map<Long, EvictionFuture> futs = new ConcurrentHashMap8<>();
+ 
+     /** Futures count modification lock. */
+     private final Lock futsCntLock = new ReentrantLock();
+ 
+     /** Futures count condition. */
+     private final Condition futsCntCond = futsCntLock.newCondition();
+ 
+     /** Active futures count. */
+     private volatile int activeFutsCnt;
+ 
+     /** Max active futures count. */
+     private int maxActiveFuts;
+ 
+     /** Generator of future IDs. */
+     private final AtomicLong idGen = new AtomicLong();
+ 
+     /** Evict backup synchronized flag. */
+     private boolean evictSync;
+ 
+     /** Evict near synchronized flag. */
+     private boolean nearSync;
+ 
+     /** Flag to hold {@code evictSync || nearSync} result. */
+     private boolean evictSyncAgr;
+ 
+     /** Policy enabled. */
+     private boolean plcEnabled;
+ 
+     /** */
+     private CacheMemoryMode memoryMode;
+ 
+     /** Backup entries worker. */
+     private BackupWorker backupWorker;
+ 
+     /** Backup entries worker thread. */
+     private IgniteThread backupWorkerThread;
+ 
+     /** Busy lock. */
+     private final GridBusyLock busyLock = new GridBusyLock();
+ 
+     /** Stopping flag. */
+     private volatile boolean stopping;
+ 
+     /** Current future. */
+     private final AtomicReference<EvictionFuture> curEvictFut = new AtomicReference<>();
+ 
+     /** First eviction flag. */
+     private volatile boolean firstEvictWarn;
+ 
+     /** {@inheritDoc} */
+     @Override public void start0() throws IgniteCheckedException {
+         CacheConfiguration cfg = cctx.config();
+ 
+         plc = cctx.isNear() ? cfg.<K, V>getNearEvictionPolicy() : cfg.<K, V>getEvictionPolicy();
+ 
+         memoryMode = cctx.config().getMemoryMode();
+ 
+         plcEnabled = plc != null && memoryMode != OFFHEAP_TIERED;
+ 
+         filter = cfg.getEvictionFilter();
+ 
+         if (cfg.getEvictMaxOverflowRatio() < 0)
+             throw new IgniteCheckedException("Configuration parameter 'maxEvictOverflowRatio' cannot be negative.");
+ 
+         if (cfg.getEvictSynchronizedKeyBufferSize() < 0)
+             throw new IgniteCheckedException("Configuration parameter 'evictSynchronizedKeyBufferSize' cannot be negative.");
+ 
+         if (!cctx.isLocal()) {
+             evictSync = cfg.isEvictSynchronized() && !cctx.isNear() && !cctx.isSwapOrOffheapEnabled();
+ 
+             nearSync = cfg.isEvictNearSynchronized() && isNearEnabled(cctx) && !cctx.isNear();
+         }
+         else {
+             if (cfg.isEvictSynchronized())
+                 U.warn(log, "Ignored 'evictSynchronized' configuration property for LOCAL cache: " + cctx.namexx());
+ 
+             if (cfg.isEvictNearSynchronized())
+                 U.warn(log, "Ignored 'evictNearSynchronized' configuration property for LOCAL cache: " + cctx.namexx());
+         }
+ 
+         if (cctx.isDht() && !nearSync && evictSync && isNearEnabled(cctx))
+             throw new IgniteCheckedException("Illegal configuration (may lead to data inconsistency) " +
+                 "[evictSync=true, evictNearSync=false]");
+ 
+         reportConfigurationProblems();
+ 
+         evictSyncAgr = evictSync || nearSync;
+ 
+         if (evictSync && !cctx.isNear() && plcEnabled) {
+             backupWorker = new BackupWorker();
+ 
+             cctx.events().addListener(
+                 new GridLocalEventListener() {
+                     @Override public void onEvent(IgniteEvent evt) {
+                         assert evt.type() == EVT_NODE_FAILED || evt.type() == EVT_NODE_LEFT ||
+                             evt.type() == EVT_NODE_JOINED;
+ 
+                         IgniteDiscoveryEvent discoEvt = (IgniteDiscoveryEvent)evt;
+ 
+                         // Notify backup worker on each topology change.
+                         if (CU.affinityNode(cctx, discoEvt.eventNode()))
+                             backupWorker.addEvent(discoEvt);
+                     }
+                 },
+                 EVT_NODE_FAILED, EVT_NODE_LEFT, EVT_NODE_JOINED);
+         }
+ 
+         if (evictSyncAgr) {
+             if (cfg.getEvictSynchronizedTimeout() <= 0)
+                 throw new IgniteCheckedException("Configuration parameter 'evictSynchronousTimeout' should be positive.");
+ 
+             if (cfg.getEvictSynchronizedConcurrencyLevel() <= 0)
+                 throw new IgniteCheckedException("Configuration parameter 'evictSynchronousConcurrencyLevel' " +
+                     "should be positive.");
+ 
+             maxActiveFuts = cfg.getEvictSynchronizedConcurrencyLevel();
+ 
+             cctx.io().addHandler(cctx.cacheId(), GridCacheEvictionRequest.class, new CI2<UUID, GridCacheEvictionRequest<K, V>>() {
+                 @Override public void apply(UUID nodeId, GridCacheEvictionRequest<K, V> msg) {
+                     processEvictionRequest(nodeId, msg);
+                 }
+             });
+ 
+             cctx.io().addHandler(cctx.cacheId(), GridCacheEvictionResponse.class, new CI2<UUID, GridCacheEvictionResponse<K, V>>() {
+                 @Override public void apply(UUID nodeId, GridCacheEvictionResponse<K, V> msg) {
+                     processEvictionResponse(nodeId, msg);
+                 }
+             });
+ 
+             cctx.events().addListener(
+                 new GridLocalEventListener() {
+                     @Override public void onEvent(IgniteEvent evt) {
+                         assert evt.type() == EVT_NODE_FAILED || evt.type() == EVT_NODE_LEFT;
+ 
+                         IgniteDiscoveryEvent discoEvt = (IgniteDiscoveryEvent)evt;
+ 
+                         for (EvictionFuture fut : futs.values())
+                             fut.onNodeLeft(discoEvt.eventNode().id());
+                     }
+                 },
+                 EVT_NODE_FAILED, EVT_NODE_LEFT);
+         }
+ 
+         if (log.isDebugEnabled())
+             log.debug("Eviction manager started on node: " + cctx.nodeId());
+     }
+ 
+     /**
+      * Outputs warnings if potential configuration problems are detected.
+      */
+     private void reportConfigurationProblems() {
+         CacheMode mode = cctx.config().getCacheMode();
+ 
+         if (plcEnabled && !cctx.isNear() && mode == PARTITIONED) {
+             if (!evictSync) {
+                 U.warn(log, "Evictions are not synchronized with other nodes in topology " +
+                     "which provides 2x-3x better performance but may cause data inconsistency if cache store " +
+                     "is not configured (consider changing 'evictSynchronized' configuration property).",
+                     "Evictions are not synchronized for cache: " + cctx.namexx());
+             }
+ 
+             if (!nearSync && isNearEnabled(cctx)) {
+                 U.warn(log, "Evictions on primary node are not synchronized with near caches on other nodes " +
+                     "which provides 2x-3x better performance but may cause data inconsistency (consider changing " +
+                     "'nearEvictSynchronized' configuration property).",
+                     "Evictions are not synchronized with near caches on other nodes for cache: " + cctx.namexx());
+             }
+         }
+     }
+ 
+     /** {@inheritDoc} */
+     @Override protected void onKernalStart0() throws IgniteCheckedException {
+         super.onKernalStart0();
+ 
+         if (plcEnabled && evictSync && !cctx.isNear()) {
+             // Add dummy event to worker.
+             ClusterNode locNode = cctx.localNode();
+ 
+             IgniteDiscoveryEvent evt = new IgniteDiscoveryEvent(locNode, "Dummy event.", EVT_NODE_JOINED, locNode);
+ 
+             evt.topologySnapshot(locNode.order(), cctx.discovery().topology(locNode.order()));
+ 
+             backupWorker.addEvent(evt);
+ 
+             backupWorkerThread = new IgniteThread(backupWorker);
+             backupWorkerThread.start();
+         }
+     }
+ 
+     /** {@inheritDoc} */
+     @Override protected void onKernalStop0(boolean cancel) {
+         super.onKernalStop0(cancel);
+ 
+         // Change stopping first.
+         stopping = true;
+ 
+         busyLock.block();
+ 
+         // Stop backup worker.
+         if (evictSync && !cctx.isNear() && backupWorker != null) {
+             backupWorker.cancel();
+ 
+             U.join(backupWorkerThread, log);
+         }
+ 
+         // Cancel all active futures.
+         for (EvictionFuture fut : futs.values())
+             fut.cancel();
+ 
+         if (log.isDebugEnabled())
+             log.debug("Eviction manager stopped on node: " + cctx.nodeId());
+     }
+ 
+     /**
+      * @return Current size of evict queue.
+      */
+     public int evictQueueSize() {
+         return bufEvictQ.sizex();
+     }
+ 
+     /**
+      * @param nodeId Sender node ID.
+      * @param res Response.
+      */
+     private void processEvictionResponse(UUID nodeId, GridCacheEvictionResponse<K, V> res) {
+         assert nodeId != null;
+         assert res != null;
+ 
+         if (log.isDebugEnabled())
+             log.debug("Processing eviction response [node=" + nodeId + ", localNode=" + cctx.nodeId() +
+                 ", res=" + res + ']');
+ 
+         if (!busyLock.enterBusy())
+             return;
+ 
+         try {
+             EvictionFuture fut = futs.get(res.futureId());
+ 
+             if (fut != null)
+                 fut.onResponse(nodeId, res);
+             else if (log.isDebugEnabled())
+                 log.debug("Eviction future for response is not found [res=" + res + ", node=" + nodeId +
+                     ", localNode=" + cctx.nodeId() + ']');
+         }
+         finally {
+             busyLock.leaveBusy();
+         }
+     }
+ 
+     /**
+      * @param nodeId Sender node ID.
+      * @param req Request.
+      */
+     private void processEvictionRequest(UUID nodeId, GridCacheEvictionRequest<K, V> req) {
+         assert nodeId != null;
+         assert req != null;
+ 
+         if (!busyLock.enterBusy())
+             return;
+ 
+         try {
+             if (req.classError() != null) {
+                 if (log.isDebugEnabled())
+                     log.debug("Class got undeployed during eviction: " + req.classError());
+ 
+                 sendEvictionResponse(nodeId, new GridCacheEvictionResponse<K, V>(cctx.cacheId(), req.futureId(), true));
+ 
+                 return;
+             }
+ 
+             long topVer = lockTopology();
+ 
+             try {
+                 if (topVer != req.topologyVersion()) {
+                     if (log.isDebugEnabled())
+                         log.debug("Topology version is different [locTopVer=" + topVer +
+                             ", rmtTopVer=" + req.topologyVersion() + ']');
+ 
+                     sendEvictionResponse(nodeId,
+                         new GridCacheEvictionResponse<K, V>(cctx.cacheId(), req.futureId(), true));
+ 
+                     return;
+                 }
+ 
+                 processEvictionRequest0(nodeId, req);
+             }
+             finally {
+                 unlockTopology();
+             }
+         }
+         finally {
+             busyLock.leaveBusy();
+         }
+     }
+ 
+     /**
+      * @param nodeId Sender node ID.
+      * @param req Request.
+      */
+     private void processEvictionRequest0(UUID nodeId, GridCacheEvictionRequest<K, V> req) {
+         if (log.isDebugEnabled())
+             log.debug("Processing eviction request [node=" + nodeId + ", localNode=" + cctx.nodeId() +
+                 ", reqSize=" + req.entries().size() + ']');
+ 
+         // Partition -> {{Key, Version}, ...}.
+         // Group DHT and replicated cache entries by their partitions.
+         Map<Integer, Collection<GridTuple3<K, GridCacheVersion, Boolean>>> dhtEntries =
+             new HashMap<>();
+ 
+         Collection<GridTuple3<K, GridCacheVersion, Boolean>> nearEntries =
+             new LinkedList<>();
+ 
+         for (GridTuple3<K, GridCacheVersion, Boolean> t : req.entries()) {
+             Boolean near = t.get3();
+ 
+             if (!near) {
+                 // Lock is required.
+                 Collection<GridTuple3<K, GridCacheVersion, Boolean>> col =
+                     F.addIfAbsent(dhtEntries, cctx.affinity().partition(t.get1()),
+                         new LinkedList<GridTuple3<K, GridCacheVersion, Boolean>>());
+ 
+                 assert col != null;
+ 
+                 col.add(t);
+             }
+             else
+                 nearEntries.add(t);
+         }
+ 
+         GridCacheEvictionResponse<K, V> res = new GridCacheEvictionResponse<>(cctx.cacheId(), req.futureId());
+ 
+         GridCacheVersion obsoleteVer = cctx.versions().next();
+ 
+         // DHT and replicated cache entries.
+         for (Map.Entry<Integer, Collection<GridTuple3<K, GridCacheVersion, Boolean>>> e : dhtEntries.entrySet()) {
+             int part = e.getKey();
+ 
+             boolean locked = lockPartition(part); // Will return false if preloading is disabled.
+ 
+             try {
+                 for (GridTuple3<K, GridCacheVersion, Boolean> t : e.getValue()) {
+                     K key = t.get1();
+                     GridCacheVersion ver = t.get2();
+                     Boolean near = t.get3();
+ 
+                     assert !near;
+ 
+                     boolean evicted = evictLocally(key, ver, near, obsoleteVer);
+ 
+                     if (log.isDebugEnabled())
+                         log.debug("Evicted key [key=" + key + ", ver=" + ver + ", near=" + near +
+                             ", evicted=" + evicted +']');
+ 
+                     if (locked && evicted)
+                         // Preloading is in progress, we need to save eviction info.
+                         saveEvictionInfo(key, ver, part);
+ 
+                     if (!evicted)
+                         res.addRejected(key);
+                 }
+             }
+             finally {
+                 if (locked)
+                     unlockPartition(part);
+             }
+         }
+ 
+         // Near entries.
+         for (GridTuple3<K, GridCacheVersion, Boolean> t : nearEntries) {
+             K key = t.get1();
+             GridCacheVersion ver = t.get2();
+             Boolean near = t.get3();
+ 
+             assert near;
+ 
+             boolean evicted = evictLocally(key, ver, near, obsoleteVer);
+ 
+             if (log.isDebugEnabled())
+                 log.debug("Evicted key [key=" + key + ", ver=" + ver + ", near=" + near +
+                     ", evicted=" + evicted +']');
+ 
+             if (!evicted)
+                 res.addRejected(key);
+         }
+ 
+         sendEvictionResponse(nodeId, res);
+     }
+ 
+     /**
+      * @param nodeId Node ID.
+      * @param res Response.
+      */
+     private void sendEvictionResponse(UUID nodeId, GridCacheEvictionResponse<K, V> res) {
+         try {
+             cctx.io().send(nodeId, res);
+ 
+             if (log.isDebugEnabled())
+                 log.debug("Sent eviction response [node=" + nodeId + ", localNode=" + cctx.nodeId() +
+                     ", res" + res + ']');
+         }
+         catch (ClusterTopologyException ignored) {
+             if (log.isDebugEnabled())
+                 log.debug("Failed to send eviction response since initiating node left grid " +
+                     "[node=" + nodeId + ", localNode=" + cctx.nodeId() + ']');
+         }
+         catch (IgniteCheckedException e) {
+             U.error(log, "Failed to send eviction response to node [node=" + nodeId +
+                 ", localNode=" + cctx.nodeId() + ", res" + res + ']', e);
+         }
+     }
+ 
+     /**
+      * @param key Key.
+      * @param ver Version.
+      * @param p Partition ID.
+      */
+     private void saveEvictionInfo(K key, GridCacheVersion ver, int p) {
+         assert cctx.preloadEnabled();
+ 
+         if (!cctx.isNear()) {
+             try {
+                 GridDhtLocalPartition<K, V> part = cctx.dht().topology().localPartition(p, -1, false);
+ 
+                 assert part != null;
+ 
+                 part.onEntryEvicted(key, ver);
+             }
+             catch (GridDhtInvalidPartitionException ignored) {
+                 if (log.isDebugEnabled())
+                     log.debug("Partition does not belong to local node [part=" + p +
+                         ", nodeId" + cctx.localNode().id() + ']');
+             }
+         }
+         else
+             assert false : "Failed to save eviction info: " + cctx.namexx();
+     }
+ 
+     /**
+      * @param p Partition ID.
+      * @return {@code True} if partition has been actually locked,
+      *      {@code false} if preloading is finished or disabled and no lock is needed.
+      */
+     private boolean lockPartition(int p) {
+         if (!cctx.preloadEnabled())
+             return false;
+ 
+         if (!cctx.isNear()) {
+             try {
+                 GridDhtLocalPartition<K, V> part = cctx.dht().topology().localPartition(p, -1, false);
+ 
+                 if (part != null && part.reserve()) {
+                     part.lock();
+ 
+                     if (part.state() != MOVING) {
+                         part.unlock();
+ 
+                         part.release();
+ 
+                         return false;
+                     }
+ 
+                     return true;
+                 }
+             }
+             catch (GridDhtInvalidPartitionException ignored) {
+                 if (log.isDebugEnabled())
+                     log.debug("Partition does not belong to local node [part=" + p +
+                         ", nodeId" + cctx.localNode().id() + ']');
+             }
+         }
+ 
+         // No lock is needed.
+         return false;
+     }
+ 
+     /**
+      * @param p Partition ID.
+      */
+     private void unlockPartition(int p) {
+         if (!cctx.preloadEnabled())
+             return;
+ 
+         if (!cctx.isNear()) {
+             try {
+                 GridDhtLocalPartition<K, V> part = cctx.dht().topology().localPartition(p, -1, false);
+ 
+                 if (part != null) {
+                     part.unlock();
+ 
+                     part.release();
+                 }
+             }
+             catch (GridDhtInvalidPartitionException ignored) {
+                 if (log.isDebugEnabled())
+                     log.debug("Partition does not belong to local node [part=" + p +
+                         ", nodeId" + cctx.localNode().id() + ']');
+             }
+         }
+     }
+ 
+     /**
+      * Locks topology (for DHT cache only) and returns its version.
+      *
+      * @return Topology version after lock.
+      */
+     private long lockTopology() {
+         if (!cctx.isNear()) {
+             cctx.dht().topology().readLock();
+ 
+             return cctx.dht().topology().topologyVersion();
+         }
+ 
+         return 0;
+     }
+ 
+     /**
+      * Unlocks topology.
+      */
+     private void unlockTopology() {
+         if (!cctx.isNear())
+             cctx.dht().topology().readUnlock();
+     }
+ 
+     /**
+      * @param key Key to evict.
+      * @param ver Entry version on initial node.
+      * @param near {@code true} if entry should be evicted from near cache.
+      * @param obsoleteVer Obsolete version.
+      * @return {@code true} if evicted successfully, {@code false} if could not be evicted.
+      */
+     private boolean evictLocally(K key, final GridCacheVersion ver, boolean near, GridCacheVersion obsoleteVer) {
+         assert key != null;
+         assert ver != null;
+         assert obsoleteVer != null;
+         assert evictSyncAgr;
+         assert !cctx.isNear() || cctx.isReplicated();
+ 
+         if (log.isDebugEnabled())
+             log.debug("Evicting key locally [key=" + key + ", ver=" + ver + ", obsoleteVer=" + obsoleteVer +
+                 ", localNode=" + cctx.localNode() + ']');
+ 
+         GridCacheAdapter<K, V> cache = near ? cctx.dht().near() : cctx.cache();
+ 
+         GridCacheEntryEx<K, V> entry = cache.peekEx(key);
+ 
+         if (entry == null)
+             return true;
+ 
+         try {
+             // If entry should be evicted from near cache it can be done safely
+             // without any consistency risks. We don't use filter in this case.
+             // If entry should be evicted from DHT cache, we do not compare versions
+             // as well because versions may change outside the transaction.
+             return evict0(cache, entry, obsoleteVer, null, false);
+         }
+         catch (IgniteCheckedException e) {
+             U.error(log, "Failed to evict entry on remote node [key=" + key + ", localNode=" + cctx.nodeId() + ']', e);
+ 
+             return false;
+         }
+     }
+ 
+     /**
+      * @param cache Cache from which to evict entry.
+      * @param entry Entry to evict.
+      * @param obsoleteVer Obsolete version.
+      * @param filter Filter.
+      * @param explicit If eviction is initiated by user.
+      * @return {@code true} if entry has been evicted.
+      * @throws IgniteCheckedException If failed to evict entry.
+      */
+     private boolean evict0(GridCacheAdapter<K, V> cache, GridCacheEntryEx<K, V> entry, GridCacheVersion obsoleteVer,
+         @Nullable IgnitePredicate<CacheEntry<K, V>>[] filter, boolean explicit) throws IgniteCheckedException {
+         assert cache != null;
+         assert entry != null;
+         assert obsoleteVer != null;
+ 
+         boolean recordable = cctx.events().isRecordable(EVT_CACHE_ENTRY_EVICTED);
+ 
+         V oldVal = recordable ? entry.rawGet() : null;
+ 
+         boolean hasVal = recordable && entry.hasValue();
+ 
+         boolean evicted = entry.evictInternal(cctx.isSwapOrOffheapEnabled(), obsoleteVer, filter);
+ 
+         if (evicted) {
+             // Remove manually evicted entry from policy.
+             if (explicit && plcEnabled)
+                 notifyPolicy(entry);
+ 
+             cache.removeEntry(entry);
+ 
++            if (cache.configuration().isStatisticsEnabled())
++                cache.metrics0().onEvict();
++
+             if (recordable)
+                 cctx.events().addEvent(entry.partition(), entry.key(), cctx.nodeId(), (IgniteUuid)null, null,
+                     EVT_CACHE_ENTRY_EVICTED, null, false, oldVal, hasVal, null, null, null);
+ 
+             if (log.isDebugEnabled())
+                 log.debug("Entry was evicted [entry=" + entry + ", localNode=" + cctx.nodeId() + ']');
+         }
+         else if (log.isDebugEnabled())
+             log.debug("Entry was not evicted [entry=" + entry + ", localNode=" + cctx.nodeId() + ']');
+ 
+         return evicted;
+     }
+ 
+     /**
+      * @param txEntry Transactional entry.
+      */
+     public void touch(IgniteTxEntry<K, V> txEntry, boolean loc) {
+         if (!plcEnabled && memoryMode != OFFHEAP_TIERED)
+             return;
+ 
+         if (!loc) {
+             if (cctx.isNear())
+                 return;
+ 
+             if (evictSync)
+                 return;
+         }
+ 
+         GridCacheEntryEx<K, V> e = txEntry.cached();
+ 
+         if (e.detached() || e.isInternal())
+             return;
+ 
+         try {
+             if (e.markObsoleteIfEmpty(null) || e.obsolete())
+                 e.context().cache().removeEntry(e);
+         }
+         catch (IgniteCheckedException ex) {
+             U.error(log, "Failed to evict entry from cache: " + e, ex);
+         }
+ 
+         if (memoryMode == OFFHEAP_TIERED) {
+             try {
+                 evict0(cctx.cache(), e, cctx.versions().next(), null, false);
+             }
+             catch (IgniteCheckedException ex) {
+                 U.error(log, "Failed to evict entry from on heap memory: " + e, ex);
+             }
+         }
+         else {
+             notifyPolicy(e);
+ 
+             if (evictSyncAgr)
+                 waitForEvictionFutures();
+         }
+     }
+ 
+     /**
+      * @param e Entry for eviction policy notification.
+      * @param topVer Topology version.
+      */
+     public void touch(GridCacheEntryEx<K, V> e, long topVer) {
+         if (e.detached() || e.isInternal())
+             return;
+ 
+         try {
+             if (e.markObsoleteIfEmpty(null) || e.obsolete())
+                 e.context().cache().removeEntry(e);
+         }
+         catch (IgniteCheckedException ex) {
+             U.error(log, "Failed to evict entry from cache: " + e, ex);
+         }
+ 
+         if (memoryMode == OFFHEAP_TIERED) {
+             try {
+                 evict0(cctx.cache(), e, cctx.versions().next(), null, false);
+             }
+             catch (IgniteCheckedException ex) {
+                 U.error(log, "Failed to evict entry from on heap memory: " + e, ex);
+             }
+ 
+             return;
+         }
+ 
+         if (!plcEnabled)
+             return;
+ 
+         // Don't track non-primary entries if evicts are synchronized.
+         if (!cctx.isNear() && evictSync && !cctx.affinity().primary(cctx.localNode(), e.partition(), topVer))
+             return;
+ 
+         if (!busyLock.enterBusy())
+             return;
+ 
+         try {
+             // Wait for futures to finish.
+             if (evictSyncAgr)
+                 waitForEvictionFutures();
+ 
+             if (log.isDebugEnabled())
+                 log.debug("Touching entry [entry=" + e + ", localNode=" + cctx.nodeId() + ']');
+ 
+             notifyPolicy(e);
+         }
+         finally {
+             busyLock.leaveBusy();
+         }
+     }
+ 
+     /**
+      * @param e Entry for eviction policy notification.
+      */
+     private void touch0(GridCacheEntryEx<K, V> e) {
+         assert evictSyncAgr;
+         assert plcEnabled;
+ 
+         // Do not wait for futures here since only limited number
+         // of entries can be passed to this method.
+         notifyPolicy(e);
+     }
+ 
+     /**
+      * @param entries Entries for eviction policy notification.
+      */
+     private void touchOnTopologyChange(Iterable<? extends GridCacheEntryEx<K, V>> entries) {
+         assert evictSync;
+         assert plcEnabled;
+ 
+         if (log.isDebugEnabled())
+             log.debug("Touching entries [entries=" + entries + ", localNode=" + cctx.nodeId() + ']');
+ 
+         for (GridCacheEntryEx<K, V> e : entries) {
+             if (e.key() instanceof GridCacheInternal)
+                 // Skip internal entry.
+                 continue;
+ 
+             // Do not wait for futures here since only limited number
+             // of entries can be passed to this method.
+             notifyPolicy(e);
+         }
+     }
+ 
+     /**
+      * Warns on first eviction.
+      */
+     private void warnFirstEvict() {
+         // Do not move warning output to synchronized block (it causes warning in IDE).
+         synchronized (this) {
+             if (firstEvictWarn)
+                 return;
+ 
+             firstEvictWarn = true;
+         }
+ 
+         U.warn(log, "Evictions started (cache may have reached its capacity)." +
+             " You may wish to increase 'maxSize' on eviction policy being used for cache: " + cctx.name(),
+             "Evictions started (cache may have reached its capacity): " + cctx.name());
+     }
+ 
+     /**
+      * @return {@code True} if either evicts or near evicts are synchronized, {@code false} otherwise.
+      */
+     public boolean evictSyncOrNearSync() {
+         return evictSyncAgr;
+     }
+ 
+     /**
+      * @param entry Entry to attempt to evict.
+      * @param obsoleteVer Obsolete version.
+      * @param filter Optional entry filter.
+      * @param explicit {@code True} if evict is called explicitly, {@code false} if it's called
+      *      from eviction policy.
+      * @return {@code True} if entry was marked for eviction.
+      * @throws IgniteCheckedException In case of error.
+      */
+     public boolean evict(@Nullable GridCacheEntryEx<K, V> entry, @Nullable GridCacheVersion obsoleteVer,
+         boolean explicit, @Nullable IgnitePredicate<CacheEntry<K, V>>[] filter) throws IgniteCheckedException {
+         if (entry == null)
+             return true;
+ 
+         // Do not evict internal entries.
+         if (entry.key() instanceof GridCacheInternal)
+             return false;
+ 
+         if (!cctx.isNear() && !explicit && !firstEvictWarn)
+             warnFirstEvict();
+ 
+         if (evictSyncAgr) {
+             assert !cctx.isNear(); // Make sure cache is not NEAR.
+ 
+             if (entry.wrap(false).backup() && evictSync)
+                 // Do not track backups if evicts are synchronized.
+                 return !explicit;
+ 
+             try {
+                 if (!cctx.isAll(entry, filter))
+                     return false;
+ 
+                 if (entry.lockedByAny())
+                     return false;
+ 
+                 // Add entry to eviction queue.
+                 enqueue(entry, filter);
+             }
+             catch (GridCacheEntryRemovedException ignored) {
+                 if (log.isDebugEnabled())
+                     log.debug("Entry got removed while evicting [entry=" + entry +
+                         ", localNode=" + cctx.nodeId() + ']');
+             }
+         }
+         else {
+             if (obsoleteVer == null)
+                 obsoleteVer = cctx.versions().next();
+ 
+             // Do not touch entry if not evicted:
+             // 1. If it is call from policy, policy tracks it on its own.
+             // 2. If it is explicit call, entry is touched on tx commit.
+             return evict0(cctx.cache(), entry, obsoleteVer, filter, explicit);
+         }
+ 
+         return true;
+     }
+ 
+     /**
+      * @param keys Keys to evict.
+      * @param obsoleteVer Obsolete version.
+      * @throws IgniteCheckedException In case of error.
+      */
+     public void batchEvict(Collection<? extends K> keys, @Nullable GridCacheVersion obsoleteVer) throws IgniteCheckedException {
+         assert !evictSyncAgr;
+         assert cctx.isSwapOrOffheapEnabled();
+ 
+         List<GridCacheEntryEx<K, V>> locked = new ArrayList<>(keys.size());
+ 
+         Collection<GridCacheBatchSwapEntry<K, V>> swapped = new ArrayList<>(keys.size());
+ 
+         boolean recordable = cctx.events().isRecordable(EVT_CACHE_ENTRY_EVICTED);
+ 
+         GridCacheAdapter<K, V> cache = cctx.cache();
+ 
+         Map<K, GridCacheEntryEx<K, V>> cached = U.newHashMap(keys.size());
+ 
+         // Get all participating entries to avoid deadlock.
+         for (K k : keys)
+             cached.put(k, cache.peekEx(k));
+ 
+         try {
+             for (K key : keys) {
+                 GridCacheEntryEx<K, V> entry = cached.get(key);
+ 
+                 // Do not evict internal entries.
+                 if (entry == null || entry.key() instanceof GridCacheInternal)
+                     continue;
+ 
+                 // Lock entry.
+                 unsafe.monitorEnter(entry);
+ 
+                 locked.add(entry);
+ 
+                 if (obsoleteVer == null)
+                     obsoleteVer = cctx.versions().next();
+ 
+                 GridCacheBatchSwapEntry<K, V> swapEntry = entry.evictInBatchInternal(obsoleteVer);
+ 
+                 if (swapEntry != null) {
+                     swapped.add(swapEntry);
+ 
+                     if (log.isDebugEnabled())
+                         log.debug("Entry was evicted [entry=" + entry + ", localNode=" + cctx.nodeId() + ']');
+                 }
+             }
+ 
+             // Batch write to swap.
+             if (swapped != null)
+                 cctx.swap().writeAll(swapped);
+         }
+         finally {
+             // Unlock entries in reverse order.
+             for (ListIterator<GridCacheEntryEx<K, V>> it = locked.listIterator(locked.size()); it.hasPrevious();) {
+                 GridCacheEntryEx<K, V> e = it.previous();
+ 
+                 unsafe.monitorExit(e);
+             }
+ 
+             // Remove entries and fire events outside the locks.
+             for (GridCacheEntryEx<K, V> entry : locked) {
+                 if (entry.obsolete()) {
+                     entry.onMarkedObsolete();
+ 
+                     cache.removeEntry(entry);
+ 
+                     if (plcEnabled)
+                         notifyPolicy(entry);
+ 
+                     if (recordable)
+                         cctx.events().addEvent(entry.partition(), entry.key(), cctx.nodeId(), (IgniteUuid)null, null,
+                             EVT_CACHE_ENTRY_EVICTED, null, false, entry.rawGet(), entry.hasValue(), null, null, null);
+                 }
+             }
+         }
+     }
+ 
+     /**
+      * Enqueues entry for synchronized eviction.
+      *
+      * @param entry Entry.
+      * @param filter Filter.
+      * @throws GridCacheEntryRemovedException If entry got removed.
+      */
+     private void enqueue(GridCacheEntryEx<K, V> entry, IgnitePredicate<CacheEntry<K, V>>[] filter)
+         throws GridCacheEntryRemovedException {
+         Node<EvictionInfo> node = entry.meta(meta);
+ 
+         if (node == null) {
+             node = bufEvictQ.addLastx(new EvictionInfo(entry, entry.version(), filter));
+ 
+             if (entry.putMetaIfAbsent(meta, node) != null)
+                 // Was concurrently added, need to clear it from queue.
+                 bufEvictQ.unlinkx(node);
+             else if (log.isDebugEnabled())
+                 log.debug("Added entry to eviction queue: " + entry);
+         }
+     }
+ 
+     /**
+      * Checks eviction queue.
+      */
+     private void checkEvictionQueue() {
+         int maxSize = maxQueueSize();
+ 
+         int bufSize = bufEvictQ.sizex();
+ 
+         if (bufSize >= maxSize) {
+             if (log.isDebugEnabled())
+                 log.debug("Processing eviction queue: " + bufSize);
+ 
+             Collection<EvictionInfo> evictInfos = new ArrayList<>(bufSize);
+ 
+             for (int i = 0; i < bufSize; i++) {
+                 EvictionInfo info = bufEvictQ.poll();
+ 
+                 if (info == null)
+                     break;
+ 
+                 evictInfos.add(info);
+             }
+ 
+             if (!evictInfos.isEmpty())
+                 addToCurrentFuture(evictInfos);
+         }
+     }
+ 
+     /**
+      * @return Max queue size.
+      */
+     private int maxQueueSize() {
+         int size = (int)(cctx.cache().size() * cctx.config().getEvictMaxOverflowRatio()) / 100;
+ 
+         if (size <= 0)
+             size = 500;
+ 
+         return Math.min(size, cctx.config().getEvictSynchronizedKeyBufferSize());
+     }
+ 
+     /**
+      * Processes eviction queue (sends required requests, etc.).
+      *
+      * @param evictInfos Eviction information to create future with.
+      */
+     private void addToCurrentFuture(Collection<EvictionInfo> evictInfos) {
+         assert !evictInfos.isEmpty();
+ 
+         while (true) {
+             EvictionFuture fut = curEvictFut.get();
+ 
+             if (fut == null) {
+                 curEvictFut.compareAndSet(null, new EvictionFuture(cctx.kernalContext()));
+ 
+                 continue;
+             }
+ 
+             if (fut.prepareLock()) {
+                 boolean added;
+ 
+                 try {
+                     added = fut.add(evictInfos);
+                 }
+                 finally {
+                     fut.prepareUnlock();
+                 }
+ 
+                 if (added) {
+                     if (fut.prepare()) {
+                         // Thread that prepares future should remove it and install listener.
+                         curEvictFut.compareAndSet(fut, null);
+ 
+                         fut.listenAsync(new CI1<IgniteFuture<?>>() {
+                             @Override public void apply(IgniteFuture<?> f) {
+                                 if (!busyLock.enterBusy()) {
+                                     if (log.isDebugEnabled())
+                                         log.debug("Will not notify eviction future completion (grid is stopping): " +
+                                             f);
+ 
+                                     return;
+                                 }
+ 
+                                 try {
+                                     long topVer = lockTopology();
+ 
+                                     try {
+                                         onFutureCompleted((EvictionFuture)f, topVer);
+                                     }
+                                     finally {
+                                         unlockTopology();
+                                     }
+                                 }
+                                 finally {
+                                     busyLock.leaveBusy();
+                                 }
+                             }
+                         });
+                     }
+ 
+                     break;
+                 }
+                 else
+                     // Infos were not added, create another future for next iteration.
+                     curEvictFut.compareAndSet(fut, new EvictionFuture(cctx.kernalContext()));
+             }
+             else
+                 // Future has not been locked, create another future for next iteration.
+                 curEvictFut.compareAndSet(fut, new EvictionFuture(cctx.kernalContext()));
+         }
+     }
+ 
+     /**
+      * @param fut Completed eviction future.
+      * @param topVer Topology version on future complete.
+      */
+     private void onFutureCompleted(EvictionFuture fut, long topVer) {
+         if (!busyLock.enterBusy())
+             return;
+ 
+         try {
+             IgniteBiTuple<Collection<EvictionInfo>, Collection<EvictionInfo>> t;
+ 
+             try {
+                 t = fut.get();
+             }
+             catch (IgniteFutureCancelledException ignored) {
+                 assert false : "Future has been cancelled, but manager is not stopping: " + fut;
+ 
+                 return;
+             }
+             catch (IgniteCheckedException e) {
+                 U.error(log, "Eviction future finished with error (all entries will be touched): " + fut, e);
+ 
+                 if (plcEnabled) {
+                     for (EvictionInfo info : fut.entries())
+                         touch0(info.entry());
+                 }
+ 
+                 return;
+             }
+ 
+             // Check if topology version is different.
+             if (fut.topologyVersion() != topVer) {
+                 if (log.isDebugEnabled())
+                     log.debug("Topology has changed, all entries will be touched: " + fut);
+ 
+                 if (plcEnabled) {
+                     for (EvictionInfo info : fut.entries())
+                         touch0(info.entry());
+                 }
+ 
+                 return;
+             }
+ 
+             // Evict remotely evicted entries.
+             GridCacheVersion obsoleteVer = null;
+ 
+             Collection<EvictionInfo> evictedEntries = t.get1();
+ 
+             for (EvictionInfo info : evictedEntries) {
+                 GridCacheEntryEx<K, V> entry = info.entry();
+ 
+                 try {
+                     // Remove readers on which the entry was evicted.
+                     for (IgniteBiTuple<ClusterNode, Long> r : fut.evictedReaders(entry.key())) {
+                         UUID readerId = r.get1().id();
+                         Long msgId = r.get2();
+ 
+                         ((GridDhtCacheEntry<K, V>)entry).removeReader(readerId, msgId);
+                     }
+ 
+                     if (obsoleteVer == null)
+                         obsoleteVer = cctx.versions().next();
+ 
+                     // Do not touch primary entries, if not evicted.
+                     // They will be touched within updating transactions.
+                     evict0(cctx.cache(), entry, obsoleteVer, versionFilter(info), false);
+                 }
+                 catch (IgniteCheckedException e) {
+                     U.error(log, "Failed to evict entry [entry=" + entry +
+                         ", localNode=" + cctx.nodeId() + ']', e);
+                 }
+                 catch (GridCacheEntryRemovedException ignored) {
+                     if (log.isDebugEnabled())
+                         log.debug("Entry was concurrently removed while evicting [entry=" + entry +
+                             ", localNode=" + cctx.nodeId() + ']');
+                 }
+             }
+ 
+             Collection<EvictionInfo> rejectedEntries = t.get2();
+ 
+             // Touch remotely rejected entries (only if policy is enabled).
+             if (plcEnabled && !rejectedEntries.isEmpty()) {
+                 for (EvictionInfo info : rejectedEntries)
+                     touch0(info.entry());
+             }
+         }
+         finally {
+             busyLock.leaveBusy();
+ 
+             // Signal on future completion.
+             signal();
+         }
+     }
+ 
+     /**
+      * This method should be called when eviction future is processed
+      * and unwind may continue.
+      */
+     private void signal() {
+         futsCntLock.lock();
+ 
+         try {
+             // Avoid volatile read on assertion.
+             int cnt = --activeFutsCnt;
+ 
+             assert cnt >= 0 : "Invalid futures count: " + cnt;
+ 
+             if (cnt < maxActiveFuts)
+                 futsCntCond.signalAll();
+         }
+         finally {
+             futsCntLock.unlock();
+         }
+     }
+ 
+     /**
+      * @param info Eviction info.
+      * @return Version aware filter.
+      */
+     private IgnitePredicate<CacheEntry<K, V>>[] versionFilter(final EvictionInfo info) {
+         // If version has changed since we started the whole process
+         // then we should not evict entry.
+         return cctx.vararg(new P1<CacheEntry<K, V>>() {
+             @Override public boolean apply(CacheEntry<K, V> e) {
+                 GridCacheVersion ver = (GridCacheVersion)e.version();
+ 
+                 return info.version().equals(ver) && F.isAll(info.filter());
+             }
+         });
+     }
+ 
+     /**
+      * Gets a collection of nodes to send eviction requests to.
+      *
+      *
+      * @param entry Entry.
+      * @param topVer Topology version.
+      * @return Tuple of two collections: dht (in case of partitioned cache) nodes
+      *      and readers (empty for replicated cache).
+      * @throws GridCacheEntryRemovedException If entry got removed during method
+      *      execution.
+      */
+     @SuppressWarnings( {"IfMayBeConditional"})
+     private IgniteBiTuple<Collection<ClusterNode>, Collection<ClusterNode>> remoteNodes(GridCacheEntryEx<K, V> entry,
+         long topVer)
+         throws GridCacheEntryRemovedException {
+         assert entry != null;
+ 
+         assert cctx.config().getCacheMode() != LOCAL;
+ 
+         Collection<ClusterNode> backups;
+ 
+         if (evictSync)
+             backups = F.view(cctx.dht().topology().nodes(entry.partition(), topVer), F0.notEqualTo(cctx.localNode()));
+         else
+             backups = Collections.emptySet();
+ 
+         Collection<ClusterNode> readers;
+ 
+         if (nearSync) {
+             readers = F.transform(((GridDhtCacheEntry<K, V>)entry).readers(), new C1<UUID, ClusterNode>() {
+                 @Nullable @Override public ClusterNode apply(UUID nodeId) {
+                     return cctx.node(nodeId);
+                 }
+             });
+         }
+         else
+             readers = Collections.emptySet();
+ 
+         return new IgnitePair<>(backups, readers);
+     }
+ 
+     /**
+      * Notifications.
+      */
+     public void unwind() {
+         if (!evictSyncAgr)
+             return;
+ 
+         if (!busyLock.enterBusy())
+             return;
+ 
+         try {
+             checkEvictionQueue();
+         }
+         finally {
+             busyLock.leaveBusy();
+         }
+     }
+ 
+     /**
+      * @param e Entry to notify eviction policy.
+      */
+     @SuppressWarnings({"IfMayBeConditional", "RedundantIfStatement"})
+     private void notifyPolicy(GridCacheEntryEx<K, V> e) {
+         assert plcEnabled;
+         assert plc != null;
+         assert !e.isInternal() : "Invalid entry for policy notification: " + e;
+ 
+         if (log.isDebugEnabled())
+             log.debug("Notifying eviction policy with entry: " + e);
+ 
+         if (filter == null || filter.evictAllowed(e.wrap(false)))
+             plc.onEntryAccessed(e.obsoleteOrDeleted(), e.evictWrap());
+     }
+ 
+     /**
+      *
+      */
+     @SuppressWarnings("TooBroadScope")
+     private void waitForEvictionFutures() {
+         if (activeFutsCnt >= maxActiveFuts) {
+             boolean interrupted = false;
+ 
+             futsCntLock.lock();
+ 
+             try {
+                 while(!stopping && activeFutsCnt >= maxActiveFuts) {
+                     try {
+                         futsCntCond.await(2000, MILLISECONDS);
+                     }
+                     catch (InterruptedException ignored) {
+                         interrupted = true;
+                     }
+                 }
+             }
+             finally {
+                 futsCntLock.unlock();
+ 
+                 if (interrupted)
+                     Thread.currentThread().interrupt();
+             }
+         }
+     }
+ 
+     /**
+      * Prints out eviction stats.
+      */
+     public void printStats() {
+         X.println("Eviction stats [grid=" + cctx.gridName() + ", cache=" + cctx.cache().name() +
+             ", buffEvictQ=" + bufEvictQ.sizex() + ']');
+     }
+ 
+     /** {@inheritDoc} */
+     @Override public void printMemoryStats() {
+         X.println(">>> ");
+         X.println(">>> Eviction manager memory stats [grid=" + cctx.gridName() + ", cache=" + cctx.name() + ']');
+         X.println(">>>   buffEvictQ size: " + bufEvictQ.sizex());
+         X.println(">>>   futsSize: " + futs.size());
+         X.println(">>>   futsCreated: " + idGen.get());
+     }
+ 
+     /**
+      *
+      */
+     private class BackupWorker extends GridWorker {
+         /** */
+         private final BlockingQueue<IgniteDiscoveryEvent> evts = new LinkedBlockingQueue<>();
+ 
+         /** */
+         private final Collection<Integer> primaryParts = new HashSet<>();
+ 
+         /**
+          *
+          */
+         private BackupWorker() {
+             super(cctx.gridName(), "cache-eviction-backup-worker", log);
+ 
+             assert plcEnabled;
+         }
+ 
+         /**
+          * @param evt New event.
+          */
+         void addEvent(IgniteDiscoveryEvent evt) {
+             assert evt != null;
+ 
+             evts.add(evt);
+         }
+ 
+         /** {@inheritDoc} */
+         @Override protected void body() throws InterruptedException, IgniteInterruptedException {
+             try {
+                 assert !cctx.isNear() && evictSync;
+ 
+                 ClusterNode loc = cctx.localNode();
+ 
+                 // Initialize.
+                 primaryParts.addAll(cctx.affinity().primaryPartitions(cctx.localNodeId(),
+                     cctx.affinity().affinityTopologyVersion()));
+ 
+                 while (!isCancelled()) {
+                     IgniteDiscoveryEvent evt = evts.take();
+ 
+                     if (log.isDebugEnabled())
+                         log.debug("Processing event: " + evt);
+ 
+                     // Remove partitions that are no longer primary.
+                     for (Iterator<Integer> it = primaryParts.iterator(); it.hasNext();) {
+                         if (!evts.isEmpty())
+                             break;
+ 
+                         if (!cctx.affinity().primary(loc, it.next(), evt.topologyVersion()))
+                             it.remove();
+                     }
+ 
+                     // Move on to next event.
+                     if (!evts.isEmpty())
+                         continue;
+ 
+                     for (GridDhtLocalPartition<K, V> part : cctx.topology().localPartitions()) {
+                         if (!evts.isEmpty())
+                             break;
+ 
+                         if (part.primary(evt.topologyVersion()) && primaryParts.add(part.id())) {
+                             if (log.isDebugEnabled())
+                                 log.debug("Touching partition entries: " + part);
+ 
+                             touchOnTopologyChange(part.entries());
+                         }
+                     }
+                 }
+             }
+             catch (InterruptedException ignored) {
+                 // No-op.
+             }
+             catch (IgniteException e) {
+                 if (!e.hasCause(InterruptedException.class))
+                     throw e;
+             }
+         }
+     }
+ 
+     /**
+      * Wrapper around an entry to be put into queue.
+      */
+     private class EvictionInfo {
+         /** Cache entry. */
+         private GridCacheEntryEx<K, V> entry;
+ 
+         /** Start version. */
+         private GridCacheVersion ver;
+ 
+         /** Filter to pass before entry will be evicted. */
+         private IgnitePredicate<CacheEntry<K, V>>[] filter;
+ 
+         /**
+          * @param entry Entry.
+          * @param ver Version.
+          * @param filter Filter.
+          */
+         EvictionInfo(GridCacheEntryEx<K, V> entry, GridCacheVersion ver,
+             IgnitePredicate<CacheEntry<K, V>>[] filter) {
+             assert entry != null;
+             assert ver != null;
+ 
+             this.entry = entry;
+             this.ver = ver;
+             this.filter = filter;
+         }
+ 
+         /**
+          * @return Entry.
+          */
+         GridCacheEntryEx<K, V> entry() {
+             return entry;
+         }
+ 
+         /**
+          * @return Version.
+          */
+         GridCacheVersion version() {
+             return ver;
+         }
+ 
+         /**
+          * @return Filter.
+          */
+         IgnitePredicate<CacheEntry<K, V>>[] filter() {
+             return filter;
+         }
+ 
+         /** {@inheritDoc} */
+         @Override public String toString() {
+             return S.toString(EvictionInfo.class, this);
+         }
+     }
+ 
+     /**
+      * Future for synchronized eviction. Result is a tuple: {evicted entries, rejected entries}.
+      */
+     private class EvictionFuture extends GridFutureAdapter<IgniteBiTuple<Collection<EvictionInfo>,
+                 Collection<EvictionInfo>>> {
+         /** */
+         private static final long serialVersionUID = 0L;
+ 
+         /** */
+         private final long id = idGen.incrementAndGet();
+ 
+         /** */
+         private ConcurrentLinkedDeque8<EvictionInfo> evictInfos = new ConcurrentLinkedDeque8<>();
+ 
+         /** */
+         private final ConcurrentMap<K, EvictionInfo> entries = new ConcurrentHashMap8<>();
+ 
+         /** */
+         private final ConcurrentMap<K, Collection<ClusterNode>> readers =
+             new ConcurrentHashMap8<>();
+ 
+         /** */
+         private final Collection<EvictionInfo> evictedEntries = new GridConcurrentHashSet<>();
+ 
+         /** */
+         private final ConcurrentMap<K, EvictionInfo> rejectedEntries = new ConcurrentHashMap8<>();
+ 
+         /** Request map. */
+         private final ConcurrentMap<UUID, GridCacheEvictionRequest<K, V>> reqMap =
+             new ConcurrentHashMap8<>();
+ 
+         /** Response map. */
+         private final ConcurrentMap<UUID, GridCacheEvictionResponse<K, V>> resMap =
+             new ConcurrentHashMap8<>();
+ 
+         /** To make sure that future is completing within a single thread. */
+         private final AtomicBoolean finishPrepare = new AtomicBoolean();
+ 
+         /** Lock to use when future is being initialized. */
+         @GridToStringExclude
+         private final ReadWriteLock prepareLock = new ReentrantReadWriteLock();
+ 
+         /** To make sure that future is completing within a single thread. */
+         private final AtomicBoolean completing = new AtomicBoolean();
+ 
+         /** Lock to use after future is initialized. */
+         @GridToStringExclude
+         private final ReadWriteLock lock = new ReentrantReadWriteLock();
+ 
+         /** Object to force future completion on elapsing eviction timeout. */
+         @GridToStringExclude
+         private GridTimeoutObject timeoutObj;
+ 
+         /** Topology version future is processed on. */
+         private long topVer;
+ 
+         /**
+          * @param ctx Context.
+          */
+         EvictionFuture(GridKernalContext ctx) {
+             super(ctx);
+         }
+ 
+         /**
+          * Required by {@code Externalizable}.
+          */
+         public EvictionFuture() {
+             assert false : "This should never happen.";
+         }
+ 
+         /**
+          * @return {@code True} if prepare lock was acquired.
+          */
+         boolean prepareLock() {
+             return prepareLock.readLock().tryLock();
+         }
+ 
+         /**
+          *
+          */
+         void prepareUnlock() {
+             prepareLock.readLock().unlock();
+         }
+ 
+         /**
+          * @param infos Infos to add.
+          * @return {@code False} if entries were not added due to capacity restrictions.
+          */
+         boolean add(Collection<EvictionInfo> infos) {
+             assert infos != null && !infos.isEmpty();
+ 
+             if (evictInfos.sizex() > maxQueueSize())
+                 return false;
+ 
+             evictInfos.addAll(infos);
+ 
+             return true;
+         }
+ 
+         /**
+          * @return {@code True} if future has been prepared by this call.
+          */
+         @SuppressWarnings("LockAcquiredButNotSafelyReleased")
+         boolean prepare() {
+             if (evictInfos.sizex() >= maxQueueSize() && finishPrepare.compareAndSet(false, true)) {
+                 // Lock will never be released intentionally.
+                 prepareLock.writeLock().lock();
+ 
+                 futsCntLock.lock();
+ 
+                 try {
+                     activeFutsCnt++;
+                 }
+                 finally {
+                     futsCntLock.unlock();
+                 }
+ 
+                 // Put future in map.
+                 futs.put(id, this);
+ 
+                 prepare0();
+ 
+                 return true;
+             }
+ 
+             return false;
+         }
+ 
+         /**
+          * Prepares future (sends all required requests).
+          */
+         private void prepare0() {
+             if (log.isDebugEnabled())
+                 log.debug("Preparing eviction future [futId=" + id + ", localNode=" + cctx.nodeId() +
+                     ", infos=" + evictInfos + ']');
+ 
+             assert evictInfos != null && !evictInfos.isEmpty();
+ 
+             topVer = lockTopology();
+ 
+             try {
+                 Collection<EvictionInfo> locals = null;
+ 
+                 for (EvictionInfo info : evictInfos) {
+                     // Queue node may have been stored in entry metadata concurrently, but we don't care
+                     // about it since we are currently processing this entry.
+                     Node<EvictionInfo> queueNode = info.entry().removeMeta(meta);
+ 
+                     if (queueNode != null)
+                         bufEvictQ.unlinkx(queueNode);
+ 
+                     IgniteBiTuple<Collection<ClusterNode>, Collection<ClusterNode>> tup;
+ 
+                     try {
+                         tup = remoteNodes(info.entry(), topVer);
+                     }
+                     catch (GridCacheEntryRemovedException ignored) {
+                         if (log.isDebugEnabled())
+                             log.debug("Entry got removed while preparing eviction future (ignoring) [entry=" +
+                                 info.entry() + ", nodeId=" + cctx.nodeId() + ']');
+ 
+                         continue;
+                     }
+ 
+                     Collection<ClusterNode> entryReaders =
+                         F.addIfAbsent(readers, info.entry().key(), new GridConcurrentHashSet<ClusterNode>());
+ 
+                     assert entryReaders != null;
+ 
+                     // Add entry readers so that we could remove them right before local eviction.
+                     entryReaders.addAll(tup.get2());
+ 
+                     Collection<ClusterNode> nodes = F.concat(true, tup.get1(), tup.get2());
+ 
+                     if (!nodes.isEmpty()) {
+                         entries.put(info.entry().key(), info);
+ 
+                         // There are remote participants.
+                         for (ClusterNode node : nodes) {
+                             GridCacheEvictionRequest<K, V> req = F.addIfAbsent(reqMap, node.id(),
+                                 new GridCacheEvictionRequest<K, V>(cctx.cacheId(), id, evictInfos.size(), topVer));
+ 
+                             assert req != null;
+ 
+                             req.addKey(info.entry().key(), info.version(), entryReaders.contains(node));
+                         }
+                     }
+                     else {
+                         if (locals == null)
+                             locals = new HashSet<>(evictInfos.size(), 1.0f);
+ 
+                         // There are no remote participants, need to keep the entry as local.
+                         locals.add(info);
+                     }
+                 }
+ 
+                 if (locals != null) {
+                     // Evict entries without remote participant nodes immediately.
+                     GridCacheVersion obsoleteVer = cctx.versions().next();
+ 
+                     for (EvictionInfo info : locals) {
+                         if (log.isDebugEnabled())
+                             log.debug("Evicting key without remote participant nodes: " + info);
+ 
+                         try {
+                             // Touch primary entry (without backup nodes) if not evicted
+                             // to keep tracking.
+                             if (!evict0(cctx.cache(), info.entry(), obsoleteVer, versionFilter(info), false) &&
+                                 plcEnabled)
+                                 touch0(info.entry());
+                         }
+                         catch (IgniteCheckedException e) {
+                             U.error(log, "Failed to evict entry: " + info.entry(), e);
+                         }
+                     }
+                 }
+ 
+                 // If there were only local entries.
+                 if (entries.isEmpty()) {
+                     complete(false);
+ 
+                     return;
+                 }
+             }
+             finally {
+                 unlockTopology();
+             }
+ 
+             // Send eviction requests.
+             for (Map.Entry<UUID, GridCacheEvictionRequest<K, V>> e : reqMap.entrySet()) {
+                 UUID nodeId = e.getKey();
+ 
+                 GridCacheEvictionRequest<K, V> req = e.getValue();
+ 
+                 if (log.isDebugEnabled())
+                     log.debug("Sending eviction request [node=" + nodeId + ", req=" + req + ']');
+ 
+                 try {
+                     cctx.io().send(nodeId, req);
+                 }
+                 catch (ClusterTopologyException ignored) {
+                     // Node left the topology.
+                     onNodeLeft(nodeId);
+                 }
+                 catch (IgniteCheckedException ex) {
+                     U.error(log, "Failed to send eviction request to node [node=" + nodeId + ", req=" + req + ']', ex);
+ 
+                     rejectEntries(nodeId);
+                 }
+             }
+ 
+             registerTimeoutObject();
+         }
+ 
+         /**
+          *
+          */
+         private void registerTimeoutObject() {
+             // Check whether future has not been completed yet.
+             if (lock.readLock().tryLock()) {
+                 try {
+                     timeoutObj = new GridTimeoutObjectAdapter(cctx.config().getEvictSynchronizedTimeout()) {
+                         @Override public void onTimeout() {
+                             complete(true);
+                         }
+                     };
+ 
+                     cctx.time().addTimeoutObject(timeoutObj);
+                 }
+                 finally {
+                     lock.readLock().unlock();
+                 }
+             }
+         }
+ 
+         /**
+          * @return Future ID.
+          */
+         long id() {
+             return id;
+         }
+ 
+         /**
+          * @return Topology version.
+          */
+         long topologyVersion() {
+             return topVer;
+         }
+ 
+         /**
+          * @return Keys to readers mapping.
+          */
+         Map<K, Collection<ClusterNode>> readers() {
+             return readers;
+         }
+ 
+         /**
+          * @return All entries associated with future that should be evicted (or rejected).
+          */
+         Collection<EvictionInfo> entries() {
+             return entries.values();
+         }
+ 
+         /**
+          * Reject all entries on behalf of specified node.
+          *
+          * @param nodeId Node ID.
+          */
+         private void rejectEntries(UUID nodeId) {
+             assert nodeId != null;
+ 
+             if (lock.readLock().tryLock()) {
+                 try {
+                     if (log.isDebugEnabled())
+                         log.debug("Rejecting entries for node: " + nodeId);
+ 
+                     GridCacheEvictionRequest<K, V> req = reqMap.remove(nodeId);
+ 
+                     for (GridTuple3<K, GridCacheVersion, Boolean> t : req.entries()) {
+                         EvictionInfo info = entries.get(t.get1());
+ 
+                         assert info != null;
+ 
+                         rejectedEntries.put(t.get1(), info);
+                     }
+                 }
+                 finally {
+                     lock.readLock().unlock();
+                 }
+             }
+ 
+             checkDone();
+         }
+ 
+         /**
+          * @param nodeId Node id that left the topology.
+          */
+         void onNodeLeft(UUID nodeId) {
+             assert nodeId != null;
+ 
+             if (lock.readLock().tryLock()) {
+                 try {
+                     // Stop waiting response from this node.
+                     reqMap.remove(nodeId);
+ 
+                     resMap.remove(nodeId);
+                 }
+                 finally {
+                     lock.readLock().unlock();
+                 }
+ 
+                 checkDone();
+             }
+         }
+ 
+         /**
+          * @param nodeId Sender node ID.
+          * @param res Response.
+          */
+         void onResponse(UUID nodeId, GridCacheEvictionResponse<K, V> res) {
+             assert nodeId != null;
+             assert res != null;
+ 
+             if (lock.readLock().tryLock()) {
+                 try {
+                     if (log.isDebugEnabled())
+                         log.debug("Entered to eviction future onResponse() [fut=" + this + ", node=" + nodeId +
+                             ", res=" + res + ']');
+ 
+                     ClusterNode node = cctx.node(nodeId);
+ 
+                     if (node != null)
+                         resMap.put(nodeId, res);
+                     else
+                         // Sender node left grid.
+                         reqMap.remove(nodeId);
+                 }
+                 finally {
+                     lock.readLock().unlock();
+                 }
+ 
+                 if (res.error())
+                     // Complete future, since there was a class loading error on at least one node.
+                     complete(false);
+                 else
+                     checkDone();
+             }
+             else {
+                 if (log.isDebugEnabled())
+                     log.debug("Ignored eviction response [fut=" + this + ", node=" + nodeId + ", res=" + res + ']');
+             }
+         }
+ 
+         /**
+          *
+          */
+         private void checkDone() {
+             if (reqMap.isEmpty() || resMap.keySet().containsAll(reqMap.keySet()))
+                 complete(false);
+         }
+ 
+         /**
+          * Completes future.
+          *
+          * @param timedOut {@code True} if future is being forcibly completed on timeout.
+          */
+         @SuppressWarnings({"LockAcquiredButNotSafelyReleased"})
+         private void complete(boolean timedOut) {
+             if (completing.compareAndSet(false, true)) {
+                 // Lock will never be released intentionally.
+                 lock.writeLock().lock();
+ 
+                 futs.remove(id);
+ 
+                 if (timeoutObj != null && !timedOut)
+                     // If future is timed out, corresponding object is already removed.
+                     cctx.time().removeTimeoutObject(timeoutObj);
+ 
+                 if (log.isDebugEnabled())
+                     log.debug("Building eviction future result [fut=" + this + ", timedOut=" + timedOut + ']');
+ 
+                 boolean err = F.forAny(resMap.values(), new P1<GridCacheEvictionResponse<K, V>>() {
+                     @Override public boolean apply(GridCacheEvictionResponse<K, V> res) {
+                         return res.error();
+                     }
+                 });
+ 
+                 if (err) {
+                     Collection<UUID> ids = F.view(resMap.keySet(), new P1<UUID>() {
+                         @Override public boolean apply(UUID e) {
+                             return resMap.get(e).error();
+                         }
+                     });
+ 
+                     assert !ids.isEmpty();
+ 
+                     U.warn(log, "Remote node(s) failed to process eviction request " +
+                         "due to topology changes " +
+                         "(some backup or remote values maybe lost): " + ids);
+                 }
+ 
+                 if (timedOut)
+                     U.warn(log, "Timed out waiting for eviction future " +
+                         "(consider changing 'evictSynchronousTimeout' and 'evictSynchronousConcurrencyLevel' " +
+                         "configuration parameters).");
+ 
+                 if (err || timedOut) {
+                     // Future has not been completed successfully, all entries should be rejected.
+                     assert evictedEntries.isEmpty();
+ 
+                     rejectedEntries.putAll(entries);
+                 }
+                 else {
+                     // Copy map to filter remotely rejected entries,
+                     // as they will be touched within corresponding txs.
+                     Map<K, EvictionInfo> rejectedEntries0 = new HashMap<>(rejectedEntries);
+ 
+                     // Future has been completed successfully - build result.
+                     for (EvictionInfo info : entries.values()) {
+                         K key = info.entry().key();
+ 
+                         if (rejectedEntries0.containsKey(key))
+                             // Was already rejected.
+                             continue;
+ 
+                         boolean rejected = false;
+ 
+                         for (GridCacheEvictionResponse<K, V> res : resMap.values()) {
+                             if (res.rejectedKeys().contains(key)) {
+                                 // Modify copied map.
+                                 rejectedEntries0.put(key, info);
+ 
+                                 rejected = true;
+ 
+                                 break;
+                             }
+                         }
+ 
+                         if (!rejected)
+                             evictedEntries.add(info);
+                     }
+                 }
+ 
+                 // Pass entries that were rejected due to topology changes
+                 // or due to timeout or class loading issues.
+                 // Remotely rejected entries will be touched within corresponding txs.
+                 onDone(F.t(evictedEntries, rejectedEntries.values()));
+             }
+         }
+ 
+         /**
+          * @param key Key.
+          * @return Reader nodes on which given key was evicted.
+          */
+         Collection<IgniteBiTuple<ClusterNode, Long>> evictedReaders(K key) {
+             Collection<ClusterNode> mappedReaders = readers.get(key);
+ 
+             if (mappedReaders == null)
+                 return Collections.emptyList();
+ 
+             Collection<IgniteBiTuple<ClusterNode, Long>> col = new LinkedList<>();
+ 
+             for (Map.Entry<UUID, GridCacheEvictionResponse<K, V>> e : resMap.entrySet()) {
+                 ClusterNode node = cctx.node(e.getKey());
+ 
+                 // If node has left or response did not arrive from near node
+                 // then just skip it.
+                 if (node == null || !mappedReaders.contains(node))
+                     continue;
+ 
+                 GridCacheEvictionResponse<K, V> res = e.getValue();
+ 
+                 if (!res.rejectedKeys().contains(key))
+                     col.add(F.t(node, res.messageId()));
+             }
+ 
+             return col;
+         }
+ 
+         /** {@inheritDoc} */
+         @SuppressWarnings("LockAcquiredButNotSafelyReleased")
+         @Override public boolean cancel() {
+             if (completing.compareAndSet(false, true)) {
+                 // Lock will never be released intentionally.
+                 lock.writeLock().lock();
+ 
+                 if (timeoutObj != null)
+                     cctx.time().removeTimeoutObject(timeoutObj);
+ 
+                 boolean b = onCancelled();
+ 
+                 assert b;
+ 
+                 return true;
+             }
+ 
+             return false;
+         }
+ 
+         /** {@inheritDoc} */
+         @Override public String toString() {
+             return S.toString(EvictionFuture.class, this);
+         }
+     }
+ }


Mime
View raw message