ignite-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From dma...@apache.org
Subject [09/11] ignite git commit: 'Single' operations optimizations for tx cache.
Date Wed, 18 Nov 2015 13:29:46 GMT
'Single' operations optimizations for tx cache.


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

Branch: refs/heads/ignite-801
Commit: 3ff71fd742efda91fef6ed6d92dac11cc6d00976
Parents: 38e66d9
Author: sboikov <sboikov@gridgain.com>
Authored: Tue Nov 17 19:38:05 2015 +0300
Committer: sboikov <sboikov@gridgain.com>
Committed: Tue Nov 17 19:38:05 2015 +0300

----------------------------------------------------------------------
 .../processors/cache/GridCacheAdapter.java      |   48 +-
 .../processors/cache/GridCacheMapEntry.java     |   11 +-
 .../cache/GridCacheSharedContext.java           |   31 +-
 .../distributed/GridDistributedTxMapping.java   |   78 -
 .../GridDistributedTxRemoteAdapter.java         |  136 +-
 .../dht/GridDhtTransactionalCacheAdapter.java   |    1 -
 .../cache/distributed/dht/GridDhtTxLocal.java   |   24 +-
 .../distributed/dht/GridDhtTxLocalAdapter.java  |   80 +-
 .../cache/distributed/dht/GridDhtTxMapping.java |  134 +-
 .../distributed/dht/GridDhtTxPrepareFuture.java |   59 +-
 .../cache/distributed/dht/GridDhtTxRemote.java  |   29 +-
 ...arOptimisticSerializableTxPrepareFuture.java |   16 -
 .../near/GridNearOptimisticTxPrepareFuture.java |  103 +-
 ...ridNearOptimisticTxPrepareFutureAdapter.java |   72 +-
 .../GridNearPessimisticTxPrepareFuture.java     |    1 -
 .../near/GridNearTransactionalCache.java        |    1 -
 .../near/GridNearTxFinishFuture.java            |   61 +-
 .../cache/distributed/near/GridNearTxLocal.java |  109 +-
 .../near/GridNearTxPrepareRequest.java          |    9 -
 .../distributed/near/GridNearTxRemote.java      |   26 +-
 .../distributed/near/IgniteTxMappings.java      |   75 +
 .../distributed/near/IgniteTxMappingsImpl.java  |   92 ++
 .../near/IgniteTxMappingsSingleImpl.java        |  101 ++
 .../cache/transactions/IgniteInternalTx.java    |   12 +-
 .../cache/transactions/IgniteTxAdapter.java     |   67 +-
 .../cache/transactions/IgniteTxHandler.java     |   16 +-
 .../IgniteTxImplicitSingleStateImpl.java        |  259 ++++
 .../transactions/IgniteTxLocalAdapter.java      | 1388 ++++++++++--------
 .../cache/transactions/IgniteTxLocalEx.java     |   30 +-
 .../cache/transactions/IgniteTxLocalState.java  |   44 +
 .../transactions/IgniteTxLocalStateAdapter.java |   41 +
 .../cache/transactions/IgniteTxManager.java     |   18 +-
 .../cache/transactions/IgniteTxMap.java         |    3 +-
 .../cache/transactions/IgniteTxRemoteEx.java    |   17 +-
 .../IgniteTxRemoteSingleStateImpl.java          |  108 ++
 .../cache/transactions/IgniteTxRemoteState.java |   34 +
 .../IgniteTxRemoteStateAdapter.java             |  110 ++
 .../transactions/IgniteTxRemoteStateImpl.java   |  124 ++
 .../cache/transactions/IgniteTxState.java       |  171 +++
 .../cache/transactions/IgniteTxStateImpl.java   |  375 +++++
 40 files changed, 2709 insertions(+), 1405 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/ignite/blob/3ff71fd7/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAdapter.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAdapter.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAdapter.java
