ignite-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From voze...@apache.org
Subject [28/51] [abbrv] ignite git commit: ignite-2523 : Created GridDhtAtomicSingleUpdateRequest optimized implementation.
Date Thu, 25 Feb 2016 12:31:27 GMT
ignite-2523 : Created GridDhtAtomicSingleUpdateRequest optimized implementation.


Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/3391c847
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/3391c847
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/3391c847

Branch: refs/heads/ignite-2523
Commit: 3391c847e9825405d8ca2f7df7e01ba432f60b1b
Parents: 6040f45
Author: Ilya Lantukh <ilantukh@gridgain.com>
Authored: Tue Feb 9 17:43:39 2016 +0300
Committer: Ilya Lantukh <ilantukh@gridgain.com>
Committed: Tue Feb 9 17:43:39 2016 +0300

----------------------------------------------------------------------
 .../communication/GridIoMessageFactory.java     |   20 +-
 .../processors/cache/GridCacheIoManager.java    |   22 +-
 .../dht/atomic/GridDhtAtomicCache.java          |   32 +-
 .../GridDhtAtomicMultipleUpdateRequest.java     | 1065 ++++++++++++++++++
 .../GridDhtAtomicMultipleUpdateResponse.java    |  297 +++++
 .../GridDhtAtomicSingleUpdateRequest.java       | 1035 +++++++++++++++++
 .../GridDhtAtomicSingleUpdateResponse.java      |  296 +++++
 .../dht/atomic/GridDhtAtomicUpdateFuture.java   |   88 +-
 .../dht/atomic/GridDhtAtomicUpdateRequest.java  | 1056 +----------------
 .../dht/atomic/GridDhtAtomicUpdateResponse.java |  300 +----
 .../GridNearAtomicSingleUpdateResponse.java     |    2 +-
 .../distributed/near/GridNearAtomicCache.java   |    2 +-
 .../GridCacheAtomicMessageCountSelfTest.java    |    6 +-
 ...eAtomicInvalidPartitionHandlingSelfTest.java |    2 +-
 14 files changed, 2911 insertions(+), 1312 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/ignite/blob/3391c847/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoMessageFactory.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoMessageFactory.java b/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoMessageFactory.java
index 4d769af..2366104 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoMessageFactory.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoMessageFactory.java
@@ -66,8 +66,10 @@ import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtTxPrep
 import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtTxPrepareResponse;
 import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtUnlockRequest;
 import org.apache.ignite.internal.processors.cache.distributed.dht.atomic.GridDhtAtomicDeferredUpdateResponse;
-import org.apache.ignite.internal.processors.cache.distributed.dht.atomic.GridDhtAtomicUpdateRequest;
-import org.apache.ignite.internal.processors.cache.distributed.dht.atomic.GridDhtAtomicUpdateResponse;
+import org.apache.ignite.internal.processors.cache.distributed.dht.atomic.GridDhtAtomicMultipleUpdateRequest;
+import org.apache.ignite.internal.processors.cache.distributed.dht.atomic.GridDhtAtomicSingleUpdateRequest;
+import org.apache.ignite.internal.processors.cache.distributed.dht.atomic.GridDhtAtomicMultipleUpdateResponse;
+import org.apache.ignite.internal.processors.cache.distributed.dht.atomic.GridDhtAtomicSingleUpdateResponse;
 import org.apache.ignite.internal.processors.cache.distributed.dht.atomic.GridNearAtomicMultipleUpdateResponse;
 import org.apache.ignite.internal.processors.cache.distributed.dht.atomic.GridNearAtomicSingleUpdateRequest;
 import org.apache.ignite.internal.processors.cache.distributed.dht.atomic.GridNearAtomicMultipleUpdateRequest;
@@ -354,12 +356,12 @@ public class GridIoMessageFactory implements MessageFactory {
                 break;
 
             case 38:
-                msg = new GridDhtAtomicUpdateRequest();
+                msg = new GridDhtAtomicMultipleUpdateRequest();
 
                 break;
 
             case 39:
-                msg = new GridDhtAtomicUpdateResponse();
+                msg = new GridDhtAtomicMultipleUpdateResponse();
 
                 break;
 
@@ -738,6 +740,16 @@ public class GridIoMessageFactory implements MessageFactory {
 
                 break;
 
+            case -25:
+                msg = new GridDhtAtomicSingleUpdateRequest();
+
+                break;
+
+            case -26:
+                msg = new GridDhtAtomicSingleUpdateResponse();
+
+                break;
+
             // [-3..119] [124] - this
             // [120..123] - DR
             // [-4..-22] - SQL

http://git-wip-us.apache.org/repos/asf/ignite/blob/3391c847/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheIoManager.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheIoManager.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheIoManager.java
index 7ba9542..a4ad500 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheIoManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheIoManager.java
@@ -42,6 +42,10 @@ import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtLockRe
 import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtTxPrepareRequest;
 import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtTxPrepareResponse;
 import org.apache.ignite.internal.processors.cache.distributed.dht.GridPartitionedSingleGetFuture;
+import org.apache.ignite.internal.processors.cache.distributed.dht.atomic.GridDhtAtomicMultipleUpdateRequest;
+import org.apache.ignite.internal.processors.cache.distributed.dht.atomic.GridDhtAtomicMultipleUpdateResponse;
+import org.apache.ignite.internal.processors.cache.distributed.dht.atomic.GridDhtAtomicSingleUpdateRequest;
+import org.apache.ignite.internal.processors.cache.distributed.dht.atomic.GridDhtAtomicSingleUpdateResponse;
 import org.apache.ignite.internal.processors.cache.distributed.dht.atomic.GridDhtAtomicUpdateRequest;
 import org.apache.ignite.internal.processors.cache.distributed.dht.atomic.GridDhtAtomicUpdateResponse;
 import org.apache.ignite.internal.processors.cache.distributed.dht.atomic.GridNearAtomicSingleUpdateRequest;
@@ -398,14 +402,22 @@ public class GridCacheIoManager extends GridCacheSharedManagerAdapter {
             case 38: {
                 GridDhtAtomicUpdateRequest req = (GridDhtAtomicUpdateRequest)msg;
 
-                GridDhtAtomicUpdateResponse res = new GridDhtAtomicUpdateResponse(
-                    ctx.cacheId(),
-                    req.futureVersion(),
-                    ctx.deploymentEnabled());
+                GridDhtAtomicUpdateResponse res;
+
+                if (req instanceof GridDhtAtomicSingleUpdateRequest)
+                    res = new GridDhtAtomicSingleUpdateResponse(
+                        ctx.cacheId(),
+                        req.futureVersion(),
+                        ctx.deploymentEnabled());
+                else
+                    res = new GridDhtAtomicMultipleUpdateResponse(
+                        ctx.cacheId(),
+                        req.futureVersion(),
+                        ctx.deploymentEnabled());
 
                 res.onError(req.classError());
 
-                sendResponseOnFailedMessage(nodeId, res, cctx, ctx.ioPolicy());
+                sendResponseOnFailedMessage(nodeId, (GridCacheMessage) res, cctx, ctx.ioPolicy());
             }
 
             break;

http://git-wip-us.apache.org/repos/asf/ignite/blob/3391c847/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java
----------------------------------------------------------------------
diff --git 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
index 629aa25..454a3aa 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
@@ -275,14 +275,26 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
             }
         });
 
