ignite-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From sboi...@apache.org
Subject [05/53] [abbrv] incubator-ignite git commit: IGNITE-49 merged from sprint-1 into ignite-49
Date Tue, 27 Jan 2015 16:40:25 GMT
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/c7a9a1db/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearGetFuture.java
----------------------------------------------------------------------
diff --cc modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearGetFuture.java
index 0000000,f2e564b..0e4e59b
mode 000000,100644..100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearGetFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearGetFuture.java
@@@ -1,0 -1,866 +1,868 @@@
+ /*
+  * 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.near;
+ 
+ import org.apache.ignite.*;
+ import org.apache.ignite.cache.*;
+ import org.apache.ignite.cluster.*;
+ import org.apache.ignite.internal.processors.cache.*;
+ import org.apache.ignite.internal.util.*;
+ import org.apache.ignite.lang.*;
+ 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.tostring.*;
+ import org.apache.ignite.internal.util.typedef.*;
+ import org.apache.ignite.internal.util.typedef.internal.*;
+ import org.jetbrains.annotations.*;
+ 
+ import java.io.*;
+ import java.util.*;
+ import java.util.concurrent.atomic.*;
+ 
+ import static org.apache.ignite.IgniteSystemProperties.*;
+ import static org.apache.ignite.transactions.IgniteTxIsolation.*;
+ import static org.apache.ignite.internal.processors.cache.GridCacheUtils.*;
+ 
+ /**
+  *
+  */
+ public final class GridNearGetFuture<K, V> extends GridCompoundIdentityFuture<Map<K, V>>
+     implements GridCacheFuture<Map<K, V>> {
+     /** */
+     private static final long serialVersionUID = 0L;
+ 
+     /** Default max remap count value. */
+     public static final int DFLT_MAX_REMAP_CNT = 3;
+ 
+     /** Logger reference. */
+     private static final AtomicReference<IgniteLogger> logRef = new AtomicReference<>();
+ 
+     /** Maximum number of attempts to remap key to the same primary node. */
+     private static final int MAX_REMAP_CNT = getInteger(GG_NEAR_GET_MAX_REMAPS, DFLT_MAX_REMAP_CNT);
+ 
+     /** Context. */
+     private GridCacheContext<K, V> cctx;
+ 
+     /** Keys. */
+     private Collection<? extends K> keys;
+ 
+     /** Reload flag. */
+     private boolean reload;
+ 
+     /** Read through flag. */
+     private boolean readThrough;
+ 
+     /** Force primary flag. */
+     private boolean forcePrimary;
+ 
+     /** Future ID. */
+     private IgniteUuid futId;
+ 
+     /** Version. */
+     private GridCacheVersion ver;
+ 
+     /** Transaction. */
+     private IgniteTxLocalEx<K, V> tx;
+ 
+     /** Filters. */
+     private IgnitePredicate<CacheEntry<K, V>>[] filters;
+ 
+     /** Logger. */
+     private IgniteLogger log;
+ 
+     /** Trackable flag. */
+     private boolean trackable;
+ 
+     /** Remap count. */
+     private AtomicInteger remapCnt = new AtomicInteger();
+ 
+     /** Subject ID. */
+     private UUID subjId;
+ 
+     /** Task name. */
+     private String taskName;
+ 
+     /** Whether to deserialize portable objects. */
+     private boolean deserializePortable;
+ 
+     /** Expiry policy. */
+     private IgniteCacheExpiryPolicy expiryPlc;
+ 
+     /**
+      * Empty constructor required for {@link Externalizable}.
+      */
+     public GridNearGetFuture() {
+         // No-op.
+     }
+ 
+     /**
+      * @param cctx Context.
+      * @param keys Keys.
+      * @param readThrough Read through flag.
+      * @param reload Reload flag.
+      * @param forcePrimary If {@code true} get will be performed on primary node even if
+      *      called on backup node.
+      * @param tx Transaction.
+      * @param filters Filters.
+      * @param subjId Subject ID.
+      * @param taskName Task name.
+      * @param deserializePortable Deserialize portable flag.
+      * @param expiryPlc Expiry policy.
+      */
+     public GridNearGetFuture(
+         GridCacheContext<K, V> cctx,
+         Collection<? extends K> keys,
+         boolean readThrough,
+         boolean reload,
+         boolean forcePrimary,
+         @Nullable IgniteTxLocalEx<K, V> tx,
+         @Nullable IgnitePredicate<CacheEntry<K, V>>[] filters,
+         @Nullable UUID subjId,
+         String taskName,
+         boolean deserializePortable,
+         @Nullable IgniteCacheExpiryPolicy expiryPlc
+     ) {
+         super(cctx.kernalContext(), CU.<K, V>mapsReducer(keys.size()));
+ 
+         assert !F.isEmpty(keys);
+ 
+         this.cctx = cctx;
+         this.keys = keys;
+         this.readThrough = readThrough;
+         this.reload = reload;
+         this.forcePrimary = forcePrimary;
+         this.filters = filters;
+         this.tx = tx;
+         this.subjId = subjId;
+         this.taskName = taskName;
+         this.deserializePortable = deserializePortable;
+         this.expiryPlc = expiryPlc;
+ 
+         futId = IgniteUuid.randomUuid();
+ 
+         ver = tx == null ? cctx.versions().next() : tx.xidVersion();
+ 
+         log = U.logger(ctx, logRef, GridNearGetFuture.class);
+     }
+ 
+     /**
+      * Initializes future.
+      */
+     public void init() {
+         long topVer = tx == null ? cctx.affinity().affinityTopologyVersion() : tx.topologyVersion();
+ 
+         map(keys, Collections.<ClusterNode, LinkedHashMap<K, Boolean>>emptyMap(), topVer);
+ 
+         markInitialized();
+     }
+ 
+     /** {@inheritDoc} */
+     @Override public boolean trackable() {
+         return trackable;
+     }
+ 
+     /** {@inheritDoc} */
+     @Override public void markNotTrackable() {
+         // Should not flip trackable flag from true to false since get future can be remapped.
+     }
+ 
+     /**
+      * @return Keys.
+      */
+     Collection<? extends K> keys() {
+         return keys;
+     }
+ 
+     /** {@inheritDoc} */
+     @Override public IgniteUuid futureId() {
+         return futId;
+     }
+ 
+     /** {@inheritDoc} */
+     @Override public GridCacheVersion version() {
+         return ver;
+     }
+ 
+     /** {@inheritDoc} */
+     @Override public Collection<? extends ClusterNode> nodes() {
+         return
+             F.viewReadOnly(futures(), new IgniteClosure<IgniteFuture<Map<K, V>>, ClusterNode>() {
+                 @Nullable @Override public ClusterNode apply(IgniteFuture<Map<K, V>> f) {
+                     if (isMini(f))
+                         return ((MiniFuture)f).node();
+ 
+                     return cctx.discovery().localNode();
+                 }
+             });
+     }
+ 
+     /** {@inheritDoc} */
+     @Override public boolean onNodeLeft(UUID nodeId) {
+         for (IgniteFuture<Map<K, V>> fut : futures())
+             if (isMini(fut)) {
+                 MiniFuture f = (MiniFuture)fut;
+ 
+                 if (f.node().id().equals(nodeId)) {
+                     f.onResult(new ClusterTopologyException("Remote node left grid (will retry): " + nodeId));
+ 
+                     return true;
+                 }
+             }
+ 
+         return false;
+     }
+ 
+     /**
+      * @param nodeId Sender.
+      * @param res Result.
+      */
+     void onResult(UUID nodeId, GridNearGetResponse<K, V> res) {
+         for (IgniteFuture<Map<K, V>> fut : futures())
+             if (isMini(fut)) {
+                 MiniFuture f = (MiniFuture)fut;
+ 
+                 if (f.futureId().equals(res.miniId())) {
+                     assert f.node().id().equals(nodeId);
+ 
+                     f.onResult(res);
+                 }
+             }
+     }
+ 
+     /** {@inheritDoc} */
+     @Override public boolean onDone(Map<K, V> res, Throwable err) {
+         if (super.onDone(res, err)) {
+             // Don't forget to clean up.
+             if (trackable)
+                 cctx.mvcc().removeFuture(this);
+ 
+             cache().dht().sendTtlUpdateRequest(expiryPlc);
+ 
+             return true;
+         }
+ 
+         return false;
+     }
+ 
+     /**
+      * @param f Future.
+      * @return {@code True} if mini-future.
+      */
+     private boolean isMini(IgniteFuture<Map<K, V>> f) {
+         return f.getClass().equals(MiniFuture.class);
+     }
+ 
+     /**
+      * @param keys Keys.
+      * @param mapped Mappings to check for duplicates.
+      * @param topVer Topology version to map on.
+      */
+     private void map(Collection<? extends K> keys,
+         Map<ClusterNode, LinkedHashMap<K, Boolean>> mapped,
+         final long topVer) {
+         Collection<ClusterNode> affNodes = CU.affinityNodes(cctx, topVer);
+ 
+         if (affNodes.isEmpty()) {
+             assert !isAffinityNode(cctx.config());
+ 
+             onDone(new ClusterTopologyException("Failed to map keys for near-only cache (all partition " +
+                 "nodes left the grid)."));
+ 
+             return;
+         }
+ 
+         Map<ClusterNode, LinkedHashMap<K, Boolean>> mappings = U.newHashMap(affNodes.size());
+ 
+         Map<K, GridCacheVersion> savedVers = null;
+ 
+         // Assign keys to primary nodes.
+         for (K key : keys) {
+             if (key != null)
+                 savedVers = map(key, mappings, topVer, mapped, savedVers);
+         }
+ 
+         if (isDone())
+             return;
+ 
+         final Map<K, GridCacheVersion> saved = savedVers;
+ 
+         final int keysSize = keys.size();
+ 
+         // Create mini futures.
+         for (Map.Entry<ClusterNode, LinkedHashMap<K, Boolean>> entry : mappings.entrySet()) {
+             final ClusterNode n = entry.getKey();
+ 
+             final LinkedHashMap<K, Boolean> mappedKeys = entry.getValue();
+ 
+             assert !mappedKeys.isEmpty();
+ 
+             // If this is the primary or backup node for the keys.
+             if (n.isLocal()) {
+                 final GridDhtFuture<Collection<GridCacheEntryInfo<K, V>>> fut =
+                     dht().getDhtAsync(n.id(),
+                         -1,
+                         mappedKeys,
+                         readThrough,
+                         reload,
+                         topVer,
+                         subjId,
+                         taskName == null ? 0 : taskName.hashCode(),
+                         deserializePortable,
+                         filters,
+                         expiryPlc);
+ 
+                 final Collection<Integer> invalidParts = fut.invalidPartitions();
+ 
+                 if (!F.isEmpty(invalidParts)) {
+                     Collection<K> remapKeys = new ArrayList<>(keysSize);
+ 
+                     for (K key : keys) {
+                         if (key != null && invalidParts.contains(cctx.affinity().partition(key)))
+                             remapKeys.add(key);
+                     }
+ 
+                     long updTopVer = ctx.discovery().topologyVersion();
+ 
+                     assert updTopVer > topVer : "Got invalid partitions for local node but topology version did " +
+                         "not change [topVer=" + topVer + ", updTopVer=" + updTopVer +
+                         ", invalidParts=" + invalidParts + ']';
+ 
+                     // Remap recursively.
+                     map(remapKeys, mappings, updTopVer);
+                 }
+ 
+                 // Add new future.
+                 add(fut.chain(new C1<IgniteFuture<Collection<GridCacheEntryInfo<K, V>>>, Map<K, V>>() {
+                     @Override public Map<K, V> apply(IgniteFuture<Collection<GridCacheEntryInfo<K, V>>> fut) {
+                         try {
+                             return loadEntries(n.id(), mappedKeys.keySet(), fut.get(), saved, topVer);
+                         }
+                         catch (Exception e) {
+                             U.error(log, "Failed to get values from dht cache [fut=" + fut + "]", e);
+ 
+                             onDone(e);
+ 
+                             return Collections.emptyMap();
+                         }
+                     }
+                 }));
+             }
+             else {
+                 if (!trackable) {
+                     trackable = true;
+ 
+                     cctx.mvcc().addFuture(this);
+                 }
+ 
+                 MiniFuture fut = new MiniFuture(n, mappedKeys, saved, topVer);
+ 
+                 GridCacheMessage<K, V> req = new GridNearGetRequest<>(
+                     cctx.cacheId(),
+                     futId,
+                     fut.futureId(),
+                     ver,
+                     mappedKeys,
+                     readThrough,
+                     reload,
+                     topVer,
+                     filters,
+                     subjId,
+                     taskName == null ? 0 : taskName.hashCode(),
+                     expiryPlc != null ? expiryPlc.forAccess() : -1L);
+ 
+                 add(fut); // Append new future.
+ 
+                 try {
+                     cctx.io().send(n, req);
+                 }
+                 catch (IgniteCheckedException e) {
+                     // Fail the whole thing.
+                     if (e instanceof ClusterTopologyException)
+                         fut.onResult((ClusterTopologyException)e);
+                     else
+                         fut.onResult(e);
+                 }
+             }
+         }
+     }
+ 
+     /**
+      * @param mappings Mappings.
+      * @param key Key to map.
+      * @param topVer Topology version
+      * @param mapped Previously mapped.
+      * @param savedVers Saved versions.
+      * @return Map.
+      */
+     private Map<K, GridCacheVersion> map(K key, Map<ClusterNode, LinkedHashMap<K, Boolean>> mappings,
+         long topVer, Map<ClusterNode, LinkedHashMap<K, Boolean>> mapped, Map<K, GridCacheVersion> savedVers) {
+         final GridNearCacheAdapter<K, V> near = cache();
+ 
+         // Allow to get cached value from the local node.
+         boolean allowLocRead = !forcePrimary || cctx.affinity().primary(cctx.localNode(), key, topVer);
+ 
+         GridCacheEntryEx<K, V> entry = allowLocRead ? near.peekEx(key) : null;
+ 
+         while (true) {
+             try {
+                 V v = null;
+ 
+                 boolean isNear = entry != null;
+ 
+                 // First we peek into near cache.
+                 if (isNear)
+                     v = entry.innerGet(tx,
+                         /*swap*/false,
+                         /*read-through*/false,
+                         /*fail-fast*/true,
+                         /*unmarshal*/true,
+                         /*metrics*/true,
+                         /*events*/true,
+                         /*temporary*/false,
+                         subjId,
+                         null,
+                         taskName,
+                         filters,
+                         expiryPlc);
+ 
+                 ClusterNode primary = null;
+ 
+                 if (v == null && allowLocRead) {
+                     GridDhtCacheAdapter<K, V> dht = cache().dht();
+ 
+                     try {
+                         entry = dht.context().isSwapOrOffheapEnabled() ? dht.entryEx(key) : dht.peekEx(key);
+ 
+                         // If near cache does not have value, then we peek DHT cache.
+                         if (entry != null) {
+                             boolean isNew = entry.isNewLocked() || !entry.valid(topVer);
+ 
+                             v = entry.innerGet(tx,
+                                 /*swap*/true,
+                                 /*read-through*/false,
+                                 /*fail-fast*/true,
+                                 /*unmarshal*/true,
+                                 /*update-metrics*/false,
+                                 /*events*/!isNear,
+                                 /*temporary*/false,
+                                 subjId,
+                                 null,
+                                 taskName,
+                                 filters,
+                                 expiryPlc);
+ 
+                             // Entry was not in memory or in swap, so we remove it from cache.
+                             if (v == null && isNew && entry.markObsoleteIfEmpty(ver))
+                                 dht.removeIfObsolete(key);
+                         }
+ 
 -                        if (v != null)
 -                            near.metrics0().onRead(true);
++                        if (v != null) {
++                            if (cctx.cache().configuration().isStatisticsEnabled())
++                                near.metrics0().onRead(true);
++                        }
+                         else {
+                             primary = cctx.affinity().primary(key, topVer);
+ 
 -                            if (!primary.isLocal())
++                            if (!primary.isLocal() && cctx.cache().configuration().isStatisticsEnabled())
+                                 near.metrics0().onRead(false);
+                         }
+                     }
+                     catch (GridDhtInvalidPartitionException ignored) {
+                         // No-op.
+                     }
+                     finally {
+                         if (entry != null && (tx == null || (!tx.implicit() && tx.isolation() == READ_COMMITTED))) {
+                             dht.context().evicts().touch(entry, topVer);
+ 
+                             entry = null;
+                         }
+                     }
+                 }
+ 
+                 if (v != null && !reload) {
+                     if (cctx.portableEnabled())
+                         v = (V)cctx.unwrapPortableIfNeeded(v, !deserializePortable);
+ 
+                     add(new GridFinishedFuture<>(cctx.kernalContext(), Collections.singletonMap(key, v)));
+                 }
+                 else {
+                     if (primary == null)
+                         primary = cctx.affinity().primary(key, topVer);
+ 
+                     GridNearCacheEntry<K, V> nearEntry = allowLocRead ? near.peekExx(key) : null;
+ 
+                     entry = nearEntry;
+ 
+                     if (savedVers == null)
+                         savedVers = U.newHashMap(3);
+ 
+                     savedVers.put(key, nearEntry == null ? null : nearEntry.dhtVersion());
+ 
+                     LinkedHashMap<K, Boolean> keys = mapped.get(primary);
+ 
+                     if (keys != null && keys.containsKey(key)) {
+                         if (remapCnt.incrementAndGet() > MAX_REMAP_CNT) {
+                             onDone(new ClusterTopologyException("Failed to remap key to a new node after " + MAX_REMAP_CNT
+                                 + " attempts (key got remapped to the same node) [key=" + key + ", node=" +
+                                 U.toShortString(primary) + ", mappings=" + mapped + ']'));
+ 
+                             return savedVers;
+                         }
+                     }
+ 
+                     // Don't add reader if transaction acquires lock anyway to avoid deadlock.
+                     boolean addRdr = tx == null || tx.optimistic();
+ 
+                     if (!addRdr && tx.readCommitted() && !tx.writeSet().contains(key))
+                         addRdr = true;
+ 
+                     LinkedHashMap<K, Boolean> old = mappings.get(primary);
+ 
+                     if (old == null)
+                         mappings.put(primary, old = new LinkedHashMap<>(3, 1f));
+ 
+                     old.put(key, addRdr);
+                 }
+ 
+                 break;
+             }
+             catch (IgniteCheckedException e) {
+                 onDone(e);
+ 
+                 break;
+             }
+             catch (GridCacheEntryRemovedException ignored) {
+                 entry = allowLocRead ? near.peekEx(key) : null;
+             }
+             catch (GridCacheFilterFailedException e) {
+                 if (log.isDebugEnabled())
+                     log.debug("Filter validation failed for entry: " + e);
+ 
+                 break;
+             }
+             finally {
+                 if (entry != null && !reload && tx == null)
+                     cctx.evicts().touch(entry, topVer);
+             }
+         }
+ 
+         return savedVers;
+     }
+ 
+     /**
+      * @return Near cache.
+      */
+     private GridNearCacheAdapter<K, V> cache() {
+         return (GridNearCacheAdapter<K, V>)cctx.cache();
+     }
+ 
+     /**
+      * @return DHT cache.
+      */
+     private GridDhtCacheAdapter<K, V> dht() {
+         return cache().dht();
+     }
+ 
+     /**
+      * @param nodeId Node id.
+      * @param keys Keys.
+      * @param infos Entry infos.
+      * @param savedVers Saved versions.
+      * @param topVer Topology version
+      * @return Result map.
+      */
+     private Map<K, V> loadEntries(UUID nodeId,
+         Collection<K> keys,
+         Collection<GridCacheEntryInfo<K, V>> infos,
+         Map<K, GridCacheVersion> savedVers,
+         long topVer) {
+         boolean empty = F.isEmpty(keys);
+ 
+         Map<K, V> map = empty ? Collections.<K, V>emptyMap() : new GridLeanMap<K, V>(keys.size());
+ 
+         if (!empty) {
+             boolean atomic = cctx.atomic();
+ 
+             GridCacheVersion ver = atomic ? null : F.isEmpty(infos) ? null : cctx.versions().next();
+ 
+             for (GridCacheEntryInfo<K, V> info : infos) {
+                 try {
+                     info.unmarshalValue(cctx, cctx.deploy().globalLoader());
+ 
+                     // Entries available locally in DHT should not be loaded into near cache for reading.
+                     if (!cctx.cache().affinity().isPrimaryOrBackup(cctx.localNode(), info.key())) {
+                         GridNearCacheEntry<K, V> entry = cache().entryExx(info.key(), topVer);
+ 
+                         GridCacheVersion saved = savedVers.get(info.key());
+ 
+                         // Load entry into cache.
+                         entry.loadedValue(tx,
+                             nodeId,
+                             info.value(),
+                             info.valueBytes(),
+                             atomic ? info.version() : ver,
+                             info.version(),
+                             saved,
+                             info.ttl(),
+                             info.expireTime(),
+                             true,
+                             topVer,
+                             subjId);
+ 
+                         cctx.evicts().touch(entry, topVer);
+                     }
+ 
+                     V val = info.value();
+ 
+                     if (cctx.portableEnabled())
+                         val = (V)cctx.unwrapPortableIfNeeded(val, !deserializePortable);
+ 
+                     map.put(info.key(), val);
+                 }
+                 catch (GridCacheEntryRemovedException ignore) {
+                     if (log.isDebugEnabled())
+                         log.debug("Got removed entry while processing get response (will not retry).");
+                 }
+                 catch (IgniteCheckedException e) {
+                     // Fail.
+                     onDone(e);
+ 
+                     return Collections.emptyMap();
+                 }
+             }
+         }
+ 
+         return map;
+     }
+ 
+     /**
+      * Mini-future for get operations. Mini-futures are only waiting on a single
+      * node as opposed to multiple nodes.
+      */
+     private class MiniFuture extends GridFutureAdapter<Map<K, V>> {
+         /** */
+         private static final long serialVersionUID = 0L;
+ 
+         /** */
+         private final IgniteUuid futId = IgniteUuid.randomUuid();
+ 
+         /** Node ID. */
+         private ClusterNode node;
+ 
+         /** Keys. */
+         @GridToStringInclude
+         private LinkedHashMap<K, Boolean> keys;
+ 
+         /** Saved entry versions. */
+         private Map<K, GridCacheVersion> savedVers;
+ 
+         /** Topology version on which this future was mapped. */
+         private long topVer;
+ 
+         /**
+          * Empty constructor required for {@link Externalizable}.
+          */
+         public MiniFuture() {
+             // No-op.
+         }
+ 
+         /**
+          * @param node Node.
+          * @param keys Keys.
+          * @param savedVers Saved entry versions.
+          * @param topVer Topology version.
+          */
+         MiniFuture(ClusterNode node, LinkedHashMap<K, Boolean> keys, Map<K, GridCacheVersion> savedVers, long topVer) {
+             super(cctx.kernalContext());
+ 
+             this.node = node;
+             this.keys = keys;
+             this.savedVers = savedVers;
+             this.topVer = topVer;
+         }
+ 
+         /**
+          * @return Future ID.
+          */
+         IgniteUuid futureId() {
+             return futId;
+         }
+ 
+         /**
+          * @return Node ID.
+          */
+         public ClusterNode node() {
+             return node;
+         }
+ 
+         /**
+          * @return Keys.
+          */
+         public Collection<K> keys() {
+             return keys.keySet();
+         }
+ 
+         /**
+          * @param e Error.
+          */
+         void onResult(Throwable e) {
+             if (log.isDebugEnabled())
+                 log.debug("Failed to get future result [fut=" + this + ", err=" + e + ']');
+ 
+             // Fail.
+             onDone(e);
+         }
+ 
+         /**
+          * @param e Topology exception.
+          */
+         void onResult(ClusterTopologyException e) {
+             if (log.isDebugEnabled())
+                 log.debug("Remote node left grid while sending or waiting for reply (will retry): " + this);
+ 
+             long updTopVer = ctx.discovery().topologyVersion();
+ 
+             if (updTopVer > topVer) {
+                 // Remap.
+                 map(keys.keySet(), F.t(node, keys), updTopVer);
+ 
+                 onDone(Collections.<K, V>emptyMap());
+             }
+             else {
+                 final RemapTimeoutObject timeout = new RemapTimeoutObject(ctx.config().getNetworkTimeout(), topVer, e);
+ 
+                 ctx.discovery().topologyFuture(topVer + 1).listenAsync(new CI1<IgniteFuture<Long>>() {
+                     @Override public void apply(IgniteFuture<Long> longIgniteFuture) {
+                         if (timeout.finish()) {
+                             ctx.timeout().removeTimeoutObject(timeout);
+ 
+                             // Remap.
+                             map(keys.keySet(), F.t(node, keys), cctx.affinity().affinityTopologyVersion());
+ 
+                             onDone(Collections.<K, V>emptyMap());
+                         }
+                     }
+                 });
+ 
+                 ctx.timeout().addTimeoutObject(timeout);
+             }
+         }
+ 
+         /**
+          * @param res Result callback.
+          */
+         void onResult(final GridNearGetResponse<K, V> res) {
+             final Collection<Integer> invalidParts = res.invalidPartitions();
+ 
+             // If error happened on remote node, fail the whole future.
+             if (res.error() != null) {
+                 onDone(res.error());
+ 
+                 return;
+             }
+ 
+             // Remap invalid partitions.
+             if (!F.isEmpty(invalidParts)) {
+                 long rmtTopVer = res.topologyVersion();
+ 
+                 assert rmtTopVer != 0;
+ 
+                 if (rmtTopVer <= topVer) {
+                     // Fail the whole get future.
+                     onDone(new IgniteCheckedException("Failed to process invalid partitions response (remote node reported " +
+                         "invalid partitions but remote topology version does not differ from local) " +
+                         "[topVer=" + topVer + ", rmtTopVer=" + rmtTopVer + ", invalidParts=" + invalidParts +
+                         ", nodeId=" + node.id() + ']'));
+ 
+                     return;
+                 }
+ 
+                 if (log.isDebugEnabled())
+                     log.debug("Remapping mini get future [invalidParts=" + invalidParts + ", fut=" + this + ']');
+ 
+                 // Need to wait for next topology version to remap.
+                 IgniteFuture<Long> topFut = ctx.discovery().topologyFuture(rmtTopVer);
+ 
+                 topFut.listenAsync(new CIX1<IgniteFuture<Long>>() {
+                     @Override public void applyx(IgniteFuture<Long> fut) throws IgniteCheckedException {
+                         long readyTopVer = fut.get();
+ 
+                         // This will append new futures to compound list.
+                         map(F.view(keys.keySet(), new P1<K>() {
+                             @Override public boolean apply(K key) {
+                                 return invalidParts.contains(cctx.affinity().partition(key));
+                             }
+                         }), F.t(node, keys), readyTopVer);
+ 
+                         // It is critical to call onDone after adding futures to compound list.
+                         onDone(loadEntries(node.id(), keys.keySet(), res.entries(), savedVers, topVer));
+                     }
+                 });
+             }
+             else
+                 onDone(loadEntries(node.id(), keys.keySet(), res.entries(), savedVers, topVer));
+         }
+ 
+         /** {@inheritDoc} */
+         @Override public String toString() {
+             return S.toString(MiniFuture.class, this);
+         }
+ 
+         /**
+          * Remap timeout object.
+          */
+         private class RemapTimeoutObject extends GridTimeoutObjectAdapter {
+             /** Finished flag. */
+             private AtomicBoolean finished = new AtomicBoolean();
+ 
+             /** Topology version to wait. */
+             private long topVer;
+ 
+             /** Exception cause. */
+             private IgniteCheckedException e;
+ 
+             /**
+              * @param timeout Timeout.
+              * @param topVer Topology version timeout was created on.
+              */
+             private RemapTimeoutObject(long timeout, long topVer, IgniteCheckedException e) {
+                 super(timeout);
+ 
+                 this.topVer = topVer;
+                 this.e = e;
+             }
+ 
+             /** {@inheritDoc} */
+             @Override public void onTimeout() {
+                 if (finish())
+                     // Fail the whole get future.
+                     onDone(new IgniteCheckedException("Failed to wait for topology version to change: " + (topVer + 1), e));
+                 // else remap happened concurrently.
+             }
+ 
+             /**
+              * @return Guard against concurrent completion.
+              */
+             public boolean finish() {
+                 return finished.compareAndSet(false, true);
+             }
+         }
+     }
+ }

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/c7a9a1db/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearLockFuture.java
----------------------------------------------------------------------
diff --cc modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearLockFuture.java
index 0000000,3dec521..2409335
mode 000000,100644..100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearLockFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearLockFuture.java
@@@ -1,0 -1,1468 +1,1470 @@@
+ /*
+  * 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.near;
+ 
+ import org.apache.ignite.*;
+ import org.apache.ignite.cache.*;
+ import org.apache.ignite.cluster.*;
+ import org.apache.ignite.internal.processors.cache.*;
+ import org.apache.ignite.internal.processors.cache.distributed.*;
+ import org.apache.ignite.lang.*;
+ import org.apache.ignite.transactions.*;
+ import org.apache.ignite.internal.managers.discovery.*;
+ 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.jdk8.backport.*;
+ import org.jetbrains.annotations.*;
+ 
+ import java.io.*;
+ import java.util.*;
+ import java.util.concurrent.atomic.*;
+ 
+ import static org.apache.ignite.events.IgniteEventType.*;
+ import static org.apache.ignite.internal.managers.communication.GridIoPolicy.*;
+ 
+ /**
+  * Cache lock future.
+  */
+ public final class GridNearLockFuture<K, V> extends GridCompoundIdentityFuture<Boolean>
+     implements GridCacheMvccFuture<K, V, Boolean> {
+     /** */
+     private static final long serialVersionUID = 0L;
+ 
+     /** Logger reference. */
+     private static final AtomicReference<IgniteLogger> logRef = new AtomicReference<>();
+ 
+     /** Cache registry. */
+     @GridToStringExclude
+     private GridCacheContext<K, V> cctx;
+ 
+     /** Lock owner thread. */
+     @GridToStringInclude
+     private long threadId;
+ 
+     /** Keys to lock. */
+     private Collection<? extends K> keys;
+ 
+     /** Future ID. */
+     private IgniteUuid futId;
+ 
+     /** Lock version. */
+     private GridCacheVersion lockVer;
+ 
+     /** Read flag. */
+     private boolean read;
+ 
+     /** Flag to return value. */
+     private boolean retval;
+ 
+     /** Error. */
+     private AtomicReference<Throwable> err = new AtomicReference<>(null);
+ 
+     /** Timed out flag. */
+     private volatile boolean timedOut;
+ 
+     /** Timeout object. */
+     @GridToStringExclude
+     private LockTimeoutObject timeoutObj;
+ 
+     /** Lock timeout. */
+     private long timeout;
+ 
+     /** Logger. */
+     @GridToStringExclude
+     private IgniteLogger log;
+ 
+     /** Filter. */
+     private IgnitePredicate<CacheEntry<K, V>>[] filter;
+ 
+     /** Transaction. */
+     @GridToStringExclude
+     private GridNearTxLocal<K, V> tx;
+ 
+     /** Topology snapshot to operate on. */
+     private AtomicReference<GridDiscoveryTopologySnapshot> topSnapshot =
+         new AtomicReference<>();
+ 
+     /** Map of current values. */
+     private Map<K, GridTuple3<GridCacheVersion, V, byte[]>> valMap;
+ 
+     /** Trackable flag. */
+     private boolean trackable = true;
+ 
+     /** Mutex. */
+     private final Object mux = new Object();
+ 
+     /** Keys locked so far. */
+     @SuppressWarnings({"FieldAccessedSynchronizedAndUnsynchronized"})
+     @GridToStringExclude
+     private List<GridDistributedCacheEntry<K, V>> entries;
+ 
+     /** TTL for read operation. */
+     private long accessTtl;
+ 
+     /**
+      * Empty constructor required by {@link Externalizable}.
+      */
+     public GridNearLockFuture() {
+         // No-op.
+     }
+ 
+     /**
+      * @param cctx Registry.
+      * @param keys Keys to lock.
+      * @param tx Transaction.
+      * @param read Read flag.
+      * @param retval Flag to return value or not.
+      * @param timeout Lock acquisition timeout.
+      * @param accessTtl TTL for read operation.
+      * @param filter Filter.
+      */
+     public GridNearLockFuture(
+         GridCacheContext<K, V> cctx,
+         Collection<? extends K> keys,
+         @Nullable GridNearTxLocal<K, V> tx,
+         boolean read,
+         boolean retval,
+         long timeout,
+         long accessTtl,
+         IgnitePredicate<CacheEntry<K, V>>[] filter) {
+         super(cctx.kernalContext(), CU.boolReducer());
+ 
+         assert keys != null;
+ 
+         this.cctx = cctx;
+         this.keys = keys;
+         this.tx = tx;
+         this.read = read;
+         this.retval = retval;
+         this.timeout = timeout;
+         this.accessTtl = accessTtl;
+         this.filter = filter;
+ 
+         threadId = tx == null ? Thread.currentThread().getId() : tx.threadId();
+ 
+         lockVer = tx != null ? tx.xidVersion() : cctx.versions().next();
+ 
+         futId = IgniteUuid.randomUuid();
+ 
+         entries = new ArrayList<>(keys.size());
+ 
+         log = U.logger(ctx, logRef, GridNearLockFuture.class);
+ 
+         if (timeout > 0) {
+             timeoutObj = new LockTimeoutObject();
+ 
+             cctx.time().addTimeoutObject(timeoutObj);
+         }
+ 
+         valMap = new ConcurrentHashMap8<>(keys.size(), 1f);
+     }
+ 
+     /**
+      * @return Participating nodes.
+      */
+     @Override public Collection<? extends ClusterNode> nodes() {
+         return
+             F.viewReadOnly(futures(), new IgniteClosure<IgniteFuture<?>, ClusterNode>() {
+                 @Nullable @Override public ClusterNode apply(IgniteFuture<?> f) {
+                     if (isMini(f))
+                         return ((MiniFuture)f).node();
+ 
+                     return cctx.discovery().localNode();
+                 }
+             });
+     }
+ 
+     /** {@inheritDoc} */
+     @Override public GridCacheVersion version() {
+         return lockVer;
+     }
+ 
+     /**
+      * @return Entries.
+      */
+     public List<GridDistributedCacheEntry<K, V>> entriesCopy() {
+         synchronized (mux) {
+             return new ArrayList<>(entries);
+         }
+     }
+ 
+     /**
+      * @return Future ID.
+      */
+     @Override public IgniteUuid futureId() {
+         return futId;
+     }
+ 
+     /** {@inheritDoc} */
+     @Override public boolean trackable() {
+         return trackable;
+     }
+ 
+     /** {@inheritDoc} */
+     @Override public void markNotTrackable() {
+         trackable = false;
+     }
+ 
+     /**
+      * @return {@code True} if transaction is not {@code null}.
+      */
+     private boolean inTx() {
+         return tx != null;
+     }
+ 
+     /**
+      * @return {@code True} if implicit-single-tx flag is set.
+      */
+     private boolean implicitSingleTx() {
+         return tx != null && tx.implicitSingle();
+     }
+ 
+     /**
+      * @return {@code True} if transaction is not {@code null} and has invalidate flag set.
+      */
+     private boolean isInvalidate() {
+         return tx != null && tx.isInvalidate();
+     }
+ 
+     /**
+      * @return {@code True} if commit is synchronous.
+      */
+     private boolean syncCommit() {
+         return tx != null && tx.syncCommit();
+     }
+ 
+     /**
+      * @return {@code True} if rollback is synchronous.
+      */
+     private boolean syncRollback() {
+         return tx != null && tx.syncRollback();
+     }
+ 
+     /**
+      * @return Transaction isolation or {@code null} if no transaction.
+      */
+     @Nullable private IgniteTxIsolation isolation() {
+         return tx == null ? null : tx.isolation();
+     }
+ 
+     /**
+      * @return {@code true} if related transaction is implicit.
+      */
+     private boolean implicitTx() {
+         return tx != null && tx.implicit();
+     }
+ 
+     /**
+      * @param cached Entry.
+      * @return {@code True} if locked.
+      * @throws GridCacheEntryRemovedException If removed.
+      */
+     private boolean locked(GridCacheEntryEx<K, V> cached) throws GridCacheEntryRemovedException {
+         // Reentry-aware check (If filter failed, lock is failed).
+         return cached.lockedLocallyByIdOrThread(lockVer, threadId) && filter(cached);
+     }
+ 
+     /**
+      * Adds entry to future.
+      *
+      * @param topVer Topology version.
+      * @param entry Entry to add.
+      * @param dhtNodeId DHT node ID.
+      * @return Lock candidate.
+      * @throws GridCacheEntryRemovedException If entry was removed.
+      */
+     @Nullable private GridCacheMvccCandidate<K> addEntry(long topVer, GridNearCacheEntry<K, V> entry, UUID dhtNodeId)
+         throws GridCacheEntryRemovedException {
+         // Check if lock acquisition is timed out.
+         if (timedOut)
+             return null;
+ 
+         // Add local lock first, as it may throw GridCacheEntryRemovedException.
+         GridCacheMvccCandidate<K> c = entry.addNearLocal(
+             dhtNodeId,
+             threadId,
+             lockVer,
+             timeout,
+             !inTx(),
+             inTx(),
+             implicitSingleTx()
+         );
+ 
+         if (inTx()) {
+             IgniteTxEntry<K, V> txEntry = tx.entry(entry.txKey());
+ 
+             txEntry.cached(entry, txEntry.keyBytes());
+         }
+ 
+         if (c != null)
+             c.topologyVersion(topVer);
+ 
+         synchronized (mux) {
+             entries.add(entry);
+         }
+ 
+         if (c == null && timeout < 0) {
+             if (log.isDebugEnabled())
+                 log.debug("Failed to acquire lock with negative timeout: " + entry);
+ 
+             onFailed(false);
+ 
+             return null;
+         }
+ 
+         // Double check if lock acquisition has already timed out.
+         if (timedOut) {
+             entry.removeLock(lockVer);
+ 
+             return null;
+         }
+ 
+         return c;
+     }
+ 
+     /**
+      * Undoes all locks.
+      *
+      * @param dist If {@code true}, then remove locks from remote nodes as well.
+      */
+     private void undoLocks(boolean dist) {
+         // Transactions will undo during rollback.
+         if (dist && tx == null)
+             cctx.nearTx().removeLocks(lockVer, keys);
+         else {
+             if (tx != null) {
+                 if (tx.setRollbackOnly()) {
+                     if (log.isDebugEnabled())
+                         log.debug("Marked transaction as rollback only because locks could not be acquired: " + tx);
+                 }
+                 else if (log.isDebugEnabled())
+                     log.debug("Transaction was not marked rollback-only while locks were not acquired: " + tx);
+             }
+ 
+             for (GridCacheEntryEx<K, V> e : entriesCopy()) {
+                 try {
+                     e.removeLock(lockVer);
+                 }
+                 catch (GridCacheEntryRemovedException ignored) {
+                     while (true) {
+                         try {
+                             e = cctx.cache().peekEx(e.key());
+ 
+                             if (e != null)
+                                 e.removeLock(lockVer);
+ 
+                             break;
+                         }
+                         catch (GridCacheEntryRemovedException ignore) {
+                             if (log.isDebugEnabled())
+                                 log.debug("Attempted to remove lock on removed entry (will retry) [ver=" +
+                                     lockVer + ", entry=" + e + ']');
+                         }
+                     }
+                 }
+             }
+         }
+ 
+         cctx.mvcc().recheckPendingLocks();
+     }
+ 
+     /**
+      *
+      * @param dist {@code True} if need to distribute lock release.
+      */
+     private void onFailed(boolean dist) {
+         undoLocks(dist);
+ 
+         complete(false);
+     }
+ 
+     /**
+      * @param success Success flag.
+      */
+     public void complete(boolean success) {
+         onComplete(success, true);
+     }
+ 
+     /**
+      * @param nodeId Left node ID
+      * @return {@code True} if node was in the list.
+      */
+     @SuppressWarnings({"ThrowableInstanceNeverThrown"})
+     @Override public boolean onNodeLeft(UUID nodeId) {
+         boolean found = false;
+ 
+         for (IgniteFuture<?> fut : futures()) {
+             if (isMini(fut)) {
+                 MiniFuture f = (MiniFuture)fut;
+ 
+                 if (f.node().id().equals(nodeId)) {
+                     if (log.isDebugEnabled())
+                         log.debug("Found mini-future for left node [nodeId=" + nodeId + ", mini=" + f + ", fut=" +
+                             this + ']');
+ 
+                     f.onResult(newTopologyException(null, nodeId));
+ 
+                     found = true;
+                 }
+             }
+         }
+ 
+         if (!found) {
+             if (log.isDebugEnabled())
+                 log.debug("Near lock future does not have mapping for left node (ignoring) [nodeId=" + nodeId +
+                     ", fut=" + this + ']');
+         }
+ 
+         return found;
+     }
+ 
+     /**
+      * @param nodeId Sender.
+      * @param res Result.
+      */
+     void onResult(UUID nodeId, GridNearLockResponse<K, V> res) {
+         if (!isDone()) {
+             if (log.isDebugEnabled())
+                 log.debug("Received lock response from node [nodeId=" + nodeId + ", res=" + res + ", fut=" + this + ']');
+ 
+             for (IgniteFuture<Boolean> fut : pending()) {
+                 if (isMini(fut)) {
+                     MiniFuture mini = (MiniFuture)fut;
+ 
+                     if (mini.futureId().equals(res.miniId())) {
+                         assert mini.node().id().equals(nodeId);
+ 
+                         if (log.isDebugEnabled())
+                             log.debug("Found mini future for response [mini=" + mini + ", res=" + res + ']');
+ 
+                         mini.onResult(res);
+ 
+                         if (log.isDebugEnabled())
+                             log.debug("Future after processed lock response [fut=" + this + ", mini=" + mini +
+                                 ", res=" + res + ']');
+ 
+                         return;
+                     }
+                 }
+             }
+ 
+             U.warn(log, "Failed to find mini future for response (perhaps due to stale message) [res=" + res +
+                 ", fut=" + this + ']');
+         }
+         else if (log.isDebugEnabled())
+             log.debug("Ignoring lock response from node (future is done) [nodeId=" + nodeId + ", res=" + res +
+                 ", fut=" + this + ']');
+     }
+ 
+     /**
+      * @param t Error.
+      */
+     private void onError(Throwable t) {
+         err.compareAndSet(null, t instanceof GridCacheLockTimeoutException ? null : t);
+     }
+ 
+     /**
+      * @param cached Entry to check.
+      * @return {@code True} if filter passed.
+      */
+     private boolean filter(GridCacheEntryEx<K, V> cached) {
+         try {
+             if (!cctx.isAll(cached, filter)) {
+                 if (log.isDebugEnabled())
+                     log.debug("Filter didn't pass for entry (will fail lock): " + cached);
+ 
+                 onFailed(true);
+ 
+                 return false;
+             }
+ 
+             return true;
+         }
+         catch (IgniteCheckedException e) {
+             onError(e);
+ 
+             return false;
+         }
+     }
+ 
+     /**
+      * Callback for whenever entry lock ownership changes.
+      *
+      * @param entry Entry whose lock ownership changed.
+      */
+     @Override public boolean onOwnerChanged(GridCacheEntryEx<K, V> entry, GridCacheMvccCandidate<K> owner) {
+         if (owner != null && owner.version().equals(lockVer)) {
+             onDone(true);
+ 
+             return true;
+         }
+ 
+         return false;
+     }
+ 
+     /**
+      * @return {@code True} if locks have been acquired.
+      */
+     private boolean checkLocks() {
+         if (!isDone() && initialized() && !hasPending()) {
+             for (int i = 0; i < entries.size(); i++) {
+                 while (true) {
+                     GridCacheEntryEx<K, V> cached = entries.get(i);
+ 
+                     try {
+                         if (!locked(cached)) {
+                             if (log.isDebugEnabled())
+                                 log.debug("Lock is still not acquired for entry (will keep waiting) [entry=" +
+                                     cached + ", fut=" + this + ']');
+ 
+                             return false;
+                         }
+ 
+                         break;
+                     }
+                     // Possible in concurrent cases, when owner is changed after locks
+                     // have been released or cancelled.
+                     catch (GridCacheEntryRemovedException ignore) {
+                         if (log.isDebugEnabled())
+                             log.debug("Got removed entry in onOwnerChanged method (will retry): " + cached);
+ 
+                         // Replace old entry with new one.
+                         entries.set(i, (GridDistributedCacheEntry<K, V>)cctx.cache().entryEx(cached.key()));
+                     }
+                 }
+             }
+ 
+             if (log.isDebugEnabled())
+                 log.debug("Local lock acquired for entries [fut=" + this + ", entries=" + entries + "]");
+ 
+             onComplete(true, true);
+ 
+             return true;
+         }
+ 
+         return false;
+     }
+ 
+     /** {@inheritDoc} */
+     @Override public boolean cancel() {
+         if (onCancelled())
+             onComplete(false, true);
+ 
+         return isCancelled();
+     }
+ 
+     /** {@inheritDoc} */
+     @Override public boolean onDone(Boolean success, Throwable err) {
+         if (log.isDebugEnabled())
+             log.debug("Received onDone(..) callback [success=" + success + ", err=" + err + ", fut=" + this + ']');
+ 
+         // If locks were not acquired yet, delay completion.
+         if (isDone() || (err == null && success && !checkLocks()))
+             return false;
+ 
+         this.err.compareAndSet(null, err instanceof GridCacheLockTimeoutException ? null : err);
+ 
+         if (err != null)
+             success = false;
+ 
+         return onComplete(success, true);
+     }
+ 
+     /**
+      * Completeness callback.
+      *
+      * @param success {@code True} if lock was acquired.
+      * @param distribute {@code True} if need to distribute lock removal in case of failure.
+      * @return {@code True} if complete by this operation.
+      */
+     private boolean onComplete(boolean success, boolean distribute) {
+         if (log.isDebugEnabled())
+             log.debug("Received onComplete(..) callback [success=" + success + ", distribute=" + distribute +
+                 ", fut=" + this + ']');
+ 
+         if (!success)
+             undoLocks(distribute);
+ 
+         if (tx != null)
+             cctx.tm().txContext(tx);
+ 
+         if (super.onDone(success, err.get())) {
+             if (log.isDebugEnabled())
+                 log.debug("Completing future: " + this);
+ 
+             // Clean up.
+             cctx.mvcc().removeFuture(this);
+ 
+             if (timeoutObj != null)
+                 cctx.time().removeTimeoutObject(timeoutObj);
+ 
+             return true;
+         }
+ 
+         return false;
+     }
+ 
+     /** {@inheritDoc} */
+     @Override public int hashCode() {
+         return futId.hashCode();
+     }
+ 
+     /** {@inheritDoc} */
+     @Override public String toString() {
+         return S.toString(GridNearLockFuture.class, this, "inTx", inTx(), "super", super.toString());
+     }
+ 
+     /**
+      * @param f Future.
+      * @return {@code True} if mini-future.
+      */
+     private boolean isMini(IgniteFuture<?> f) {
+         return f.getClass().equals(MiniFuture.class);
+     }
+ 
+     /**
+      * Basically, future mapping consists from two parts. First, we must determine the topology version this future
+      * will map on. Locking is performed within a user transaction, we must continue to map keys on the same
+      * topology version as it started. If topology version is undefined, we get current topology future and wait
+      * until it completes so the topology is ready to use.
+      * <p/>
+      * During the second part we map keys to primary nodes using topology snapshot we obtained during the first
+      * part. Note that if primary node leaves grid, the future will fail and transaction will be rolled back.
+      */
+     void map() {
+         // Obtain the topology version to use.
+         GridDiscoveryTopologySnapshot snapshot = tx != null ? tx.topologySnapshot() :
+             cctx.mvcc().lastExplicitLockTopologySnapshot(Thread.currentThread().getId());
+ 
+         if (snapshot != null) {
+             // Continue mapping on the same topology version as it was before.
+             topSnapshot.compareAndSet(null, snapshot);
+ 
+             map(keys);
+ 
+             markInitialized();
+ 
+             return;
+         }
+ 
+         // Must get topology snapshot and map on that version.
+         mapOnTopology();
+     }
+ 
+     /**
+      * Acquires topology future and checks it completeness under the read lock. If it is not complete,
+      * will asynchronously wait for it's completeness and then try again.
+      */
+     void mapOnTopology() {
+         // We must acquire topology snapshot from the topology version future.
+         try {
+             cctx.topology().readLock();
+ 
+             try {
+                 GridDhtTopologyFuture fut = cctx.topologyVersionFuture();
+ 
+                 if (fut.isDone()) {
+                     GridDiscoveryTopologySnapshot snapshot = fut.topologySnapshot();
+ 
+                     if (tx != null) {
+                         tx.topologyVersion(snapshot.topologyVersion());
+                         tx.topologySnapshot(snapshot);
+                     }
+ 
+                     topSnapshot.compareAndSet(null, snapshot);
+ 
+                     map(keys);
+ 
+                     markInitialized();
+                 }
+                 else {
+                     fut.listenAsync(new CI1<IgniteFuture<Long>>() {
+                         @Override public void apply(IgniteFuture<Long> t) {
+                             mapOnTopology();
+                         }
+                     });
+                 }
+             }
+             finally {
+                 cctx.topology().readUnlock();
+             }
+         }
+         catch (IgniteCheckedException e) {
+             onDone(e);
+         }
+     }
+ 
+     /**
+      * Maps keys to nodes. Note that we can not simply group keys by nodes and send lock request as
+      * such approach does not preserve order of lock acquisition. Instead, keys are split in continuous
+      * groups belonging to one primary node and locks for these groups are acquired sequentially.
+      *
+      * @param keys Keys.
+      */
+     private void map(Iterable<? extends K> keys) {
+         try {
+             GridDiscoveryTopologySnapshot snapshot = topSnapshot.get();
+ 
+             assert snapshot != null;
+ 
+             long topVer = snapshot.topologyVersion();
+ 
+             assert topVer > 0;
+ 
+             if (CU.affinityNodes(cctx, topVer).isEmpty()) {
+                 onDone(new ClusterTopologyException("Failed to map keys for near-only cache (all " +
+                     "partition nodes left the grid)."));
+ 
+                 return;
+             }
+ 
+             ConcurrentLinkedDeque8<GridNearLockMapping<K, V>> mappings =
+                 new ConcurrentLinkedDeque8<>();
+ 
+             // Assign keys to primary nodes.
+             GridNearLockMapping<K, V> map = null;
+ 
+             for (K key : keys) {
+                 GridNearLockMapping<K, V> updated = map(key, map, topVer);
+ 
+                 // If new mapping was created, add to collection.
+                 if (updated != map) {
+                     mappings.add(updated);
+ 
+                     if (tx != null && updated.node().isLocal())
+                         tx.nearLocallyMapped(true);
+                 }
+ 
+                 map = updated;
+             }
+ 
+             if (isDone()) {
+                 if (log.isDebugEnabled())
+                     log.debug("Abandoning (re)map because future is done: " + this);
+ 
+                 return;
+             }
+ 
+             if (log.isDebugEnabled())
+                 log.debug("Starting (re)map for mappings [mappings=" + mappings + ", fut=" + this + ']');
+ 
+             // Create mini futures.
+             for (Iterator<GridNearLockMapping<K, V>> iter = mappings.iterator(); iter.hasNext(); ) {
+                 GridNearLockMapping<K, V> mapping = iter.next();
+ 
+                 ClusterNode node = mapping.node();
+                 Collection<K> mappedKeys = mapping.mappedKeys();
+ 
+                 assert !mappedKeys.isEmpty();
+ 
+                 GridNearLockRequest<K, V> req = null;
+ 
+                 Collection<K> distributedKeys = new ArrayList<>(mappedKeys.size());
+ 
+                 boolean explicit = false;
+ 
+                 for (K key : mappedKeys) {
+                     IgniteTxKey<K> txKey = cctx.txKey(key);
+ 
+                     while (true) {
+                         GridNearCacheEntry<K, V> entry = null;
+ 
+                         try {
+                             entry = cctx.near().entryExx(key, topVer);
+ 
+                             if (!cctx.isAll(entry.wrap(false), filter)) {
+                                 if (log.isDebugEnabled())
+                                     log.debug("Entry being locked did not pass filter (will not lock): " + entry);
+ 
+                                 onComplete(false, false);
+ 
+                                 return;
+                             }
+ 
+                             // Removed exception may be thrown here.
+                             GridCacheMvccCandidate<K> cand = addEntry(topVer, entry, node.id());
+ 
+                             if (isDone()) {
+                                 if (log.isDebugEnabled())
+                                     log.debug("Abandoning (re)map because future is done after addEntry attempt " +
+                                         "[fut=" + this + ", entry=" + entry + ']');
+ 
+                                 return;
+                             }
+ 
+                             if (cand != null) {
+                                 if (tx == null && !cand.reentry())
+                                     cctx.mvcc().addExplicitLock(threadId, cand, snapshot);
+ 
+                                 GridTuple3<GridCacheVersion, V, byte[]> val = entry.versionedValue();
+ 
+                                 if (val == null) {
+                                     GridDhtCacheEntry<K, V> dhtEntry = dht().peekExx(key);
+ 
+                                     try {
+                                         if (dhtEntry != null)
+                                             val = dhtEntry.versionedValue(topVer);
+                                     }
+                                     catch (GridCacheEntryRemovedException ignored) {
+                                         assert dhtEntry.obsolete() : " Got removed exception for non-obsolete entry: "
+                                             + dhtEntry;
+ 
+                                         if (log.isDebugEnabled())
+                                             log.debug("Got removed exception for DHT entry in map (will ignore): "
+                                                 + dhtEntry);
+                                     }
+                                 }
+ 
+                                 GridCacheVersion dhtVer = null;
+ 
+                                 if (val != null) {
+                                     dhtVer = val.get1();
+ 
+                                     valMap.put(key, val);
+                                 }
+ 
+                                 if (!cand.reentry()) {
+                                     if (req == null) {
+                                         req = new GridNearLockRequest<>(
+                                             cctx.cacheId(),
+                                             topVer,
+                                             cctx.nodeId(),
+                                             threadId,
+                                             futId,
+                                             lockVer,
+                                             inTx(),
+                                             implicitTx(),
+                                             implicitSingleTx(),
+                                             read,
+                                             isolation(),
+                                             isInvalidate(),
+                                             timeout,
+                                             mappedKeys.size(),
+                                             inTx() ? tx.size() : mappedKeys.size(),
+                                             inTx() && tx.syncCommit(),
+                                             inTx() ? tx.groupLockKey() : null,
+                                             inTx() && tx.partitionLock(),
+                                             inTx() ? tx.subjectId() : null,
+                                             inTx() ? tx.taskNameHash() : 0,
+                                             read ? accessTtl : -1L);
+ 
+                                         mapping.request(req);
+                                     }
+ 
+                                     distributedKeys.add(key);
+ 
+                                     IgniteTxEntry<K, V> writeEntry = tx != null ? tx.writeMap().get(txKey) : null;
+ 
+                                     if (tx != null)
+                                         tx.addKeyMapping(txKey, mapping.node());
+ 
+                                     req.addKeyBytes(
+                                         key,
+                                         node.isLocal() ? null : entry.getOrMarshalKeyBytes(),
+                                         retval && dhtVer == null,
+                                         dhtVer, // Include DHT version to match remote DHT entry.
+                                         writeEntry,
+                                         inTx() ? tx.entry(txKey).drVersion() : null,
+                                         cctx);
+ 
+                                     // Clear transfer required flag since we are sending message.
+                                     if (writeEntry != null)
+                                         writeEntry.transferRequired(false);
+                                 }
+ 
+                                 if (cand.reentry())
+                                     explicit = tx != null && !entry.hasLockCandidate(tx.xidVersion());
+                             }
+                             else
+                                 // Ignore reentries within transactions.
+                                 explicit = tx != null && !entry.hasLockCandidate(tx.xidVersion());
+ 
+                             if (explicit)
+                                 tx.addKeyMapping(txKey, mapping.node());
+ 
+                             break;
+                         }
+                         catch (GridCacheEntryRemovedException ignored) {
+                             assert entry.obsolete() : "Got removed exception on non-obsolete entry: " + entry;
+ 
+                             if (log.isDebugEnabled())
+                                 log.debug("Got removed entry in lockAsync(..) method (will retry): " + entry);
+                         }
+                     }
+ 
+                     // Mark mapping explicit lock flag.
+                     if (explicit) {
+                         boolean marked = tx != null && tx.markExplicit(node.id());
+ 
+                         assert tx == null || marked;
+                     }
+                 }
+ 
+                 if (!distributedKeys.isEmpty())
+                     mapping.distributedKeys(distributedKeys);
+                 else {
+                     assert mapping.request() == null;
+ 
+                     iter.remove();
+                 }
+             }
+ 
+             cctx.mvcc().recheckPendingLocks();
+ 
+             proceedMapping(mappings);
+         }
+         catch (IgniteCheckedException ex) {
+             onError(ex);
+         }
+     }
+ 
+     /**
+      * Gets next near lock mapping and either acquires dht locks locally or sends near lock request to
+      * remote primary node.
+      *
+      * @param mappings Queue of mappings.
+      * @throws IgniteCheckedException If mapping can not be completed.
+      */
+     private void proceedMapping(final ConcurrentLinkedDeque8<GridNearLockMapping<K, V>> mappings)
+         throws IgniteCheckedException {
+         GridNearLockMapping<K, V> map = mappings.poll();
+ 
+         // If there are no more mappings to process, complete the future.
+         if (map == null)
+             return;
+ 
+         final GridNearLockRequest<K, V> req = map.request();
+         final Collection<K> mappedKeys = map.distributedKeys();
+         final ClusterNode node = map.node();
+ 
+         if (filter != null && filter.length != 0)
+             req.filter(filter, cctx);
+ 
+         if (node.isLocal()) {
+             req.miniId(IgniteUuid.randomUuid());
+ 
+             if (log.isDebugEnabled())
+                 log.debug("Before locally locking near request: " + req);
+ 
+             IgniteFuture<GridNearLockResponse<K, V>> fut = dht().lockAllAsync(cctx, cctx.localNode(), req, filter);
+ 
+             // Add new future.
+             add(new GridEmbeddedFuture<>(
+                 cctx.kernalContext(),
+                 fut,
+                 new C2<GridNearLockResponse<K, V>, Exception, Boolean>() {
+                     @Override public Boolean apply(GridNearLockResponse<K, V> res, Exception e) {
+                         if (CU.isLockTimeoutOrCancelled(e) ||
+                             (res != null && CU.isLockTimeoutOrCancelled(res.error())))
+                             return false;
+ 
+                         if (e != null) {
+                             onError(e);
+ 
+                             return false;
+                         }
+ 
+                         if (res == null) {
+                             onError(new IgniteCheckedException("Lock response is null for future: " + this));
+ 
+                             return false;
+                         }
+ 
+                         if (res.error() != null) {
+                             onError(res.error());
+ 
+                             return false;
+                         }
+ 
+                         if (log.isDebugEnabled())
+                             log.debug("Acquired lock for local DHT mapping [locId=" + cctx.nodeId() +
+                                 ", mappedKeys=" + mappedKeys + ", fut=" + GridNearLockFuture.this + ']');
+ 
+                         try {
+                             int i = 0;
+ 
+                             for (K k : mappedKeys) {
+                                 while (true) {
+                                     GridNearCacheEntry<K, V> entry = cctx.near().entryExx(k, req.topologyVersion());
+ 
+                                     try {
+                                         GridTuple3<GridCacheVersion, V, byte[]> oldValTup = valMap.get(entry.key());
+ 
+                                         boolean hasBytes = entry.hasValue();
+                                         V oldVal = entry.rawGet();
+                                         V newVal = res.value(i);
+                                         byte[] newBytes = res.valueBytes(i);
+ 
+                                         GridCacheVersion dhtVer = res.dhtVersion(i);
+                                         GridCacheVersion mappedVer = res.mappedVersion(i);
+ 
+                                         // On local node don't record twice if DHT cache already recorded.
+                                         boolean record = retval && oldValTup != null && oldValTup.get1().equals(dhtVer);
+ 
+                                         if (newVal == null) {
+                                             if (oldValTup != null) {
+                                                 if (oldValTup.get1().equals(dhtVer)) {
+                                                     newVal = oldValTup.get2();
+ 
+                                                     newBytes = oldValTup.get3();
+                                                 }
+ 
+                                                 oldVal = oldValTup.get2();
+                                             }
+                                         }
+ 
+                                         // Lock is held at this point, so we can set the
+                                         // returned value if any.
+                                         entry.resetFromPrimary(newVal, newBytes, lockVer, dhtVer, node.id());
+ 
+                                         entry.readyNearLock(lockVer, mappedVer, res.committedVersions(),
+                                             res.rolledbackVersions(), res.pending());
+ 
+                                         if (inTx() && implicitTx() && tx.onePhaseCommit()) {
+                                             boolean pass = res.filterResult(i);
+ 
+                                             tx.entry(cctx.txKey(k)).filters(pass ? CU.<K, V>empty() : CU.<K, V>alwaysFalse());
+                                         }
+ 
+                                         if (record) {
+                                             if (cctx.events().isRecordable(EVT_CACHE_OBJECT_READ))
+                                                 cctx.events().addEvent(
+                                                     entry.partition(),
+                                                     entry.key(),
+                                                     tx,
+                                                     null,
+                                                     EVT_CACHE_OBJECT_READ,
+                                                     newVal,
+                                                     newVal != null,
+                                                     oldVal,
+                                                     hasBytes,
+                                                     CU.subjectId(tx, cctx.shared()),
+                                                     null,
+                                                     inTx() ? tx.resolveTaskName() : null);
+ 
 -                                            cctx.cache().metrics0().onRead(oldVal != null);
++                                            if (cctx.cache().configuration().isStatisticsEnabled())
++                                                cctx.cache().metrics0().onRead(oldVal != null);
+                                         }
+ 
+                                         if (log.isDebugEnabled())
+                                             log.debug("Processed response for entry [res=" + res +
+                                                 ", entry=" + entry + ']');
+ 
+                                         break; // Inner while loop.
+                                     }
+                                     catch (GridCacheEntryRemovedException ignored) {
+                                         if (log.isDebugEnabled())
+                                             log.debug("Failed to add candidates because entry was " +
+                                                 "removed (will renew).");
+ 
+                                         // Replace old entry with new one.
+                                         entries.set(i, (GridDistributedCacheEntry<K, V>)
+                                             cctx.cache().entryEx(entry.key()));
+                                     }
+                                 }
+ 
+                                 i++; // Increment outside of while loop.
+                             }
+ 
+                             // Proceed and add new future (if any) before completing embedded future.
+                             proceedMapping(mappings);
+                         }
+                         catch (IgniteCheckedException ex) {
+                             onError(ex);
+ 
+                             return false;
+                         }
+ 
+                         return true;
+                     }
+                 }
+             ));
+         }
+         else {
+             final MiniFuture fut = new MiniFuture(node, mappedKeys, mappings);
+ 
+             req.miniId(fut.futureId());
+ 
+             add(fut); // Append new future.
+ 
+             IgniteFuture<?> txSync = null;
+ 
+             if (inTx())
+                 txSync = cctx.tm().awaitFinishAckAsync(node.id(), tx.threadId());
+ 
+             if (txSync == null || txSync.isDone()) {
+                 try {
+                     if (log.isDebugEnabled())
+                         log.debug("Sending near lock request [node=" + node.id() + ", req=" + req + ']');
+ 
+                     cctx.io().send(node, req, cctx.system() ? UTILITY_CACHE_POOL : SYSTEM_POOL);
+                 }
+                 catch (ClusterTopologyException ex) {
+                     assert fut != null;
+ 
+                     fut.onResult(ex);
+                 }
+             }
+             else {
+                 txSync.listenAsync(new CI1<IgniteFuture<?>>() {
+                     @Override public void apply(IgniteFuture<?> t) {
+                         try {
+                             if (log.isDebugEnabled())
+                                 log.debug("Sending near lock request [node=" + node.id() + ", req=" + req + ']');
+ 
+                             cctx.io().send(node, req, cctx.system() ? UTILITY_CACHE_POOL : SYSTEM_POOL);
+                         }
+                         catch (ClusterTopologyException ex) {
+                             assert fut != null;
+ 
+                             fut.onResult(ex);
+                         }
+                         catch (IgniteCheckedException e) {
+                             onError(e);
+                         }
+                     }
+                 });
+             }
+         }
+     }
+ 
+     /**
+      * @param mapping Mappings.
+      * @param key Key to map.
+      * @param topVer Topology version.
+      * @return Near lock mapping.
+      * @throws IgniteCheckedException If mapping for key failed.
+      */
+     private GridNearLockMapping<K, V> map(K key, @Nullable GridNearLockMapping<K, V> mapping,
+         long topVer) throws IgniteCheckedException {
+         assert mapping == null || mapping.node() != null;
+ 
+         ClusterNode primary = cctx.affinity().primary(key, topVer);
+ 
+         if (cctx.discovery().node(primary.id()) == null)
+             // If primary node left the grid before lock acquisition, fail the whole future.
+             throw newTopologyException(null, primary.id());
+ 
+         if (inTx() && tx.groupLock() && !primary.isLocal())
+             throw new IgniteCheckedException("Failed to start group lock transaction (local node is not primary for " +
+                 " key) [key=" + key + ", primaryNodeId=" + primary.id() + ']');
+ 
+         if (mapping == null || !primary.id().equals(mapping.node().id()))
+             mapping = new GridNearLockMapping<>(primary, key);
+         else
+             mapping.addKey(key);
+ 
+         return mapping;
+     }
+ 
+     /**
+      * @return DHT cache.
+      */
+     private GridDhtTransactionalCacheAdapter<K, V> dht() {
+         return cctx.nearTx().dht();
+     }
+ 
+     /**
+      * Creates new topology exception for cases when primary node leaves grid during mapping.
+      *
+      * @param nested Optional nested exception.
+      * @param nodeId Node ID.
+      * @return Topology exception with user-friendly message.
+      */
+     private ClusterTopologyException newTopologyException(@Nullable Throwable nested, UUID nodeId) {
+         return new ClusterTopologyException("Failed to acquire lock for keys (primary node left grid, " +
+             "retry transaction if possible) [keys=" + keys + ", node=" + nodeId + ']', nested);
+     }
+ 
+     /**
+      * Lock request timeout object.
+      */
+     private class LockTimeoutObject extends GridTimeoutObjectAdapter {
+         /**
+          * Default constructor.
+          */
+         LockTimeoutObject() {
+             super(timeout);
+         }
+ 
+         /** {@inheritDoc} */
+         @SuppressWarnings({"ThrowableInstanceNeverThrown"})
+         @Override public void onTimeout() {
+             if (log.isDebugEnabled())
+                 log.debug("Timed out waiting for lock response: " + this);
+ 
+             timedOut = true;
+ 
+             onComplete(false, true);
+         }
+ 
+         /** {@inheritDoc} */
+         @Override public String toString() {
+             return S.toString(LockTimeoutObject.class, this);
+         }
+     }
+ 
+     /**
+      * Mini-future for get operations. Mini-futures are only waiting on a single
+      * node as opposed to multiple nodes.
+      */
+     private class MiniFuture extends GridFutureAdapter<Boolean> {
+         /** */
+         private static final long serialVersionUID = 0L;
+ 
+         /** */
+         private final IgniteUuid futId = IgniteUuid.randomUuid();
+ 
+         /** Node ID. */
+         @GridToStringExclude
+         private ClusterNode node;
+ 
+         /** Keys. */
+         @GridToStringInclude
+         private Collection<K> keys;
+ 
+         /** Mappings to proceed. */
+         @GridToStringExclude
+         private ConcurrentLinkedDeque8<GridNearLockMapping<K, V>> mappings;
+ 
+         /** */
+         private AtomicBoolean rcvRes = new AtomicBoolean(false);
+ 
+         /**
+          * Empty constructor required for {@link Externalizable}.
+          */
+         public MiniFuture() {
+             // No-op.
+         }
+ 
+         /**
+          * @param node Node.
+          * @param keys Keys.
+          * @param mappings Mappings to proceed.
+          */
+         MiniFuture(ClusterNode node, Collection<K> keys,
+             ConcurrentLinkedDeque8<GridNearLockMapping<K, V>> mappings) {
+             super(cctx.kernalContext());
+ 
+             this.node = node;
+             this.keys = keys;
+             this.mappings = mappings;
+         }
+ 
+         /**
+          * @return Future ID.
+          */
+         IgniteUuid futureId() {
+             return futId;
+         }
+ 
+         /**
+          * @return Node ID.
+          */
+         public ClusterNode node() {
+             return node;
+         }
+ 
+         /**
+          * @return Keys.
+          */
+         public Collection<K> keys() {
+             return keys;
+         }
+ 
+         /**
+          * @param e Error.
+          */
+         void onResult(Throwable e) {
+             if (rcvRes.compareAndSet(false, true)) {
+                 if (log.isDebugEnabled())
+                     log.debug("Failed to get future result [fut=" + this + ", err=" + e + ']');
+ 
+                 // Fail.
+                 onDone(e);
+             }
+             else
+                 U.warn(log, "Received error after another result has been processed [fut=" + GridNearLockFuture.this +
+                     ", mini=" + this + ']', e);
+         }
+ 
+         /**
+          * @param e Node left exception.
+          */
+         void onResult(ClusterTopologyException e) {
+             if (isDone())
+                 return;
+ 
+             if (rcvRes.compareAndSet(false, true)) {
+                 if (log.isDebugEnabled())
+                     log.debug("Remote node left grid while sending or waiting for reply (will fail): " + this);
+ 
+                 if (tx != null)
+                     tx.removeMapping(node.id());
+ 
+                 // Primary node left the grid, so fail the future.
+                 GridNearLockFuture.this.onDone(newTopologyException(e, node.id()));
+ 
+                 onDone(true);
+             }
+         }
+ 
+         /**
+          * @param res Result callback.
+          */
+         void onResult(GridNearLockResponse<K, V> res) {
+             if (rcvRes.compareAndSet(false, true)) {
+                 if (res.error() != null) {
+                     if (log.isDebugEnabled())
+                         log.debug("Finishing mini future with an error due to error in response [miniFut=" + this +
+                             ", res=" + res + ']');
+ 
+                     // Fail.
+                     if (res.error() instanceof GridCacheLockTimeoutException)
+                         onDone(false);
+                     else
+                         onDone(res.error());
+ 
+                     return;
+                 }
+ 
+                 int i = 0;
+ 
+                 long topVer = topSnapshot.get().topologyVersion();
+ 
+                 for (K k : keys) {
+                     while (true) {
+                         GridNearCacheEntry<K, V> entry = cctx.near().entryExx(k, topVer);
+ 
+                         try {
+                             if (res.dhtVersion(i) == null) {
+                                 onDone(new IgniteCheckedException("Failed to receive DHT version from remote node " +
+                                     "(will fail the lock): " + res));
+ 
+                                 return;
+                             }
+ 
+                             GridTuple3<GridCacheVersion, V, byte[]> oldValTup = valMap.get(entry.key());
+ 
+                             V oldVal = entry.rawGet();
+                             boolean hasOldVal = false;
+                             V newVal = res.value(i);
+                             byte[] newBytes = res.valueBytes(i);
+ 
+                             boolean readRecordable = false;
+ 
+                             if (retval) {
+                                 readRecordable = cctx.events().isRecordable(EVT_CACHE_OBJECT_READ);
+ 
+                                 if (readRecordable)
+                                     hasOldVal = entry.hasValue();
+                             }
+ 
+                             GridCacheVersion dhtVer = res.dhtVersion(i);
+                             GridCacheVersion mappedVer = res.mappedVersion(i);
+ 
+                             if (newVal == null) {
+                                 if (oldValTup != null) {
+                                     if (oldValTup.get1().equals(dhtVer)) {
+                                         newVal = oldValTup.get2();
+ 
+                                         newBytes = oldValTup.get3();
+                                     }
+ 
+                                     oldVal = oldValTup.get2();
+                                 }
+                             }
+ 
+                             // Lock is held at this point, so we can set the
+                             // returned value if any.
+                             entry.resetFromPrimary(newVal, newBytes, lockVer, dhtVer, node.id());
+ 
+                             if (inTx() && implicitTx() && tx.onePhaseCommit()) {
+                                 boolean pass = res.filterResult(i);
+ 
+                                 tx.entry(cctx.txKey(k)).filters(pass ? CU.<K, V>empty() : CU.<K, V>alwaysFalse());
+                             }
+ 
+                             entry.readyNearLock(lockVer, mappedVer, res.committedVersions(), res.rolledbackVersions(),
+                                 res.pending());
+ 
+                             if (retval) {
+                                 if (readRecordable)
+                                     cctx.events().addEvent(
+                                         entry.partition(),
+                                         entry.key(),
+                                         tx,
+                                         null,
+                                         EVT_CACHE_OBJECT_READ,
+                                         newVal,
+                                         newVal != null || newBytes != null,
+                                         oldVal,
+                                         hasOldVal,
+                                         CU.subjectId(tx, cctx.shared()),
+                                         null,
+                                         inTx() ? tx.resolveTaskName() : null);
+ 
 -                                cctx.cache().metrics0().onRead(false);
++                                if (cctx.cache().configuration().isStatisticsEnabled())
++                                    cctx.cache().metrics0().onRead(false);
+                             }
+ 
+                             if (log.isDebugEnabled())
+                                 log.debug("Processed response for entry [res=" + res + ", entry=" + entry + ']');
+ 
+                             break; // Inner while loop.
+                         }
+                         catch (GridCacheEntryRemovedException ignored) {
+                             if (log.isDebugEnabled())
+                                 log.debug("Failed to add candidates because entry was removed (will renew).");
+ 
+                             // Replace old entry with new one.
+                             entries.set(i, (GridDistributedCacheEntry<K, V>)cctx.cache().entryEx(entry.key()));
+                         }
+                         catch (IgniteCheckedException e) {
+                             onDone(e);
+ 
+                             return;
+                         }
+                     }
+ 
+                     i++;
+                 }
+ 
+                 try {
+                     proceedMapping(mappings);
+                 }
+                 catch (IgniteCheckedException e) {
+                     onDone(e);
+                 }
+ 
+                 onDone(true);
+             }
+         }
+ 
+         /** {@inheritDoc} */
+         @Override public String toString() {
+             return S.toString(MiniFuture.class, this, "node", node.id(), "super", super.toString());
+         }
+     }
+ }


Mime
View raw message