ignite-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From sboi...@apache.org
Subject [07/53] [abbrv] incubator-ignite git commit: IGNITE-49 merged from sprint-1 into ignite-49
Date Tue, 27 Jan 2015 16:40:27 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/dht/atomic/GridDhtAtomicCache.java
----------------------------------------------------------------------
diff --cc modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java
index 0000000,683b7b9..9b63363
mode 000000,100644..100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java
@@@ -1,0 -1,3005 +1,3007 @@@
+ /*
+  * Licensed to the Apache Software Foundation (ASF) under one or more
+  * contributor license agreements.  See the NOTICE file distributed with
+  * this work for additional information regarding copyright ownership.
+  * The ASF licenses this file to You under the Apache License, Version 2.0
+  * (the "License"); you may not use this file except in compliance with
+  * the License.  You may obtain a copy of the License at
+  *
+  *      http://www.apache.org/licenses/LICENSE-2.0
+  *
+  * Unless required by applicable law or agreed to in writing, software
+  * distributed under the License is distributed on an "AS IS" BASIS,
+  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+  * See the License for the specific language governing permissions and
+  * limitations under the License.
+  */
+ 
+ package org.apache.ignite.internal.processors.cache.distributed.dht.atomic;
+ 
+ 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.plugin.security.*;
+ import org.apache.ignite.transactions.*;
+ import org.apache.ignite.internal.managers.communication.*;
+ import org.apache.ignite.internal.processors.cache.distributed.dht.*;
+ import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.*;
+ import org.apache.ignite.internal.processors.cache.distributed.near.*;
+ import org.apache.ignite.internal.processors.cache.dr.*;
+ 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 sun.misc.*;
+ 
+ import javax.cache.expiry.*;
+ import javax.cache.processor.*;
+ import java.io.*;
+ import java.util.*;
+ import java.util.concurrent.*;
+ import java.util.concurrent.atomic.*;
+ import java.util.concurrent.locks.*;
+ 
+ import static org.apache.ignite.IgniteSystemProperties.*;
+ import static org.apache.ignite.cache.CacheAtomicWriteOrderMode.*;
+ import static org.apache.ignite.cache.GridCachePeekMode.*;
+ import static org.apache.ignite.cache.CacheWriteSynchronizationMode.*;
+ import static org.apache.ignite.internal.processors.cache.GridCacheOperation.*;
+ import static org.apache.ignite.internal.processors.cache.GridCacheUtils.*;
+ import static org.apache.ignite.internal.processors.dr.GridDrType.*;
+ 
+ /**
+  * Non-transactional partitioned cache.
+  */
+ @GridToStringExclude
+ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
+     /** */
+     private static final long serialVersionUID = 0L;
+ 
+     /** Deferred update response buffer size. */
+     private static final int DEFERRED_UPDATE_RESPONSE_BUFFER_SIZE =
+         Integer.getInteger(GG_ATOMIC_DEFERRED_ACK_BUFFER_SIZE, 256);
+ 
+     /** Deferred update response timeout. */
+     private static final int DEFERRED_UPDATE_RESPONSE_TIMEOUT =
+         Integer.getInteger(GG_ATOMIC_DEFERRED_ACK_TIMEOUT, 500);
+ 
+     /** Unsafe instance. */
+     private static final Unsafe UNSAFE = GridUnsafe.unsafe();
+ 
+     /** Update reply closure. */
+     private CI2<GridNearAtomicUpdateRequest<K, V>, GridNearAtomicUpdateResponse<K, V>> updateReplyClos;
+ 
+     /** Pending  */
+     private ConcurrentMap<UUID, DeferredResponseBuffer> pendingResponses = new ConcurrentHashMap8<>();
+ 
+     /** */
+     private GridNearAtomicCache<K, V> near;
+ 
+     /**
+      * Empty constructor required by {@link Externalizable}.
+      */
+     public GridDhtAtomicCache() {
+         // No-op.
+     }
+ 
+     /**
+      * @param ctx Cache context.
+      */
+     public GridDhtAtomicCache(GridCacheContext<K, V> ctx) {
+         super(ctx);
+     }
+ 
+     /**
+      * @param ctx Cache context.
+      * @param map Cache concurrent map.
+      */
+     public GridDhtAtomicCache(GridCacheContext<K, V> ctx, GridCacheConcurrentMap<K, V> map) {
+         super(ctx, map);
+     }
+ 
+     /** {@inheritDoc} */
+     @Override public boolean isDhtAtomic() {
+         return true;
+     }
+ 
+     /** {@inheritDoc} */
+     @Override protected void init() {
+         map.setEntryFactory(new GridCacheMapEntryFactory<K, V>() {
+             /** {@inheritDoc} */
+             @Override public GridCacheMapEntry<K, V> create(GridCacheContext<K, V> ctx, long topVer, K key, int hash,
+                 V val, GridCacheMapEntry<K, V> next, long ttl, int hdrId) {
+                 return new GridDhtAtomicCacheEntry<>(ctx, topVer, key, hash, val, next, ttl, hdrId);
+             }
+         });
+ 
+         updateReplyClos = new CI2<GridNearAtomicUpdateRequest<K, V>, GridNearAtomicUpdateResponse<K, V>>() {
+             @Override public void apply(GridNearAtomicUpdateRequest<K, V> req, GridNearAtomicUpdateResponse<K, V> res) {
+                 if (ctx.config().getAtomicWriteOrderMode() == CLOCK) {
+                     // Always send reply in CLOCK ordering mode.
+                     sendNearUpdateReply(res.nodeId(), res);
+ 
+                     return;
+                 }
+ 
+                 // Request should be for primary keys only in PRIMARY ordering mode.
+                 assert req.hasPrimary();
+ 
+                 if (req.writeSynchronizationMode() != FULL_ASYNC)
+                     sendNearUpdateReply(res.nodeId(), res);
+                 else {
+                     if (!F.isEmpty(res.remapKeys()))
+                         // Remap keys on primary node in FULL_ASYNC mode.
+                         remapToNewPrimary(req);
+                     else if (res.error() != null) {
+                         U.error(log, "Failed to process write update request in FULL_ASYNC mode for keys: " +
+                             res.failedKeys(), res.error());
+                     }
+                 }
+             }
+         };
+     }
+ 
+     /** {@inheritDoc} */
+     @SuppressWarnings({"IfMayBeConditional", "SimplifiableIfStatement"})
+     @Override public void start() throws IgniteCheckedException {
+         super.start();
+ 
 -        resetMetrics();
++        CacheMetricsAdapter m = new CacheMetricsAdapter(ctx);
++
++        if (ctx.dht().near() != null)
++            m.delegate(ctx.dht().near().metrics0());
++
++        metrics = m;
+ 
+         preldr = new GridDhtPreloader<>(ctx);
+ 
+         preldr.start();
+ 
+         ctx.io().addHandler(ctx.cacheId(), GridNearGetRequest.class, new CI2<UUID, GridNearGetRequest<K, V>>() {
+             @Override public void apply(UUID nodeId, GridNearGetRequest<K, V> req) {
+                 processNearGetRequest(nodeId, req);
+             }
+         });
+ 
+         ctx.io().addHandler(ctx.cacheId(), GridNearAtomicUpdateRequest.class, new CI2<UUID, GridNearAtomicUpdateRequest<K, V>>() {
+             @Override public void apply(UUID nodeId, GridNearAtomicUpdateRequest<K, V> req) {
+                 processNearAtomicUpdateRequest(nodeId, req);
+             }
+         });
+ 
+         ctx.io().addHandler(ctx.cacheId(), GridNearAtomicUpdateResponse.class, new CI2<UUID, GridNearAtomicUpdateResponse<K, V>>() {
+             @Override public void apply(UUID nodeId, GridNearAtomicUpdateResponse<K, V> res) {
+                 processNearAtomicUpdateResponse(nodeId, res);
+             }
+         });
+ 
+         ctx.io().addHandler(ctx.cacheId(), GridDhtAtomicUpdateRequest.class, new CI2<UUID, GridDhtAtomicUpdateRequest<K, V>>() {
+             @Override public void apply(UUID nodeId, GridDhtAtomicUpdateRequest<K, V> req) {
+                 processDhtAtomicUpdateRequest(nodeId, req);
+             }
+         });
+ 
+         ctx.io().addHandler(ctx.cacheId(), GridDhtAtomicUpdateResponse.class, new CI2<UUID, GridDhtAtomicUpdateResponse<K, V>>() {
+             @Override public void apply(UUID nodeId, GridDhtAtomicUpdateResponse<K, V> res) {
+                 processDhtAtomicUpdateResponse(nodeId, res);
+             }
+         });
+ 
+         ctx.io().addHandler(ctx.cacheId(), GridDhtAtomicDeferredUpdateResponse.class,
+             new CI2<UUID, GridDhtAtomicDeferredUpdateResponse<K, V>>() {
+                 @Override public void apply(UUID nodeId, GridDhtAtomicDeferredUpdateResponse<K, V> res) {
+                     processDhtAtomicDeferredUpdateResponse(nodeId, res);
+                 }
+             });
+ 
+         if (near == null) {
+             ctx.io().addHandler(ctx.cacheId(), GridNearGetResponse.class, new CI2<UUID, GridNearGetResponse<K, V>>() {
+                 @Override public void apply(UUID nodeId, GridNearGetResponse<K, V> res) {
+                     processNearGetResponse(nodeId, res);
+                 }
+             });
+         }
+ 
+         ctx.io().addDisconnectListener(new GridDisconnectListener() {
+             @Override public void onNodeDisconnected(UUID nodeId) {
+                 scheduleAtomicFutureRecheck();
+             }
+         });
+     }
+ 
 -    /** {@inheritDoc} */
 -    @Override public void resetMetrics() {
 -        GridCacheMetricsAdapter m = new GridCacheMetricsAdapter();
 -
 -        if (ctx.dht().near() != null)
 -            m.delegate(ctx.dht().near().metrics0());
 -
 -        metrics = m;
 -
 -        ctx.dr().resetMetrics();
 -    }
 -
