ignite-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From agoncha...@apache.org
Subject [02/50] [abbrv] ignite git commit: IGNITE-264 - WIP
Date Thu, 03 Sep 2015 01:03:11 GMT
IGNITE-264 - WIP


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

Branch: refs/heads/ignite-264
Commit: 9b0bd71b5844af5fd3a89655c1e3833e3e4cfcc4
Parents: 4bf51e7
Author: Alexey Goncharuk <agoncharuk@gridgain.com>
Authored: Thu Feb 26 11:20:34 2015 -0800
Committer: Alexey Goncharuk <agoncharuk@gridgain.com>
Committed: Thu Feb 26 11:20:34 2015 -0800

----------------------------------------------------------------------
 ...ridCacheOptimisticCheckPreparedTxFuture.java | 38 ++++++++--
 ...idCacheOptimisticCheckPreparedTxRequest.java | 44 +++++++++--
 .../dht/GridClientPartitionTopology.java        | 20 +++++
 .../dht/GridDhtPartitionTopology.java           |  7 ++
 .../dht/GridDhtPartitionTopologyImpl.java       | 20 +++++
 .../distributed/dht/GridDhtTxFinishFuture.java  |  5 +-
 .../distributed/dht/GridDhtTxPrepareFuture.java | 33 ++++++--
 .../cache/distributed/dht/GridDhtTxRemote.java  |  2 +-
 .../cache/transactions/IgniteTxAdapter.java     |  1 +
 .../cache/transactions/IgniteTxEntry.java       | 19 ++++-
 .../cache/transactions/IgniteTxHandler.java     | 80 ++++++++++++++++----
 .../cache/transactions/IgniteTxManager.java     | 23 ++++++
 .../GridTransactionalCacheQueueImpl.java        | 14 ++--
 .../jdk8/backport/ConcurrentLinkedHashMap.java  |  7 ++
 .../GridConcurrentLinkedHashMapSelfTest.java    | 18 +++++
 15 files changed, 287 insertions(+), 44 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/ignite/blob/9b0bd71b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridCacheOptimisticCheckPreparedTxFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridCacheOptimisticCheckPreparedTxFuture.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridCacheOptimisticCheckPreparedTxFuture.java
index 75b2683..429023b 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridCacheOptimisticCheckPreparedTxFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridCacheOptimisticCheckPreparedTxFuture.java
@@ -22,6 +22,7 @@ import org.apache.ignite.cluster.*;
 import org.apache.ignite.internal.*;
 import org.apache.ignite.internal.cluster.*;
 import org.apache.ignite.internal.processors.cache.*;
+import org.apache.ignite.internal.processors.cache.distributed.dht.*;
 import org.apache.ignite.internal.processors.cache.transactions.*;
 import org.apache.ignite.internal.processors.cache.version.*;
 import org.apache.ignite.internal.util.*;
@@ -155,7 +156,7 @@ public class GridCacheOptimisticCheckPreparedTxFuture<K, V> extends
GridCompound
 
                     GridCacheOptimisticCheckPreparedTxRequest<K, V>
                         req = new GridCacheOptimisticCheckPreparedTxRequest<>(tx,
-                        nodeTransactions(id), futureId(), fut.futureId());
+                        nodeTransactions(id), futureId(), fut.futureId(), false);
 
                     try {
                         cctx.io().send(id, req, tx.ioPolicy());
@@ -176,7 +177,7 @@ public class GridCacheOptimisticCheckPreparedTxFuture<K, V> extends
GridCompound
                 add(fut);
 
                 GridCacheOptimisticCheckPreparedTxRequest<K, V> req = new GridCacheOptimisticCheckPreparedTxRequest<>(
-                    tx, nodeTransactions(nodeId), futureId(), fut.futureId());
+                    tx, nodeTransactions(nodeId), futureId(), fut.futureId(), false);
 
                 try {
                     cctx.io().send(nodeId, req, tx.ioPolicy());
@@ -192,6 +193,32 @@ public class GridCacheOptimisticCheckPreparedTxFuture<K, V> extends
GridCompound
             }
         }
 