index 419ccec..cbb7486 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAdapter.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAdapter.java
@@ -1854,7 +1854,7 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
 
         V prevVal = syncOp(new SyncOp<V>(true) {
             @Override public V op(IgniteTxLocalAdapter tx) throws IgniteCheckedException {
-                return (V)tx.putAllAsync(ctx, F.t(key, val), true, filter).get().value();
+                return (V)tx.putAsync(ctx, key, val, true, filter).get().value();
             }
 
             @Override public String toString() {
@@ -1909,7 +1909,7 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
 
         return asyncOp(new AsyncOp<V>() {
             @Override public IgniteInternalFuture<V> op(IgniteTxLocalAdapter tx) {
-                return tx.putAllAsync(ctx, F.t(key, val), true, filter)
+                return tx.putAsync(ctx, key, val, true, filter)
                     .chain((IgniteClosure<IgniteInternalFuture<GridCacheReturn>, V>)RET2VAL);
             }
 
@@ -1948,7 +1948,7 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
 
         Boolean stored = syncOp(new SyncOp<Boolean>(true) {
             @Override public Boolean op(IgniteTxLocalAdapter tx) throws IgniteCheckedException {
-                return tx.putAllAsync(ctx, F.t(key, val), false, filter).get().success();
+                return tx.putAsync(ctx, key, val, false, filter).get().success();
             }
 
             @Override public String toString() {
@@ -2013,10 +2013,8 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
         return syncOp(new SyncOp<EntryProcessorResult<T>>(true) {
             @Nullable @Override public EntryProcessorResult<T> op(IgniteTxLocalAdapter tx)
                 throws IgniteCheckedException {
-                Map<? extends K, EntryProcessor<K, V, Object>> invokeMap =
-                    Collections.singletonMap(key, (EntryProcessor<K, V, Object>) entryProcessor);
-
-                IgniteInternalFuture<GridCacheReturn> fut = tx.invokeAsync(ctx, invokeMap, args);
+                IgniteInternalFuture<GridCacheReturn> fut =
+                    tx.invokeAsync(ctx, key, (EntryProcessor<K, V, Object>)entryProcessor, args);
 
                 Map<K, EntryProcessorResult<T>> resMap = fut.get().value();
 
@@ -2240,8 +2238,8 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
 
         return asyncOp(new AsyncOp<Boolean>() {
             @Override public IgniteInternalFuture<Boolean> op(IgniteTxLocalAdapter tx) {
-                return tx.putAllAsync(ctx, F.t(key, val), false, filter).chain(
-                    (IgniteClosure<IgniteInternalFuture<GridCacheReturn>, Boolean>) RET2FLAG);
+                return tx.putAsync(ctx, key, val, false, filter).chain(
+                    (IgniteClosure<IgniteInternalFuture<GridCacheReturn>, Boolean>)RET2FLAG);
             }
 
             @Override public String toString() {
@@ -2275,7 +2273,7 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
 
         return syncOp(new SyncOp<V>(true) {
             @Override public V op(IgniteTxLocalAdapter tx) throws IgniteCheckedException {
-                return (V)tx.putAllAsync(ctx, F.t(key, val), true, ctx.noValArray()).get().value();
+                return (V)tx.putAsync(ctx, key, val, true, ctx.noValArray()).get().value();
             }
 
             @Override public String toString() {
@@ -2299,8 +2297,8 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
 
         IgniteInternalFuture<V> fut = asyncOp(new AsyncOp<V>() {
             @Override public IgniteInternalFuture<V> op(IgniteTxLocalAdapter tx) {
-                return tx.putAllAsync(ctx, F.t(key, val), true, ctx.noValArray())
-                    .chain((IgniteClosure<IgniteInternalFuture<GridCacheReturn>, V>) RET2VAL);
+                return tx.putAsync(ctx, key, val, true, ctx.noValArray())
+                    .chain((IgniteClosure<IgniteInternalFuture<GridCacheReturn>, V>)RET2VAL);
             }
 
             @Override public String toString() {
@@ -2329,7 +2327,7 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
 
         Boolean stored = syncOp(new SyncOp<Boolean>(true) {
             @Override public Boolean op(IgniteTxLocalAdapter tx) throws IgniteCheckedException {
-                return tx.putAllAsync(ctx, F.t(key, val), false, ctx.noValArray()).get().success();
+                return tx.putAsync(ctx, key, val, false, ctx.noValArray()).get().success();
             }
 
             @Override public String toString() {
@@ -2358,7 +2356,7 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
 
         IgniteInternalFuture<Boolean> fut = asyncOp(new AsyncOp<Boolean>() {
             @Override public IgniteInternalFuture<Boolean> op(IgniteTxLocalAdapter tx) {
-                return tx.putAllAsync(ctx, F.t(key, val), false, ctx.noValArray()).chain(
+                return tx.putAsync(ctx, key, val, false, ctx.noValArray()).chain(
                     (IgniteClosure<IgniteInternalFuture<GridCacheReturn>, Boolean>)RET2FLAG);
             }
 
@@ -2384,7 +2382,7 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
 
         return syncOp(new SyncOp<V>(true) {
             @Override public V op(IgniteTxLocalAdapter tx) throws IgniteCheckedException {
-                return (V)tx.putAllAsync(ctx, F.t(key, val), true, ctx.hasValArray()).get().value();
+                return (V)tx.putAsync(ctx, key, val, true, ctx.hasValArray()).get().value();
             }
 
             @Override public String toString() {
@@ -2408,7 +2406,7 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
 
         IgniteInternalFuture<V> fut = asyncOp(new AsyncOp<V>() {
             @Override public IgniteInternalFuture<V> op(IgniteTxLocalAdapter tx) {
-                return tx.putAllAsync(ctx, F.t(key, val), true, ctx.hasValArray()).chain(
+                return tx.putAsync(ctx, key, val, true, ctx.hasValArray()).chain(
                     (IgniteClosure<IgniteInternalFuture<GridCacheReturn>, V>)RET2VAL);
             }
 
@@ -2434,7 +2432,7 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
 
         return syncOp(new SyncOp<Boolean>(true) {
             @Override public Boolean op(IgniteTxLocalAdapter tx) throws IgniteCheckedException {
-                return tx.putAllAsync(ctx, F.t(key, val), false, ctx.hasValArray()).get().success();
+                return tx.putAsync(ctx, key, val, false, ctx.hasValArray()).get().success();
             }
 
             @Override public String toString() {
@@ -2454,7 +2452,7 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
 
         return asyncOp(new AsyncOp<Boolean>() {
             @Override public IgniteInternalFuture<Boolean> op(IgniteTxLocalAdapter tx) {
-                return tx.putAllAsync(ctx, F.t(key, val), false, ctx.hasValArray()).chain(
+                return tx.putAsync(ctx, key, val, false, ctx.hasValArray()).chain(
                     (IgniteClosure<IgniteInternalFuture<GridCacheReturn>, Boolean>) RET2FLAG);
             }
 
@@ -2481,7 +2479,7 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
                 if (ctx.deploymentEnabled())
                     ctx.deploy().registerClass(oldVal);
 
-                return tx.putAllAsync(ctx, F.t(key, newVal), false, ctx.equalsValArray(oldVal)).get()
+                return tx.putAsync(ctx, key, newVal, false, ctx.equalsValArray(oldVal)).get()
                     .success();
             }
 
@@ -2518,7 +2516,7 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
                     }
                 }
 
-                return tx.putAllAsync(ctx, F.t(key, newVal), false, ctx.equalsValArray(oldVal)).chain(
+                return tx.putAsync(ctx, key, newVal, false, ctx.equalsValArray(oldVal)).chain(
                     (IgniteClosure<IgniteInternalFuture<GridCacheReturn>, Boolean>)RET2FLAG);
             }
 
@@ -2883,8 +2881,9 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
                 if (ctx.deploymentEnabled())
                     ctx.deploy().registerClass(oldVal);
 
-                return tx.putAllAsync(ctx,
-                        F.t(key, newVal),
+                return tx.putAsync(ctx,
+                        key,
+                        newVal,
                         true,
                         ctx.equalsValArray(oldVal)).get();
             }
@@ -2945,8 +2944,9 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
                     return new GridFinishedFuture<>(e);
                 }
 
-                return (IgniteInternalFuture)tx.putAllAsync(ctx,
-                    F.t(key, newVal),
+                return (IgniteInternalFuture)tx.putAsync(ctx,
+                    key,
+                    newVal,
                     true,
                     ctx.equalsValArray(oldVal));
             }

http://git-wip-us.apache.org/repos/asf/ignite/blob/3ff71fd7/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMapEntry.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMapEntry.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMapEntry.java
index df9f5c4..0786a50 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMapEntry.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMapEntry.java
@@ -74,6 +74,7 @@ import static org.apache.ignite.events.EventType.EVT_CACHE_OBJECT_EXPIRED;
 import static org.apache.ignite.events.EventType.EVT_CACHE_OBJECT_PUT;
 import static org.apache.ignite.events.EventType.EVT_CACHE_OBJECT_READ;
 import static org.apache.ignite.events.EventType.EVT_CACHE_OBJECT_REMOVED;
+import static org.apache.ignite.internal.processors.cache.GridCacheOperation.DELETE;
 import static org.apache.ignite.internal.processors.dr.GridDrType.DR_NONE;
 
 /**
@@ -4014,7 +4015,15 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
 
         IgniteInternalTx tx = cctx.tm().localTxx();
 
-        return tx == null || !tx.removed(txKey());
+        if (tx != null) {
+            IgniteTxEntry e = tx.entry(txKey());
+
+            boolean rmvd = e != null && e.op() == DELETE;
+
+            return !rmvd;
+        }
+
+        return true;
     }
 
     /** {@inheritDoc} */

http://git-wip-us.apache.org/repos/asf/ignite/blob/3ff71fd7/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheSharedContext.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheSharedContext.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheSharedContext.java
index b37742c..4293b90 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheSharedContext.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheSharedContext.java
@@ -582,12 +582,7 @@ public class GridCacheSharedContext<K, V> {
      * @throws IgniteCheckedException If failed.
      */
     public void endTx(IgniteInternalTx tx) throws IgniteCheckedException {
-        Collection<Integer> cacheIds = tx.activeCacheIds();
-
-        if (!cacheIds.isEmpty()) {
-            for (Integer cacheId : cacheIds)
-                cacheContext(cacheId).cache().awaitLastFut();
-        }
+        tx.txState().awaitLastFut(this);
 
         tx.close();
     }
@@ -596,22 +591,17 @@ public class GridCacheSharedContext<K, V> {
      * @param tx Transaction to commit.
      * @return Commit future.
      */
+    @SuppressWarnings("unchecked")
     public IgniteInternalFuture<IgniteInternalTx> commitTxAsync(IgniteInternalTx tx) {
-        Collection<Integer> cacheIds = tx.activeCacheIds();
-
-        if (cacheIds.isEmpty())
-            return tx.commitAsync();
-        else if (cacheIds.size() == 1) {
-            int cacheId = F.first(cacheIds);
+        GridCacheContext ctx = tx.txState().singleCacheContext(this);
 
-            return cacheContext(cacheId).cache().commitTxAsync(tx);
-        }
-        else {
-            for (Integer cacheId : cacheIds)
-                cacheContext(cacheId).cache().awaitLastFut();
+        if (ctx == null) {
+            tx.txState().awaitLastFut(this);
 
             return tx.commitAsync();
         }
+        else
+            return ctx.cache().commitTxAsync(tx);
     }
 
     /**
@@ -620,12 +610,7 @@ public class GridCacheSharedContext<K, V> {
      * @return Rollback future.
      */
     public IgniteInternalFuture rollbackTxAsync(IgniteInternalTx tx) throws IgniteCheckedException {
-        Collection<Integer> cacheIds = tx.activeCacheIds();
-
-        if (!cacheIds.isEmpty()) {
-            for (Integer cacheId : cacheIds)
-                cacheContext(cacheId).cache().awaitLastFut();
-        }
+        tx.txState().awaitLastFut(this);
 
         return tx.rollbackAsync();
     }

http://git-wip-us.apache.org/repos/asf/ignite/blob/3ff71fd7/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedTxMapping.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedTxMapping.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedTxMapping.java
index 2d2d935..8c9f181 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedTxMapping.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedTxMapping.java
@@ -24,9 +24,7 @@ import java.io.ObjectOutput;
 import java.util.Collection;
 import java.util.Iterator;
 import java.util.LinkedHashSet;
-import java.util.UUID;
 import org.apache.ignite.cluster.ClusterNode;
-import org.apache.ignite.internal.processors.cache.GridCacheEntryEx;
 import org.apache.ignite.internal.processors.cache.transactions.IgniteTxEntry;
 import org.apache.ignite.internal.processors.cache.transactions.IgniteTxKey;
 import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
@@ -59,15 +57,9 @@ public class GridDistributedTxMapping implements Externalizable {
     /** DHT version. */
     private GridCacheVersion dhtVer;
 
-    /** Copy on remove flag. */
-    private boolean readOnly;
-
     /** {@code True} if this is last mapping for node. */
     private boolean last;
 
-    /** IDs of backup nodes receiving last prepare request during this mapping. */
-    private Collection<UUID> lastBackups;
-
     /** {@code True} if mapping is for near caches, {@code false} otherwise. */
     private boolean near;
 
@@ -91,20 +83,6 @@ public class GridDistributedTxMapping implements Externalizable {
     }
 
     /**
-     * @return IDs of backup nodes receiving last prepare request during this mapping.
-     */
-    @Nullable public Collection<UUID> lastBackups() {
-        return lastBackups;
-    }
-
-    /**
-     * @param lastBackups IDs of backup nodes receiving last prepare request during this mapping.
-     */
-    public void lastBackups(@Nullable Collection<UUID> lastBackups) {
-        this.lastBackups = lastBackups;
-    }
-
-    /**
      * @return {@code True} if this is last mapping for node.
      */
     public boolean last() {
@@ -161,17 +139,6 @@ public class GridDistributedTxMapping implements Externalizable {
     }
 
     /**
-     * @param entries Mapped entries.
-     * @param readOnly Flag indicating that passed in collection is read-only.
-     */
-    public void entries(Collection<IgniteTxEntry> entries, boolean readOnly) {
-        this.entries = entries;
-
-        // Set copy on remove flag as passed in collection is unmodifiable.
-        this.readOnly = true;
-    }
-
-    /**
      * @return {@code True} if lock is explicit.
      */
     public boolean explicitLock() {
@@ -221,8 +188,6 @@ public class GridDistributedTxMapping implements Externalizable {
      * @param entry Adds entry.
      */
     public void add(IgniteTxEntry entry) {
-        ensureModifiable();
-
         entries.add(entry);
     }
 
@@ -231,48 +196,16 @@ public class GridDistributedTxMapping implements Externalizable {
      * @return {@code True} if entry was removed.
      */
     public boolean removeEntry(IgniteTxEntry entry) {
-        ensureModifiable();
-
         return entries.remove(entry);
     }
 
     /**
-     * @param parts Evicts partitions from mapping.
-     */
-    public void evictPartitions(@Nullable int[] parts) {
-        if (!F.isEmpty(parts)) {
-            ensureModifiable();
-
-            evictPartitions(parts, entries);
-        }
-    }
-
-    /**
-     * @param parts Partitions.
-     * @param c Collection.
-     */
-    private void evictPartitions(int[] parts, Collection<IgniteTxEntry> c) {
-        assert parts != null;
-
-        for (Iterator<IgniteTxEntry> it = c.iterator(); it.hasNext();) {
-            IgniteTxEntry e = it.next();
-
-            GridCacheEntryEx cached = e.cached();
-
-            if (U.containsIntArray(parts, cached.partition()))
-                it.remove();
-        }
-    }
-
-    /**
      * @param keys Keys to evict readers for.
      */
     public void evictReaders(@Nullable Collection<IgniteTxKey> keys) {
         if (keys == null || keys.isEmpty())
             return;
 
-        ensureModifiable();
-
         evictReaders(keys, entries);
     }
 
@@ -293,17 +226,6 @@ public class GridDistributedTxMapping implements Externalizable {
     }
 
     /**
-     * Copies collection of entries if it is read-only.
-     */
-    private void ensureModifiable() {
-        if (readOnly) {
-            entries = new LinkedHashSet<>(entries);
-
-            readOnly = false;
-        }
-    }
-
-    /**
      * Whether empty or not.
      *
      * @return Empty or not.

http://git-wip-us.apache.org/repos/asf/ignite/blob/3ff71fd7/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedTxRemoteAdapter.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedTxRemoteAdapter.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedTxRemoteAdapter.java
index 93303c8..0d49584 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedTxRemoteAdapter.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedTxRemoteAdapter.java
@@ -45,6 +45,8 @@ import org.apache.ignite.internal.processors.cache.transactions.IgniteTxAdapter;
 import org.apache.ignite.internal.processors.cache.transactions.IgniteTxEntry;
 import org.apache.ignite.internal.processors.cache.transactions.IgniteTxKey;
 import org.apache.ignite.internal.processors.cache.transactions.IgniteTxRemoteEx;
+import org.apache.ignite.internal.processors.cache.transactions.IgniteTxRemoteState;
+import org.apache.ignite.internal.processors.cache.transactions.IgniteTxState;
 import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
 import org.apache.ignite.internal.processors.cache.version.GridCacheVersionConflictContext;
 import org.apache.ignite.internal.transactions.IgniteTxHeuristicCheckedException;
@@ -85,18 +87,6 @@ public class GridDistributedTxRemoteAdapter extends IgniteTxAdapter
     /** */
     private static final long serialVersionUID = 0L;
 
-    /** Read set. */
-    @GridToStringInclude
-    protected Map<IgniteTxKey, IgniteTxEntry> readMap;
-
-    /** Write map. */
-    @GridToStringInclude
-    protected Map<IgniteTxKey, IgniteTxEntry> writeMap;
-
-    /** Remote thread ID. */
-    @GridToStringInclude
-    private long rmtThreadId;
-
     /** Explicit versions. */
     @GridToStringInclude
     private List<GridCacheVersion> explicitVers;
@@ -109,6 +99,10 @@ public class GridDistributedTxRemoteAdapter extends IgniteTxAdapter
     @GridToStringInclude
     private AtomicBoolean commitAllowed = new AtomicBoolean(false);
 
+    /** */
+    @GridToStringInclude
+    protected IgniteTxRemoteState txState;
+
     /**
      * Empty constructor required for {@link Externalizable}.
      */
@@ -119,7 +113,6 @@ public class GridDistributedTxRemoteAdapter extends IgniteTxAdapter
     /**
      * @param ctx Cache registry.
      * @param nodeId Node ID.
-     * @param rmtThreadId Remote thread ID.
      * @param xidVer XID version.
      * @param commitVer Commit version.
      * @param sys System flag.
@@ -135,7 +128,6 @@ public class GridDistributedTxRemoteAdapter extends IgniteTxAdapter
     public GridDistributedTxRemoteAdapter(
         GridCacheSharedContext<?, ?> ctx,
         UUID nodeId,
-        long rmtThreadId,
         GridCacheVersion xidVer,
         GridCacheVersion commitVer,
         boolean sys,
@@ -163,7 +155,6 @@ public class GridDistributedTxRemoteAdapter extends IgniteTxAdapter
             subjId,
             taskNameHash);
 
-        this.rmtThreadId = rmtThreadId;
         this.invalidate = invalidate;
 
         commitVersion(commitVer);
@@ -173,6 +164,11 @@ public class GridDistributedTxRemoteAdapter extends IgniteTxAdapter
     }
 
     /** {@inheritDoc} */
+    @Override public IgniteTxState txState() {
+        return txState;
+    }
+
+    /** {@inheritDoc} */
     @Override public UUID eventNodeId() {
         return nodeId;
     }
@@ -188,11 +184,6 @@ public class GridDistributedTxRemoteAdapter extends IgniteTxAdapter
     }
 
     /** {@inheritDoc} */
-    @Override public Collection<Integer> activeCacheIds() {
-        return Collections.emptyList();
-    }
-
-    /** {@inheritDoc} */
     @Override public boolean activeCachesDeploymentEnabled() {
         return false;
     }
@@ -201,14 +192,7 @@ public class GridDistributedTxRemoteAdapter extends IgniteTxAdapter
      * @return Checks if transaction has no entries.
      */
     @Override public boolean empty() {
-        return readMap.isEmpty() && writeMap.isEmpty();
-    }
-
-    /** {@inheritDoc} */
-    @Override public boolean removed(IgniteTxKey key) {
-        IgniteTxEntry e = writeMap.get(key);
-
-        return e != null && e.op() == DELETE;
+        return txState.empty();
     }
 
     /** {@inheritDoc} */
@@ -218,12 +202,12 @@ public class GridDistributedTxRemoteAdapter extends IgniteTxAdapter
 
     /** {@inheritDoc} */
     @Override public Map<IgniteTxKey, IgniteTxEntry> writeMap() {
-        return writeMap;
+        return txState.writeMap();
     }
 
     /** {@inheritDoc} */
     @Override public Map<IgniteTxKey, IgniteTxEntry> readMap() {
-        return readMap;
+        return txState.readMap();
     }
 
     /** {@inheritDoc} */
@@ -245,12 +229,7 @@ public class GridDistributedTxRemoteAdapter extends IgniteTxAdapter
 
     /** {@inheritDoc} */
     @Override public IgniteTxEntry entry(IgniteTxKey key) {
-        IgniteTxEntry e = writeMap == null ? null : writeMap.get(key);
-
-        if (e == null)
-            e = readMap == null ? null : readMap.get(key);
-
-        return e;
+        return txState.entry(key);
     }
 
     /**
@@ -259,8 +238,7 @@ public class GridDistributedTxRemoteAdapter extends IgniteTxAdapter
      * @param key key to be removed.
      */
     public void clearEntry(IgniteTxKey key) {
-        readMap.remove(key);
-        writeMap.remove(key);
+        txState.clearEntry(key);
     }
 
     /**
@@ -268,13 +246,19 @@ public class GridDistributedTxRemoteAdapter extends IgniteTxAdapter
      * @param committedVers Committed versions.
      * @param rolledbackVers Rolled back versions.
      */
-    @Override public void doneRemote(GridCacheVersion baseVer, Collection<GridCacheVersion> committedVers,
-        Collection<GridCacheVersion> rolledbackVers, Collection<GridCacheVersion> pendingVers) {
+    @Override public void doneRemote(GridCacheVersion baseVer,
+        Collection<GridCacheVersion> committedVers,
+        Collection<GridCacheVersion> rolledbackVers,
+        Collection<GridCacheVersion> pendingVers) {
+        Map<IgniteTxKey, IgniteTxEntry> readMap = txState.readMap();
+
         if (readMap != null && !readMap.isEmpty()) {
             for (IgniteTxEntry txEntry : readMap.values())
                 doneRemote(txEntry, baseVer, committedVers, rolledbackVers, pendingVers);
         }
 
+        Map<IgniteTxKey, IgniteTxEntry> writeMap = txState.writeMap();
+
         if (writeMap != null && !writeMap.isEmpty()) {
             for (IgniteTxEntry txEntry : writeMap.values())
                 doneRemote(txEntry, baseVer, committedVers, rolledbackVers, pendingVers);
@@ -290,8 +274,10 @@ public class GridDistributedTxRemoteAdapter extends IgniteTxAdapter
      * @param rolledbackVers Rolled back versions relative to base version.
      * @param pendingVers Pending versions.
      */
-    private void doneRemote(IgniteTxEntry txEntry, GridCacheVersion baseVer,
-        Collection<GridCacheVersion> committedVers, Collection<GridCacheVersion> rolledbackVers,
+    private void doneRemote(IgniteTxEntry txEntry,
+        GridCacheVersion baseVer,
+        Collection<GridCacheVersion> committedVers,
+        Collection<GridCacheVersion> rolledbackVers,
         Collection<GridCacheVersion> pendingVers) {
         while (true) {
             GridDistributedCacheEntry entry = (GridDistributedCacheEntry)txEntry.cached();
@@ -337,59 +323,9 @@ public class GridDistributedTxRemoteAdapter extends IgniteTxAdapter
         return started;
     }
 
-    /**
-     * @return Remote node thread ID.
-     */
-    @Override public long remoteThreadId() {
-        return rmtThreadId;
-    }
-
-    /**
-     * @param e Transaction entry to set.
-     * @return {@code True} if value was set.
-     */
-    @Override public boolean setWriteValue(IgniteTxEntry e) {
-        checkInternal(e.txKey());
-
-        IgniteTxEntry entry = writeMap.get(e.txKey());
-
-        if (entry == null) {
-            IgniteTxEntry rmv = readMap.remove(e.txKey());
-
-            if (rmv != null) {
-                e.cached(rmv.cached());
-
-                writeMap.put(e.txKey(), e);
-            }
-            // If lock is explicit.
-            else {
-                e.cached(e.context().cache().entryEx(e.key()));
-
-                // explicit lock.
-                writeMap.put(e.txKey(), e);
-            }
-        }
-        else {
-            // Copy values.
-            entry.value(e.value(), e.hasWriteValue(), e.hasReadValue());
-            entry.entryProcessors(e.entryProcessors());
-            entry.op(e.op());
-            entry.ttl(e.ttl());
-            entry.explicitVersion(e.explicitVersion());
-
-            // Conflict resolution stuff.
-            entry.conflictVersion(e.conflictVersion());
-            entry.conflictExpireTime(e.conflictExpireTime());
-        }
-
-        addExplicit(e);
-
-        return true;
-    }
-
     /** {@inheritDoc} */
     @Override public boolean hasWriteKey(IgniteTxKey key) {
-        return writeMap.containsKey(key);
+        return txState.hasWriteKey(key);
     }
 
     /** {@inheritDoc} */
@@ -400,27 +336,27 @@ public class GridDistributedTxRemoteAdapter extends IgniteTxAdapter
 
     /** {@inheritDoc} */
     @Override public Set<IgniteTxKey> readSet() {
-        return readMap.keySet();
+        return txState.readSet();
     }
 
     /** {@inheritDoc} */
     @Override public Set<IgniteTxKey> writeSet() {
-        return writeMap.keySet();
+        return txState.writeSet();
     }
 
     /** {@inheritDoc} */
     @Override public Collection<IgniteTxEntry> allEntries() {
-        return F.concat(false, writeEntries(), readEntries());
+        return txState.allEntries();
     }
 
     /** {@inheritDoc} */
     @Override public Collection<IgniteTxEntry> writeEntries() {
-        return writeMap.values();
+        return txState.writeEntries();
     }
 
     /** {@inheritDoc} */
     @Override public Collection<IgniteTxEntry> readEntries() {
-        return readMap.values();
+        return txState.readEntries();
     }
 
     /**
@@ -459,7 +395,7 @@ public class GridDistributedTxRemoteAdapter extends IgniteTxAdapter
     @SuppressWarnings({"CatchGenericClass"})
     private void commitIfLocked() throws IgniteCheckedException {
         if (state() == COMMITTING) {
-            for (IgniteTxEntry txEntry : writeMap.values()) {
+            for (IgniteTxEntry txEntry : writeEntries()) {
                 assert txEntry != null : "Missing transaction entry for tx: " + this;
 
                 while (true) {
@@ -494,6 +430,8 @@ public class GridDistributedTxRemoteAdapter extends IgniteTxAdapter
             if (commitAllowed.compareAndSet(false, true)) {
                 IgniteCheckedException err = null;
 
+                Map<IgniteTxKey, IgniteTxEntry> writeMap = txState.writeMap();
+
                 if (!F.isEmpty(writeMap)) {
                     // Register this transaction as completed prior to write-phase to
                     // ensure proper lock ordering for removed entries.

http://git-wip-us.apache.org/repos/asf/ignite/blob/3ff71fd7/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTransactionalCacheAdapter.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTransactionalCacheAdapter.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTransactionalCacheAdapter.java
index 3069afd..fe91e5b 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTransactionalCacheAdapter.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTransactionalCacheAdapter.java
@@ -228,7 +228,6 @@ public abstract class GridDhtTransactionalCacheAdapter<K, V> extends GridDhtCach
                                     req.futureId(),
                                     nodeId,
                                     req.nearXidVersion(),
-                                    req.threadId(),
                                     req.topologyVersion(),
                                     req.version(),
                                     /*commitVer*/null,

http://git-wip-us.apache.org/repos/asf/ignite/blob/3ff71fd7/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxLocal.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxLocal.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxLocal.java
index 44f34aa..2bed931 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxLocal.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxLocal.java
@@ -223,11 +223,6 @@ public class GridDhtTxLocal extends GridDhtTxLocalAdapter implements GridCacheMa
     }
 
     /** {@inheritDoc} */
-    @Override protected IgniteUuid nearMiniId() {
-        return nearMiniId;
-    }
-
-    /** {@inheritDoc} */
     @Override public boolean dht() {
         return true;
     }
@@ -307,8 +302,7 @@ public class GridDhtTxLocal extends GridDhtTxLocalAdapter implements GridCacheMa
                 0,
                 nearMiniId,
                 null,
-                true,
-                null);
+                true);
         }
 
         // For pessimistic mode we don't distribute prepare request.
@@ -322,8 +316,7 @@ public class GridDhtTxLocal extends GridDhtTxLocalAdapter implements GridCacheMa
                 nearMiniId,
                 Collections.<IgniteTxKey, GridCacheVersion>emptyMap(),
                 true,
-                needReturnValue(),
-                null)))
+                needReturnValue())))
                 return prepFut.get();
         }
         else
@@ -378,7 +371,6 @@ public class GridDhtTxLocal extends GridDhtTxLocalAdapter implements GridCacheMa
      * @param nearMiniId Near mini future ID.
      * @param txNodes Transaction nodes mapping.
      * @param last {@code True} if this is last prepare request.
-     * @param lastBackups IDs of backup nodes receiving last prepare request.
      * @return Future that will be completed when locks are acquired.
      */
     public IgniteInternalFuture<GridNearTxPrepareResponse> prepareAsync(
@@ -388,8 +380,7 @@ public class GridDhtTxLocal extends GridDhtTxLocalAdapter implements GridCacheMa
         long msgId,
         IgniteUuid nearMiniId,
         Map<UUID, Collection<UUID>> txNodes,
-        boolean last,
-        Collection<UUID> lastBackups
+        boolean last
     ) {
         // In optimistic mode prepare still can be called explicitly from salvageTx.
         GridDhtTxPrepareFuture fut = prepFut.get();
@@ -404,8 +395,7 @@ public class GridDhtTxLocal extends GridDhtTxLocalAdapter implements GridCacheMa
                 nearMiniId,
                 verMap,
                 last,
-                needReturnValue(),
-                lastBackups))) {
+                needReturnValue()))) {
                 GridDhtTxPrepareFuture f = prepFut.get();
 
                 assert f.nearMiniId().equals(nearMiniId) : "Wrong near mini id on existing future " +
@@ -443,13 +433,15 @@ public class GridDhtTxLocal extends GridDhtTxLocalAdapter implements GridCacheMa
         }
 
         try {
-            if (reads != null)
+            if (reads != null) {
                 for (IgniteTxEntry e : reads)
                     addEntry(msgId, e);
+            }
 
-            if (writes != null)
+            if (writes != null) {
                 for (IgniteTxEntry e : writes)
                     addEntry(msgId, e);
+            }
 
             userPrepare();
 

http://git-wip-us.apache.org/repos/asf/ignite/blob/3ff71fd7/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxLocalAdapter.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxLocalAdapter.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxLocalAdapter.java
index 6de8795..70ebf3f 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxLocalAdapter.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxLocalAdapter.java
@@ -86,9 +86,6 @@ public abstract class GridDhtTxLocalAdapter extends IgniteTxLocalAdapter {
     protected volatile boolean mapped;
 
     /** */
-    private long dhtThreadId;
-
-    /** */
     protected boolean explicitLock;
 
     /** Versions of pending locks for entries of this tx. */
@@ -159,7 +156,6 @@ public abstract class GridDhtTxLocalAdapter extends IgniteTxLocalAdapter {
         this.explicitLock = explicitLock;
 
         threadId = Thread.currentThread().getId();
-        dhtThreadId = threadId;
     }
 
     /**
@@ -216,11 +212,6 @@ public abstract class GridDhtTxLocalAdapter extends IgniteTxLocalAdapter {
     protected abstract IgniteUuid nearFutureId();
 
     /**
-     * @return Near future mini ID.
-     */
-    protected abstract IgniteUuid nearMiniId();
-
-    /**
      * Adds reader to cached entry.
      *
      * @param msgId Message ID.
@@ -260,13 +251,6 @@ public abstract class GridDhtTxLocalAdapter extends IgniteTxLocalAdapter {
     }
 
     /**
-     * @return DHT thread ID.
-     */
-    long dhtThreadId() {
-        return dhtThreadId;
-    }
-
-    /**
      * Map explicit locks.
      */
     protected void mapExplicitLocks() {
@@ -355,22 +339,6 @@ public abstract class GridDhtTxLocalAdapter extends IgniteTxLocalAdapter {
     }
 
     /**
-     * @param nodeId Node ID.
-     * @return Mapping.
-     */
-    GridDistributedTxMapping dhtMapping(UUID nodeId) {
-        return dhtMap.get(nodeId);
-    }
-
-    /**
-     * @param nodeId Node ID.
-     * @return Mapping.
-     */
-    GridDistributedTxMapping nearMapping(UUID nodeId) {
-        return nearMap.get(nodeId);
-    }
-
-    /**
      * @param mappings Mappings to add.
      */
     void addDhtNodeEntryMapping(Map<ClusterNode, List<GridDhtCacheEntry>> mappings) {
@@ -385,19 +353,6 @@ public abstract class GridDhtTxLocalAdapter extends IgniteTxLocalAdapter {
     }
 
     /**
-     * @param mappings Mappings to add.
-     */
-    public void addDhtMapping(Map<UUID, GridDistributedTxMapping> mappings) {
-        addMapping0(mappings, dhtMap);
-    }
-
-    /**
-     * @param mappings Mappings to add.
-     */
-    public void addNearMapping(Map<UUID, GridDistributedTxMapping> mappings) {
-        addMapping0(mappings, nearMap);
-    }
-    /**
      * @param nodeId Node ID.
      * @return {@code True} if mapping was removed.
      */
@@ -435,7 +390,7 @@ public abstract class GridDhtTxLocalAdapter extends IgniteTxLocalAdapter {
             if (log.isDebugEnabled())
                 log.debug("Removing mapping for entry [nodeId=" + nodeId + ", entry=" + entry + ']');
 
-            IgniteTxEntry txEntry = txMap.get(entry.txKey());
+            IgniteTxEntry txEntry = entry(entry.txKey());
 
             if (txEntry == null)
                 return false;
@@ -469,7 +424,7 @@ public abstract class GridDhtTxLocalAdapter extends IgniteTxLocalAdapter {
             List<GridDhtCacheEntry> entries = mapping.getValue();
 
             for (GridDhtCacheEntry entry : entries) {
-                IgniteTxEntry txEntry = txMap.get(entry.txKey());
+                IgniteTxEntry txEntry = entry(entry.txKey());
 
                 if (txEntry != null) {
                     if (m == null)
@@ -481,26 +436,6 @@ public abstract class GridDhtTxLocalAdapter extends IgniteTxLocalAdapter {
         }
     }
 
-    /**
-     * @param mappings Mappings to add.
-     * @param dst Map to add to.
-     */
-    private void addMapping0(
-        Map<UUID, GridDistributedTxMapping> mappings,
-        Map<UUID, GridDistributedTxMapping> dst
-    ) {
-        for (Map.Entry<UUID, GridDistributedTxMapping> entry : mappings.entrySet()) {
-            GridDistributedTxMapping targetMapping = dst.get(entry.getKey());
-
-            if (targetMapping == null)
-                dst.put(entry.getKey(), entry.getValue());
-            else {
-                for (IgniteTxEntry txEntry : entry.getValue().entries())
-                    targetMapping.add(txEntry);
-            }
-        }
-    }
-
     /** {@inheritDoc} */
     @Override public void addInvalidPartition(GridCacheContext ctx, int part) {
         assert false : "DHT transaction encountered invalid partition [part=" + part + ", tx=" + this + ']';
@@ -529,7 +464,7 @@ public abstract class GridDhtTxLocalAdapter extends IgniteTxLocalAdapter {
         GridDhtCacheAdapter dhtCache = cacheCtx.isNear() ? cacheCtx.near().dht() : cacheCtx.dht();
 
         try {
-            IgniteTxEntry existing = txMap.get(e.txKey());
+            IgniteTxEntry existing = entry(e.txKey());
 
             if (existing != null) {
                 // Must keep NOOP operation if received READ because it means that the lock was sent to a backup node.
@@ -569,7 +504,7 @@ public abstract class GridDhtTxLocalAdapter extends IgniteTxLocalAdapter {
                     existing.explicitVersion(dhtVer);
                 }
 
-                txMap.put(existing.txKey(), existing);
+                txState.addEntry(existing);
 
                 if (log.isDebugEnabled())
                     log.debug("Added entry to transaction: " + existing);
@@ -705,7 +640,6 @@ public abstract class GridDhtTxLocalAdapter extends IgniteTxLocalAdapter {
                 passedKeys,
                 read,
                 needRetVal,
-                skipped,
                 accessTtl,
                 null,
                 skipStore);
@@ -723,7 +657,6 @@ public abstract class GridDhtTxLocalAdapter extends IgniteTxLocalAdapter {
      * @param passedKeys Passed keys.
      * @param read {@code True} if read.
      * @param needRetVal Return value flag.
-     * @param skipped Skipped keys.
      * @param accessTtl TTL for read operation.
      * @param filter Entry write filter.
      * @param skipStore Skip store flag.
@@ -735,13 +668,11 @@ public abstract class GridDhtTxLocalAdapter extends IgniteTxLocalAdapter {
         final Collection<KeyCacheObject> passedKeys,
         final boolean read,
         final boolean needRetVal,
-        final Set<KeyCacheObject> skipped,
         final long accessTtl,
         @Nullable final CacheEntryPredicate[] filter,
         boolean skipStore) {
         if (log.isDebugEnabled())
-            log.debug("Before acquiring transaction lock on keys [passedKeys=" + passedKeys + ", skipped=" +
-                skipped + ']');
+            log.debug("Before acquiring transaction lock on keys [keys=" + passedKeys + ']');
 
         if (passedKeys.isEmpty())
             return new GridFinishedFuture<>(ret);
@@ -768,7 +699,6 @@ public abstract class GridDhtTxLocalAdapter extends IgniteTxLocalAdapter {
 
                     postLockWrite(cacheCtx,
                         passedKeys,
-                        skipped,
                         ret,
                         /*remove*/false,
                         /*retval*/false,

http://git-wip-us.apache.org/repos/asf/ignite/blob/3ff71fd7/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxMapping.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxMapping.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxMapping.java
index 55cbe96..9ec35b6 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxMapping.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxMapping.java
@@ -17,18 +17,15 @@
 
 package org.apache.ignite.internal.processors.cache.distributed.dht;
 
-import java.util.ArrayList;
 import java.util.Collection;
-import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
-import java.util.Set;
 import java.util.UUID;
 import org.apache.ignite.cluster.ClusterNode;
-import org.apache.ignite.internal.processors.cache.distributed.GridDistributedTxMapping;
 import org.apache.ignite.internal.util.GridLeanMap;
+import org.apache.ignite.internal.util.GridLeanSet;
 import org.apache.ignite.internal.util.typedef.F;
-import org.jetbrains.annotations.Nullable;
+import org.apache.ignite.internal.util.typedef.internal.U;
 
 /**
  * DHT transaction mapping.
@@ -37,12 +34,6 @@ public class GridDhtTxMapping {
     /** Transaction nodes mapping (primary node -> related backup nodes). */
     private final Map<UUID, Collection<UUID>> txNodes = new GridLeanMap<>();
 
-    /** */
-    private final List<TxMapping> mappings = new ArrayList<>();
-
-    /** */
-    private TxMapping last;
-
     /**
      * Adds information about next mapping.
      *
@@ -50,24 +41,26 @@ public class GridDhtTxMapping {
      */
     @SuppressWarnings("ConstantConditions")
     public void addMapping(List<ClusterNode> nodes) {
-        ClusterNode primary = F.first(nodes);
+        assert !F.isEmpty(nodes) : nodes;
 
-        Collection<ClusterNode> backups = F.view(nodes, F.notEqualTo(primary));
+        ClusterNode primary = nodes.get(0);
 
-        if (last == null || !last.primary.equals(primary.id())) {
-            last = new TxMapping(primary, backups);
+        int size = nodes.size();
 
-            mappings.add(last);
-        }
-        else
-            last.add(backups);
+        if (size > 1) {
+            Collection<UUID> backups = txNodes.get(primary.id());
 
-        Collection<UUID> storedBackups = txNodes.get(last.primary);
+            if (backups == null) {
+                backups = U.newHashSet(size - 1);
 
-        if (storedBackups == null)
-            txNodes.put(last.primary, storedBackups = new HashSet<>());
+                txNodes.put(primary.id(), backups);
+            }
 
-        storedBackups.addAll(last.backups);
+            for (int i = 1; i < size; i++)
+                backups.add(nodes.get(i).id());
+        }
+        else
+            txNodes.put(primary.id(), new GridLeanSet<UUID>());
     }
 
     /**
@@ -76,99 +69,4 @@ public class GridDhtTxMapping {
     public Map<UUID, Collection<UUID>> transactionNodes() {
         return txNodes;
     }
-
-    /**
-     * For each mapping sets flags indicating if mapping is last for node.
-     *
-     * @param mappings Mappings.
-     */
-    public void initLast(Collection<GridDistributedTxMapping> mappings) {
-        assert this.mappings.size() == mappings.size();
-
-        int idx = 0;
-
-        for (GridDistributedTxMapping map : mappings) {
-            TxMapping mapping = this.mappings.get(idx);
-
-            map.lastBackups(lastBackups(mapping, idx));
-
-            boolean last = true;
-
-            for (int i = idx + 1; i < this.mappings.size(); i++) {
-                TxMapping nextMap = this.mappings.get(i);
-
-                if (nextMap.primary.equals(mapping.primary)) {
-                    last = false;
-
-                    break;
-                }
-            }
-
-            map.last(last);
-
-            idx++;
-        }
-    }
-
-    /**
-     * @param mapping Mapping.
-     * @param idx Mapping index.
-     * @return IDs of backup nodes receiving last prepare request during this mapping.
-     */
-    @Nullable private Collection<UUID> lastBackups(TxMapping mapping, int idx) {
-        Collection<UUID> res = null;
-
-        for (UUID backup : mapping.backups) {
-            boolean foundNext = false;
-
-            for (int i = idx + 1; i < mappings.size(); i++) {
-                TxMapping nextMap = mappings.get(i);
-
-                if (nextMap.primary.equals(mapping.primary) && nextMap.backups.contains(backup)) {
-                    foundNext = true;
-
-                    break;
-                }
-            }
-
-            if (!foundNext) {
-                if (res == null)
-                    res = new ArrayList<>(mapping.backups.size());
-
-                res.add(backup);
-            }
-        }
-
-        return res;
-    }
-
-    /**
-     */
-    private static class TxMapping {
-        /** */
-        private final UUID primary;
-
-        /** */
-        private final Set<UUID> backups;
-
-        /**
-         * @param primary Primary node.
-         * @param backups Backup nodes.
-         */
-        private TxMapping(ClusterNode primary, Iterable<ClusterNode> backups) {
-            this.primary = primary.id();
-
-            this.backups = new HashSet<>();
-
-            add(backups);
-        }
-
-        /**
-         * @param backups Backup nodes.
-         */
-        private void add(Iterable<ClusterNode> backups) {
-            for (ClusterNode n : backups)
-                this.backups.add(n.id());
-        }
-    }
 }
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/ignite/blob/3ff71fd7/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxPrepareFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxPrepareFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxPrepareFuture.java
index 1d6f633..a67950d 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxPrepareFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxPrepareFuture.java
@@ -165,9 +165,6 @@ public final class GridDhtTxPrepareFuture extends GridCompoundFuture<IgniteInter
     /** {@code True} if this is last prepare operation for node. */
     private boolean last;
 
-    /** IDs of backup nodes receiving last prepare request during this prepare. */
-    private Collection<UUID> lastBackups;
-
     /** Needs return value flag. */
     private boolean retVal;
 
@@ -197,7 +194,6 @@ public final class GridDhtTxPrepareFuture extends GridCompoundFuture<IgniteInter
      * @param dhtVerMap DHT versions map.
      * @param last {@code True} if this is last prepare operation for node.
      * @param retVal Return value flag.
-     * @param lastBackups IDs of backup nodes receiving last prepare request during this prepare.
      */
     public GridDhtTxPrepareFuture(
         GridCacheSharedContext cctx,
@@ -205,8 +201,7 @@ public final class GridDhtTxPrepareFuture extends GridCompoundFuture<IgniteInter
         IgniteUuid nearMiniId,
         Map<IgniteTxKey, GridCacheVersion> dhtVerMap,
         boolean last,
-        boolean retVal,
-        Collection<UUID> lastBackups
+        boolean retVal
     ) {
         super(REDUCER);
 
@@ -214,7 +209,6 @@ public final class GridDhtTxPrepareFuture extends GridCompoundFuture<IgniteInter
         this.tx = tx;
         this.dhtVerMap = dhtVerMap;
         this.last = last;
-        this.lastBackups = lastBackups;
 
         futId = IgniteUuid.randomUuid();
 
@@ -864,14 +858,6 @@ public final class GridDhtTxPrepareFuture extends GridCompoundFuture<IgniteInter
     }
 
     /**
-     * @param backupId Backup node ID.
-     * @return {@code True} if backup node receives last prepare request for this transaction.
-     */
-    private boolean lastBackup(UUID backupId) {
-        return lastBackups != null && lastBackups.contains(backupId);
-    }
-
-    /**
      * Checks if this transaction needs previous value for the given tx entry. Will use passed in map to store
      * required key or will create new map if passed in map is {@code null}.
      *
@@ -1022,18 +1008,15 @@ public final class GridDhtTxPrepareFuture extends GridCompoundFuture<IgniteInter
             tx.writeVersion(cctx.versions().next(tx.topologyVersion()));
 
             {
-                Map<UUID, GridDistributedTxMapping> futDhtMap = new HashMap<>();
-                Map<UUID, GridDistributedTxMapping> futNearMap = new HashMap<>();
-
                 // Assign keys to primary nodes.
                 if (!F.isEmpty(writes)) {
                     for (IgniteTxEntry write : writes)
-                        map(tx.entry(write.txKey()), futDhtMap, futNearMap);
+                        map(tx.entry(write.txKey()));
                 }
 
                 if (!F.isEmpty(reads)) {
                     for (IgniteTxEntry read : reads)
-                        map(tx.entry(read.txKey()), futDhtMap, futNearMap);
+                        map(tx.entry(read.txKey()));
                 }
             }
 
@@ -1225,14 +1208,8 @@ public final class GridDhtTxPrepareFuture extends GridCompoundFuture<IgniteInter
 
     /**
      * @param entry Transaction entry.
-     * @param futDhtMap DHT mapping.
-     * @param futNearMap Near mapping.
      */
-    private void map(
-        IgniteTxEntry entry,
-        Map<UUID, GridDistributedTxMapping> futDhtMap,
-        Map<UUID, GridDistributedTxMapping> futNearMap
-    ) {
+    private void map(IgniteTxEntry entry) {
         if (entry.cached().isLocal())
             return;
 
@@ -1258,26 +1235,25 @@ public final class GridDhtTxPrepareFuture extends GridCompoundFuture<IgniteInter
                     log.debug("Mapping entry to DHT nodes [nodes=" + U.toShortString(dhtNodes) +
                         ", entry=" + entry + ']');
 
-                Collection<UUID> readers = cached.readers();
+                // Exclude local node.
+                map(entry, F.view(dhtNodes, F.remoteNodes(cctx.localNodeId())), dhtMap);
 
-                Collection<ClusterNode> nearNodes = null;
+                Collection<UUID> readers = cached.readers();
 
                 if (!F.isEmpty(readers)) {
-                    nearNodes = cctx.discovery().nodes(readers, F0.not(F.idForNodeId(tx.nearNodeId())));
+                    Collection<ClusterNode> nearNodes =
+                        cctx.discovery().nodes(readers, F0.not(F.idForNodeId(tx.nearNodeId())));
 
                     if (log.isDebugEnabled())
                         log.debug("Mapping entry to near nodes [nodes=" + U.toShortString(nearNodes) +
                             ", entry=" + entry + ']');
+
+                    // Exclude DHT nodes.
+                    map(entry, F.view(nearNodes, F0.notIn(dhtNodes)), nearMap);
                 }
                 else if (log.isDebugEnabled())
                     log.debug("Entry has no near readers: " + entry);
 
-                // Exclude local node.
-                map(entry, F.view(dhtNodes, F.remoteNodes(cctx.localNodeId())), dhtMap, futDhtMap);
-
-                // Exclude DHT nodes.
-                map(entry, F.view(nearNodes, F0.notIn(dhtNodes)), nearMap, futNearMap);
-
                 break;
             }
             catch (GridCacheEntryRemovedException ignore) {
@@ -1292,13 +1268,11 @@ public final class GridDhtTxPrepareFuture extends GridCompoundFuture<IgniteInter
      * @param entry Entry.
      * @param nodes Nodes.
      * @param globalMap Map.
-     * @param locMap Exclude map.
      */
     private void map(
         IgniteTxEntry entry,
         Iterable<ClusterNode> nodes,
-        Map<UUID, GridDistributedTxMapping> globalMap,
-        Map<UUID, GridDistributedTxMapping> locMap
+        Map<UUID, GridDistributedTxMapping> globalMap
     ) {
         if (nodes != null) {
             for (ClusterNode n : nodes) {
@@ -1321,13 +1295,6 @@ public final class GridDhtTxPrepareFuture extends GridCompoundFuture<IgniteInter
                     globalMap.put(n.id(), global = new GridDistributedTxMapping(n));
 
                 global.add(entry);
-
-                GridDistributedTxMapping loc = locMap.get(n.id());
-
-                if (loc == null)
-                    locMap.put(n.id(), loc = new GridDistributedTxMapping(n));
-
-                loc.add(entry);
             }
         }
     }

http://git-wip-us.apache.org/repos/asf/ignite/blob/3ff71fd7/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxRemote.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxRemote.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxRemote.java
index e268a88..0cbe10f 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxRemote.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxRemote.java
@@ -36,6 +36,8 @@ import org.apache.ignite.internal.processors.cache.KeyCacheObject;
 import org.apache.ignite.internal.processors.cache.distributed.GridDistributedTxRemoteAdapter;
 import org.apache.ignite.internal.processors.cache.transactions.IgniteTxEntry;
 import org.apache.ignite.internal.processors.cache.transactions.IgniteTxKey;
+import org.apache.ignite.internal.processors.cache.transactions.IgniteTxRemoteSingleStateImpl;
+import org.apache.ignite.internal.processors.cache.transactions.IgniteTxRemoteStateImpl;
 import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
 import org.apache.ignite.internal.util.tostring.GridToStringBuilder;
 import org.apache.ignite.internal.util.typedef.T2;
@@ -77,7 +79,6 @@ public class GridDhtTxRemote extends GridDistributedTxRemoteAdapter {
      * @param nearNodeId Near node ID.
      * @param rmtFutId Remote future ID.
      * @param nodeId Node ID.
-     * @param rmtThreadId Remote thread ID.
      * @param topVer Topology version.
      * @param xidVer XID version.
      * @param commitVer Commit version.
@@ -96,7 +97,6 @@ public class GridDhtTxRemote extends GridDistributedTxRemoteAdapter {
         UUID nearNodeId,
         IgniteUuid rmtFutId,
         UUID nodeId,
-        long rmtThreadId,
         AffinityTopologyVersion topVer,
         GridCacheVersion xidVer,
         GridCacheVersion commitVer,
@@ -110,12 +110,12 @@ public class GridDhtTxRemote extends GridDistributedTxRemoteAdapter {
         GridCacheVersion nearXidVer,
         Map<UUID, Collection<UUID>> txNodes,
         @Nullable UUID subjId,
-        int taskNameHash
+        int taskNameHash,
+        boolean single
     ) {
         super(
             ctx,
             nodeId,
-            rmtThreadId,
             xidVer,
             commitVer,
             sys,
@@ -137,9 +137,10 @@ public class GridDhtTxRemote extends GridDistributedTxRemoteAdapter {
         this.nearXidVer = nearXidVer;
         this.txNodes = txNodes;
 
-        readMap = Collections.emptyMap();
-
-        writeMap = new ConcurrentLinkedHashMap<>(U.capacity(txSize), 0.75f, 1);
+        txState = single ? new IgniteTxRemoteSingleStateImpl() :
+            new IgniteTxRemoteStateImpl(
+            Collections.<IgniteTxKey, IgniteTxEntry>emptyMap(),
+            new ConcurrentLinkedHashMap<IgniteTxKey, IgniteTxEntry>(U.capacity(txSize), 0.75f, 1));
 
         topologyVersion(topVer);
     }
@@ -151,7 +152,6 @@ public class GridDhtTxRemote extends GridDistributedTxRemoteAdapter {
      * @param rmtFutId Remote future ID.
      * @param nodeId Node ID.
      * @param nearXidVer Near transaction ID.
-     * @param rmtThreadId Remote thread ID.
      * @param topVer Topology version.
      * @param xidVer XID version.
      * @param commitVer Commit version.
@@ -169,7 +169,6 @@ public class GridDhtTxRemote extends GridDistributedTxRemoteAdapter {
         IgniteUuid rmtFutId,
         UUID nodeId,
         GridCacheVersion nearXidVer,
-        long rmtThreadId,
         AffinityTopologyVersion topVer,
         GridCacheVersion xidVer,
         GridCacheVersion commitVer,
@@ -186,7 +185,6 @@ public class GridDhtTxRemote extends GridDistributedTxRemoteAdapter {
         super(
             ctx,
             nodeId,
-            rmtThreadId,
             xidVer,
             commitVer,
             sys,
@@ -207,8 +205,9 @@ public class GridDhtTxRemote extends GridDistributedTxRemoteAdapter {
         this.nearNodeId = nearNodeId;
         this.rmtFutId = rmtFutId;
 
-        readMap = Collections.emptyMap();
-        writeMap = new ConcurrentLinkedHashMap<>(U.capacity(txSize), 0.75f, 1);
+        txState = new IgniteTxRemoteStateImpl(
+            Collections.<IgniteTxKey, IgniteTxEntry>emptyMap(),
+            new ConcurrentLinkedHashMap<IgniteTxKey, IgniteTxEntry>(U.capacity(txSize), 0.75f, 1));
 
         topologyVersion(topVer);
     }
@@ -280,6 +279,8 @@ public class GridDhtTxRemote extends GridDistributedTxRemoteAdapter {
     @Override public void addInvalidPartition(GridCacheContext cacheCtx, int part) {
         super.addInvalidPartition(cacheCtx, part);
 
+        Map<IgniteTxKey, IgniteTxEntry> writeMap = txState.writeMap();
+
         for (Iterator<IgniteTxEntry> it = writeMap.values().iterator(); it.hasNext();) {
             IgniteTxEntry e = it.next();
 
@@ -312,7 +313,7 @@ public class GridDhtTxRemote extends GridDistributedTxRemoteAdapter {
             // Initialize cache entry.
             entry.cached(cached);
 
-            writeMap.put(entry.txKey(), entry);
+            txState.addWriteEntry(entry.txKey(), entry);
 
             addExplicit(entry);
         }
@@ -356,7 +357,7 @@ public class GridDhtTxRemote extends GridDistributedTxRemoteAdapter {
 
         txEntry.entryProcessors(entryProcessors);
 
-        writeMap.put(key, txEntry);
+        txState.addWriteEntry(key, txEntry);
     }
 
     /** {@inheritDoc} */

http://git-wip-us.apache.org/repos/asf/ignite/blob/3ff71fd7/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearOptimisticSerializableTxPrepareFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearOptimisticSerializableTxPrepareFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearOptimisticSerializableTxPrepareFuture.java
index 5488bb1..29774a5 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearOptimisticSerializableTxPrepareFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearOptimisticSerializableTxPrepareFuture.java
@@ -29,7 +29,6 @@ import org.apache.ignite.cluster.ClusterNode;
 import org.apache.ignite.cluster.ClusterTopologyException;
 import org.apache.ignite.internal.IgniteInternalFuture;
 import org.apache.ignite.internal.cluster.ClusterTopologyCheckedException;
-import org.apache.ignite.internal.cluster.ClusterTopologyServerNotFoundException;
 import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
 import org.apache.ignite.internal.processors.cache.GridCacheContext;
 import org.apache.ignite.internal.processors.cache.GridCacheEntryEx;
@@ -57,7 +56,6 @@ import org.apache.ignite.internal.util.typedef.CI1;
 import org.apache.ignite.internal.util.typedef.F;
 import org.apache.ignite.internal.util.typedef.P1;
 import org.apache.ignite.internal.util.typedef.X;
-import org.apache.ignite.internal.util.typedef.internal.CU;
 import org.apache.ignite.internal.util.typedef.internal.S;
 import org.apache.ignite.internal.util.typedef.internal.U;
 import org.apache.ignite.lang.IgniteBiTuple;
@@ -326,19 +324,6 @@ public class GridNearOptimisticSerializableTxPrepareFuture extends GridNearOptim
 
         txMapping = new GridDhtTxMapping();
 
-        if (!F.isEmpty(reads) || !F.isEmpty(writes)) {
-            for (int cacheId : tx.activeCacheIds()) {
-                GridCacheContext<?, ?> cacheCtx = cctx.cacheContext(cacheId);
-
-                if (CU.affinityNodes(cacheCtx, topVer).isEmpty()) {
-                    onDone(new ClusterTopologyServerNotFoundException("Failed to map keys for cache (all " +
-                        "partition nodes left the grid): " + cacheCtx.name()));
-
-                    return;
-                }
-            }
-        }
-
         Map<IgniteBiTuple<ClusterNode, Boolean>, GridDistributedTxMapping> mappings = new HashMap<>();
 
         for (IgniteTxEntry write : writes)
@@ -437,7 +422,6 @@ public class GridNearOptimisticSerializableTxPrepareFuture extends GridNearOptim
             m.near(),
             txMapping.transactionNodes(),
             m.last(),
-            m.lastBackups(),
             tx.onePhaseCommit(),
             tx.needReturnValue() && tx.implicit(),
             tx.implicitSingle(),

http://git-wip-us.apache.org/repos/asf/ignite/blob/3ff71fd7/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearOptimisticTxPrepareFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearOptimisticTxPrepareFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearOptimisticTxPrepareFuture.java
index 0002180..791d2f3 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearOptimisticTxPrepareFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearOptimisticTxPrepareFuture.java
@@ -19,7 +19,9 @@ package org.apache.ignite.internal.processors.cache.distributed.near;
 
 import java.util.ArrayDeque;
 import java.util.Collection;
+import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
 import java.util.Queue;
 import java.util.UUID;
 import java.util.concurrent.atomic.AtomicBoolean;
@@ -28,7 +30,6 @@ import org.apache.ignite.cluster.ClusterNode;
 import org.apache.ignite.cluster.ClusterTopologyException;
 import org.apache.ignite.internal.IgniteInternalFuture;
 import org.apache.ignite.internal.cluster.ClusterTopologyCheckedException;
-import org.apache.ignite.internal.cluster.ClusterTopologyServerNotFoundException;
 import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
 import org.apache.ignite.internal.processors.cache.GridCacheContext;
 import org.apache.ignite.internal.processors.cache.GridCacheEntryEx;
@@ -50,7 +51,6 @@ import org.apache.ignite.internal.util.typedef.C1;
 import org.apache.ignite.internal.util.typedef.CI1;
 import org.apache.ignite.internal.util.typedef.F;
 import org.apache.ignite.internal.util.typedef.X;
-import org.apache.ignite.internal.util.typedef.internal.CU;
 import org.apache.ignite.internal.util.typedef.internal.S;
 import org.apache.ignite.internal.util.typedef.internal.U;
 import org.apache.ignite.lang.IgniteClosure;
@@ -267,7 +267,12 @@ public class GridNearOptimisticTxPrepareFuture extends GridNearOptimisticTxPrepa
                 return;
             }
 
-            prepare(tx.writeEntries(), topLocked);
+            IgniteTxEntry singleWrite = tx.singleWrite();
+
+            if (singleWrite != null)
+                prepareSingle(singleWrite, topLocked);
+            else
+                prepare(tx.writeEntries(), topLocked);
 
             markInitialized();
         }
@@ -280,6 +285,46 @@ public class GridNearOptimisticTxPrepareFuture extends GridNearOptimisticTxPrepa
     }
 
     /**
+     * @param write Write.
+     * @param topLocked {@code True} if thread already acquired lock preventing topology change.
+     */
+    private void prepareSingle(IgniteTxEntry write, boolean topLocked) {
+        AffinityTopologyVersion topVer = tx.topologyVersion();
+
+        assert topVer.topologyVersion() > 0;
+
+        txMapping = new GridDhtTxMapping();
+
+        GridDistributedTxMapping mapping = map(write, topVer, null, topLocked);
+
+        if (mapping.node().isLocal()) {
+            if (write.context().isNear())
+                tx.nearLocallyMapped(true);
+            else if (write.context().isColocated())
+                tx.colocatedLocallyMapped(true);
+        }
+
+        if (isDone()) {
+            if (log.isDebugEnabled())
+                log.debug("Abandoning (re)map because future is done: " + this);
+
+            return;
+        }
+
+        tx.addSingleEntryMapping(mapping, write);
+
+        cctx.mvcc().recheckPendingLocks();
+
+        mapping.last(true);
+
+        tx.transactionNodes(txMapping.transactionNodes());
+
+        checkOnePhase();
+
+        proceedPrepare(mapping, null);
+    }
+
+    /**
      * @param writes Write entries.
      * @param topLocked {@code True} if thread already acquired lock preventing topology change.
      * @throws IgniteCheckedException If failed.
@@ -294,30 +339,26 @@ public class GridNearOptimisticTxPrepareFuture extends GridNearOptimisticTxPrepa
 
         txMapping = new GridDhtTxMapping();
 
-        Queue<GridDistributedTxMapping> mappings = new ArrayDeque<>();
-
-        if (!F.isEmpty(writes)) {
-            for (int cacheId : tx.activeCacheIds()) {
-                GridCacheContext<?, ?> cacheCtx = cctx.cacheContext(cacheId);
-
-                if (CU.affinityNodes(cacheCtx, topVer).isEmpty()) {
-                    onDone(new ClusterTopologyServerNotFoundException("Failed to map keys for cache (all " +
-                        "partition nodes left the grid): " + cacheCtx.name()));
-
-                    return;
-                }
-            }
-        }
+        Map<UUID, GridDistributedTxMapping> map = new HashMap<>();
 
         // Assign keys to primary nodes.
         GridDistributedTxMapping cur = null;
 
+        Queue<GridDistributedTxMapping> mappings = new ArrayDeque<>();
+
         for (IgniteTxEntry write : writes) {
             GridDistributedTxMapping updated = map(write, topVer, cur, topLocked);
 
             if (cur != updated) {
                 mappings.offer(updated);
 
+                updated.last(true);
+
+                GridDistributedTxMapping prev = map.put(updated.node().id(), updated);
+
+                if (prev != null)
+                    prev.last(false);
+
                 if (updated.node().isLocal()) {
                     if (write.context().isNear())
                         tx.nearLocallyMapped(true);
@@ -340,8 +381,6 @@ public class GridNearOptimisticTxPrepareFuture extends GridNearOptimisticTxPrepa
 
         cctx.mvcc().recheckPendingLocks();
 
-        txMapping.initLast(mappings);
-
         tx.transactionNodes(txMapping.transactionNodes());
 
         checkOnePhase();
@@ -355,14 +394,24 @@ public class GridNearOptimisticTxPrepareFuture extends GridNearOptimisticTxPrepa
      * @param mappings Queue of mappings.
      */
     private void proceedPrepare(final Queue<GridDistributedTxMapping> mappings) {
-        if (isDone())
-            return;
-
         final GridDistributedTxMapping m = mappings.poll();
 
         if (m == null)
             return;
 
+        proceedPrepare(m, mappings);
+    }
+
+    /**
+     * Continues prepare after previous mapping successfully finished.
+     *
+     * @param m Mapping.
+     * @param mappings Queue of mappings.
+     */
+    private void proceedPrepare(GridDistributedTxMapping m, @Nullable final Queue<GridDistributedTxMapping> mappings) {
+        if (isDone())
+            return;
+
         assert !m.empty();
 
         final ClusterNode n = m.node();
@@ -376,7 +425,6 @@ public class GridNearOptimisticTxPrepareFuture extends GridNearOptimisticTxPrepa
             m.near(),
             txMapping.transactionNodes(),
             m.last(),
-            m.lastBackups(),
             tx.onePhaseCommit(),
             tx.needReturnValue() && tx.implicit(),
             tx.implicitSingle(),
@@ -457,7 +505,14 @@ public class GridNearOptimisticTxPrepareFuture extends GridNearOptimisticTxPrepa
     ) {
         GridCacheContext cacheCtx = entry.context();
 
-        List<ClusterNode> nodes = cacheCtx.affinity().nodes(entry.key(), topVer);
+        List<ClusterNode> nodes;
+
+        GridCacheEntryEx cached0 = entry.cached();
+
+        if (cached0.isDht())
+            nodes = cacheCtx.affinity().nodes(cached0.partition(), topVer);
+        else
+            nodes = cacheCtx.affinity().nodes(entry.key(), topVer);
 
         txMapping.addMapping(nodes);
 

http://git-wip-us.apache.org/repos/asf/ignite/blob/3ff71fd7/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearOptimisticTxPrepareFutureAdapter.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearOptimisticTxPrepareFutureAdapter.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearOptimisticTxPrepareFutureAdapter.java
index fd9183e..6b7244a 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearOptimisticTxPrepareFutureAdapter.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearOptimisticTxPrepareFutureAdapter.java
@@ -20,13 +20,11 @@ package org.apache.ignite.internal.processors.cache.distributed.near;
 import org.apache.ignite.IgniteCheckedException;
 import org.apache.ignite.internal.IgniteInternalFuture;
 import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
-import org.apache.ignite.internal.processors.cache.GridCacheContext;
 import org.apache.ignite.internal.processors.cache.GridCacheSharedContext;
 import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtTopologyFuture;
 import org.apache.ignite.internal.processors.cache.transactions.IgniteInternalTx;
 import org.apache.ignite.internal.util.lang.GridPlainRunnable;
 import org.apache.ignite.internal.util.typedef.CI1;
-import org.apache.ignite.internal.util.typedef.internal.U;
 import org.jetbrains.annotations.Nullable;
 
 /**
@@ -75,56 +73,14 @@ public abstract class GridNearOptimisticTxPrepareFutureAdapter extends GridNearT
      * @return Topology ready future.
      */
     protected final GridDhtTopologyFuture topologyReadLock() {
-        if (tx.activeCacheIds().isEmpty())
-            return cctx.exchange().lastTopologyFuture();
-
-        GridCacheContext<?, ?> nonLocCtx = null;
-
-        for (int cacheId : tx.activeCacheIds()) {
-            GridCacheContext<?, ?> cacheCtx = cctx.cacheContext(cacheId);
-
-            if (!cacheCtx.isLocal()) {
-                nonLocCtx = cacheCtx;
-
-                break;
-            }
-        }
-
-        if (nonLocCtx == null)
-            return cctx.exchange().lastTopologyFuture();
-
-        nonLocCtx.topology().readLock();
-
-        if (nonLocCtx.topology().stopping()) {
-            onDone(new IgniteCheckedException("Failed to perform cache operation (cache is stopped): " +
-                nonLocCtx.name()));
-
-            return null;
-        }
-
-        return nonLocCtx.topology().topologyVersionFuture();
+        return tx.txState().topologyReadLock(cctx, this);
     }
 
     /**
      * Releases topology read lock.
      */
     protected final void topologyReadUnlock() {
-        if (!tx.activeCacheIds().isEmpty()) {
-            GridCacheContext<?, ?> nonLocCtx = null;
-
-            for (int cacheId : tx.activeCacheIds()) {
-                GridCacheContext<?, ?> cacheCtx = cctx.cacheContext(cacheId);
-
-                if (!cacheCtx.isLocal()) {
-                    nonLocCtx = cacheCtx;
-
-                    break;
-                }
-            }
-
-            if (nonLocCtx != null)
-                nonLocCtx.topology().readUnlock();
-        }
+        tx.txState().topologyReadUnlock(cctx);
     }
 
     /**
@@ -160,28 +116,10 @@ public abstract class GridNearOptimisticTxPrepareFutureAdapter extends GridNearT
         }
 
         if (topVer != null) {
-            StringBuilder invalidCaches = null;
-
-            for (Integer cacheId : tx.activeCacheIds()) {
-                GridCacheContext ctx = cctx.cacheContext(cacheId);
-
-                assert ctx != null : cacheId;
-
-                Throwable err = topFut.validateCache(ctx);
-
-                if (err != null) {
-                    if (invalidCaches != null)
-                        invalidCaches.append(", ");
-                    else
-                        invalidCaches = new StringBuilder();
-
-                    invalidCaches.append(U.maskName(ctx.name()));
-                }
-            }
+            IgniteCheckedException err = tx.txState().validateTopology(cctx, topFut);
 
-            if (invalidCaches != null) {
-                onDone(new IgniteCheckedException("Failed to perform cache operation (cache topology is not valid): " +
-                    invalidCaches.toString()));
+            if (err != null) {
+                onDone(err);
 
                 return;
             }

http://git-wip-us.apache.org/repos/asf/ignite/blob/3ff71fd7/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearPessimisticTxPrepareFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearPessimisticTxPrepareFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearPessimisticTxPrepareFuture.java
index 11d31b2..1554a62 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearPessimisticTxPrepareFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearPessimisticTxPrepareFuture.java
@@ -200,7 +200,6 @@ public class GridNearPessimisticTxPrepareFuture extends GridNearTxPrepareFutureA
                 m.near(),
                 txMapping.transactionNodes(),
                 true,
-                txMapping.transactionNodes().get(node.id()),
                 tx.onePhaseCommit(),
                 tx.needReturnValue() && tx.implicit(),
                 tx.implicitSingle(),

http://git-wip-us.apache.org/repos/asf/ignite/blob/3ff71fd7/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTransactionalCache.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTransactionalCache.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTransactionalCache.java
index 5ab85b2..e8546ef 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTransactionalCache.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTransactionalCache.java
@@ -315,7 +315,6 @@ public class GridNearTransactionalCache<K, V> extends GridNearCacheAdapter<K, V>
                                         nodeId,
                                         req.nearNodeId(),
                                         req.nearXidVersion(),
-                                        req.threadId(),
                                         req.version(),
                                         null,
                                         ctx.systemTx(),

http://git-wip-us.apache.org/repos/asf/ignite/blob/3ff71fd7/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxFinishFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxFinishFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxFinishFuture.java
index a9dbda2..31aa8c3 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxFinishFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxFinishFuture.java
@@ -88,7 +88,7 @@ public final class GridNearTxFinishFuture<K, V> extends GridCompoundIdentityFutu
     private boolean commit;
 
     /** Node mappings. */
-    private Map<UUID, GridDistributedTxMapping> mappings;
+    private IgniteTxMappings mappings;
 
     /** Trackable flag. */
     private boolean trackable = true;
@@ -347,8 +347,16 @@ public final class GridNearTxFinishFuture<K, V> extends GridCompoundIdentityFutu
 
         try {
             if (tx.finish(commit) || (!commit && tx.state() == UNKNOWN)) {
-                if ((tx.onePhaseCommit() && needFinishOnePhase()) || (!tx.onePhaseCommit() && mappings != null))
-                    finish(mappings.values());
+                if ((tx.onePhaseCommit() && needFinishOnePhase()) || (!tx.onePhaseCommit() && mappings != null)) {
+                    if (mappings.single()) {
+                        GridDistributedTxMapping mapping = mappings.singleMapping();
+
+                        if (mapping != null)
+                            finish(mapping);
+                    }
+                    else
+                        finish(mappings.mappings());
+                }
 
                 markInitialized();
 
@@ -381,11 +389,10 @@ public final class GridNearTxFinishFuture<K, V> extends GridCompoundIdentityFutu
      *
      */
     private void checkBackup() {
-        assert mappings.size() <= 1;
+        GridDistributedTxMapping mapping = mappings.singleMapping();
 
-        for (Map.Entry<UUID, GridDistributedTxMapping> entry : mappings.entrySet()) {
-            UUID nodeId = entry.getKey();
-            GridDistributedTxMapping mapping = entry.getValue();
+        if (mapping != null) {
+            UUID nodeId = mapping.node().id();
 
             Collection<UUID> backups = tx.transactionNodes().get(nodeId);
 
@@ -482,25 +489,13 @@ public final class GridNearTxFinishFuture<K, V> extends GridCompoundIdentityFutu
      *
      */
     private boolean needFinishOnePhase() {
-        if (F.isEmpty(tx.mappings()))
+        if (tx.mappings().empty())
             return false;
 
-        assert tx.mappings().size() == 1;
-
-        boolean finish = false;
-
-        for (Integer cacheId : tx.activeCacheIds()) {
-            GridCacheContext<K, V> cacheCtx = cctx.cacheContext(cacheId);
-
-            if (cacheCtx.isNear()) {
-                finish = true;
-
-                break;
-            }
-        }
+        boolean finish = tx.txState().hasNearCache(cctx);
 
         if (finish) {
-            GridDistributedTxMapping mapping = F.first(tx.mappings().values());
+            GridDistributedTxMapping mapping = tx.mappings().singleMapping();
 
             if (FINISH_NEAR_ONE_PHASE_SINCE.compareTo(mapping.node().version()) > 0)
                 finish = false;
@@ -520,18 +515,16 @@ public final class GridNearTxFinishFuture<K, V> extends GridCompoundIdentityFutu
 
         finishOnePhaseCalled = true;
 
-        // No need to send messages as transaction was already committed on remote node.
-        // Finish local mapping only as we need send commit message to backups.
-        for (GridDistributedTxMapping m : mappings.values()) {
-            if (m.node().isLocal()) {
-                IgniteInternalFuture<IgniteInternalTx> fut = cctx.tm().txHandler().finishColocatedLocal(commit, tx);
+        GridDistributedTxMapping locMapping = mappings.localMapping();
 
-                // Add new future.
-                if (fut != null)
-                    add(fut);
+        if (locMapping != null) {
+            // No need to send messages as transaction was already committed on remote node.
+            // Finish local mapping only as we need send commit message to backups.
+            IgniteInternalFuture<IgniteInternalTx> fut = cctx.tm().txHandler().finishColocatedLocal(commit, tx);
 
-                break;
-            }
+            // Add new future.
+            if (fut != null)
+                add(fut);
         }
     }
 
@@ -544,7 +537,9 @@ public final class GridNearTxFinishFuture<K, V> extends GridCompoundIdentityFutu
 
             mapping.dhtVersion(xidVer, xidVer);
 
-            tx.readyNearLocks(mapping, Collections.<GridCacheVersion>emptyList(), Collections.<GridCacheVersion>emptyList(),
+            tx.readyNearLocks(mapping,
+                Collections.<GridCacheVersion>emptyList(),
+                Collections.<GridCacheVersion>emptyList(),
                 Collections.<GridCacheVersion>emptyList());
         }
     }


Mime
View raw message