+     /**
+      * @param near Near cache.
+      */
+     public void near(GridNearAtomicCache<K, V> near) {
+         this.near = near;
+     }
+ 
+     /** {@inheritDoc} */
+     @Override public GridNearCacheAdapter<K, V> near() {
+         return near;
+     }
+ 
+     /** {@inheritDoc} */
+     @Override public CacheEntry<K, V> entry(K key) {
+         return new GridDhtCacheEntryImpl<>(ctx.projectionPerCall(), ctx, key, null);
+     }
+ 
+     /** {@inheritDoc} */
+     @Override public V peek(K key, @Nullable Collection<GridCachePeekMode> modes) throws IgniteCheckedException {
+         GridTuple<V> val = null;
+ 
+         if (ctx.isReplicated() || !modes.contains(NEAR_ONLY)) {
+             try {
+                 val = peek0(true, key, modes, ctx.tm().txx());
+             }
+             catch (GridCacheFilterFailedException ignored) {
+                 if (log.isDebugEnabled())
+                     log.debug("Filter validation failed for key: " + key);
+ 
+                 return null;
+             }
+         }
+ 
+         return val != null ? val.get() : null;
+     }
+ 
+     /** {@inheritDoc} */
+     @Override public IgniteFuture<Map<K, V>> getAllAsync(
+         @Nullable final Collection<? extends K> keys,
+         final boolean forcePrimary,
+         boolean skipTx,
+         @Nullable final GridCacheEntryEx<K, V> entry,
+         @Nullable UUID subjId,
+         final String taskName,
+         final boolean deserializePortable,
+         @Nullable final IgnitePredicate<CacheEntry<K, V>>[] filter
+     ) {
+         GridCacheProjectionImpl<K, V> prj = ctx.projectionPerCall();
+ 
+         subjId = ctx.subjectIdPerCall(null, prj);
+ 
+         final UUID subjId0 = subjId;
+ 
+         final ExpiryPolicy expiryPlc = prj != null ? prj.expiry() : null;
+ 
+         return asyncOp(new CO<IgniteFuture<Map<K, V>>>() {
+             @Override public IgniteFuture<Map<K, V>> apply() {
+                 return getAllAsync0(keys,
+                     false,
+                     forcePrimary,
+                     filter,
+                     subjId0,
+                     taskName,
+                     deserializePortable,
+                     expiryPlc);
+             }
+         });
+     }
+ 
+     /** {@inheritDoc} */
+     @Override public V put(K key, V val, @Nullable GridCacheEntryEx<K, V> cached, long ttl,
+         @Nullable IgnitePredicate<CacheEntry<K, V>>[] filter) throws IgniteCheckedException {
+         return putAsync(key, val, cached, ttl, filter).get();
+     }
+ 
+     /** {@inheritDoc} */
+     @Override public boolean putx(K key, V val, @Nullable GridCacheEntryEx<K, V> cached,
+         long ttl, @Nullable IgnitePredicate<CacheEntry<K, V>>... filter) throws IgniteCheckedException {
+         return putxAsync(key, val, cached, ttl, filter).get();
+     }
+ 
+     /** {@inheritDoc} */
+     @Override public boolean putx(K key, V val,
+         IgnitePredicate<CacheEntry<K, V>>[] filter) throws IgniteCheckedException {
+         return putxAsync(key, val, filter).get();
+     }
+ 
+     /** {@inheritDoc} */
+     @SuppressWarnings("unchecked")
+     @Override public IgniteFuture<V> putAsync(K key, V val, @Nullable GridCacheEntryEx<K, V> entry,
+         long ttl, @Nullable IgnitePredicate<CacheEntry<K, V>>... filter) {
+         return updateAllAsync0(F0.asMap(key, val),
+             null,
+             null,
+             null,
+             null,
+             true,
+             false,
+             entry,
+             filter);
+     }
+ 
+     /** {@inheritDoc} */
+     @SuppressWarnings("unchecked")
+     @Override public IgniteFuture<Boolean> putxAsync(K key, V val, @Nullable GridCacheEntryEx<K, V> entry, long ttl,
+         @Nullable IgnitePredicate<CacheEntry<K, V>>... filter) {
+         return updateAllAsync0(F0.asMap(key, val),
+             null,
+             null,
+             null,
+             null,
+             false,
+             false,
+             entry,
+             filter);
+     }
+ 
+     /** {@inheritDoc} */
+     @Override public V putIfAbsent(K key, V val) throws IgniteCheckedException {
+         return putIfAbsentAsync(key, val).get();
+     }
+ 
+     /** {@inheritDoc} */
+     @Override public IgniteFuture<V> putIfAbsentAsync(K key, V val) {
+         return putAsync(key, val, ctx.noPeekArray());
+     }
+ 
+     /** {@inheritDoc} */
+     @Override public boolean putxIfAbsent(K key, V val) throws IgniteCheckedException {
+         return putxIfAbsentAsync(key, val).get();
+     }
+ 
+     /** {@inheritDoc} */
+     @Override public IgniteFuture<Boolean> putxIfAbsentAsync(K key, V val) {
+         return putxAsync(key, val, ctx.noPeekArray());
+     }
+ 
+     /** {@inheritDoc} */
+     @Override public V replace(K key, V val) throws IgniteCheckedException {
+         return replaceAsync(key, val).get();
+     }
+ 
+     /** {@inheritDoc} */
+     @Override public IgniteFuture<V> replaceAsync(K key, V val) {
+         return putAsync(key, val, ctx.hasPeekArray());
+     }
+ 
+     /** {@inheritDoc} */
+     @Override public boolean replacex(K key, V val) throws IgniteCheckedException {
+         return replacexAsync(key, val).get();
+     }
+ 
+     /** {@inheritDoc} */
+     @Override public IgniteFuture<Boolean> replacexAsync(K key, V val) {
+         return putxAsync(key, val, ctx.hasPeekArray());
+     }
+ 
+     /** {@inheritDoc} */
+     @Override public boolean replace(K key, V oldVal, V newVal) throws IgniteCheckedException {
+         return replaceAsync(key, oldVal, newVal).get();
+     }
+ 
+     /** {@inheritDoc} */
+     @Override public IgniteFuture<Boolean> replaceAsync(K key, V oldVal, V newVal) {
+         return putxAsync(key, newVal, ctx.equalsPeekArray(oldVal));
+     }
+ 
+     /** {@inheritDoc} */
+     @Override public GridCacheReturn<V> removex(K key, V val) throws IgniteCheckedException {
+         return removexAsync(key, val).get();
+     }
+ 
+     /** {@inheritDoc} */
+     @Override public GridCacheReturn<V> replacex(K key, V oldVal, V newVal) throws IgniteCheckedException {
+         return replacexAsync(key, oldVal, newVal).get();
+     }
+ 
+     /** {@inheritDoc} */
+     @SuppressWarnings("unchecked")
+     @Override public IgniteFuture<GridCacheReturn<V>> removexAsync(K key, V val) {
+         return removeAllAsync0(F.asList(key), null, null, true, true, ctx.equalsPeekArray(val));
+     }
+ 
+     /** {@inheritDoc} */
+     @SuppressWarnings("unchecked")
+     @Override public IgniteFuture<GridCacheReturn<V>> replacexAsync(K key, V oldVal, V newVal) {
+         return updateAllAsync0(F.asMap(key, newVal),
+             null,
+             null,
+             null,
+             null,
+             true,
+             true,
+             null,
+             ctx.equalsPeekArray(oldVal));
+     }
+ 
+     /** {@inheritDoc} */
+     @Override public void putAll(Map<? extends K, ? extends V> m,
+         IgnitePredicate<CacheEntry<K, V>>[] filter) throws IgniteCheckedException {
+         putAllAsync(m, filter).get();
+     }
+ 
+     /** {@inheritDoc} */
+     @Override public IgniteFuture<?> putAllAsync(Map<? extends K, ? extends V> m,
+         @Nullable IgnitePredicate<CacheEntry<K, V>>[] filter) {
+         return updateAllAsync0(m,
+             null,
+             null,
+             null,
+             null,
+             false,
+             false,
+             null,
+             filter);
+     }
+ 
+     /** {@inheritDoc} */
+     @Override public void putAllDr(Map<? extends K, GridCacheDrInfo<V>> drMap) throws IgniteCheckedException {
+         putAllDrAsync(drMap).get();
+     }
+ 
+     /** {@inheritDoc} */
+     @Override public IgniteFuture<?> putAllDrAsync(Map<? extends K, GridCacheDrInfo<V>> drMap) {
+         ctx.dr().onReceiveCacheEntriesReceived(drMap.size());
+ 
+         return updateAllAsync0(null,
+             null,
+             null,
+             drMap,
+             null,
+             false,
+             false,
+             null,
+             null);
+     }
+ 
+     /** {@inheritDoc} */
+     @Override public V remove(K key, @Nullable GridCacheEntryEx<K, V> entry,
+         @Nullable IgnitePredicate<CacheEntry<K, V>>... filter) throws IgniteCheckedException {
+         return removeAsync(key, entry, filter).get();
+     }
+ 
+     /** {@inheritDoc} */
+     @SuppressWarnings("unchecked")
+     @Override public IgniteFuture<V> removeAsync(K key, @Nullable GridCacheEntryEx<K, V> entry,
+         @Nullable IgnitePredicate<CacheEntry<K, V>>... filter) {
+         return removeAllAsync0(Collections.singletonList(key), null, entry, true, false, filter);
+     }
+ 
+     /** {@inheritDoc} */
+     @Override public void removeAll(Collection<? extends K> keys,
+         IgnitePredicate<CacheEntry<K, V>>... filter) throws IgniteCheckedException {
+         removeAllAsync(keys, filter).get();
+     }
+ 
+     /** {@inheritDoc} */
+     @Override public IgniteFuture<?> removeAllAsync(Collection<? extends K> keys,
+         IgnitePredicate<CacheEntry<K, V>>[] filter) {
+         A.notNull(keys, "keys");
+ 
+         return removeAllAsync0(keys, null, null, false, false, filter);
+     }
+ 
+     /** {@inheritDoc} */
+     @Override public boolean removex(K key, @Nullable GridCacheEntryEx<K, V> entry,
+         @Nullable IgnitePredicate<CacheEntry<K, V>>... filter) throws IgniteCheckedException {
+         return removexAsync(key, entry, filter).get();
+     }
+ 
+     /** {@inheritDoc} */
+     @SuppressWarnings("unchecked")
+     @Override public IgniteFuture<Boolean> removexAsync(K key, @Nullable GridCacheEntryEx<K, V> entry,
+         @Nullable IgnitePredicate<CacheEntry<K, V>>... filter) {
+         return removeAllAsync0(Collections.singletonList(key), null, entry, false, false, filter);
+     }
+ 
+     /** {@inheritDoc} */
+     @Override public boolean remove(K key, V val) throws IgniteCheckedException {
+         return removeAsync(key, val).get();
+     }
+ 
+     /** {@inheritDoc} */
+     @Override public IgniteFuture<Boolean> removeAsync(K key, V val) {
+         return removexAsync(key, ctx.equalsPeekArray(val));
+     }
+ 
+     /** {@inheritDoc} */
+     @Override public void removeAll(IgnitePredicate<CacheEntry<K, V>>[] filter) throws IgniteCheckedException {
+         removeAllAsync(filter).get();
+     }
+ 
+     /** {@inheritDoc} */
+     @Override public IgniteFuture<?> removeAllAsync(IgnitePredicate<CacheEntry<K, V>>[] filter) {
+         return removeAllAsync(keySet(filter), filter);
+     }
+ 
+     /** {@inheritDoc} */
+     @Override public void removeAllDr(Map<? extends K, GridCacheVersion> drMap) throws IgniteCheckedException {
+         removeAllDrAsync(drMap).get();
+     }
+ 
+     /** {@inheritDoc} */
+     @Override public IgniteFuture<?> removeAllDrAsync(Map<? extends K, GridCacheVersion> drMap) {
+         ctx.dr().onReceiveCacheEntriesReceived(drMap.size());
+ 
+         return removeAllAsync0(null, drMap, null, false, false, null);
+     }
+ 
+     /**
+      * @return {@code True} if store write-through enabled.
+      */
+     private boolean writeThrough() {
+         return ctx.writeThrough() && ctx.store().configured();
+     }
+ 
+     /**
+      * @param op Operation closure.
+      * @return Future.
+      */
+     @SuppressWarnings("unchecked")
+     protected <T> IgniteFuture<T> asyncOp(final CO<IgniteFuture<T>> op) {
+         IgniteFuture<T> fail = asyncOpAcquire();
+ 
+         if (fail != null)
+             return fail;
+ 
+         FutureHolder holder = lastFut.get();
+ 
+         holder.lock();
+ 
+         try {
+             IgniteFuture fut = holder.future();
+ 
+             if (fut != null && !fut.isDone()) {
+                 IgniteFuture<T> f = new GridEmbeddedFuture<>(fut,
+                     new C2<T, Exception, IgniteFuture<T>>() {
+                         @Override public IgniteFuture<T> apply(T t, Exception e) {
+                             return op.apply();
+                         }
+                     }, ctx.kernalContext());
+ 
+                 saveFuture(holder, f);
+ 
+                 return f;
+             }
+ 
+             IgniteFuture<T> f = op.apply();
+ 
+             saveFuture(holder, f);
+ 
+             return f;
+         }
+         finally {
+             holder.unlock();
+         }
+     }
+ 
+     /** {@inheritDoc} */
+     @Override protected IgniteFuture<Boolean> lockAllAsync(Collection<? extends K> keys,
+         long timeout,
+         @Nullable IgniteTxLocalEx<K, V> tx,
+         boolean isInvalidate,
+         boolean isRead,
+         boolean retval,
+         @Nullable IgniteTxIsolation isolation,
+         long accessTtl,
+         IgnitePredicate<CacheEntry<K, V>>[] filter) {
+         return new FinishedLockFuture(new UnsupportedOperationException("Locks are not supported for " +
+             "CacheAtomicityMode.ATOMIC mode (use CacheAtomicityMode.TRANSACTIONAL instead)"));
+     }
+ 
+     /** {@inheritDoc} */
+     @Override public <T> EntryProcessorResult<T> invoke(K key, EntryProcessor<K, V, T> entryProcessor, Object... args)
+         throws IgniteCheckedException {
+         return invokeAsync(key, entryProcessor, args).get();
+     }
+ 
+     /** {@inheritDoc} */
+     @Override public <T> Map<K, EntryProcessorResult<T>> invokeAll(Set<? extends K> keys,
+         EntryProcessor<K, V, T> entryProcessor,
+         Object... args)
+         throws IgniteCheckedException {
+         return invokeAllAsync(keys, entryProcessor, args).get();
+     }
+ 
+     /** {@inheritDoc} */
+     @SuppressWarnings("unchecked")
+     @Override public <T> IgniteFuture<EntryProcessorResult<T>> invokeAsync(K key,
+         EntryProcessor<K, V, T> entryProcessor,
+         Object... args) {
+         A.notNull(key, "key", entryProcessor, "entryProcessor");
+ 
+         if (keyCheck)
+             validateCacheKey(key);
+ 
+         ctx.denyOnLocalRead();
+ 
+         Map<? extends K, EntryProcessor> invokeMap =
+             Collections.singletonMap(key, (EntryProcessor)entryProcessor);
+ 
+         IgniteFuture<Map<K, EntryProcessorResult<T>>> fut = updateAllAsync0(null,
+             invokeMap,
+             args,
+             null,
+             null,
+             true,
+             false,
+             null,
+             null);
+ 
+         return fut.chain(new CX1<IgniteFuture<Map<K, EntryProcessorResult<T>>>, EntryProcessorResult<T>>() {
+             @Override public EntryProcessorResult<T> applyx(IgniteFuture<Map<K, EntryProcessorResult<T>>> fut)
+                 throws IgniteCheckedException {
+                 Map<K, EntryProcessorResult<T>> resMap = fut.get();
+ 
+                 if (resMap != null) {
+                     assert resMap.isEmpty() || resMap.size() == 1 : resMap.size();
+ 
+                     return resMap.isEmpty() ? null : resMap.values().iterator().next();
+                 }
+ 
+                 return null;
+             }
+         });
+     }
+ 
+     /** {@inheritDoc} */
+     @SuppressWarnings("unchecked")
+     @Override public <T> IgniteFuture<Map<K, EntryProcessorResult<T>>> invokeAllAsync(Set<? extends K> keys,
+         final EntryProcessor<K, V, T> entryProcessor,
+         Object... args) {
+         A.notNull(keys, "keys", entryProcessor, "entryProcessor");
+ 
+         if (keyCheck)
+             validateCacheKeys(keys);
+ 
+         ctx.denyOnLocalRead();
+ 
+         Map<? extends K, EntryProcessor> invokeMap = F.viewAsMap(keys, new C1<K, EntryProcessor>() {
+             @Override public EntryProcessor apply(K k) {
+                 return entryProcessor;
+             }
+         });
+ 
+         return updateAllAsync0(null,
+             invokeMap,
+             args,
+             null,
+             null,
+             true,
+             false,
+             null,
+             null);
+     }
+ 
+     /** {@inheritDoc} */
+     @Override public <T> Map<K, EntryProcessorResult<T>> invokeAll(
+         Map<? extends K, ? extends EntryProcessor<K, V, T>> map,
+         Object... args) throws IgniteCheckedException {
+         return invokeAllAsync(map, args).get();
+     }
+ 
+     /** {@inheritDoc} */
+     @Override public <T> IgniteFuture<Map<K, EntryProcessorResult<T>>> invokeAllAsync(
+         Map<? extends K, ? extends EntryProcessor<K, V, T>> map,
+         Object... args) {
+         A.notNull(map, "map");
+ 
+         if (keyCheck)
+             validateCacheKeys(map.keySet());
+ 
+         ctx.denyOnLocalRead();
+ 
+         return updateAllAsync0(null,
+             map,
+             args,
+             null,
+             null,
+             true,
+             false,
+             null,
+             null);
+     }
+ 
+     /**
+      * Entry point for all public API put/transform methods.
+      *
+      * @param map Put map. Either {@code map}, {@code invokeMap} or {@code drMap} should be passed.
+      * @param invokeMap Invoke map. Either {@code map}, {@code invokeMap} or {@code drMap} should be passed.
+      * @param invokeArgs Optional arguments for EntryProcessor.
+      * @param drPutMap DR put map.
+      * @param drRmvMap DR remove map.
+      * @param retval Return value required flag.
+      * @param rawRetval Return {@code GridCacheReturn} instance.
+      * @param cached Cached cache entry for key. May be passed if and only if map size is {@code 1}.
+      * @param filter Cache entry filter for atomic updates.
+      * @return Completion future.
+      */
+     private IgniteFuture updateAllAsync0(
+         @Nullable final Map<? extends K, ? extends V> map,
+         @Nullable final Map<? extends K, ? extends EntryProcessor> invokeMap,
+         @Nullable Object[] invokeArgs,
+         @Nullable final Map<? extends K, GridCacheDrInfo<V>> drPutMap,
+         @Nullable final Map<? extends K, GridCacheVersion> drRmvMap,
+         final boolean retval,
+         final boolean rawRetval,
+         @Nullable GridCacheEntryEx<K, V> cached,
+         @Nullable final IgnitePredicate<CacheEntry<K, V>>[] filter
+     ) {
+         if (map != null && keyCheck)
+             validateCacheKeys(map.keySet());
+ 
+         ctx.checkSecurity(GridSecurityPermission.CACHE_PUT);
+ 
+         GridCacheProjectionImpl<K, V> prj = ctx.projectionPerCall();
+ 
+         UUID subjId = ctx.subjectIdPerCall(null, prj);
+ 
+         int taskNameHash = ctx.kernalContext().job().currentTaskNameHash();
+ 
+         final GridNearAtomicUpdateFuture<K, V> updateFut = new GridNearAtomicUpdateFuture<>(
+             ctx,
+             this,
+             ctx.config().getWriteSynchronizationMode(),
+             invokeMap != null ? TRANSFORM : UPDATE,
+             map != null ? map.keySet() : invokeMap != null ? invokeMap.keySet() : drPutMap != null ?
+                 drPutMap.keySet() : drRmvMap.keySet(),
+             map != null ? map.values() : invokeMap != null ? invokeMap.values() : null,
+             invokeArgs,
+             drPutMap != null ? drPutMap.values() : null,
+             drRmvMap != null ? drRmvMap.values() : null,
+             retval,
+             rawRetval,
+             cached,
+             prj != null ? prj.expiry() : null,
+             filter,
+             subjId,
+             taskNameHash);
+ 
+         return asyncOp(new CO<IgniteFuture<Object>>() {
+             @Override public IgniteFuture<Object> apply() {
+                 updateFut.map();
+ 
+                 return updateFut;
+             }
+         });
+     }
+ 
+     /**
+      * Entry point for all public API remove methods.
+      *
+      * @param keys Keys to remove.
+      * @param drMap DR map.
+      * @param cached Cached cache entry for key. May be passed if and only if keys size is {@code 1}.
+      * @param retval Return value required flag.
+      * @param rawRetval Return {@code GridCacheReturn} instance.
+      * @param filter Cache entry filter for atomic removes.
+      * @return Completion future.
+      */
+     private IgniteFuture removeAllAsync0(
+         @Nullable final Collection<? extends K> keys,
+         @Nullable final Map<? extends K, GridCacheVersion> drMap,
+         @Nullable GridCacheEntryEx<K, V> cached,
+         final boolean retval,
+         boolean rawRetval,
+         @Nullable final IgnitePredicate<CacheEntry<K, V>>[] filter
+     ) {
++        final boolean statsEnabled = ctx.config().isStatisticsEnabled();
++
++        final long start = statsEnabled ? System.nanoTime() : 0L;
++
+         assert keys != null || drMap != null;
+ 
+         if (keyCheck)
+             validateCacheKeys(keys);
+ 
+         ctx.checkSecurity(GridSecurityPermission.CACHE_REMOVE);
+ 
+         GridCacheProjectionImpl<K, V> prj = ctx.projectionPerCall();
+ 
+         UUID subjId = ctx.subjectIdPerCall(null, prj);
+ 
+         int taskNameHash = ctx.kernalContext().job().currentTaskNameHash();
+ 
+         final GridNearAtomicUpdateFuture<K, V> updateFut = new GridNearAtomicUpdateFuture<>(
+             ctx,
+             this,
+             ctx.config().getWriteSynchronizationMode(),
+             DELETE,
+             keys != null ? keys : drMap.keySet(),
+             null,
+             null,
+             null,
+             keys != null ? null : drMap.values(),
+             retval,
+             rawRetval,
+             cached,
+             (filter != null && prj != null) ? prj.expiry() : null,
+             filter,
+             subjId,
+             taskNameHash);
+ 
++        if (statsEnabled)
++            updateFut.listenAsync(new UpdateRemoveTimeStatClosure<>(metrics0(), start));
++
+         return asyncOp(new CO<IgniteFuture<Object>>() {
+             @Override public IgniteFuture<Object> apply() {
+                 updateFut.map();
+ 
+                 return updateFut;
+             }
+         });
+     }
+ 
+     /**
+      * Entry point to all public API get methods.
+      *
+      * @param keys Keys to remove.
+      * @param reload Reload flag.
+      * @param forcePrimary Force primary flag.
+      * @param filter Filter.
+      * @param subjId Subject ID.
+      * @param taskName Task name.
+      * @param deserializePortable Deserialize portable flag.
+      * @param expiryPlc Expiry policy.
+      * @return Get future.
+      */
+     private IgniteFuture<Map<K, V>> getAllAsync0(@Nullable Collection<? extends K> keys,
+         boolean reload,
+         boolean forcePrimary,
+         @Nullable IgnitePredicate<CacheEntry<K, V>>[] filter,
+         UUID subjId,
+         String taskName,
+         boolean deserializePortable,
+         @Nullable ExpiryPolicy expiryPlc) {
+         ctx.checkSecurity(GridSecurityPermission.CACHE_READ);
+ 
+         if (F.isEmpty(keys))
+             return new GridFinishedFuture<>(ctx.kernalContext(), Collections.<K, V>emptyMap());
+ 
+         if (keyCheck)
+             validateCacheKeys(keys);
+ 
+         long topVer = ctx.affinity().affinityTopologyVersion();
+ 
+         final GetExpiryPolicy expiry = accessExpiryPolicy(expiryPlc);
+ 
+         // Optimisation: try to resolve value locally and escape 'get future' creation.
+         if (!reload && !forcePrimary) {
+             Map<K, V> locVals = new HashMap<>(keys.size(), 1.0f);
+ 
+             boolean success = true;
+ 
+             // Optimistically expect that all keys are available locally (avoid creation of get future).
+             for (K key : keys) {
+                 GridCacheEntryEx<K, V> entry = null;
+ 
+                 while (true) {
+                     try {
+                         entry = ctx.isSwapOrOffheapEnabled() ? entryEx(key) : peekEx(key);
+ 
+                         // If our DHT cache do has value, then we peek it.
+                         if (entry != null) {
+                             boolean isNew = entry.isNewLocked();
+ 
+                             V v = entry.innerGet(null,
+                                 /*swap*/true,
+                                 /*read-through*/false,
+                                 /*fail-fast*/true,
+                                 /*unmarshal*/true,
 -                                /**update-metrics*/true,
++                                /**update-metrics*/false,
+                                 /*event*/true,
+                                 /*temporary*/false,
+                                 subjId,
+                                 null,
+                                 taskName,
+                                 filter,
+                                 expiry);
+ 
+                             // Entry was not in memory or in swap, so we remove it from cache.
+                             if (v == null) {
+                                 GridCacheVersion obsoleteVer = context().versions().next();
+ 
+                                 if (isNew && entry.markObsoleteIfEmpty(obsoleteVer))
+                                     removeIfObsolete(key);
+ 
+                                 success = false;
+                             }
+                             else {
+                                 if (ctx.portableEnabled() && deserializePortable) {
+                                     key = (K)ctx.unwrapPortableIfNeeded(key, false);
+                                     v = (V)ctx.unwrapPortableIfNeeded(v, false);
+                                 }
+ 
+                                 locVals.put(key, v);
+                             }
+                         }
+                         else
+                             success = false;
+ 
+                         break; // While.
+                     }
+                     catch (GridCacheEntryRemovedException ignored) {
+                         // No-op, retry.
+                     }
+                     catch (GridCacheFilterFailedException ignored) {
+                         // No-op, skip the key.
+                         break;
+                     }
+                     catch (GridDhtInvalidPartitionException ignored) {
+                         success = false;
+ 
+                         break; // While.
+                     }
+                     catch (IgniteCheckedException e) {
+                         return new GridFinishedFuture<>(ctx.kernalContext(), e);
+                     }
+                     finally {
+                         if (entry != null)
+                             ctx.evicts().touch(entry, topVer);
+                     }
+                 }
+ 
+                 if (!success)
+                     break;
++                else
++                    metrics0().onRead(true);
+             }
+ 
+             if (success) {
+                 sendTtlUpdateRequest(expiry);
+ 
+                 return ctx.wrapCloneMap(new GridFinishedFuture<>(ctx.kernalContext(), locVals));
+             }
+         }
+ 
+         if (expiry != null)
+             expiry.reset();
+ 
+         // Either reload or not all values are available locally.
+         GridPartitionedGetFuture<K, V> fut = new GridPartitionedGetFuture<>(ctx,
+             keys,
+             topVer,
+             true,
+             reload,
+             forcePrimary,
+             filter,
+             subjId,
+             taskName,
+             deserializePortable,
+             expiry);
+ 
+         fut.init();
+ 
+         return ctx.wrapCloneMap(fut);
+     }
+ 
+     /**
+      * Executes local update.
+      *
+      * @param nodeId Node ID.
+      * @param req Update request.
+      * @param cached Cached entry if updating single local entry.
+      * @param completionCb Completion callback.
+      */
+     public void updateAllAsyncInternal(
+         final UUID nodeId,
+         final GridNearAtomicUpdateRequest<K, V> req,
+         @Nullable final GridCacheEntryEx<K, V> cached,
+         final CI2<GridNearAtomicUpdateRequest<K, V>, GridNearAtomicUpdateResponse<K, V>> completionCb
+     ) {
+         IgniteFuture<Object> forceFut = preldr.request(req.keys(), req.topologyVersion());
+ 
+         if (forceFut.isDone())
+             updateAllAsyncInternal0(nodeId, req, completionCb);
+         else {
+             forceFut.listenAsync(new CI1<IgniteFuture<Object>>() {
+                 @Override public void apply(IgniteFuture<Object> t) {
+                     updateAllAsyncInternal0(nodeId, req, completionCb);
+                 }
+             });
+         }
+     }
+ 
+     /**
+      * Executes local update after preloader fetched values.
+      *
+      * @param nodeId Node ID.
+      * @param req Update request.
+      * @param completionCb Completion callback.
+      */
+     public void updateAllAsyncInternal0(
+         UUID nodeId,
+         GridNearAtomicUpdateRequest<K, V> req,
+         CI2<GridNearAtomicUpdateRequest<K, V>, GridNearAtomicUpdateResponse<K, V>> completionCb
+     ) {
+         GridNearAtomicUpdateResponse<K, V> res = new GridNearAtomicUpdateResponse<>(ctx.cacheId(), nodeId,
+             req.futureVersion());
+ 
+         List<K> keys = req.keys();
+ 
+         assert !req.returnValue() || (req.operation() == TRANSFORM || keys.size() == 1);
+ 
+         GridDhtAtomicUpdateFuture<K, V> dhtFut = null;
+ 
+         boolean remap = false;
+ 
+         String taskName = ctx.kernalContext().task().resolveTaskName(req.taskNameHash());
+ 
+         IgniteCacheExpiryPolicy expiry = null;
+ 
+         try {
+             // If batch store update is enabled, we need to lock all entries.
+             // First, need to acquire locks on cache entries, then check filter.
+             List<GridDhtCacheEntry<K, V>> locked = lockEntries(keys, req.topologyVersion());
+             Collection<IgniteBiTuple<GridDhtCacheEntry<K, V>, GridCacheVersion>> deleted = null;
+ 
+             try {
+                 topology().readLock();
+ 
+                 try {
+                     // Do not check topology version for CLOCK versioning since
+                     // partition exchange will wait for near update future.
+                     if (topology().topologyVersion() == req.topologyVersion() ||
+                         ctx.config().getAtomicWriteOrderMode() == CLOCK) {
+                         ClusterNode node = ctx.discovery().node(nodeId);
+ 
+                         if (node == null) {
+                             U.warn(log, "Node originated update request left grid: " + nodeId);
+ 
+                             return;
+                         }
+ 
+                         checkClearForceTransformBackups(req, locked);
+ 
+                         boolean hasNear = U.hasNearCache(node, name());
+ 
+                         GridCacheVersion ver = req.updateVersion();
+ 
+                         if (ver == null) {
+                             // Assign next version for update inside entries lock.
+                             ver = ctx.versions().next(req.topologyVersion());
+ 
+                             if (hasNear)
+                                 res.nearVersion(ver);
+                         }
+ 
+                         assert ver != null : "Got null version for update request: " + req;
+ 
+                         if (log.isDebugEnabled())
+                             log.debug("Using cache version for update request on primary node [ver=" + ver +
+                                 ", req=" + req + ']');
+ 
+                         dhtFut = createDhtFuture(ver, req, res, completionCb, false);
+ 
+                         GridCacheReturn<Object> retVal = null;
+ 
+                         boolean replicate = ctx.isDrEnabled();
+ 
+                         ExpiryPolicy plc = req.expiry() != null ? req.expiry() : ctx.expiry();
+ 
+                         if (plc != null)
+                             expiry = new UpdateExpiryPolicy(plc);
+ 
+                         if (writeThrough() && keys.size() > 1 && !ctx.dr().receiveEnabled()) {
+                             // This method can only be used when there are no replicated entries in the batch.
+                             UpdateBatchResult<K, V> updRes = updateWithBatch(node,
+                                 hasNear,
+                                 req,
+                                 res,
+                                 locked,
+                                 ver,
+                                 dhtFut,
+                                 completionCb,
+                                 replicate,
+                                 taskName,
+                                 expiry);
+ 
+                             deleted = updRes.deleted();
+                             dhtFut = updRes.dhtFuture();
+ 
+                             if (req.operation() == TRANSFORM)
+                                 retVal = new GridCacheReturn<>((Object)updRes.invokeResults(), true);
+                         }
+                         else {
+                             UpdateSingleResult<K, V> updRes = updateSingle(node,
+                                 hasNear,
+                                 req,
+                                 res,
+                                 locked,
+                                 ver,
+                                 dhtFut,
+                                 completionCb,
+                                 replicate,
+                                 taskName,
+                                 expiry);
+ 
+                             retVal = updRes.returnValue();
+                             deleted = updRes.deleted();
+                             dhtFut = updRes.dhtFuture();
+                         }
+ 
+                         if (retVal == null)
+                             retVal = new GridCacheReturn<>(null, true);
+ 
+                         res.returnValue(retVal);
+                     }
+                     else
+                         // Should remap all keys.
+                         remap = true;
+                 }
+                 finally {
+                     topology().readUnlock();
+                 }
+             }
+             catch (GridCacheEntryRemovedException e) {
+                 assert false : "Entry should not become obsolete while holding lock.";
+ 
+                 e.printStackTrace();
+             }
+             finally {
+                 unlockEntries(locked, req.topologyVersion());
+ 
+                 // Enqueue if necessary after locks release.
+                 if (deleted != null) {
+                     assert !deleted.isEmpty();
+                     assert ctx.deferredDelete();
+ 
+                     for (IgniteBiTuple<GridDhtCacheEntry<K, V>, GridCacheVersion> e : deleted)
+                         ctx.onDeferredDelete(e.get1(), e.get2());
+                 }
+             }
+         }
+         catch (GridDhtInvalidPartitionException ignore) {
+             assert ctx.config().getAtomicWriteOrderMode() == PRIMARY;
+ 
+             if (log.isDebugEnabled())
+                 log.debug("Caught invalid partition exception for cache entry (will remap update request): " + req);
+ 
+             remap = true;
+         }
+ 
+         if (remap) {
+             assert dhtFut == null;
+ 
+             res.remapKeys(req.keys());
+ 
+             completionCb.apply(req, res);
+         }
+         else {
+             // If there are backups, map backup update future.
+             if (dhtFut != null)
+                 dhtFut.map();
+                 // Otherwise, complete the call.
+             else
+                 completionCb.apply(req, res);
+         }
+ 
+         sendTtlUpdateRequest(expiry);
+     }
+ 
+     /**
+      * Updates locked entries using batched write-through.
+      *
+      * @param node Sender node.
+      * @param hasNear {@code True} if originating node has near cache.
+      * @param req Update request.
+      * @param res Update response.
+      * @param locked Locked entries.
+      * @param ver Assigned version.
+      * @param dhtFut Optional DHT future.
+      * @param completionCb Completion callback to invoke when DHT future is completed.
+      * @param replicate Whether replication is enabled.
+      * @param taskName Task name.
+      * @param expiry Expiry policy.
+      * @return Deleted entries.
+      * @throws GridCacheEntryRemovedException Should not be thrown.
+      */
+     @SuppressWarnings("unchecked")
+     private UpdateBatchResult<K, V> updateWithBatch(
+         ClusterNode node,
+         boolean hasNear,
+         GridNearAtomicUpdateRequest<K, V> req,
+         GridNearAtomicUpdateResponse<K, V> res,
+         List<GridDhtCacheEntry<K, V>> locked,
+         GridCacheVersion ver,
+         @Nullable GridDhtAtomicUpdateFuture<K, V> dhtFut,
+         CI2<GridNearAtomicUpdateRequest<K, V>, GridNearAtomicUpdateResponse<K, V>> completionCb,
+         boolean replicate,
+         String taskName,
+         @Nullable IgniteCacheExpiryPolicy expiry
+     ) throws GridCacheEntryRemovedException {
+         assert !ctx.dr().receiveEnabled(); // Cannot update in batches during DR due to possible conflicts.
+         assert !req.returnValue() || req.operation() == TRANSFORM; // Should not request return values for putAll.
+ 
+         if (!F.isEmpty(req.filter()) && ctx.loadPreviousValue()) {
+             try {
+                 reloadIfNeeded(locked);
+             }
+             catch (IgniteCheckedException e) {
+                 res.addFailedKeys(req.keys(), e);
+ 
+                 return new UpdateBatchResult<>();
+             }
+         }
+ 
+         int size = req.keys().size();
+ 
+         Map<K, V> putMap = null;
+ 
+         Map<K, EntryProcessor<K, V, ?>> entryProcessorMap = null;
+ 
+         Collection<K> rmvKeys = null;
+ 
+         UpdateBatchResult<K, V> updRes = new UpdateBatchResult<>();
+ 
+         List<GridDhtCacheEntry<K, V>> filtered = new ArrayList<>(size);
+ 
+         GridCacheOperation op = req.operation();
+ 
+         Map<K, EntryProcessorResult> invokeResMap =
+             op == TRANSFORM ? U.<K, EntryProcessorResult>newHashMap(size) : null;
+ 
+         int firstEntryIdx = 0;
+ 
+         boolean intercept = ctx.config().getInterceptor() != null;
+ 
+         for (int i = 0; i < locked.size(); i++) {
+             GridDhtCacheEntry<K, V> entry = locked.get(i);
+ 
+             if (entry == null)
+                 continue;
+ 
+             try {
+                 if (!checkFilter(entry, req, res)) {
+                     if (expiry != null && entry.hasValue()) {
+                         long ttl = expiry.forAccess();
+ 
+                         if (ttl != -1L) {
+                             entry.updateTtl(null, ttl);
+ 
+                             expiry.ttlUpdated(entry.key(),
+                                 entry.getOrMarshalKeyBytes(),
+                                 entry.version(),
+                                 entry.readers());
+                         }
+                     }
+ 
+                     if (log.isDebugEnabled())
+                         log.debug("Entry did not pass the filter (will skip write) [entry=" + entry +
+                             ", filter=" + Arrays.toString(req.filter()) + ", res=" + res + ']');
+ 
+                     if (hasNear)
+                         res.addSkippedIndex(i);
+ 
+                     firstEntryIdx++;
+ 
+                     continue;
+                 }
+ 
+                 if (op == TRANSFORM) {
+                     EntryProcessor<K, V, ?> entryProcessor = req.entryProcessor(i);
+ 
+                     V old = entry.innerGet(
+                         null,
+                         /*read swap*/true,
+                         /*read through*/true,
+                         /*fail fast*/false,
+                         /*unmarshal*/true,
+                         /*metrics*/true,
+                         /*event*/true,
+                         /*temporary*/true,
+                         req.subjectId(),
+                         entryProcessor,
+                         taskName,
+                         CU.<K, V>empty(),
+                         null);
+ 
+                     CacheInvokeEntry<K, V> invokeEntry = new CacheInvokeEntry<>(entry.key(), old);
+ 
+                     V updated;
+                     CacheInvokeResult invokeRes = null;
+ 
+                     try {
+                         Object computed = entryProcessor.process(invokeEntry, req.invokeArguments());
+ 
+                         updated = ctx.unwrapTemporary(invokeEntry.getValue());
+ 
+                         if (computed != null)
+                             invokeRes = new CacheInvokeResult<>(ctx.unwrapTemporary(computed));
+                     }
+                     catch (Exception e) {
+                         invokeRes = new CacheInvokeResult<>(e);
+ 
+                         updated = old;
+                     }
+ 
+                     if (invokeRes != null)
+                         invokeResMap.put(entry.key(), invokeRes);
+ 
+                     if (updated == null) {
+                         if (intercept) {
+                             IgniteBiTuple<Boolean, ?> interceptorRes = ctx.config().getInterceptor().onBeforeRemove(
+                                 entry.key(), old);
+ 
+                             if (ctx.cancelRemove(interceptorRes))
+                                 continue;
+                         }
+ 
+                         // Update previous batch.
+                         if (putMap != null) {
+                             dhtFut = updatePartialBatch(
+                                 hasNear,
+                                 firstEntryIdx,
+                                 filtered,
+                                 ver,
+                                 node,
+                                 putMap,
+                                 null,
+                                 entryProcessorMap,
+                                 dhtFut,
+                                 completionCb,
+                                 req,
+                                 res,
+                                 replicate,
+                                 updRes,
+                                 taskName,
+                                 expiry);
+ 
+                             firstEntryIdx = i + 1;
+ 
+                             putMap = null;
+                             entryProcessorMap = null;
+ 
+                             filtered = new ArrayList<>();
+                         }
+ 
+                         // Start collecting new batch.
+                         if (rmvKeys == null)
+                             rmvKeys = new ArrayList<>(size);
+ 
+                         rmvKeys.add(entry.key());
+                     }
+                     else {
+                         if (intercept) {
+                             updated = (V)ctx.config().getInterceptor().onBeforePut(entry.key(), old, updated);
+ 
+                             if (updated == null)
+                                 continue;
+                         }
+ 
+                         // Update previous batch.
+                         if (rmvKeys != null) {
+                             dhtFut = updatePartialBatch(
+                                 hasNear,
+                                 firstEntryIdx,
+                                 filtered,
+                                 ver,
+                                 node,
+                                 null,
+                                 rmvKeys,
+                                 entryProcessorMap,
+                                 dhtFut,
+                                 completionCb,
+                                 req,
+                                 res,
+                                 replicate,
+                                 updRes,
+                                 taskName,
+                                 expiry);
+ 
+                             firstEntryIdx = i + 1;
+ 
+                             rmvKeys = null;
+                             entryProcessorMap = null;
+ 
+                             filtered = new ArrayList<>();
+                         }
+ 
+                         if (putMap == null)
+                             putMap = new LinkedHashMap<>(size, 1.0f);
+ 
+                         putMap.put(entry.key(), ctx.<V>unwrapTemporary(updated));
+                     }
+ 
+                     if (entryProcessorMap == null)
+                         entryProcessorMap = new HashMap<>();
+ 
+                     entryProcessorMap.put(entry.key(), entryProcessor);
+                 }
+                 else if (op == UPDATE) {
+                     V updated = req.value(i);
+ 
+                     if (intercept) {
+                         V old = entry.innerGet(
+                              null,
+                             /*read swap*/true,
+                             /*read through*/ctx.loadPreviousValue(),
+                             /*fail fast*/false,
+                             /*unmarshal*/true,
+                             /*metrics*/true,
+                             /*event*/true,
+                             /*temporary*/true,
+                             req.subjectId(),
+                             null,
+                             taskName,
+                             CU.<K, V>empty(),
+                             null);
+ 
+                         updated = (V)ctx.config().getInterceptor().onBeforePut(entry.key(), old, updated);
+ 
+                         if (updated == null)
+                             continue;
+ 
+                         updated = ctx.unwrapTemporary(updated);
+                     }
+ 
+                     assert updated != null;
+ 
+                     if (putMap == null)
+                         putMap = new LinkedHashMap<>(size, 1.0f);
+ 
+                     putMap.put(entry.key(), updated);
+                 }
+                 else {
+                     assert op == DELETE;
+ 
+                     if (intercept) {
+                         V old = entry.innerGet(
+                             null,
+                             /*read swap*/true,
+                             /*read through*/ctx.loadPreviousValue(),
+                             /*fail fast*/false,
+                             /*unmarshal*/true,
+                             /*metrics*/true,
+                             /*event*/true,
+                             /*temporary*/true,
+                             req.subjectId(),
+                             null,
+                             taskName,
+                             CU.<K, V>empty(),
+                             null);
+ 
+                         IgniteBiTuple<Boolean, ?> interceptorRes = ctx.config().getInterceptor().onBeforeRemove(
+                             entry.key(), old);
+ 
+                         if (ctx.cancelRemove(interceptorRes))
+                             continue;
+                     }
+ 
+                     if (rmvKeys == null)
+                         rmvKeys = new ArrayList<>(size);
+ 
+                     rmvKeys.add(entry.key());
+                 }
+ 
+                 filtered.add(entry);
+             }
+             catch (IgniteCheckedException e) {
+                 res.addFailedKey(entry.key(), e);
+             }
+             catch (GridCacheFilterFailedException ignore) {
+                 assert false : "Filter should never fail with failFast=false and empty filter.";
+             }
+         }
+ 
+         // Store final batch.
+         if (putMap != null || rmvKeys != null) {
+             dhtFut = updatePartialBatch(
+                 hasNear,
+                 firstEntryIdx,
+                 filtered,
+                 ver,
+                 node,
+                 putMap,
+                 rmvKeys,
+                 entryProcessorMap,
+                 dhtFut,
+                 completionCb,
+                 req,
+                 res,
+                 replicate,
+                 updRes,
+                 taskName,
+                 expiry);
+         }
+         else
+             assert filtered.isEmpty();
+ 
+         updRes.dhtFuture(dhtFut);
+ 
+         updRes.invokeResult(invokeResMap);
+ 
+         return updRes;
+     }
+ 
+     /**
+      * @param entries Entries.
+      * @throws IgniteCheckedException If failed.
+      */
+     private void reloadIfNeeded(final List<GridDhtCacheEntry<K, V>> entries) throws IgniteCheckedException {
+         Map<K, Integer> needReload = null;
+ 
+         for (int i = 0; i < entries.size(); i++) {
+             GridDhtCacheEntry<K, V> entry = entries.get(i);
+ 
+             if (entry == null)
+                 continue;
+ 
+             V val = entry.rawGetOrUnmarshal(false);
+ 
+             if (val == null) {
+                 if (needReload == null)
+                     needReload = new HashMap<>(entries.size(), 1.0f);
+ 
+                 needReload.put(entry.key(), i);
+             }
+         }
+ 
+         if (needReload != null) {
+             final Map<K, Integer> idxMap = needReload;
+ 
+             ctx.store().loadAllFromStore(null, needReload.keySet(), new CI2<K, V>() {
+                 @Override public void apply(K k, V v) {
+                     Integer idx = idxMap.get(k);
+ 
+                     if (idx != null) {
+                         GridDhtCacheEntry<K, V> entry = entries.get(idx);
+                         try {
+                             GridCacheVersion ver = entry.version();
+ 
+                             entry.versionedValue(v, null, ver);
+                         }
+                         catch (GridCacheEntryRemovedException e) {
+                             assert false : "Entry should not get obsolete while holding lock [entry=" + entry +
+                                 ", e=" + e + ']';
+                         }
+                         catch (IgniteCheckedException e) {
+                             throw new IgniteException(e);
+                         }
+                     }
+                 }
+             });
+         }
+     }
+ 
+     /**
+      * Updates locked entries one-by-one.
+      *
+      * @param node Originating node.
+      * @param hasNear {@code True} if originating node has near cache.
+      * @param req Update request.
+      * @param res Update response.
+      * @param locked Locked entries.
+      * @param ver Assigned update version.
+      * @param dhtFut Optional DHT future.
+      * @param completionCb Completion callback to invoke when DHT future is completed.
+      * @param replicate Whether DR is enabled for that cache.
+      * @param taskName Task name.
+      * @param expiry Expiry policy.
+      * @return Return value.
+      * @throws GridCacheEntryRemovedException Should be never thrown.
+      */
+     private UpdateSingleResult<K, V> updateSingle(
+         ClusterNode node,
+         boolean hasNear,
+         GridNearAtomicUpdateRequest<K, V> req,
+         GridNearAtomicUpdateResponse<K, V> res,
+         List<GridDhtCacheEntry<K, V>> locked,
+         GridCacheVersion ver,
+         @Nullable GridDhtAtomicUpdateFuture<K, V> dhtFut,
+         CI2<GridNearAtomicUpdateRequest<K, V>, GridNearAtomicUpdateResponse<K, V>> completionCb,
+         boolean replicate,
+         String taskName,
+         @Nullable IgniteCacheExpiryPolicy expiry
+     ) throws GridCacheEntryRemovedException {
+         GridCacheReturn<Object> retVal = null;
+         Collection<IgniteBiTuple<GridDhtCacheEntry<K, V>, GridCacheVersion>> deleted = null;
+ 
+         List<K> keys = req.keys();
+ 
+         long topVer = req.topologyVersion();
+ 
+         boolean checkReaders = hasNear || ctx.discovery().hasNearCache(name(), topVer);
+ 
+         boolean readersOnly = false;
+ 
+         boolean intercept = ctx.config().getInterceptor() != null;
+ 
+         Map<K, EntryProcessorResult<?>> computedMap = null;
+ 
+         // Avoid iterator creation.
+         for (int i = 0; i < keys.size(); i++) {
+             K k = keys.get(i);
+ 
+             GridCacheOperation op = req.operation();
+ 
+             // We are holding java-level locks on entries at this point.
+             // No GridCacheEntryRemovedException can be thrown.
+             try {
+                 GridDhtCacheEntry<K, V> entry = locked.get(i);
+ 
+                 if (entry == null)
+                     continue;
+ 
+                 GridCacheVersion newDrVer = req.drVersion(i);
+                 long newDrTtl = req.drTtl(i);
+                 long newDrExpireTime = req.drExpireTime(i);
+ 
+                 assert !(newDrVer instanceof GridCacheVersionEx) : newDrVer; // Plain version is expected here.
+ 
+                 if (newDrVer == null)
+                     newDrVer = ver;
+ 
+                 boolean primary = !req.fastMap() || ctx.affinity().primary(ctx.localNode(), entry.key(),
+                     req.topologyVersion());
+ 
+                 byte[] newValBytes = req.valueBytes(i);
+ 
+                 Object writeVal = req.writeValue(i);
+ 
+                 Collection<UUID> readers = null;
+                 Collection<UUID> filteredReaders = null;
+ 
+                 if (checkReaders) {
+                     readers = entry.readers();
+                     filteredReaders = F.view(entry.readers(), F.notEqualTo(node.id()));
+                 }
+ 
+                 GridCacheUpdateAtomicResult<K, V> updRes = entry.innerUpdate(
+                     ver,
+                     node.id(),
+                     locNodeId,
+                     op,
+                     writeVal,
+                     newValBytes,
+                     req.invokeArguments(),
+                     primary && writeThrough(),
+                     req.returnValue(),
+                     expiry,
+                     true,
+                     true,
+                     primary,
+                     ctx.config().getAtomicWriteOrderMode() == CLOCK, // Check version in CLOCK mode on primary node.
+                     req.filter(),
+                     replicate ? primary ? DR_PRIMARY : DR_BACKUP : DR_NONE,
+                     newDrTtl,
+                     newDrExpireTime,
+                     newDrVer,
+                     true,
+                     intercept,
+                     req.subjectId(),
+                     taskName);
+ 
+                 if (dhtFut == null && !F.isEmpty(filteredReaders)) {
+                     dhtFut = createDhtFuture(ver, req, res, completionCb, true);
+ 
+                     readersOnly = true;
+                 }
+ 
+                 if (dhtFut != null) {
+                     if (updRes.sendToDht()) { // Send to backups even in case of remove-remove scenarios.
+                         GridDrResolveResult<V> ctx = updRes.drResolveResult();
+ 
+                         long ttl = updRes.newTtl();
+                         long expireTime = updRes.drExpireTime();
+ 
+                         if (ctx == null)
+                             newDrVer = null;
+                         else if (ctx.isMerge()) {
+                             newDrVer = null; // DR version is discarded in case of merge.
+                             newValBytes = null; // Value has been changed.
+                         }
+ 
+                         EntryProcessor<K, V, ?> entryProcessor = null;
+ 
+                         if (req.forceTransformBackups() && op == TRANSFORM)
+                             entryProcessor = (EntryProcessor<K, V, ?>)writeVal;
+ 
+                         if (!readersOnly) {
+                             dhtFut.addWriteEntry(entry,
+                                 updRes.newValue(),
+                                 newValBytes,
+                                 entryProcessor,
+                                 updRes.newTtl(),
+                                 expireTime,
+                                 newDrVer);
+                         }
+ 
+                         if (!F.isEmpty(filteredReaders))
+                             dhtFut.addNearWriteEntries(filteredReaders,
+                                 entry,
+                                 updRes.newValue(),
+                                 newValBytes,
+                                 entryProcessor,
+                                 ttl,
+                                 expireTime);
+                     }
+                     else {
+                         if (log.isDebugEnabled())
+                             log.debug("Entry did not pass the filter or conflict resolution (will skip write) " +
+                                 "[entry=" + entry + ", filter=" + Arrays.toString(req.filter()) + ']');
+                     }
+                 }
+ 
+                 if (hasNear) {
+                     if (primary && updRes.sendToDht()) {
+                         if (!ctx.affinity().belongs(node, entry.partition(), topVer)) {
+                             GridDrResolveResult<V> ctx = updRes.drResolveResult();
+ 
+                             long ttl = updRes.newTtl();
+                             long expireTime = updRes.drExpireTime();
+ 
+                             if (ctx != null && ctx.isMerge())
+                                 newValBytes = null;
+ 
+                             // If put the same value as in request then do not need to send it back.
+                             if (op == TRANSFORM || writeVal != updRes.newValue()) {
+                                 res.addNearValue(i,
+                                     updRes.newValue(),
+                                     newValBytes,
+                                     ttl,
+                                     expireTime);
+                             }
+                             else
+                                 res.addNearTtl(i, ttl, expireTime);
+ 
+                             if (updRes.newValue() != null || newValBytes != null) {
+                                 IgniteFuture<Boolean> f = entry.addReader(node.id(), req.messageId(), topVer);
+ 
+                                 assert f == null : f;
+                             }
+                         }
+                         else if (F.contains(readers, node.id())) // Reader became primary or backup.
+                             entry.removeReader(node.id(), req.messageId());
+                         else
+                             res.addSkippedIndex(i);
+                     }
+                     else
+                         res.addSkippedIndex(i);
+                 }
+ 
+                 if (updRes.removeVersion() != null) {
+                     if (deleted == null)
+                         deleted = new ArrayList<>(keys.size());
+ 
+                     deleted.add(F.t(entry, updRes.removeVersion()));
+                 }
+ 
+                 if (op == TRANSFORM) {
+                     assert req.returnValue();
+ 
+                     if (updRes.computedResult() != null) {
+                         if (retVal == null) {
+                             computedMap = U.newHashMap(keys.size());
+ 
+                             retVal = new GridCacheReturn<>((Object)computedMap, updRes.success());
+                         }
+ 
+                         computedMap.put(k, updRes.computedResult());
+                     }
+                 }
+                 else {
+                     // Create only once.
+                     if (retVal == null) {
+                         Object ret = updRes.oldValue();
+ 
+                         retVal = new GridCacheReturn<>(req.returnValue() ? ret : null, updRes.success());
+                     }
+                 }
+             }
+             catch (IgniteCheckedException e) {
+                 res.addFailedKey(k, e);
+             }
+         }
+ 
+         return new UpdateSingleResult<>(retVal, deleted, dhtFut);
+     }
+ 
+     /**
+      * @param hasNear {@code True} if originating node has near cache.
+      * @param firstEntryIdx Index of the first entry in the request keys collection.
+      * @param entries Entries to update.
+      * @param ver Version to set.
+      * @param node Originating node.
+      * @param putMap Values to put.
+      * @param rmvKeys Keys to remove.
+      * @param entryProcessorMap Entry processors.
+      * @param dhtFut DHT update future if has backups.
+      * @param completionCb Completion callback to invoke when DHT future is completed.
+      * @param req Request.
+      * @param res Response.
+      * @param replicate Whether replication is enabled.
+      * @param batchRes Batch update result.
+      * @param taskName Task name.
+      * @param expiry Expiry policy.
+      * @return Deleted entries.
+      */
+     @SuppressWarnings("ForLoopReplaceableByForEach")
+     @Nullable private GridDhtAtomicUpdateFuture<K, V> updatePartialBatch(
+         boolean hasNear,
+         int firstEntryIdx,
+         List<GridDhtCacheEntry<K, V>> entries,
+         final GridCacheVersion ver,
+         ClusterNode node,
+         @Nullable Map<K, V> putMap,
+         @Nullable Collection<K> rmvKeys,
+         @Nullable Map<K, EntryProcessor<K, V, ?>> entryProcessorMap,
+         @Nullable GridDhtAtomicUpdateFuture<K, V> dhtFut,
+         CI2<GridNearAtomicUpdateRequest<K, V>, GridNearAtomicUpdateResponse<K, V>> completionCb,
+         final GridNearAtomicUpdateRequest<K, V> req,
+         final GridNearAtomicUpdateResponse<K, V> res,
+         boolean replicate,
+         UpdateBatchResult<K, V> batchRes,
+         String taskName,
+         @Nullable IgniteCacheExpiryPolicy expiry
+     ) {
+         assert putMap == null ^ rmvKeys == null;
+ 
+         assert req.drVersions() == null : "updatePartialBatch cannot be called when there are DR entries in the batch.";
+ 
+         long topVer = req.topologyVersion();
+ 
+         boolean checkReaders = hasNear || ctx.discovery().hasNearCache(name(), topVer);
+ 
+         CacheStorePartialUpdateException storeErr = null;
+ 
+         try {
+             GridCacheOperation op;
+ 
+             if (putMap != null) {
+                 // If fast mapping, filter primary keys for write to store.
+                 Map<K, V> storeMap = req.fastMap() ?
+                     F.view(putMap, new P1<K>() {
+                         @Override public boolean apply(K key) {
+                             return ctx.affinity().primary(ctx.localNode(), key, req.topologyVersion());
+                         }
+                     }) :
+                     putMap;
+ 
+                 try {
+                     ctx.store().putAllToStore(null, F.viewReadOnly(storeMap, new C1<V, IgniteBiTuple<V, GridCacheVersion>>() {
+                         @Override public IgniteBiTuple<V, GridCacheVersion> apply(V v) {
+                             return F.t(v, ver);
+                         }
+                     }));
+                 }
+                 catch (CacheStorePartialUpdateException e) {
+                     storeErr = e;
+                 }
+ 
+                 op = UPDATE;
+             }
+             else {
+                 // If fast mapping, filter primary keys for write to store.
+                 Collection<K> storeKeys = req.fastMap() ?
+                     F.view(rmvKeys, new P1<K>() {
+                         @Override public boolean apply(K key) {
+                             return ctx.affinity().primary(ctx.localNode(), key, req.topologyVersion());
+                         }
+                     }) :
+                     rmvKeys;
+ 
+                 try {
+                     ctx.store().removeAllFromStore(null, storeKeys);
+                 }
+                 catch (CacheStorePartialUpdateException e) {
+                     storeErr = e;
+                 }
+ 
+                 op = DELETE;
+             }
+ 
+             boolean intercept = ctx.config().getInterceptor() != null;
+ 
+             // Avoid iterator creation.
+             for (int i = 0; i < entries.size(); i++) {
+                 GridDhtCacheEntry<K, V> entry = entries.get(i);
+ 
+                 assert Thread.holdsLock(entry);
+ 
+                 if (entry.obsolete()) {
+                     assert req.operation() == DELETE : "Entry can become obsolete only after remove: " + entry;
+ 
+                     continue;
+                 }
+ 
+                 if (storeErr != null && storeErr.failedKeys().contains(entry.key()))
+                     continue;
+ 
+                 try {
+                     // We are holding java-level locks on entries at this point.
+                     V writeVal = op == UPDATE ? putMap.get(entry.key()) : null;
+ 
+                     assert writeVal != null || op == DELETE : "null write value found.";
+ 
+                     boolean primary = !req.fastMap() || ctx.affinity().primary(ctx.localNode(), entry.key(),
+                         req.topologyVersion());
+ 
+                     Collection<UUID> readers = null;
+                     Collection<UUID> filteredReaders = null;
+ 
+                     if (checkReaders) {
+                         readers = entry.readers();
+                         filteredReaders = F.view(entry.readers(), F.notEqualTo(node.id()));
+                     }
+ 
+                     GridCacheUpdateAtomicResult<K, V> updRes = entry.innerUpdate(
+                         ver,
+                         node.id(),
+                         locNodeId,
+                         op,
+                         writeVal,
+                         null,
+                         null,
+                         false,
+                         false,
+                         expiry,
+                         true,
+                         true,
+                         primary,
+                         ctx.config().getAtomicWriteOrderMode() == CLOCK, // Check version in CLOCK mode on primary node.
+                         null,
+                         replicate ? primary ? DR_PRIMARY : DR_BACKUP : DR_NONE,
+                         -1L,
+                         -1L,
+                         null,
+                         false,
+                         false,
+                         req.subjectId(),
+                         taskName);
+ 
+                     assert updRes.newTtl() == -1L || expiry != null;
+ 
+                     if (intercept) {
+                         if (op == UPDATE)
+                             ctx.config().getInterceptor().onAfterPut(entry.key(), updRes.newValue());
+                         else {
+                             assert op == DELETE : op;
+ 
+                             // Old value should be already loaded for 'CacheInterceptor.onBeforeRemove'.
+                             ctx.config().<K, V>getInterceptor().onAfterRemove(entry.key(), updRes.oldValue());
+                         }
+                     }
+ 
+                     batchRes.addDeleted(entry, updRes, entries);
+ 
+                     if (dhtFut == null && !F.isEmpty(filteredReaders)) {
+                         dhtFut = createDhtFuture(ver, req, res, completionCb, true);
+ 
+                         batchRes.readersOnly(true);
+                     }
+ 
+                     if (dhtFut != null) {
+                         GridCacheValueBytes valBytesTuple = op == DELETE ? GridCacheValueBytes.nil():
+                             entry.valueBytes();
+ 
+                         byte[] valBytes = valBytesTuple.getIfMarshaled();
+ 
+                         EntryProcessor<K, V, ?> entryProcessor =
+                             entryProcessorMap == null ? null : entryProcessorMap.get(entry.key());
+ 
+                         if (!batchRes.readersOnly())
+                             dhtFut.addWriteEntry(entry,
+                                 writeVal,
+                                 valBytes,
+                                 entryProcessor,
+                                 updRes.newTtl(),
+                                 -1,
+                                 null);
+ 
+                         if (!F.isEmpty(filteredReaders))
+                             dhtFut.addNearWriteEntries(filteredReaders,
+                                 entry,
+                                 writeVal,
+                                 valBytes,
+                                 entryProcessor,
+                                 updRes.newTtl(),
+                                 -1);
+                     }
+ 
+                     if (hasNear) {
+                         if (primary) {
+                             if (!ctx.affinity().belongs(node, entry.partition(), topVer)) {
+                                 int idx = firstEntryIdx + i;
+ 
+                                 if (req.operation() == TRANSFORM) {
+                                     GridCacheValueBytes valBytesTuple = entry.valueBytes();
+ 
+                                     byte[] valBytes = valBytesTuple.getIfMarshaled();
+ 
+                                     res.addNearValue(idx,
+                                         writeVal,
+                                         valBytes,
+                                         updRes.newTtl(),
+                                         -1);
+                                 }
+                                 else
+                                     res.addNearTtl(idx, updRes.newTtl(), -1);
+ 
+                                 if (writeVal != null || !entry.valueBytes().isNull()) {
+                                     IgniteFuture<Boolean> f = entry.addReader(node.id(), req.messageId(), topVer);
+ 
+                                     assert f == null : f;
+                                 }
+                             } else if (readers.contains(node.id())) // Reader became primary or backup.
+                                 entry.removeReader(node.id(), req.messageId());
+                             else
+                                 res.addSkippedIndex(firstEntryIdx + i);
+                         }
+                         else
+                             res.addSkippedIndex(firstEntryIdx + i);
+                     }
+                 }
+                 catch (GridCacheEntryRemovedException e) {
+                     assert false : "Entry cannot become obsolete while holding lock.";
+ 
+                     e.printStackTrace();
+                 }
+             }
+         }
+         catch (IgniteCheckedException e) {
+             res.addFailedKeys(putMap != null ? putMap.keySet() : rmvKeys, e);
+         }
+ 
+         if (storeErr != null)
+             res.addFailedKeys((Collection<K>)storeErr.failedKeys(), storeErr.getCause());
+ 
+         return dhtFut;
+     }
+ 
+     /**
+      * Acquires java-level locks on cache entries. Returns collection of locked entries.
+      *
+      * @param keys Keys to lock.
+      * @param topVer Topology version to lock on.
+      * @return Collection of locked entries.
+      * @throws GridDhtInvalidPartitionException If entry does not belong to local node. If exception is thrown,
+      *      locks are released.
+      */
+     @SuppressWarnings("ForLoopReplaceableByForEach")
+     private List<GridDhtCacheEntry<K, V>> lockEntries(List<K> keys, long topVer)
+         throws GridDhtInvalidPartitionException {
+         if (keys.size() == 1) {
+             K key = keys.get(0);
+ 
+             while (true) {
+                 try {
+                     GridDhtCacheEntry<K, V> entry = entryExx(key, topVer);
+ 
+                     UNSAFE.monitorEnter(entry);
+ 
+                     if (entry.obsolete())
+                         UNSAFE.monitorExit(entry);
+                     else
+                         return Collections.singletonList(entry);
+                 }
+                 catch (GridDhtInvalidPartitionException e) {
+                     // Ignore invalid partition exception in CLOCK ordering mode.
+                     if (ctx.config().getAtomicWriteOrderMode() == CLOCK)
+                         return Collections.singletonList(null);
+                     else
+                         throw e;
+                 }
+             }
+         }
+         else {
+             List<GridDhtCacheEntry<K, V>> locked = new ArrayList<>(keys.size());
+ 
+             while (true) {
+                 for (K key : keys) {
+                     try {
+                         GridDhtCacheEntry<K, V> entry = entryExx(key, topVer);
+ 
+                         locked.add(entry);
+                     }
+                     catch (GridDhtInvalidPartitionException e) {
+                         // Ignore invalid partition exception in CLOCK ordering mode.
+                         if (ctx.config().getAtomicWriteOrderMode() == CLOCK)
+                             locked.add(null);
+                         else
+                             throw e;
+                     }
+                 }
+ 
+                 boolean retry = false;
+ 
+                 for (int i = 0; i < locked.size(); i++) {
+                     GridCacheMapEntry<K, V> entry = locked.get(i);
+ 
+                     if (entry == null)
+                         continue;
+ 
+                     UNSAFE.monitorEnter(entry);
+ 
+                     if (entry.obsolete()) {
+                         // Unlock all locked.
+                         for (int j = 0; j <= i; j++) {
+                             if (locked.get(j) != null)
+                                 UNSAFE.monitorExit(locked.get(j));
+                         }
+ 
+                         // Clear entries.
+                         locked.clear();
+ 
+                         // Retry.
+                         retry = true;
+ 
+                         break;
+                     }
+                 }
+ 
+                 if (!retry)
+                     return locked;
+             }
+         }
+     }
+ 
+     /**
+      * Releases java-level locks on cache entries.
+      *
+      * @param locked Locked entries.
+      * @param topVer Topology version.
+      */
+     private void unlockEntries(Collection<GridDhtCacheEntry<K, V>> locked, long topVer) {
+         // Process deleted entries before locks release.
+         assert ctx.deferredDelete();
+ 
+         // Entries to skip eviction manager notification for.
+         // Enqueue entries while holding locks.
+         Collection<K> skip = null;
+ 
+         for (GridCacheMapEntry<K, V> entry : locked) {
+             if (entry != null && entry.deleted()) {
+                 if (skip == null)
+                     skip = new HashSet<>(locked.size(), 1.0f);
+ 
+                 skip.add(entry.key());
+             }
+         }
+ 
+         // Release locks.
+         for (GridCacheMapEntry<K, V> entry : locked) {
+             if (entry != null)
+                 UNSAFE.monitorExit(entry);
+         }
+ 
+         // Try evict partitions.
+         for (GridDhtCacheEntry<K, V> entry : locked) {
+             if (entry != null)
+                 entry.onUnlock();
+         }
+ 
+         if (skip != null && skip.size() == locked.size())
+             // Optimization.
+             return;
+ 
+         // Must touch all entries since update may have deleted entries.
+         // Eviction manager will remove empty entries.
+         for (GridCacheMapEntry<K, V> entry : locked) {
+             if (entry != null && (skip == null || !skip.contains(entry.key())))
+                 ctx.evicts().touch(entry, topVer);
+         }
+     }
+ 
+     /**
+      * Checks if future timeout happened.
+      */
+     private void scheduleAtomicFutureRecheck() {
+         final long timeout = ctx.kernalContext().config().getNetworkTimeout();
+ 
+         ctx.time().addTimeoutObject(new GridTimeoutObjectAdapter(timeout * 2) {
+             @Override public void onTimeout() {
+                 boolean leave = false;
+ 
+                 try {
+                     ctx.gate().enter();
+ 
+                     leave = true;
+ 
+                     for (GridCacheAtomicFuture fut : ctx.mvcc().atomicFutures())
+                         fut.checkTimeout(timeout);
+                 }
+                 catch (IllegalStateException ignored) {
+                     if (log.isDebugEnabled())
+                         log.debug("Will not check pending atomic update futures for timeout (Grid is stopping).");
+                 }
+                 finally {
+                     if (leave)
+                         ctx.gate().leave();
+                 }
+             }
+         });
+     }
+ 
+     /**
+      * @param entry Entry to check.
+      * @param req Update request.
+      * @param res Update response. If filter evaluation failed, key will be added to failed keys and method
+      *      will return false.
+      * @return {@code True} if filter evaluation succeeded.
+      */
+     private boolean checkFilter(GridCacheEntryEx<K, V> entry, GridNearAtomicUpdateRequest<K, V> req,
+         GridNearAtomicUpdateResponse<K, V> res) {
+         try {
+             return ctx.isAll(entry.wrapFilterLocked(), req.filter());
+         }
+         catch (IgniteCheckedException e) {
+             res.addFailedKey(entry.key(), e);
+ 
+             return false;
+         }
+     }
+ 
+     /**
+      * @param req Request to remap.
+      */
+     private void remapToNewPrimary(GridNearAtomicUpdateRequest<K, V> req) {
+         if (log.isDebugEnabled())
+             log.debug("Remapping near update request locally: " + req);
+ 
+         Collection<?> vals;
+         Collection<GridCacheDrInfo<V>> drPutVals;
+         Collection<GridCacheVersion> drRmvVals;
+ 
+         if (req.drVersions() == null) {
+             vals = req.values();
+ 
+             drPutVals = null;
+             drRmvVals = null;
+         }
+         else if (req.operation() == UPDATE) {
+             int size = req.keys().size();
+ 
+             drPutVals = new ArrayList<>(size);
+ 
+             for (int i = 0; i < size; i++) {
+                 long ttl = req.drTtl(i);
+ 
+                 if (ttl == -1L)
+                     drPutVals.add(new GridCacheDrInfo<>(req.value(i), req.drVersion(i)));
+                 else
+                     drPutVals.add(new GridCacheDrExpirationInfo<>(req.value(i), req.drVersion(i), ttl,
+                         req.drExpireTime(i)));
+             }
+ 
+             vals = null;
+             drRmvVals = null;
+         }
+         else {
+             assert req.operation() == DELETE;
+ 
+             drRmvVals = req.drVersions();
+ 
+             vals = null;
+             drPutVals = null;
+         }
+ 
+         final GridNearAtomicUpdateFuture<K, V> updateFut = new GridNearAtomicUpdateFuture<>(
+             ctx,
+             this,
+             ctx.config().getWriteSynchronizationMode(),
+             req.operation(),
+             req.keys(),
+             vals,
+             req.invokeArguments(),
+             drPutVals,
+             drRmvVals,
+             req.returnValue(),
+             false,
+             null,
+             req.expiry(),
+             req.filter(),
+             req.subjectId(),
+             req.taskNameHash());
+ 
+         updateFut.

<TRUNCATED>

Mime
View raw message