+        // Specifically check originating near node.
+        if (tx instanceof GridDhtTxRemote) {
+            UUID nearNodeId = ((GridDhtTxRemote)tx).nearNodeId();
+
+            if (cctx.localNodeId().equals(nearNodeId))
+                add(cctx.tm().nearTxCommitted(tx.nearXidVersion()));
+            else {
+                MiniFuture fut = new MiniFuture(nearNodeId);
+
+                add(fut);
+
+                GridCacheOptimisticCheckPreparedTxRequest<K, V> req = new GridCacheOptimisticCheckPreparedTxRequest<>(
+                    tx, 1, futureId(), fut.futureId(), true);
+
+                try {
+                    cctx.io().send(nearNodeId, req, tx.ioPolicy());
+                }
+                catch (ClusterTopologyCheckedException ignored) {
+                    fut.onNodeLeft();
+                }
+                catch (IgniteCheckedException e) {
+                    fut.onError(e);
+                }
+            }
+        }
+
         markInitialized();
     }
 
@@ -258,14 +285,11 @@ public class GridCacheOptimisticCheckPreparedTxFuture<K, V> extends
GridCompound
             if (isMini(fut)) {
                 MiniFuture f = (MiniFuture)fut;
 
-                if (f.nodeId().equals(nodeId)) {
+                if (f.nodeId().equals(nodeId))
                     f.onNodeLeft();
-
-                    return true;
-                }
             }
 
-        return false;
+        return true;
     }
 
     /** {@inheritDoc} */

http://git-wip-us.apache.org/repos/asf/ignite/blob/9b0bd71b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridCacheOptimisticCheckPreparedTxRequest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridCacheOptimisticCheckPreparedTxRequest.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridCacheOptimisticCheckPreparedTxRequest.java
index 636c5b5..4b61b7e 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridCacheOptimisticCheckPreparedTxRequest.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridCacheOptimisticCheckPreparedTxRequest.java
@@ -49,6 +49,9 @@ public class GridCacheOptimisticCheckPreparedTxRequest<K, V> extends
GridDistrib
     /** System transaction flag. */
     private boolean sys;
 
+    /** Near check falg. */
+    private boolean nearCheck;
+
     /**
      * Empty constructor required by {@link Externalizable}
      */
@@ -62,8 +65,13 @@ public class GridCacheOptimisticCheckPreparedTxRequest<K, V> extends
GridDistrib
      * @param futId Future ID.
      * @param miniId Mini future ID.
      */