-        ctx.io().addHandler(ctx.cacheId(), GridDhtAtomicUpdateRequest.class, new CI2<UUID, GridDhtAtomicUpdateRequest>() {
-            @Override public void apply(UUID nodeId, GridDhtAtomicUpdateRequest req) {
+        ctx.io().addHandler(ctx.cacheId(), GridDhtAtomicMultipleUpdateRequest.class, new CI2<UUID, GridDhtAtomicMultipleUpdateRequest>() {
+            @Override public void apply(UUID nodeId, GridDhtAtomicMultipleUpdateRequest req) {
                 processDhtAtomicUpdateRequest(nodeId, req);
             }
         });
 
-        ctx.io().addHandler(ctx.cacheId(), GridDhtAtomicUpdateResponse.class, new CI2<UUID, GridDhtAtomicUpdateResponse>() {
-            @Override public void apply(UUID nodeId, GridDhtAtomicUpdateResponse res) {
+        ctx.io().addHandler(ctx.cacheId(), GridDhtAtomicSingleUpdateRequest.class, new CI2<UUID, GridDhtAtomicSingleUpdateRequest>() {
+            @Override public void apply(UUID nodeId, GridDhtAtomicSingleUpdateRequest req) {
+                processDhtAtomicUpdateRequest(nodeId, req);
+            }
+        });
+
+        ctx.io().addHandler(ctx.cacheId(), GridDhtAtomicMultipleUpdateResponse.class, new CI2<UUID, GridDhtAtomicMultipleUpdateResponse>() {
+            @Override public void apply(UUID nodeId, GridDhtAtomicMultipleUpdateResponse res) {
+                processDhtAtomicUpdateResponse(nodeId, res);
+            }
+        });
+
+        ctx.io().addHandler(ctx.cacheId(), GridDhtAtomicSingleUpdateResponse.class, new CI2<UUID, GridDhtAtomicSingleUpdateResponse>() {
+            @Override public void apply(UUID nodeId, GridDhtAtomicSingleUpdateResponse res) {
                 processDhtAtomicUpdateResponse(nodeId, res);
             }
         });
@@ -2774,8 +2786,14 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
         GridCacheVersion ver = req.writeVersion();
 
         // Always send update reply.
-        GridDhtAtomicUpdateResponse res = new GridDhtAtomicUpdateResponse(ctx.cacheId(), req.futureVersion(),
-            ctx.deploymentEnabled());
+        GridDhtAtomicUpdateResponse res;
+
+        if (req instanceof GridDhtAtomicSingleUpdateRequest)
+            res = new GridDhtAtomicSingleUpdateResponse(ctx.cacheId(), req.futureVersion(),
+                ctx.deploymentEnabled());
+        else
+            res = new GridDhtAtomicMultipleUpdateResponse(ctx.cacheId(), req.futureVersion(),
+                ctx.deploymentEnabled());
 
         Boolean replicate = ctx.isDrEnabled();
 
@@ -2871,7 +2889,7 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
 
         try {
             if (res.failedKeys() != null || res.nearEvicted() != null || req.writeSynchronizationMode() == FULL_SYNC)
-                ctx.io().send(nodeId, res, ctx.ioPolicy());
+                ctx.io().send(nodeId, (GridCacheMessage) res, ctx.ioPolicy());
             else {
                 // No failed keys and sync mode is not FULL_SYNC, thus sending deferred response.
                 sendDeferredUpdateResponse(nodeId, req.futureVersion());

http://git-wip-us.apache.org/repos/asf/ignite/blob/3391c847/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicMultipleUpdateRequest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicMultipleUpdateRequest.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicMultipleUpdateRequest.java
new file mode 100644
index 0000000..40d68fa
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicMultipleUpdateRequest.java
@@ -0,0 +1,1065 @@
+/*
+ * 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 java.io.Externalizable;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+import java.util.UUID;
+import javax.cache.processor.EntryProcessor;
+import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.cache.CacheWriteSynchronizationMode;
+import org.apache.ignite.internal.GridDirectCollection;
+import org.apache.ignite.internal.GridDirectTransient;
+import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
+import org.apache.ignite.internal.processors.cache.CacheObject;
+import org.apache.ignite.internal.processors.cache.GridCacheContext;
+import org.apache.ignite.internal.processors.cache.GridCacheDeployable;
+import org.apache.ignite.internal.processors.cache.GridCacheMessage;
+import org.apache.ignite.internal.processors.cache.GridCacheSharedContext;
+import org.apache.ignite.internal.processors.cache.KeyCacheObject;
+import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
+import org.apache.ignite.internal.util.GridLongList;
+import org.apache.ignite.internal.util.tostring.GridToStringInclude;
+import org.apache.ignite.internal.util.typedef.internal.CU;
+import org.apache.ignite.internal.util.typedef.internal.S;
+import org.apache.ignite.plugin.extensions.communication.MessageCollectionItemType;
+import org.apache.ignite.plugin.extensions.communication.MessageReader;
+import org.apache.ignite.plugin.extensions.communication.MessageWriter;
+import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * Lite dht cache backup update request.
+ */
+public class GridDhtAtomicMultipleUpdateRequest extends GridCacheMessage implements GridCacheDeployable, GridDhtAtomicUpdateRequest {
+    /** */
+    private static final long serialVersionUID = 0L;
+
+    /** Message index. */
+    public static final int CACHE_MSG_IDX = nextIndexId();
+
+    /** Node ID. */
+    private UUID nodeId;
+
+    /** Future version. */
+    private GridCacheVersion futVer;
+
+    /** Write version. */
+    private GridCacheVersion writeVer;
+
+    /** Topology version. */
+    private AffinityTopologyVersion topVer;
+
+    /** Keys to update. */
+    @GridToStringInclude
+    @GridDirectCollection(KeyCacheObject.class)
+    private List<KeyCacheObject> keys;
+
+    /** Values to update. */
+    @GridToStringInclude
+    @GridDirectCollection(CacheObject.class)
+    private List<CacheObject> vals;
+
+    /** Previous values. */
+    @GridToStringInclude
+    @GridDirectCollection(CacheObject.class)
+    private List<CacheObject> prevVals;
+
+    /** Conflict versions. */
+    @GridDirectCollection(GridCacheVersion.class)
+    private List<GridCacheVersion> conflictVers;
+
+    /** TTLs. */
+    private GridLongList ttls;
+
+    /** Conflict expire time. */
+    private GridLongList conflictExpireTimes;
+
+    /** Near TTLs. */
+    private GridLongList nearTtls;
+
+    /** Near expire times. */
+    private GridLongList nearExpireTimes;
+
+    /** Write synchronization mode. */
+    private CacheWriteSynchronizationMode syncMode;
+
+    /** Near cache keys to update. */
+    @GridToStringInclude
+    @GridDirectCollection(KeyCacheObject.class)
+    private List<KeyCacheObject> nearKeys;
+
+    /** Values to update. */
+    @GridToStringInclude
+    @GridDirectCollection(CacheObject.class)
+    private List<CacheObject> nearVals;
+
+    /** Force transform backups flag. */
+    private boolean forceTransformBackups;
+
+    /** Entry processors. */
+    @GridDirectTransient
+    private List<EntryProcessor<Object, Object, Object>> entryProcessors;
+
+    /** Entry processors bytes. */
+    @GridDirectCollection(byte[].class)
+    private List<byte[]> entryProcessorsBytes;
+
+    /** Near entry processors. */
+    @GridDirectTransient
+    private List<EntryProcessor<Object, Object, Object>> nearEntryProcessors;
+
+    /** Near entry processors bytes. */
+    @GridDirectCollection(byte[].class)
+    private List<byte[]> nearEntryProcessorsBytes;
+
+    /** Optional arguments for entry processor. */
+    @GridDirectTransient
+    private Object[] invokeArgs;
+
+    /** Entry processor arguments bytes. */
+    private byte[][] invokeArgsBytes;
+
+    /** Subject ID. */
+    private UUID subjId;
+
+    /** Task name hash. */
+    private int taskNameHash;
+
+    /** Partition. */
+    private GridLongList updateCntrs;
+
+    /** On response flag. Access should be synced on future. */
+    @GridDirectTransient
+    private boolean onRes;
+
+    /** */
+    @GridDirectTransient
+    private List<Integer> partIds;
+
+    /** */
+    @GridDirectTransient
+    private List<CacheObject> locPrevVals;
+
+    /** Keep binary flag. */
+    private boolean keepBinary;
+
+    /**
+     * Empty constructor required by {@link Externalizable}.
+     */
+    public GridDhtAtomicMultipleUpdateRequest() {
+        // No-op.
+    }
+
+    /**
+     * Constructor.
+     *
+     * @param cacheId Cache ID.
+     * @param nodeId Node ID.
+     * @param futVer Future version.
+     * @param writeVer Write version for cache values.
+     * @param invokeArgs Optional arguments for entry processor.
+     * @param syncMode Cache write synchronization mode.
+     * @param topVer Topology version.
+     * @param forceTransformBackups Force transform backups flag.
+     * @param subjId Subject ID.
+     * @param taskNameHash Task name hash code.
+     * @param addDepInfo Deployment info.
+     */
+    public GridDhtAtomicMultipleUpdateRequest(
+        int cacheId,
+        UUID nodeId,
+        GridCacheVersion futVer,
+        GridCacheVersion writeVer,
+        CacheWriteSynchronizationMode syncMode,
+        @NotNull AffinityTopologyVersion topVer,
+        boolean forceTransformBackups,
+        UUID subjId,
+        int taskNameHash,
+        Object[] invokeArgs,
+        boolean addDepInfo,
+        boolean keepBinary
+    ) {
+        assert invokeArgs == null || forceTransformBackups;
+
+        this.cacheId = cacheId;
+        this.nodeId = nodeId;
+        this.futVer = futVer;
+        this.writeVer = writeVer;
+        this.syncMode = syncMode;
+        this.topVer = topVer;
+        this.forceTransformBackups = forceTransformBackups;
+        this.subjId = subjId;
+        this.taskNameHash = taskNameHash;
+        this.invokeArgs = invokeArgs;
+        this.addDepInfo = addDepInfo;
+        this.keepBinary = keepBinary;
+
+        keys = new ArrayList<>();
+        partIds = new ArrayList<>();
+        locPrevVals = new ArrayList<>();
+
+        if (forceTransformBackups) {
+            entryProcessors = new ArrayList<>();
+            entryProcessorsBytes = new ArrayList<>();
+        }
+        else
+            vals = new ArrayList<>();
+    }
+
+    /**
+     * @return Force transform backups flag.
+     */
+    @Override public boolean forceTransformBackups() {
+        return forceTransformBackups;
+    }
+
+    /**
+     * @param key Key to add.
+     * @param val Value, {@code null} if should be removed.
+     * @param entryProcessor Entry processor.
+     * @param ttl TTL (optional).
+     * @param conflictExpireTime Conflict expire time (optional).
+     * @param conflictVer Conflict version (optional).
+     * @param addPrevVal If {@code true} adds previous value.
+     * @param prevVal Previous value.
+     */
+    @Override public void addWriteValue(KeyCacheObject key,
+        @Nullable CacheObject val,
+        EntryProcessor<Object, Object, Object> entryProcessor,
+        long ttl,
+        long conflictExpireTime,
+        @Nullable GridCacheVersion conflictVer,
+        boolean addPrevVal,
+        int partId,
+        @Nullable CacheObject prevVal,
+        @Nullable Long updateIdx) {
+        keys.add(key);
+
+        partIds.add(partId);
+
+        locPrevVals.add(prevVal);
+
+        if (forceTransformBackups) {
+            assert entryProcessor != null;
+
+            entryProcessors.add(entryProcessor);
+        }
+        else
+            vals.add(val);
+
+        if (addPrevVal) {
+            if (prevVals == null)
+                prevVals = new ArrayList<>();
+
+            prevVals.add(prevVal);
+        }
+
+        if (updateIdx != null) {
+            if (updateCntrs == null)
+                updateCntrs = new GridLongList();
+
+            updateCntrs.add(updateIdx);
+        }
+
+        // In case there is no conflict, do not create the list.
+        if (conflictVer != null) {
+            if (conflictVers == null) {
+                conflictVers = new ArrayList<>();
+
+                for (int i = 0; i < keys.size() - 1; i++)
+                    conflictVers.add(null);
+            }
+
+            conflictVers.add(conflictVer);
+        }
+        else if (conflictVers != null)
+            conflictVers.add(null);
+
+        if (ttl >= 0) {
+            if (ttls == null) {
+                ttls = new GridLongList(keys.size());
+
+                for (int i = 0; i < keys.size() - 1; i++)
+                    ttls.add(CU.TTL_NOT_CHANGED);
+            }
+        }
+
+        if (ttls != null)
+            ttls.add(ttl);
+
+        if (conflictExpireTime >= 0) {
+            if (conflictExpireTimes == null) {
+                conflictExpireTimes = new GridLongList(keys.size());
+
+                for (int i = 0; i < keys.size() - 1; i++)
+                    conflictExpireTimes.add(CU.EXPIRE_TIME_CALCULATE);
+            }
+        }
+
+        if (conflictExpireTimes != null)
+            conflictExpireTimes.add(conflictExpireTime);
+    }
+
+    /**
+     * @param key Key to add.
+     * @param val Value, {@code null} if should be removed.
+     * @param entryProcessor Entry processor.
+     * @param ttl TTL.
+     * @param expireTime Expire time.
+     */
+    @Override public void addNearWriteValue(KeyCacheObject key,
+        @Nullable CacheObject val,
+        EntryProcessor<Object, Object, Object> entryProcessor,
+        long ttl,
+        long expireTime) {
+        if (nearKeys == null) {
+            nearKeys = new ArrayList<>();
+
+            if (forceTransformBackups) {
+                nearEntryProcessors = new ArrayList<>();
+                nearEntryProcessorsBytes = new ArrayList<>();
+            }
+            else
+                nearVals = new ArrayList<>();
+        }
+
+        nearKeys.add(key);
+
+        if (forceTransformBackups) {
+            assert entryProcessor != null;
+
+            nearEntryProcessors.add(entryProcessor);
+        }
+        else
+            nearVals.add(val);
+
+        if (ttl >= 0) {
+            if (nearTtls == null) {
+                nearTtls = new GridLongList(nearKeys.size());
+
+                for (int i = 0; i < nearKeys.size() - 1; i++)
+                    nearTtls.add(CU.TTL_NOT_CHANGED);
+            }
+        }
+
+        if (nearTtls != null)
+            nearTtls.add(ttl);
+
+        if (expireTime >= 0) {
+            if (nearExpireTimes == null) {
+                nearExpireTimes = new GridLongList(nearKeys.size());
+
+                for (int i = 0; i < nearKeys.size() - 1; i++)
+                    nearExpireTimes.add(CU.EXPIRE_TIME_CALCULATE);
+            }
+        }
+
+        if (nearExpireTimes != null)
+            nearExpireTimes.add(expireTime);
+    }
+
+    /** {@inheritDoc} */
+    @Override public int lookupIndex() {
+        return CACHE_MSG_IDX;
+    }
+
+    /**
+     * @return Node ID.
+     */
+    @Override public UUID nodeId() {
+        return nodeId;
+    }
+
+    /**
+     * @return Subject ID.
+     */
+    @Override public UUID subjectId() {
+        return subjId;
+    }
+
+    /**
+     * @return Task name.
+     */
+    @Override public int taskNameHash() {
+        return taskNameHash;
+    }
+
+    /**
+     * @return Keys size.
+     */
+    @Override public int size() {
+        return keys.size();
+    }
+
+    /**
+     * @return Keys size.
+     */
+    @Override public int nearSize() {
+        return nearKeys != null ? nearKeys.size() : 0;
+    }
+
+    /**
+     * @return Version assigned on primary node.
+     */
+    @Override public GridCacheVersion futureVersion() {
+        return futVer;
+    }
+
+    /**
+     * @return Write version.
+     */
+    @Override public GridCacheVersion writeVersion() {
+        return writeVer;
+    }
+
+    /**
+     * @return Cache write synchronization mode.
+     */
+    @Override public CacheWriteSynchronizationMode writeSynchronizationMode() {
+        return syncMode;
+    }
+
+    /**
+     * @return Topology version.
+     */
+    @Override public AffinityTopologyVersion topologyVersion() {
+        return topVer;
+    }
+
+    /**
+     * @return Keys.
+     */
+    @Override public Collection<KeyCacheObject> keys() {
+        return keys;
+    }
+
+    /**
+     * @param idx Key index.
+     * @return Key.
+     */
+    @Override public KeyCacheObject key(int idx) {
+        return keys.get(idx);
+    }
+
+    /**
+     * @param idx Partition index.
+     * @return Partition id.
+     */
+    @Override public int partitionId(int idx) {
+        return partIds.get(idx);
+    }
+
+    /**
+     * @param updCntr Update counter.
+     * @return Update counter.
+     */
+    @Override public Long updateCounter(int updCntr) {
+        if (updateCntrs != null && updCntr < updateCntrs.size())
+            return updateCntrs.get(updCntr);
+
+        return null;
+    }
+
+    /**
+     * @param idx Near key index.
+     * @return Key.
+     */
+    @Override public KeyCacheObject nearKey(int idx) {
+        return nearKeys.get(idx);
+    }
+
+    /**
+     * @return Keep binary flag.
+     */
+    @Override public boolean keepBinary() {
+        return keepBinary;
+    }
+
+    /**
+     * @param idx Key index.
+     * @return Value.
+     */
+    @Override @Nullable public CacheObject value(int idx) {
+        if (vals != null)
+            return vals.get(idx);
+
+        return null;
+    }
+
+    /**
+     * @param idx Key index.
+     * @return Value.
+     */
+    @Override @Nullable public CacheObject previousValue(int idx) {
+        if (prevVals != null)
+            return prevVals.get(idx);
+
+        return null;
+    }
+
+    /**
+     * @param idx Key index.
+     * @return Value.
+     */
+    @Override @Nullable public CacheObject localPreviousValue(int idx) {
+        return locPrevVals.get(idx);
+    }
+
+    /**
+     * @param idx Key index.
+     * @return Entry processor.
+     */
+    @Override @Nullable public EntryProcessor<Object, Object, Object> entryProcessor(int idx) {
+        return entryProcessors == null ? null : entryProcessors.get(idx);
+    }
+
+    /**
+     * @param idx Near key index.
+     * @return Value.
+     */
+    @Override @Nullable public CacheObject nearValue(int idx) {
+        if (nearVals != null)
+            return nearVals.get(idx);
+
+        return null;
+    }
+
+    /**
+     * @param idx Key index.
+     * @return Transform closure.
+     */
+    @Override @Nullable public EntryProcessor<Object, Object, Object> nearEntryProcessor(int idx) {
+        return nearEntryProcessors == null ? null : nearEntryProcessors.get(idx);
+    }
+
+    /**
+     * @param idx Index.
+     * @return Conflict version.
+     */
+    @Override @Nullable public GridCacheVersion conflictVersion(int idx) {
+        if (conflictVers != null) {
+            assert idx >= 0 && idx < conflictVers.size();
+
+            return conflictVers.get(idx);
+        }
+
+        return null;
+    }
+
+    /**
+     * @param idx Index.
+     * @return TTL.
+     */
+    @Override public long ttl(int idx) {
+        if (ttls != null) {
+            assert idx >= 0 && idx < ttls.size();
+
+            return ttls.get(idx);
+        }
+
+        return CU.TTL_NOT_CHANGED;
+    }
+
+    /**
+     * @param idx Index.
+     * @return TTL for near cache update.
+     */
+    @Override public long nearTtl(int idx) {
+        if (nearTtls != null) {
+            assert idx >= 0 && idx < nearTtls.size();
+
+            return nearTtls.get(idx);
+        }
+
+        return CU.TTL_NOT_CHANGED;
+    }
+
+    /**
+     * @param idx Index.
+     * @return Conflict expire time.
+     */
+    @Override public long conflictExpireTime(int idx) {
+        if (conflictExpireTimes != null) {
+            assert idx >= 0 && idx < conflictExpireTimes.size();
+
+            return conflictExpireTimes.get(idx);
+        }
+
+        return CU.EXPIRE_TIME_CALCULATE;
+    }
+
+    /**
+     * @param idx Index.
+     * @return Expire time for near cache update.
+     */
+    @Override public long nearExpireTime(int idx) {
+        if (nearExpireTimes != null) {
+            assert idx >= 0 && idx < nearExpireTimes.size();
+
+            return nearExpireTimes.get(idx);
+        }
+
+        return CU.EXPIRE_TIME_CALCULATE;
+    }
+
+    /**
+     * @return {@code True} if on response flag changed.
+     */
+    @Override public boolean onResponse() {
+        return !onRes && (onRes = true);
+    }
+
+    /**
+     * @return Optional arguments for entry processor.
+     */
+    @Override @Nullable public Object[] invokeArguments() {
+        return invokeArgs;
+    }
+
+    /** {@inheritDoc} */
+    @Override public void prepareMarshal(GridCacheSharedContext ctx) throws IgniteCheckedException {
+        super.prepareMarshal(ctx);
+
+        GridCacheContext cctx = ctx.cacheContext(cacheId);
+
+        prepareMarshalCacheObjects(keys, cctx);
+
+        prepareMarshalCacheObjects(vals, cctx);
+
+        prepareMarshalCacheObjects(nearKeys, cctx);
+
+        prepareMarshalCacheObjects(nearVals, cctx);
+
+        prepareMarshalCacheObjects(prevVals, cctx);
+
+        if (forceTransformBackups) {
+            // force addition of deployment info for entry processors if P2P is enabled globally.
+            if (!addDepInfo && ctx.deploymentEnabled())
+                addDepInfo = true;
+
+            if (invokeArgsBytes == null)
+                invokeArgsBytes = marshalInvokeArguments(invokeArgs, cctx);
+
+            if (entryProcessorsBytes == null)
+                entryProcessorsBytes = marshalCollection(entryProcessors, cctx);
+
+            if (nearEntryProcessorsBytes == null)
+                nearEntryProcessorsBytes = marshalCollection(nearEntryProcessors, cctx);
+        }
+    }
+
+    /** {@inheritDoc} */
+    @Override public void finishUnmarshal(GridCacheSharedContext ctx, ClassLoader ldr) throws IgniteCheckedException {
+        super.finishUnmarshal(ctx, ldr);
+
+        GridCacheContext cctx = ctx.cacheContext(cacheId);
+
+        finishUnmarshalCacheObjects(keys, cctx, ldr);
+
+        finishUnmarshalCacheObjects(vals, cctx, ldr);
+
+        finishUnmarshalCacheObjects(nearKeys, cctx, ldr);
+
+        finishUnmarshalCacheObjects(nearVals, cctx, ldr);
+
+        finishUnmarshalCacheObjects(prevVals, cctx, ldr);
+
+        if (forceTransformBackups) {
+            if (entryProcessors == null)
+                entryProcessors = unmarshalCollection(entryProcessorsBytes, ctx, ldr);
+
+            if (invokeArgs == null)
+                invokeArgs = unmarshalInvokeArguments(invokeArgsBytes, ctx, ldr);
+
+            if (nearEntryProcessors == null)
+                nearEntryProcessors = unmarshalCollection(nearEntryProcessorsBytes, ctx, ldr);
+        }
+    }
+
+    /** {@inheritDoc} */
+    @Override public boolean addDeploymentInfo() {
+        return addDepInfo;
+    }
+
+    /** {@inheritDoc} */
+    @Override public boolean writeTo(ByteBuffer buf, MessageWriter writer) {
+        writer.setBuffer(buf);
+
+        if (!super.writeTo(buf, writer))
+            return false;
+
+        if (!writer.isHeaderWritten()) {
+            if (!writer.writeHeader(directType(), fieldsCount()))
+                return false;
+
+            writer.onHeaderWritten();
+        }
+
+        switch (writer.state()) {
+            case 3:
+                if (!writer.writeMessage("conflictExpireTimes", conflictExpireTimes))
+                    return false;
+
+                writer.incrementState();
+
+            case 4:
+                if (!writer.writeCollection("conflictVers", conflictVers, MessageCollectionItemType.MSG))
+                    return false;
+
+                writer.incrementState();
+
+            case 5:
+                if (!writer.writeCollection("entryProcessorsBytes", entryProcessorsBytes, MessageCollectionItemType.BYTE_ARR))
+                    return false;
+
+                writer.incrementState();
+
+            case 6:
+                if (!writer.writeBoolean("forceTransformBackups", forceTransformBackups))
+                    return false;
+
+                writer.incrementState();
+
+            case 7:
+                if (!writer.writeMessage("futVer", futVer))
+                    return false;
+
+                writer.incrementState();
+
+            case 8:
+                if (!writer.writeObjectArray("invokeArgsBytes", invokeArgsBytes, MessageCollectionItemType.BYTE_ARR))
+                    return false;
+
+                writer.incrementState();
+
+            case 9:
+                if (!writer.writeBoolean("keepBinary", keepBinary))
+                    return false;
+
+                writer.incrementState();
+
+            case 10:
+                if (!writer.writeCollection("keys", keys, MessageCollectionItemType.MSG))
+                    return false;
+
+                writer.incrementState();
+
+            case 11:
+                if (!writer.writeCollection("nearEntryProcessorsBytes", nearEntryProcessorsBytes, MessageCollectionItemType.BYTE_ARR))
+                    return false;
+
+                writer.incrementState();
+
+            case 12:
+                if (!writer.writeMessage("nearExpireTimes", nearExpireTimes))
+                    return false;
+
+                writer.incrementState();
+
+            case 13:
+                if (!writer.writeCollection("nearKeys", nearKeys, MessageCollectionItemType.MSG))
+                    return false;
+
+                writer.incrementState();
+
+            case 14:
+                if (!writer.writeMessage("nearTtls", nearTtls))
+                    return false;
+
+                writer.incrementState();
+
+            case 15:
+                if (!writer.writeCollection("nearVals", nearVals, MessageCollectionItemType.MSG))
+                    return false;
+
+                writer.incrementState();
+
+            case 16:
+                if (!writer.writeCollection("prevVals", prevVals, MessageCollectionItemType.MSG))
+                    return false;
+
+                writer.incrementState();
+
+            case 17:
+                if (!writer.writeUuid("subjId", subjId))
+                    return false;
+
+                writer.incrementState();
+
+            case 18:
+                if (!writer.writeByte("syncMode", syncMode != null ? (byte)syncMode.ordinal() : -1))
+                    return false;
+
+                writer.incrementState();
+
+            case 19:
+                if (!writer.writeInt("taskNameHash", taskNameHash))
+                    return false;
+
+                writer.incrementState();
+
+            case 20:
+                if (!writer.writeMessage("topVer", topVer))
+                    return false;
+
+                writer.incrementState();
+
+            case 21:
+                if (!writer.writeMessage("ttls", ttls))
+                    return false;
+
+                writer.incrementState();
+
+            case 22:
+                if (!writer.writeMessage("updateCntrs", updateCntrs))
+                    return false;
+
+                writer.incrementState();
+
+            case 23:
+                if (!writer.writeCollection("vals", vals, MessageCollectionItemType.MSG))
+                    return false;
+
+                writer.incrementState();
+
+            case 24:
+                if (!writer.writeMessage("writeVer", writeVer))
+                    return false;
+
+                writer.incrementState();
+
+        }
+
+        return true;
+    }
+
+    /** {@inheritDoc} */
+    @Override public boolean readFrom(ByteBuffer buf, MessageReader reader) {
+        reader.setBuffer(buf);
+
+        if (!reader.beforeMessageRead())
+            return false;
+
+        if (!super.readFrom(buf, reader))
+            return false;
+
+        switch (reader.state()) {
+            case 3:
+                conflictExpireTimes = reader.readMessage("conflictExpireTimes");
+
+                if (!reader.isLastRead())
+                    return false;
+
+                reader.incrementState();
+
+            case 4:
+                conflictVers = reader.readCollection("conflictVers", MessageCollectionItemType.MSG);
+
+                if (!reader.isLastRead())
+                    return false;
+
+                reader.incrementState();
+
+            case 5:
+                entryProcessorsBytes = reader.readCollection("entryProcessorsBytes", MessageCollectionItemType.BYTE_ARR);
+
+                if (!reader.isLastRead())
+                    return false;
+
+                reader.incrementState();
+
+            case 6:
+                forceTransformBackups = reader.readBoolean("forceTransformBackups");
+
+                if (!reader.isLastRead())
+                    return false;
+
+                reader.incrementState();
+
+            case 7:
+                futVer = reader.readMessage("futVer");
+
+                if (!reader.isLastRead())
+                    return false;
+
+                reader.incrementState();
+
+            case 8:
+                invokeArgsBytes = reader.readObjectArray("invokeArgsBytes", MessageCollectionItemType.BYTE_ARR, byte[].class);
+
+                if (!reader.isLastRead())
+                    return false;
+
+                reader.incrementState();
+
+            case 9:
+                keepBinary = reader.readBoolean("keepBinary");
+
+                if (!reader.isLastRead())
+                    return false;
+
+                reader.incrementState();
+
+            case 10:
+                keys = reader.readCollection("keys", MessageCollectionItemType.MSG);
+
+                if (!reader.isLastRead())
+                    return false;
+
+                reader.incrementState();
+
+            case 11:
+                nearEntryProcessorsBytes = reader.readCollection("nearEntryProcessorsBytes", MessageCollectionItemType.BYTE_ARR);
+
+                if (!reader.isLastRead())
+                    return false;
+
+                reader.incrementState();
+
+            case 12:
+                nearExpireTimes = reader.readMessage("nearExpireTimes");
+
+                if (!reader.isLastRead())
+                    return false;
+
+                reader.incrementState();
+
+            case 13:
+                nearKeys = reader.readCollection("nearKeys", MessageCollectionItemType.MSG);
+
+                if (!reader.isLastRead())
+                    return false;
+
+                reader.incrementState();
+
+            case 14:
+                nearTtls = reader.readMessage("nearTtls");
+
+                if (!reader.isLastRead())
+                    return false;
+
+                reader.incrementState();
+
+            case 15:
+                nearVals = reader.readCollection("nearVals", MessageCollectionItemType.MSG);
+
+                if (!reader.isLastRead())
+                    return false;
+
+                reader.incrementState();
+
+            case 16:
+                prevVals = reader.readCollection("prevVals", MessageCollectionItemType.MSG);
+
+                if (!reader.isLastRead())
+                    return false;
+
+                reader.incrementState();
+
+            case 17:
+                subjId = reader.readUuid("subjId");
+
+                if (!reader.isLastRead())
+                    return false;
+
+                reader.incrementState();
+
+            case 18:
+                byte syncModeOrd;
+
+                syncModeOrd = reader.readByte("syncMode");
+
+                if (!reader.isLastRead())
+                    return false;
+
+                syncMode = CacheWriteSynchronizationMode.fromOrdinal(syncModeOrd);
+
+                reader.incrementState();
+
+            case 19:
+                taskNameHash = reader.readInt("taskNameHash");
+
+                if (!reader.isLastRead())
+                    return false;
+
+                reader.incrementState();
+
+            case 20:
+                topVer = reader.readMessage("topVer");
+
+                if (!reader.isLastRead())
+                    return false;
+
+                reader.incrementState();
+
+            case 21:
+                ttls = reader.readMessage("ttls");
+
+                if (!reader.isLastRead())
+                    return false;
+
+                reader.incrementState();
+
+            case 22:
+                updateCntrs = reader.readMessage("updateCntrs");
+
+                if (!reader.isLastRead())
+                    return false;
+
+                reader.incrementState();
+
+            case 23:
+                vals = reader.readCollection("vals", MessageCollectionItemType.MSG);
+
+                if (!reader.isLastRead())
+                    return false;
+
+                reader.incrementState();
+
+            case 24:
+                writeVer = reader.readMessage("writeVer");
+
+                if (!reader.isLastRead())
+                    return false;
+
+                reader.incrementState();
+
+        }
+
+        return reader.afterMessageRead(GridDhtAtomicMultipleUpdateRequest.class);
+    }
+
+    /** {@inheritDoc} */
+    @Override public byte directType() {
+        return 38;
+    }
+
+    /** {@inheritDoc} */
+    @Override public byte fieldsCount() {
+        return 25;
+    }
+
+    /** {@inheritDoc} */
+    @Override public String toString() {
+        return S.toString(GridDhtAtomicMultipleUpdateRequest.class, this, "super", super.toString());
+    }
+}

http://git-wip-us.apache.org/repos/asf/ignite/blob/3391c847/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicMultipleUpdateResponse.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicMultipleUpdateResponse.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicMultipleUpdateResponse.java
new file mode 100644
index 0000000..cd0b9a6
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicMultipleUpdateResponse.java
@@ -0,0 +1,297 @@
+/*
+ * 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 java.io.Externalizable;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.internal.GridDirectCollection;
+import org.apache.ignite.internal.GridDirectTransient;
+import org.apache.ignite.internal.processors.cache.GridCacheContext;
+import org.apache.ignite.internal.processors.cache.GridCacheDeployable;
+import org.apache.ignite.internal.processors.cache.GridCacheMessage;
+import org.apache.ignite.internal.processors.cache.GridCacheSharedContext;
+import org.apache.ignite.internal.processors.cache.KeyCacheObject;
+import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
+import org.apache.ignite.internal.util.tostring.GridToStringInclude;
+import org.apache.ignite.internal.util.typedef.internal.S;
+import org.apache.ignite.plugin.extensions.communication.MessageCollectionItemType;
+import org.apache.ignite.plugin.extensions.communication.MessageReader;
+import org.apache.ignite.plugin.extensions.communication.MessageWriter;
+
+/**
+ * DHT atomic cache backup update response.
+ */
+public class GridDhtAtomicMultipleUpdateResponse extends GridCacheMessage implements GridCacheDeployable, GridDhtAtomicUpdateResponse {
+    /** */
+    private static final long serialVersionUID = 0L;
+
+    /** Message index. */
+    public static final int CACHE_MSG_IDX = nextIndexId();
+
+    /** Future version. */
+    private GridCacheVersion futVer;
+
+    /** Failed keys. */
+    @GridToStringInclude
+    @GridDirectCollection(KeyCacheObject.class)
+    private List<KeyCacheObject> failedKeys;
+
+    /** Update error. */
+    @GridDirectTransient
+    private IgniteCheckedException err;
+
+    /** Serialized update error. */
+    private byte[] errBytes;
+
+    /** Evicted readers. */
+    @GridToStringInclude
+    @GridDirectCollection(KeyCacheObject.class)
+    private List<KeyCacheObject> nearEvicted;
+
+    /**
+     * Empty constructor required by {@link Externalizable}.
+     */
+    public GridDhtAtomicMultipleUpdateResponse() {
+        // No-op.
+    }
+
+    /**
+     * @param cacheId Cache ID.
+     * @param futVer Future version.
+     * @param addDepInfo Deployment info.
+     */
+    public GridDhtAtomicMultipleUpdateResponse(int cacheId, GridCacheVersion futVer, boolean addDepInfo) {
+        this.cacheId = cacheId;
+        this.futVer = futVer;
+        this.addDepInfo = addDepInfo;
+    }
+
+    /** {@inheritDoc} */
+    @Override public int lookupIndex() {
+        return CACHE_MSG_IDX;
+    }
+
+    /**
+     * @return Future version.
+     */
+    @Override public GridCacheVersion futureVersion() {
+        return futVer;
+    }
+
+    /**
+     * Sets update error.
+     *
+     * @param err Error.
+     */
+    @Override public void onError(IgniteCheckedException err) {
+        this.err = err;
+    }
+
+    /** {@inheritDoc} */
+    @Override public IgniteCheckedException error() {
+        return err;
+    }
+
+    /**
+     * @return Failed keys.
+     */
+    @Override public Collection<KeyCacheObject> failedKeys() {
+        return failedKeys;
+    }
+
+    /**
+     * Adds key to collection of failed keys.
+     *
+     * @param key Key to add.
+     * @param e Error cause.
+     */
+    @Override public void addFailedKey(KeyCacheObject key, Throwable e) {
+        if (failedKeys == null)
+            failedKeys = new ArrayList<>();
+
+        failedKeys.add(key);
+
+        if (err == null)
+            err = new IgniteCheckedException("Failed to update keys on primary node.");
+
+        err.addSuppressed(e);
+    }
+
+    /**
+     * @return Evicted readers.
+     */
+    @Override public Collection<KeyCacheObject> nearEvicted() {
+        return nearEvicted;
+    }
+
+    /**
+     * Adds near evicted key..
+     *
+     * @param key Evicted key.
+     */
+    @Override public void addNearEvicted(KeyCacheObject key) {
+        if (nearEvicted == null)
+            nearEvicted = new ArrayList<>();
+
+        nearEvicted.add(key);
+    }
+
+    /** {@inheritDoc} */
+    @Override public void prepareMarshal(GridCacheSharedContext ctx) throws IgniteCheckedException {
+        super.prepareMarshal(ctx);
+
+        GridCacheContext cctx = ctx.cacheContext(cacheId);
+
+        prepareMarshalCacheObjects(failedKeys, cctx);
+
+        prepareMarshalCacheObjects(nearEvicted, cctx);
+
+        if (errBytes == null)
+            errBytes = ctx.marshaller().marshal(err);
+    }
+
+    /** {@inheritDoc} */
+    @Override public void finishUnmarshal(GridCacheSharedContext ctx, ClassLoader ldr) throws IgniteCheckedException {
+        super.finishUnmarshal(ctx, ldr);
+
+        GridCacheContext cctx = ctx.cacheContext(cacheId);
+
+        finishUnmarshalCacheObjects(failedKeys, cctx, ldr);
+
+        finishUnmarshalCacheObjects(nearEvicted, cctx, ldr);
+
+        if (errBytes != null && err == null)
+            err = ctx.marshaller().unmarshal(errBytes, ldr);
+    }
+
+    /** {@inheritDoc} */
+    @Override public boolean addDeploymentInfo() {
+        return addDepInfo;
+    }
+
+    /** {@inheritDoc} */
+    @Override public boolean writeTo(ByteBuffer buf, MessageWriter writer) {
+        writer.setBuffer(buf);
+
+        if (!super.writeTo(buf, writer))
+            return false;
+
+        if (!writer.isHeaderWritten()) {
+            if (!writer.writeHeader(directType(), fieldsCount()))
+                return false;
+
+            writer.onHeaderWritten();
+        }
+
+        switch (writer.state()) {
+            case 3:
+                if (!writer.writeByteArray("errBytes", errBytes))
+                    return false;
+
+                writer.incrementState();
+
+            case 4:
+                if (!writer.writeCollection("failedKeys", failedKeys, MessageCollectionItemType.MSG))
+                    return false;
+
+                writer.incrementState();
+
+            case 5:
+                if (!writer.writeMessage("futVer", futVer))
+                    return false;
+
+                writer.incrementState();
+
+            case 6:
+                if (!writer.writeCollection("nearEvicted", nearEvicted, MessageCollectionItemType.MSG))
+                    return false;
+
+                writer.incrementState();
+
+        }
+
+        return true;
+    }
+
+    /** {@inheritDoc} */
+    @Override public boolean readFrom(ByteBuffer buf, MessageReader reader) {
+        reader.setBuffer(buf);
+
+        if (!reader.beforeMessageRead())
+            return false;
+
+        if (!super.readFrom(buf, reader))
+            return false;
+
+        switch (reader.state()) {
+            case 3:
+                errBytes = reader.readByteArray("errBytes");
+
+                if (!reader.isLastRead())
+                    return false;
+
+                reader.incrementState();
+
+            case 4:
+                failedKeys = reader.readCollection("failedKeys", MessageCollectionItemType.MSG);
+
+                if (!reader.isLastRead())
+                    return false;
+
+                reader.incrementState();
+
+            case 5:
+                futVer = reader.readMessage("futVer");
+
+                if (!reader.isLastRead())
+                    return false;
+
+                reader.incrementState();
+
+            case 6:
+                nearEvicted = reader.readCollection("nearEvicted", MessageCollectionItemType.MSG);
+
+                if (!reader.isLastRead())
+                    return false;
+
+                reader.incrementState();
+
+        }
+
+        return reader.afterMessageRead(GridDhtAtomicMultipleUpdateResponse.class);
+    }
+
+    /** {@inheritDoc} */
+    @Override public byte directType() {
+        return 39;
+    }
+
+    /** {@inheritDoc} */
+    @Override public byte fieldsCount() {
+        return 7;
+    }
+
+    /** {@inheritDoc} */
+    @Override public String toString() {
+        return S.toString(GridDhtAtomicMultipleUpdateResponse.class, this);
+    }
+}

http://git-wip-us.apache.org/repos/asf/ignite/blob/3391c847/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicSingleUpdateRequest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicSingleUpdateRequest.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicSingleUpdateRequest.java
new file mode 100644
index 0000000..c842270
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicSingleUpdateRequest.java
@@ -0,0 +1,1035 @@
+/*
+ *
+ *  * 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.IgniteCheckedException;
+import org.apache.ignite.cache.CacheWriteSynchronizationMode;
+import org.apache.ignite.internal.GridDirectCollection;
+import org.apache.ignite.internal.GridDirectTransient;
+import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
+import org.apache.ignite.internal.processors.cache.CacheObject;
+import org.apache.ignite.internal.processors.cache.GridCacheContext;
+import org.apache.ignite.internal.processors.cache.GridCacheDeployable;
+import org.apache.ignite.internal.processors.cache.GridCacheMessage;
+import org.apache.ignite.internal.processors.cache.GridCacheSharedContext;
+import org.apache.ignite.internal.processors.cache.KeyCacheObject;
+import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
+import org.apache.ignite.internal.util.GridLongList;
+import org.apache.ignite.internal.util.tostring.GridToStringInclude;
+import org.apache.ignite.internal.util.typedef.internal.CU;
+import org.apache.ignite.internal.util.typedef.internal.S;
+import org.apache.ignite.plugin.extensions.communication.MessageCollectionItemType;
+import org.apache.ignite.plugin.extensions.communication.MessageReader;
+import org.apache.ignite.plugin.extensions.communication.MessageWriter;
+import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
+import javax.cache.processor.EntryProcessor;
+import java.io.Externalizable;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.UUID;
+
+public class GridDhtAtomicSingleUpdateRequest extends GridCacheMessage implements GridCacheDeployable, GridDhtAtomicUpdateRequest {
+    /** */
+    private static final long serialVersionUID = 0L;
+
+    /** Message index. */
+    public static final int CACHE_MSG_IDX = nextIndexId();
+
+    /** Node ID. */
+    private UUID nodeId;
+
+    /** Future version. */
+    private GridCacheVersion futVer;
+
+    /** Write version. */
+    private GridCacheVersion writeVer;
+
+    /** Topology version. */
+    private AffinityTopologyVersion topVer;
+
+    @GridToStringInclude
+    private KeyCacheObject key;
+
+    @GridToStringInclude
+    private CacheObject val;
+
+    @GridToStringInclude
+    private CacheObject prevVal;
+
+    @GridToStringInclude
+    private GridCacheVersion conflictVer;
+
+    private long ttl = CU.TTL_NOT_CHANGED;
+
+    private long conflictExpireTime = CU.EXPIRE_TIME_CALCULATE;
+
+    private long nearTtl = CU.TTL_NOT_CHANGED;
+
+    private long nearExpireTime = CU.EXPIRE_TIME_CALCULATE;
+
+    @GridToStringInclude
+    private KeyCacheObject nearKey;
+
+    @GridToStringInclude
+    private CacheObject nearVal;
+
+    @GridDirectTransient
+    private EntryProcessor<Object, Object, Object> entryProcessor;
+
+    private byte[] entryProcessorBytes;
+
+    @GridDirectTransient
+    private EntryProcessor<Object, Object, Object> nearEntryProcessor;
+
+    private byte[] nearEntryProcessorBytes;
+
+    private long updateCntr = -1;
+
+    @GridDirectTransient
+    private int partId;
+
+    @GridDirectTransient
+    private CacheObject locPrevVal;
+
+    /** Keys to update. */
+    @GridToStringInclude
+    @GridDirectCollection(KeyCacheObject.class)
+    private List<KeyCacheObject> keys;
+
+    /** Values to update. */
+    @GridToStringInclude
+    @GridDirectCollection(CacheObject.class)
+    private List<CacheObject> vals;
+
+    /** Previous values. */
+    @GridToStringInclude
+    @GridDirectCollection(CacheObject.class)
+    private List<CacheObject> prevVals;
+
+    /** Conflict versions. */
+    @GridDirectCollection(GridCacheVersion.class)
+    private List<GridCacheVersion> conflictVers;
+
+    /** TTLs. */
+    private GridLongList ttls;
+
+    /** Conflict expire time. */
+    private GridLongList conflictExpireTimes;
+
+    /** Near TTLs. */
+    private GridLongList nearTtls;
+
+    /** Near expire times. */
+    private GridLongList nearExpireTimes;
+
+    /** Write synchronization mode. */
+    private CacheWriteSynchronizationMode syncMode;
+
+    /** Near cache keys to update. */
+    @GridToStringInclude
+    @GridDirectCollection(KeyCacheObject.class)
+    private List<KeyCacheObject> nearKeys;
+
+    /** Values to update. */
+    @GridToStringInclude
+    @GridDirectCollection(CacheObject.class)
+    private List<CacheObject> nearVals;
+
+    /** Force transform backups flag. */
+    private boolean forceTransformBackups;
+
+    /** Entry processors. */
+    @GridDirectTransient
+    private List<EntryProcessor<Object, Object, Object>> entryProcessors;
+
+    /** Entry processors bytes. */
+    @GridDirectCollection(byte[].class)
+    private List<byte[]> entryProcessorsBytes;
+
+    /** Near entry processors. */
+    @GridDirectTransient
+    private List<EntryProcessor<Object, Object, Object>> nearEntryProcessors;
+
+    /** Near entry processors bytes. */
+    @GridDirectCollection(byte[].class)
+    private List<byte[]> nearEntryProcessorsBytes;
+
+    /** Optional arguments for entry processor. */
+    @GridDirectTransient
+    private Object[] invokeArgs;
+
+    /** Entry processor arguments bytes. */
+    private byte[][] invokeArgsBytes;
+
+    /** Subject ID. */
+    private UUID subjId;
+
+    /** Task name hash. */
+    private int taskNameHash;
+
+    /** Partition. */
+    private GridLongList updateCntrs;
+
+    /** On response flag. Access should be synced on future. */
+    @GridDirectTransient
+    private boolean onRes;
+
+    /** */
+    @GridDirectTransient
+    private List<Integer> partIds;
+
+    /** */
+    @GridDirectTransient
+    private List<CacheObject> locPrevVals;
+
+    /** Keep binary flag. */
+    private boolean keepBinary;
+
+    /**
+     * Empty constructor required by {@link Externalizable}.
+     */
+    public GridDhtAtomicSingleUpdateRequest() {
+        // No-op.
+    }
+
+    /**
+     * Constructor.
+     *
+     * @param cacheId Cache ID.
+     * @param nodeId Node ID.
+     * @param futVer Future version.
+     * @param writeVer Write version for cache values.
+     * @param invokeArgs Optional arguments for entry processor.
+     * @param syncMode Cache write synchronization mode.
+     * @param topVer Topology version.
+     * @param forceTransformBackups Force transform backups flag.
+     * @param subjId Subject ID.
+     * @param taskNameHash Task name hash code.
+     * @param addDepInfo Deployment info.
+     */
+    public GridDhtAtomicSingleUpdateRequest(
+        int cacheId,
+        UUID nodeId,
+        GridCacheVersion futVer,
+        GridCacheVersion writeVer,
+        CacheWriteSynchronizationMode syncMode,
+        @NotNull AffinityTopologyVersion topVer,
+        boolean forceTransformBackups,
+        UUID subjId,
+        int taskNameHash,
+        Object[] invokeArgs,
+        boolean addDepInfo,
+        boolean keepBinary
+    ) {
+        assert invokeArgs == null || forceTransformBackups;
+
+        this.cacheId = cacheId;
+        this.nodeId = nodeId;
+        this.futVer = futVer;
+        this.writeVer = writeVer;
+        this.syncMode = syncMode;
+        this.topVer = topVer;
+        this.forceTransformBackups = forceTransformBackups;
+        this.subjId = subjId;
+        this.taskNameHash = taskNameHash;
+        this.invokeArgs = invokeArgs;
+        this.addDepInfo = addDepInfo;
+        this.keepBinary = keepBinary;
+
+        keys = new ArrayList<>();
+        partIds = new ArrayList<>();
+        locPrevVals = new ArrayList<>();
+
+        if (forceTransformBackups) {
+            entryProcessors = new ArrayList<>();
+            entryProcessorsBytes = new ArrayList<>();
+        }
+        else
+            vals = new ArrayList<>();
+    }
+
+    /**
+     * @return Force transform backups flag.
+     */
+    @Override public boolean forceTransformBackups() {
+        return forceTransformBackups;
+    }
+
+    /**
+     * @param key Key to add.
+     * @param val Value, {@code null} if should be removed.
+     * @param entryProcessor Entry processor.
+     * @param ttl TTL (optional).
+     * @param conflictExpireTime Conflict expire time (optional).
+     * @param conflictVer Conflict version (optional).
+     * @param addPrevVal If {@code true} adds previous value.
+     * @param prevVal Previous value.
+     */
+    @Override public void addWriteValue(KeyCacheObject key,
+        @Nullable CacheObject val,
+        EntryProcessor<Object, Object, Object> entryProcessor,
+        long ttl,
+        long conflictExpireTime,
+        @Nullable GridCacheVersion conflictVer,
+        boolean addPrevVal,
+        int partId,
+        @Nullable CacheObject prevVal,
+        @Nullable Long updateIdx) {
+        this.key = key;
+
+        this.partId = partId;
+
+        this.locPrevVal = prevVal;
+
+        if (forceTransformBackups) {
+            assert entryProcessor != null;
+
+            this.entryProcessor = entryProcessor;
+        }
+        else
+            this.val = val;
+
+        if (addPrevVal)
+            this.prevVal = prevVal;
+
+        if (updateIdx != null)
+            updateCntr = updateIdx;
+
+        this.conflictVer = conflictVer;
+
+        if (ttl >= 0)
+            this.ttl = ttl;
+
+        if (conflictExpireTime >= 0)
+            this.conflictExpireTime = conflictExpireTime;
+    }
+
+    /**
+     * @param key Key to add.
+     * @param val Value, {@code null} if should be removed.
+     * @param entryProcessor Entry processor.
+     * @param ttl TTL.
+     * @param expireTime Expire time.
+     */
+    @Override public void addNearWriteValue(KeyCacheObject key,
+        @Nullable CacheObject val,
+        EntryProcessor<Object, Object, Object> entryProcessor,
+        long ttl,
+        long expireTime) {
+
+        nearKey = key;
+
+        if (forceTransformBackups) {
+            assert entryProcessor != null;
+
+            nearEntryProcessor = entryProcessor;
+        }
+        else
+            nearVal = val;
+
+        if (ttl >= 0)
+            nearTtl = ttl;
+
+        if (expireTime >= 0)
+            nearExpireTime = expireTime;
+    }
+
+    /** {@inheritDoc} */
+    @Override public int lookupIndex() {
+        return CACHE_MSG_IDX;
+    }
+
+    /**
+     * @return Node ID.
+     */
+    @Override public UUID nodeId() {
+        return nodeId;
+    }
+
+    /**
+     * @return Subject ID.
+     */
+    @Override public UUID subjectId() {
+        return subjId;
+    }
+
+    /**
+     * @return Task name.
+     */
+    @Override public int taskNameHash() {
+        return taskNameHash;
+    }
+
+    /**
+     * @return Keys size.
+     */
+    @Override public int size() {
+        return key != null ? 1 : 0;
+    }
+
+    /**
+     * @return Keys size.
+     */
+    @Override public int nearSize() {
+        return nearKey != null ? 1 : 0;
+    }
+
+    /**
+     * @return Version assigned on primary node.
+     */
+    @Override public GridCacheVersion futureVersion() {
+        return futVer;
+    }
+
+    /**
+     * @return Write version.
+     */
+    @Override public GridCacheVersion writeVersion() {
+        return writeVer;
+    }
+
+    /**
+     * @return Cache write synchronization mode.
+     */
+    @Override public CacheWriteSynchronizationMode writeSynchronizationMode() {
+        return syncMode;
+    }
+
+    /**
+     * @return Topology version.
+     */
+    @Override public AffinityTopologyVersion topologyVersion() {
+        return topVer;
+    }
+
+    /**
+     * @return Keys.
+     */
+    @Override public Collection<KeyCacheObject> keys() {
+        return Collections.singletonList(key);
+    }
+
+    /**
+     * @param idx Key index.
+     * @return Key.
+     */
+    @Override public KeyCacheObject key(int idx) {
+        assert idx == 0;
+
+        return key;
+    }
+
+    /**
+     * @param idx Partition index.
+     * @return Partition id.
+     */
+    @Override public int partitionId(int idx) {
+        assert idx == 0;
+
+        return partId;
+    }
+
+    /**
+     * @param updCntr Update counter.
+     * @return Update counter.
+     */
+    @Override public Long updateCounter(int updCntr) {
+        if (updCntr != 0)
+            return null;
+
+        if (updateCntr == -1)
+            return null;
+
+        return updateCntr;
+    }
+
+    /**
+     * @param idx Near key index.
+     * @return Key.
+     */
+    @Override public KeyCacheObject nearKey(int idx) {
+        assert idx == 0;
+
+        return nearKey;
+    }
+
+    /**
+     * @return Keep binary flag.
+     */
+    @Override public boolean keepBinary() {
+        return keepBinary;
+    }
+
+    /**
+     * @param idx Key index.
+     * @return Value.
+     */
+    @Override @Nullable public CacheObject value(int idx) {
+        assert idx == 0;
+
+        return val;
+    }
+
+    /**
+     * @param idx Key index.
+     * @return Value.
+     */
+    @Override @Nullable public CacheObject previousValue(int idx) {
+        assert idx == 0;
+
+        return prevVal;
+    }
+
+    /**
+     * @param idx Key index.
+     * @return Value.
+     */
+    @Override @Nullable public CacheObject localPreviousValue(int idx) {
+        assert idx == 0;
+
+        return locPrevVal;
+    }
+
+    /**
+     * @param idx Key index.
+     * @return Entry processor.
+     */
+    @Override @Nullable public EntryProcessor<Object, Object, Object> entryProcessor(int idx) {
+        assert idx == 0;
+
+        return entryProcessor;
+    }
+
+    /**
+     * @param idx Near key index.
+     * @return Value.
+     */
+    @Override @Nullable public CacheObject nearValue(int idx) {
+        assert idx == 0;
+
+        return nearVal;
+    }
+
+    /**
+     * @param idx Key index.
+     * @return Transform closure.
+     */
+    @Override @Nullable public EntryProcessor<Object, Object, Object> nearEntryProcessor(int idx) {
+        assert idx == 0;
+
+        return nearEntryProcessor;
+    }
+
+    /**
+     * @param idx Index.
+     * @return Conflict version.
+     */
+    @Override @Nullable public GridCacheVersion conflictVersion(int idx) {
+        assert idx == 0;
+
+        return conflictVer;
+    }
+
+    /**
+     * @param idx Index.
+     * @return TTL.
+     */
+    @Override public long ttl(int idx) {
+        assert idx == 0;
+
+        return ttl;
+    }
+
+    /**
+     * @param idx Index.
+     * @return TTL for near cache update.
+     */
+    @Override public long nearTtl(int idx) {
+        assert idx == 0;
+
+        return nearTtl;
+    }
+
+    /**
+     * @param idx Index.
+     * @return Conflict expire time.
+     */
+    @Override public long conflictExpireTime(int idx) {
+        assert idx == 0;
+
+        return conflictExpireTime;
+    }
+
+    /**
+     * @param idx Index.
+     * @return Expire time for near cache update.
+     */
+    @Override public long nearExpireTime(int idx) {
+        assert idx == 0;
+
+        return nearExpireTime;
+    }
+
+    /**
+     * @return {@code True} if on response flag changed.
+     */
+    @Override public boolean onResponse() {
+        return !onRes && (onRes = true);
+    }
+
+    /**
+     * @return Optional arguments for entry processor.
+     */
+    @Override @Nullable public Object[] invokeArguments() {
+        return invokeArgs;
+    }
+
+    /** {@inheritDoc} */
+    @Override public void prepareMarshal(GridCacheSharedContext ctx) throws IgniteCheckedException {
+        super.prepareMarshal(ctx);
+
+        GridCacheContext cctx = ctx.cacheContext(cacheId);
+
+        prepareMarshalCacheObject(key, cctx);
+
+        prepareMarshalCacheObject(val, cctx);
+
+        prepareMarshalCacheObject(nearKey, cctx);
+
+        prepareMarshalCacheObject(nearVal, cctx);
+
+        prepareMarshalCacheObject(prevVal, cctx);
+
+        if (forceTransformBackups) {
+            // force addition of deployment info for entry processors if P2P is enabled globally.
+            if (!addDepInfo && ctx.deploymentEnabled())
+                addDepInfo = true;
+
+            if (invokeArgsBytes == null)
+                invokeArgsBytes = marshalInvokeArguments(invokeArgs, cctx);
+
+            if (entryProcessorBytes == null)
+                entryProcessorBytes = marshal(entryProcessor, cctx);
+
+            if (nearEntryProcessorBytes == null)
+                nearEntryProcessorBytes = marshal(nearEntryProcessor, cctx);
+        }
+    }
+
+    /** {@inheritDoc} */
+    @Override public void finishUnmarshal(GridCacheSharedContext ctx, ClassLoader ldr) throws IgniteCheckedException {
+        super.finishUnmarshal(ctx, ldr);
+
+        GridCacheContext cctx = ctx.cacheContext(cacheId);
+
+        finishUnmarshalCacheObject(key, cctx, ldr);
+
+        finishUnmarshalCacheObject(val, cctx, ldr);
+
+        finishUnmarshalCacheObject(nearKey, cctx, ldr);
+
+        finishUnmarshalCacheObject(nearVal, cctx, ldr);
+
+        finishUnmarshalCacheObject(prevVal, cctx, ldr);
+
+        if (forceTransformBackups) {
+            if (entryProcessor == null)
+                entryProcessor = unmarshal(entryProcessorBytes, ctx, ldr);
+
+            if (invokeArgs == null)
+                invokeArgs = unmarshalInvokeArguments(invokeArgsBytes, ctx, ldr);
+
+            if (nearEntryProcessor == null)
+                nearEntryProcessor = unmarshal(nearEntryProcessorBytes, ctx, ldr);
+        }
+    }
+
+    /** {@inheritDoc} */
+    @Override public boolean addDeploymentInfo() {
+        return addDepInfo;
+    }
+
+    /** {@inheritDoc} */
+    @Override public boolean writeTo(ByteBuffer buf, MessageWriter writer) {
+        writer.setBuffer(buf);
+
+        if (!super.writeTo(buf, writer))
+            return false;
+
+        if (!writer.isHeaderWritten()) {
+            if (!writer.writeHeader(directType(), fieldsCount()))
+                return false;
+
+            writer.onHeaderWritten();
+        }
+
+        switch (writer.state()) {
+            case 3:
+                if (!writer.writeLong("conflictExpireTime", conflictExpireTime))
+                    return false;
+
+                writer.incrementState();
+
+            case 4:
+                if (!writer.writeMessage("conflictVer", conflictVer))
+                    return false;
+
+                writer.incrementState();
+
+            case 5:
+                if (!writer.writeByteArray("entryProcessorBytes", entryProcessorBytes))
+                    return false;
+
+                writer.incrementState();
+
+            case 6:
+                if (!writer.writeBoolean("forceTransformBackups", forceTransformBackups))
+                    return false;
+
+                writer.incrementState();
+
+            case 7:
+                if (!writer.writeMessage("futVer", futVer))
+                    return false;
+
+                writer.incrementState();
+
+            case 8:
+                if (!writer.writeObjectArray("invokeArgsBytes", invokeArgsBytes, MessageCollectionItemType.BYTE_ARR))
+                    return false;
+
+                writer.incrementState();
+
+            case 9:
+                if (!writer.writeBoolean("keepBinary", keepBinary))
+                    return false;
+
+                writer.incrementState();
+
+            case 10:
+                if (!writer.writeMessage("key", key))
+                    return false;
+
+                writer.incrementState();
+
+            case 11:
+                if (!writer.writeByteArray("nearEntryProcessorBytes", nearEntryProcessorBytes))
+                    return false;
+
+                writer.incrementState();
+
+            case 12:
+                if (!writer.writeLong("nearExpireTime", nearExpireTime))
+                    return false;
+
+                writer.incrementState();
+
+            case 13:
+                if (!writer.writeMessage("nearKey", nearKey))
+                    return false;
+
+                writer.incrementState();
+
+            case 14:
+                if (!writer.writeLong("nearTtl", nearTtl))
+                    return false;
+
+                writer.incrementState();
+
+            case 15:
+                if (!writer.writeMessage("nearVal", nearVal))
+                    return false;
+
+                writer.incrementState();
+
+            case 16:
+                if (!writer.writeMessage("prevVal", prevVal))
+                    return false;
+
+                writer.incrementState();
+
+            case 17:
+                if (!writer.writeUuid("subjId", subjId))
+                    return false;
+
+                writer.incrementState();
+
+            case 18:
+                if (!writer.writeByte("syncMode", syncMode != null ? (byte)syncMode.ordinal() : -1))
+                    return false;
+
+                writer.incrementState();
+
+            case 19:
+                if (!writer.writeInt("taskNameHash", taskNameHash))
+                    return false;
+
+                writer.incrementState();
+
+            case 20:
+                if (!writer.writeMessage("topVer", topVer))
+                    return false;
+
+                writer.incrementState();
+
+            case 21:
+                if (!writer.writeLong("ttl", ttl))
+                    return false;
+
+                writer.incrementState();
+
+            case 22:
+                if (!writer.writeLong("updateCntr", updateCntr))
+                    return false;
+
+                writer.incrementState();
+
+            case 23:
+                if (!writer.writeMessage("val", val))
+                    return false;
+
+                writer.incrementState();
+
+            case 24:
+                if (!writer.writeMessage("writeVer", writeVer))
+                    return false;
+
+                writer.incrementState();
+
+        }
+
+        return true;
+    }
+
+    /** {@inheritDoc} */
+    @Override public boolean readFrom(ByteBuffer buf, MessageReader reader) {
+        reader.setBuffer(buf);
+
+        if (!reader.beforeMessageRead())
+            return false;
+
+        if (!super.readFrom(buf, reader))
+            return false;
+
+        switch (reader.state()) {
+            case 3:
+                conflictExpireTime = reader.readLong("conflictExpireTime");
+
+                if (!reader.isLastRead())
+                    return false;
+
+                reader.incrementState();
+
+            case 4:
+                conflictVer = reader.readMessage("conflictVer");
+
+                if (!reader.isLastRead())
+                    return false;
+
+                reader.incrementState();
+
+            case 5:
+                entryProcessorBytes = reader.readByteArray("entryProcessorBytes");
+
+                if (!reader.isLastRead())
+                    return false;
+
+                reader.incrementState();
+
+            case 6:
+                forceTransformBackups = reader.readBoolean("forceTransformBackups");
+
+                if (!reader.isLastRead())
+                    return false;
+
+                reader.incrementState();
+
+            case 7:
+                futVer = reader.readMessage("futVer");
+
+                if (!reader.isLastRead())
+                    return false;
+
+                reader.incrementState();
+
+            case 8:
+                invokeArgsBytes = reader.readObjectArray("invokeArgsBytes", MessageCollectionItemType.BYTE_ARR, byte[].class);
+
+                if (!reader.isLastRead())
+                    return false;
+
+                reader.incrementState();
+
+            case 9:
+                keepBinary = reader.readBoolean("keepBinary");
+
+                if (!reader.isLastRead())
+                    return false;
+
+                reader.incrementState();
+
+            case 10:
+                key = reader.readMessage("key");
+
+                if (!reader.isLastRead())
+                    return false;
+
+                reader.incrementState();
+
+            case 11:
+                nearEntryProcessorBytes = reader.readByteArray("nearEntryProcessorBytes");
+
+                if (!reader.isLastRead())
+                    return false;
+
+                reader.incrementState();
+
+            case 12:
+                nearExpireTime = reader.readLong("nearExpireTime");
+
+                if (!reader.isLastRead())
+                    return false;
+
+                reader.incrementState();
+
+            case 13:
+                nearKey = reader.readMessage("nearKey");
+
+                if (!reader.isLastRead())
+                    return false;
+
+                reader.incrementState();
+
+            case 14:
+                nearTtl = reader.readLong("nearTtl");
+
+                if (!reader.isLastRead())
+                    return false;
+
+                reader.incrementState();
+
+            case 15:
+                nearVal = reader.readMessage("nearVal");
+
+                if (!reader.isLastRead())
+                    return false;
+
+                reader.incrementState();
+
+            case 16:
+                prevVal = reader.readMessage("prevVal");
+
+                if (!reader.isLastRead())
+                    return false;
+
+                reader.incrementState();
+
+            case 17:
+                subjId = reader.readUuid("subjId");
+
+                if (!reader.isLastRead())
+                    return false;
+
+                reader.incrementState();
+
+            case 18:
+                byte syncModeOrd;
+
+                syncModeOrd = reader.readByte("syncMode");
+
+                if (!reader.isLastRead())
+                    return false;
+
+                syncMode = CacheWriteSynchronizationMode.fromOrdinal(syncModeOrd);
+
+                reader.incrementState();
+
+            case 19:
+                taskNameHash = reader.readInt("taskNameHash");
+
+                if (!reader.isLastRead())
+                    return false;
+
+                reader.incrementState();
+
+            case 20:
+                topVer = reader.readMessage("topVer");
+
+                if (!reader.isLastRead())
+                    return false;
+
+                reader.incrementState();
+
+            case 21:
+                ttl = reader.readLong("ttl");
+
+                if (!reader.isLastRead())
+                    return false;
+
+                reader.incrementState();
+
+            case 22:
+                updateCntr = reader.readLong("updateCntr");
+
+                if (!reader.isLastRead())
+                    return false;
+
+                reader.incrementState();
+
+            case 23:
+                val = reader.readMessage("val");
+
+                if (!reader.isLastRead())
+                    return false;
+
+                reader.incrementState();
+
+            case 24:
+                writeVer = reader.readMessage("writeVer");
+
+                if (!reader.isLastRead())
+                    return false;
+
+                reader.incrementState();
+
+        }
+
+        return reader.afterMessageRead(GridDhtAtomicSingleUpdateRequest.class);
+    }
+
+    /** {@inheritDoc} */
+    @Override public byte directType() {
+        return -25;
+    }
+
+    /** {@inheritDoc} */
+    @Override public byte fieldsCount() {
+        return 25;
+    }
+
+    /** {@inheritDoc} */
+    @Override public String toString() {
+        return S.toString(GridDhtAtomicSingleUpdateRequest.class, this, "super", super.toString());
+    }
+}
\ No newline at end of file


Mime
View raw message