-    public GridCacheOptimisticCheckPreparedTxRequest(IgniteInternalTx<K, V> tx, int
txNum, IgniteUuid futId,
-        IgniteUuid miniId) {
+    public GridCacheOptimisticCheckPreparedTxRequest(
+        IgniteInternalTx<K, V> tx,
+        int txNum,
+        IgniteUuid futId,
+        IgniteUuid miniId,
+        boolean nearCheck
+    ) {
         super(tx.xidVersion(), 0);
 
         nearXidVer = tx.nearXidVersion();
@@ -72,6 +80,7 @@ public class GridCacheOptimisticCheckPreparedTxRequest<K, V> extends
GridDistrib
         this.futId = futId;
         this.miniId = miniId;
         this.txNum = txNum;
+        this.nearCheck = nearCheck;
     }
 
     /**
@@ -109,6 +118,13 @@ public class GridCacheOptimisticCheckPreparedTxRequest<K, V> extends
GridDistrib
         return sys;
     }
 
+    /**
+     * @return Near check flag.
+     */
+    public boolean nearCheck() {
+        return nearCheck;
+    }
+
     /** {@inheritDoc} */
     @Override public boolean writeTo(ByteBuffer buf, MessageWriter writer) {
         writer.setBuffer(buf);
@@ -137,18 +153,24 @@ public class GridCacheOptimisticCheckPreparedTxRequest<K, V> extends
GridDistrib
                 writer.incrementState();
 
             case 8:
-                if (!writer.writeMessage("nearXidVer", nearXidVer))
+                if (!writer.writeBoolean("nearCheck", nearCheck))
                     return false;
 
                 writer.incrementState();
 
             case 9:
-                if (!writer.writeBoolean("sys", sys))
+                if (!writer.writeMessage("nearXidVer", nearXidVer))
                     return false;
 
                 writer.incrementState();
 
             case 10:
+                if (!writer.writeBoolean("sys", sys))
+                    return false;
+
+                writer.incrementState();
+
+            case 11:
                 if (!writer.writeInt("txNum", txNum))
                     return false;
 
@@ -187,7 +209,7 @@ public class GridCacheOptimisticCheckPreparedTxRequest<K, V> extends
GridDistrib
                 reader.incrementState();
 
             case 8:
-                nearXidVer = reader.readMessage("nearXidVer");
+                nearCheck = reader.readBoolean("nearCheck");
 
                 if (!reader.isLastRead())
                     return false;
@@ -195,7 +217,7 @@ public class GridCacheOptimisticCheckPreparedTxRequest<K, V> extends
GridDistrib
                 reader.incrementState();
 
             case 9:
-                sys = reader.readBoolean("sys");
+                nearXidVer = reader.readMessage("nearXidVer");
 
                 if (!reader.isLastRead())
                     return false;
@@ -203,6 +225,14 @@ public class GridCacheOptimisticCheckPreparedTxRequest<K, V> extends
GridDistrib
                 reader.incrementState();
 
             case 10:
+                sys = reader.readBoolean("sys");
+
+                if (!reader.isLastRead())
+                    return false;
+
+                reader.incrementState();
+
+            case 11:
                 txNum = reader.readInt("txNum");
 
                 if (!reader.isLastRead())
@@ -222,7 +252,7 @@ public class GridCacheOptimisticCheckPreparedTxRequest<K, V> extends
GridDistrib
 
     /** {@inheritDoc} */
     @Override public byte fieldsCount() {
-        return 11;
+        return 12;
     }
 
     /** {@inheritDoc} */

http://git-wip-us.apache.org/repos/asf/ignite/blob/9b0bd71b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridClientPartitionTopology.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridClientPartitionTopology.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridClientPartitionTopology.java
index 239efc3..98fbd47 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridClientPartitionTopology.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridClientPartitionTopology.java
@@ -304,6 +304,26 @@ public class GridClientPartitionTopology<K, V> implements GridDhtPartitionTopolo
     }
 
     /** {@inheritDoc} */
+    @Override public GridDhtPartitionState partitionState(UUID nodeId, int part) {
+        lock.readLock().lock();
+
+        try {
+            GridDhtPartitionMap partMap = node2part.get(nodeId);
+
+            if (partMap != null) {
+                GridDhtPartitionState state = partMap.get(part);
+
+                return state == null ? EVICTED : state;
+            }
+
+            return EVICTED;
+        }
+        finally {
+            lock.readLock().unlock();
+        }
+    }
+
+    /** {@inheritDoc} */
     @Override public Collection<ClusterNode> nodes(int p, long topVer) {
         lock.readLock().lock();
 

http://git-wip-us.apache.org/repos/asf/ignite/blob/9b0bd71b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtPartitionTopology.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtPartitionTopology.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtPartitionTopology.java
index d9a20ae..314d70e 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtPartitionTopology.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtPartitionTopology.java
@@ -118,6 +118,13 @@ public interface GridDhtPartitionTopology<K, V> {
     public GridDhtPartitionMap localPartitionMap();
 
     /**
+     * @param nodeId Node ID.
+     * @param part Partition.
+     * @return Partition state.
+     */
+    public GridDhtPartitionState partitionState(UUID nodeId, int part);
+
+    /**
      * @return Current update sequence.
      */
     public long updateSequence();

http://git-wip-us.apache.org/repos/asf/ignite/blob/9b0bd71b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtPartitionTopologyImpl.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtPartitionTopologyImpl.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtPartitionTopologyImpl.java
index acf00eb..40fde60 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtPartitionTopologyImpl.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtPartitionTopologyImpl.java
@@ -572,6 +572,26 @@ class GridDhtPartitionTopologyImpl<K, V> implements GridDhtPartitionTopology<K,
     }
 
     /** {@inheritDoc} */
+    @Override public GridDhtPartitionState partitionState(UUID nodeId, int part) {
+        lock.readLock().lock();
+
+        try {
+            GridDhtPartitionMap partMap = node2part.get(nodeId);
+
+            if (partMap != null) {
+                GridDhtPartitionState state = partMap.get(part);
+
+                return state == null ? EVICTED : state;
+            }
+
+            return EVICTED;
+        }
+        finally {
+            lock.readLock().unlock();
+        }
+    }
+
+    /** {@inheritDoc} */
     @Override public Collection<ClusterNode> nodes(int p, long topVer) {
         Collection<ClusterNode> affNodes = cctx.affinity().nodes(p, topVer);
 

http://git-wip-us.apache.org/repos/asf/ignite/blob/9b0bd71b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxFinishFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxFinishFuture.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxFinishFuture.java
index bb80480..dc4d15c 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxFinishFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxFinishFuture.java
@@ -325,6 +325,8 @@ public final class GridDhtTxFinishFuture<K, V> extends GridCompoundIdentityFutur
                 tx.subjectId(),
                 tx.taskNameHash());
 
+            req.writeVersion(tx.writeVersion());
+
             try {
                 cctx.io().send(n, req, tx.ioPolicy());
 
@@ -371,8 +373,7 @@ public final class GridDhtTxFinishFuture<K, V> extends GridCompoundIdentityFutur
                     tx.subjectId(),
                     tx.taskNameHash());
 
-                if (tx.onePhaseCommit())
-                    req.writeVersion(tx.writeVersion());
+                req.writeVersion(tx.writeVersion());
 
                 try {
                     cctx.io().send(nearMapping.node(), req, tx.ioPolicy());

http://git-wip-us.apache.org/repos/asf/ignite/blob/9b0bd71b/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 2b1d2f0..33ee64c 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
@@ -294,7 +294,7 @@ public final class GridDhtTxPrepareFuture<K, V> extends GridCompoundIdentityFutu
 
                 boolean hasFilters = !F.isEmptyOrNulls(txEntry.filters()) && !F.isAlwaysTrue(txEntry.filters());
 
-                if (hasFilters || retVal || txEntry.op() == GridCacheOperation.DELETE) {
+                if (hasFilters || retVal || txEntry.op() == DELETE || txEntry.op() == TRANSFORM)
{
                     cached.unswap(true, retVal);
 
                     V val = cached.innerGet(
@@ -311,14 +311,13 @@ public final class GridDhtTxPrepareFuture<K, V> extends GridCompoundIdentityFutu
                         null,
                         null);
 
-                    if (retVal) {
+                    if (retVal || txEntry.op() == TRANSFORM) {
                         if (!F.isEmpty(txEntry.entryProcessors())) {
                             K key = txEntry.key();
 
                             Object procRes = null;
                             Exception err = null;
 
-
                             for (T2<EntryProcessor<K, V, ?>, Object[]> t : txEntry.entryProcessors())
{
                                 try {
                                     CacheInvokeEntry<K, V> invokeEntry = new CacheInvokeEntry<>(txEntry.context(),
key, val);
@@ -336,6 +335,8 @@ public final class GridDhtTxPrepareFuture<K, V> extends GridCompoundIdentityFutu
                                 }
                             }
 
+                            txEntry.entryProcessorCalculatedValue(val);
+
                             if (err != null || procRes != null)
                                 ret.addEntryProcessResult(key,
                                     err == null ? new CacheInvokeResult<>(procRes)
: new CacheInvokeResult<>(err));
@@ -360,7 +361,7 @@ public final class GridDhtTxPrepareFuture<K, V> extends GridCompoundIdentityFutu
                         ret.success(false);
                     }
                     else
-                        ret.success(txEntry.op() != GridCacheOperation.DELETE || cached.hasValue());
+                        ret.success(txEntry.op() != DELETE || cached.hasValue());
                 }
             }
             catch (IgniteCheckedException e) {
@@ -1002,7 +1003,8 @@ public final class GridDhtTxPrepareFuture<K, V> extends GridCompoundIdentityFutu
     private boolean map(
         IgniteTxEntry<K, V> entry,
         Map<UUID, GridDistributedTxMapping<K, V>> futDhtMap,
-        Map<UUID, GridDistributedTxMapping<K, V>> futNearMap) {
+        Map<UUID, GridDistributedTxMapping<K, V>> futNearMap
+    ) {
         if (entry.cached().isLocal())
             return false;
 
@@ -1069,14 +1071,31 @@ public final class GridDhtTxPrepareFuture<K, V> extends GridCompoundIdentityFutu
      * @param locMap Exclude map.
      * @return {@code True} if mapped.
      */
-    private boolean map(IgniteTxEntry<K, V> entry, Iterable<ClusterNode> nodes,
-        Map<UUID, GridDistributedTxMapping<K, V>> globalMap, Map<UUID, GridDistributedTxMapping<K,
V>> locMap) {
+    private boolean map(
+        IgniteTxEntry<K, V> entry,
+        Iterable<ClusterNode> nodes,
+        Map<UUID, GridDistributedTxMapping<K, V>> globalMap,
+        Map<UUID, GridDistributedTxMapping<K, V>> locMap
+    ) {
         boolean ret = false;
 
         if (nodes != null) {
             for (ClusterNode n : nodes) {
                 GridDistributedTxMapping<K, V> global = globalMap.get(n.id());
 
+                if (!F.isEmpty(entry.entryProcessors())) {
+                    GridDhtPartitionState state = entry.context().topology().partitionState(n.id(),
+                        entry.cached().partition());
+
+                    if (state != GridDhtPartitionState.OWNING && state != GridDhtPartitionState.EVICTED)
{
+                        V procVal = entry.entryProcessorCalculatedValue();
+
+                        entry.op(procVal == null ? DELETE : UPDATE);
+                        entry.value(procVal, true, false);
+                        entry.entryProcessors(null);
+                    }
+                }
+
                 if (global == null)
                     globalMap.put(n.id(), global = new GridDistributedTxMapping<>(n));
 

http://git-wip-us.apache.org/repos/asf/ignite/blob/9b0bd71b/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 3366033..ed37ae4 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
@@ -203,7 +203,7 @@ public class GridDhtTxRemote<K, V> extends GridDistributedTxRemoteAdapter<K,
V>
     /**
      * @return Near node ID.
      */
-    UUID nearNodeId() {
+    public UUID nearNodeId() {
         return nearNodeId;
     }
 

http://git-wip-us.apache.org/repos/asf/ignite/blob/9b0bd71b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxAdapter.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxAdapter.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxAdapter.java
index 8410ef7..e11a5b2 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxAdapter.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxAdapter.java
@@ -160,6 +160,7 @@ public abstract class IgniteTxAdapter<K, V> extends GridMetadataAwareAdapter
     private AtomicBoolean preparing = new AtomicBoolean();
 
     /** */
+    @GridToStringInclude
     private Set<Integer> invalidParts = new GridLeanSet<>();
 
     /**

http://git-wip-us.apache.org/repos/asf/ignite/blob/9b0bd71b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxEntry.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxEntry.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxEntry.java
index 6bb768c..286f896 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxEntry.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxEntry.java
@@ -83,6 +83,9 @@ public class IgniteTxEntry<K, V> implements GridPeerDeployAware, Externalizable,
     @GridToStringInclude
     private Collection<T2<EntryProcessor<K, V, ?>, Object[]>> entryProcessorsCol;
 
+    /** Transient field for calculated entry processor value. */
+    private V entryProcessorCalcVal;
+
     /** Transform closure bytes. */
     @GridToStringExclude
     private byte[] transformClosBytes;
@@ -420,6 +423,20 @@ public class IgniteTxEntry<K, V> implements GridPeerDeployAware,
Externalizable,
     }
 
     /**
+     * @return Entry processor calculated value.
+     */
+    public V entryProcessorCalculatedValue() {
+        return entryProcessorCalcVal;
+    }
+
+    /**
+     * @param entryProcessorCalcVal Entry processor calculated value.
+     */
+    public void entryProcessorCalculatedValue(V entryProcessorCalcVal) {
+        this.entryProcessorCalcVal = entryProcessorCalcVal;
+    }
+
+    /**
      * @return Underlying cache entry.
      */
     public GridCacheEntryEx<K, V> cached() {
@@ -902,7 +919,7 @@ public class IgniteTxEntry<K, V> implements GridPeerDeployAware,
Externalizable,
     @Override public String toString() {
         return GridToStringBuilder.toString(IgniteTxEntry.class, this,
             "keyBytesSize", keyBytes == null ? "null" : Integer.toString(keyBytes.length),
-            "xidVer", tx == null ? "null" : tx.xidVersion());
+            "xidVer", tx == null ? "null" : tx.xidVersion(), "hc", System.identityHashCode(this));
     }
 
     /**

http://git-wip-us.apache.org/repos/asf/ignite/blob/9b0bd71b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxHandler.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxHandler.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxHandler.java
index 90af0b2..ca58d6b 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxHandler.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxHandler.java
@@ -798,15 +798,14 @@ public class IgniteTxHandler<K, V> {
 
         try {
             if (req.commit() || req.isSystemInvalidate()) {
-                if (tx.commitVersion(req.commitVersion())) {
-                    tx.invalidate(req.isInvalidate());
-                    tx.systemInvalidate(req.isSystemInvalidate());
+                tx.commitVersion(req.commitVersion());
+                tx.invalidate(req.isInvalidate());
+                tx.systemInvalidate(req.isSystemInvalidate());
 
-                    // Complete remote candidates.
-                    tx.doneRemote(req.version());
+                // Complete remote candidates.
+                tx.doneRemote(req.version());
 
-                    tx.commit();
-                }
+                tx.commit();
             }
             else {
                 tx.doneRemote(req.version());
@@ -986,8 +985,14 @@ public class IgniteTxHandler<K, V> {
             // in prepare phase will get properly ordered as well.
             tx.prepare();
 
-            if (req.last())
+            if (req.last()) {
+                assert !F.isEmpty(req.transactionNodes()) :
+                    "Received last prepare request with empty transaction nodes: " + req;
+
+                tx.transactionNodes(req.transactionNodes());
+
                 tx.state(PREPARED);
+            }
 
             res.invalidPartitions(tx.invalidPartitions());
 
@@ -1085,20 +1090,69 @@ public class IgniteTxHandler<K, V> {
      * @param nodeId Node ID.
      * @param req Request.
      */
-    protected void processCheckPreparedTxRequest(UUID nodeId, GridCacheOptimisticCheckPreparedTxRequest<K,
V> req) {
+    protected void processCheckPreparedTxRequest(
+        final UUID nodeId,
+        final GridCacheOptimisticCheckPreparedTxRequest<K, V> req
+    ) {
         if (log.isDebugEnabled())
             log.debug("Processing check prepared transaction requests [nodeId=" + nodeId
+ ", req=" + req + ']');
 
-        boolean prepared = ctx.tm().txsPreparedOrCommitted(req.nearXidVersion(), req.transactions());
+        if (req.nearCheck()) {
+            IgniteInternalFuture<Boolean> fut = ctx.tm().nearTxCommitted(req.nearXidVersion());
+
+            fut.listenAsync(new CI1<IgniteInternalFuture<Boolean>>() {
+                @Override public void apply(IgniteInternalFuture<Boolean> f) {
+                    try {
+                        boolean prepared = f.get();
 
-        GridCacheOptimisticCheckPreparedTxResponse<K, V> res =
-            new GridCacheOptimisticCheckPreparedTxResponse<>(req.version(), req.futureId(),
req.miniId(), prepared);
+                        sendCheckPrepareTxResponse(nodeId,
+                            new GridCacheOptimisticCheckPreparedTxResponse<K, V>(
+                                req.version(),
+                                req.futureId(),
+                                req.miniId(),
+                                prepared),
+                            req.system());
 
+                    }
+                    catch (IgniteCheckedException e) {
+                        U.error(log, "Failed to wait for transaction check prepared future
" +
+                            "(will send rolled back response): " + req.nearXidVersion(),
e);
+
+                        sendCheckPrepareTxResponse(nodeId,
+                            new GridCacheOptimisticCheckPreparedTxResponse<K, V>(
+                                req.version(),
+                                req.futureId(),
+                                req.miniId(),
+                                false),
+                            req.system());
+                    }
+                }
+            });
+        }
+        else {
+            boolean prepared = ctx.tm().txsPreparedOrCommitted(req.nearXidVersion(), req.transactions());
+
+            sendCheckPrepareTxResponse(nodeId,
+                new GridCacheOptimisticCheckPreparedTxResponse<K, V>(req.version(),
req.futureId(), req.miniId(), prepared),
+                req.system());
+        }
+    }
+
+    /**
+     * @param nodeId Node ID.
+     * @param res Response to send.
+     * @param sys System pool flag.
+     */
+    private void sendCheckPrepareTxResponse(
+        UUID nodeId,
+        GridCacheOptimisticCheckPreparedTxResponse<K, V> res,
+        boolean sys
+    ) {
         try {
             if (log.isDebugEnabled())
                 log.debug("Sending check prepared transaction response [nodeId=" + nodeId
+ ", res=" + res + ']');
 
-            ctx.io().send(nodeId, res, req.system() ? UTILITY_CACHE_POOL : SYSTEM_POOL);
+            ctx.io().send(nodeId, res, sys ? UTILITY_CACHE_POOL : SYSTEM_POOL);
         }
         catch (ClusterTopologyCheckedException ignored) {
             if (log.isDebugEnabled())

http://git-wip-us.apache.org/repos/asf/ignite/blob/9b0bd71b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxManager.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxManager.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxManager.java
index e347cce..974144a 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxManager.java
@@ -1653,6 +1653,29 @@ public class IgniteTxManager<K, V> extends GridCacheSharedManagerAdapter<K,
V> {
     }
 
     /**
+     * @param nearVer Near version to check.
+     * @return Future.
+     */
+    public IgniteInternalFuture<Boolean> nearTxCommitted(GridCacheVersion nearVer)
{
+        for (final IgniteInternalTx<K, V> tx : txs()) {
+            if (tx.near() && tx.xidVersion().equals(nearVer)) {
+                return tx.done() ?
+                    new GridFinishedFutureEx<>(tx.state() == COMMITTED) :
+                    tx.finishFuture().chain(new C1<IgniteInternalFuture<IgniteInternalTx>,
Boolean>() {
+                        @Override public Boolean apply(IgniteInternalFuture<IgniteInternalTx>
f) {
+                            return tx.state() == COMMITTED;
+                        }
+                    });
+            }
+        }
+
+        // Transaction was not found. Check committed versions buffer.
+        Boolean res = completedVers.get(nearVer);
+
+        return new GridFinishedFutureEx<>(res != null && res);
+    }
+
+    /**
      * Gets local transaction for pessimistic tx recovery.
      *
      * @param nearXidVer Near tx ID.

http://git-wip-us.apache.org/repos/asf/ignite/blob/9b0bd71b/modules/core/src/main/java/org/apache/ignite/internal/processors/datastructures/GridTransactionalCacheQueueImpl.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/datastructures/GridTransactionalCacheQueueImpl.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/datastructures/GridTransactionalCacheQueueImpl.java
index 801e27f..37817c4 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/datastructures/GridTransactionalCacheQueueImpl.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/datastructures/GridTransactionalCacheQueueImpl.java
@@ -21,7 +21,9 @@ import org.apache.ignite.*;
 import org.apache.ignite.internal.cluster.*;
 import org.apache.ignite.internal.processors.cache.*;
 import org.apache.ignite.internal.processors.cache.transactions.*;
+import org.apache.ignite.internal.transactions.*;
 import org.apache.ignite.internal.util.typedef.internal.*;
+import org.apache.ignite.transactions.*;
 import org.jetbrains.annotations.*;
 
 import java.util.*;
@@ -30,7 +32,7 @@ import static org.apache.ignite.transactions.TransactionConcurrency.*;
 import static org.apache.ignite.transactions.TransactionIsolation.*;
 
 /**
- * {@link org.apache.ignite.IgniteQueue} implementation using transactional cache.
+ * {@link IgniteQueue} implementation using transactional cache.
  */
 public class GridTransactionalCacheQueueImpl<T> extends GridCacheQueueAdapter<T>
{
     /**
@@ -72,7 +74,7 @@ public class GridTransactionalCacheQueueImpl<T> extends GridCacheQueueAdapter<T>
                         break;
                     }
                 }
-                catch (ClusterTopologyCheckedException e) {
+                catch (ClusterTopologyCheckedException | TransactionRollbackException | IgniteTxRollbackCheckedException
e) {
                     if (e instanceof ClusterGroupEmptyCheckedException)
                         throw e;
 
@@ -119,14 +121,14 @@ public class GridTransactionalCacheQueueImpl<T> extends GridCacheQueueAdapter<T>
 
                     break;
                 }
-                catch (ClusterTopologyCheckedException e) {
+                catch (ClusterTopologyCheckedException | TransactionRollbackException | IgniteTxRollbackCheckedException
e) {
                     if (e instanceof ClusterGroupEmptyCheckedException)
                         throw e;
 
                     if (cnt++ == MAX_UPDATE_RETRIES)
                         throw e;
                     else {
-                        U.warn(log, "Failed to add item, will retry [err=" + e + ']');
+                        U.warn(log, "Failed to poll item, will retry [err=" + e + ']');
 
                         U.sleep(RETRY_DELAY);
                     }
@@ -176,7 +178,7 @@ public class GridTransactionalCacheQueueImpl<T> extends GridCacheQueueAdapter<T>
 
                     break;
                 }
-                catch (ClusterTopologyCheckedException e) {
+                catch (ClusterTopologyCheckedException | TransactionRollbackException | IgniteTxRollbackCheckedException
e) {
                     if (e instanceof ClusterGroupEmptyCheckedException)
                         throw e;
 
@@ -219,7 +221,7 @@ public class GridTransactionalCacheQueueImpl<T> extends GridCacheQueueAdapter<T>
 
                     break;
                 }
-                catch (ClusterTopologyCheckedException e) {
+                catch (ClusterTopologyCheckedException | TransactionRollbackException | IgniteTxRollbackCheckedException
e) {
                     if (e instanceof ClusterGroupEmptyCheckedException)
                         throw e;
 

http://git-wip-us.apache.org/repos/asf/ignite/blob/9b0bd71b/modules/core/src/main/java/org/jdk8/backport/ConcurrentLinkedHashMap.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/jdk8/backport/ConcurrentLinkedHashMap.java b/modules/core/src/main/java/org/jdk8/backport/ConcurrentLinkedHashMap.java
index 8992e77..e3f16a2 100644
--- a/modules/core/src/main/java/org/jdk8/backport/ConcurrentLinkedHashMap.java
+++ b/modules/core/src/main/java/org/jdk8/backport/ConcurrentLinkedHashMap.java
@@ -724,6 +724,13 @@ public class ConcurrentLinkedHashMap<K, V> extends AbstractMap<K,
V> implements
                     if (!onlyIfAbsent) {
                         e.val = val;
 
+                        HashEntry<K, V> qEntry = (HashEntry<K, V>)e.node.item();
+
+                        if (qEntry != null && qEntry != e)
+                            qEntry.val = val;
+
+                        ((HashEntry<K, V>)e.node.item).val = val;
+
                         modified = true;
                     }
                 }

http://git-wip-us.apache.org/repos/asf/ignite/blob/9b0bd71b/modules/core/src/test/java/org/apache/ignite/lang/utils/GridConcurrentLinkedHashMapSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/lang/utils/GridConcurrentLinkedHashMapSelfTest.java
b/modules/core/src/test/java/org/apache/ignite/lang/utils/GridConcurrentLinkedHashMapSelfTest.java
index 75ac896..49f24b0 100644
--- a/modules/core/src/test/java/org/apache/ignite/lang/utils/GridConcurrentLinkedHashMapSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/lang/utils/GridConcurrentLinkedHashMapSelfTest.java
@@ -176,6 +176,24 @@ public class GridConcurrentLinkedHashMapSelfTest extends GridCommonAbstractTest
     }
 
     /**
+     * @throws Exception If failed.
+     */
+    public void testRehash() throws Exception {
+        Map<Integer, Date> map = new ConcurrentLinkedHashMap<>(10);
+
+        for (int i = 0; i < 100; i++)
+            // Will initiate rehash in the middle.
+            map.put(i, new Date(0));
+
+        for (int i = 0; i < 100; i++)
+            // Will initiate rehash in the middle.
+            map.put(i, new Date(1));
+
+        for (Date date : map.values())
+            assertEquals(1L, date.getTime());
+    }
+
+    /**
      *
      */
     public void testDescendingMethods() {


Mime
View raw message