ignite-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From agoncha...@apache.org
Subject [13/49] incubator-ignite git commit: GG-9655 - Fixing tests after merge.
Date Sun, 15 Feb 2015 08:18:33 GMT
GG-9655 - Fixing tests after merge.


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

Branch: refs/heads/sprint-1
Commit: 9a995a3aba1cc3cef0dd8027407d234dcf87c885
Parents: 61c102c
Author: Alexey Goncharuk <agoncharuk@gridgain.com>
Authored: Fri Jan 30 16:41:37 2015 -0800
Committer: Alexey Goncharuk <agoncharuk@gridgain.com>
Committed: Fri Jan 30 16:41:37 2015 -0800

----------------------------------------------------------------------
 .../distributed/dht/GridDhtTxLocalAdapter.java  |  1 +
 .../distributed/dht/GridDhtTxPrepareFuture.java | 74 ++++++++----------
 .../distributed/near/GridNearLockRequest.java   |  5 --
 .../cache/distributed/near/GridNearTxLocal.java | 11 ++-
 .../near/GridNearTxPrepareFuture.java           | 81 ++++++++------------
 .../transactions/IgniteTxLocalAdapter.java      |  8 +-
 .../cache/GridCacheAbstractFullApiSelfTest.java |  5 +-
 7 files changed, 76 insertions(+), 109 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/9a995a3a/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 33509ab..ae8e3c2 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
@@ -451,6 +451,7 @@ public abstract class GridDhtTxLocalAdapter<K, V> extends IgniteTxLocalAdapter<K
                 entry.valueBytes(e.valueBytes());
                 entry.ttl(e.ttl());
                 entry.filters(e.filters());
+                entry.expiry(e.expiry());
                 entry.drExpireTime(e.drExpireTime());
                 entry.drVersion(e.drVersion());
             }

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/9a995a3a/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 407b8fe..256c9f9 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
@@ -116,6 +116,12 @@ public final class GridDhtTxPrepareFuture<K, V> extends GridCompoundIdentityFutu
     /** Keys that did not pass the filter. */
     private Collection<IgniteTxKey<K>> filterFailedKeys;
 
+    /** Keys that should be locked. */
+    private GridConcurrentHashSet<IgniteTxKey<K>> lockKeys = new GridConcurrentHashSet<>();
+
+    /** Locks ready flag. */
+    private volatile boolean locksReady;
+
     /**
      * Empty constructor required for {@link Externalizable}.
      */
@@ -209,9 +215,9 @@ public final class GridDhtTxPrepareFuture<K, V> extends GridCompoundIdentityFutu
         if (log.isDebugEnabled())
             log.debug("Transaction future received owner changed callback: " + entry);
 
-        boolean ret = tx.hasWriteKey(entry.txKey());
+        boolean rmv = lockKeys.remove(entry.txKey());
 
-        return ret && mapIfLocked();
+        return rmv && mapIfLocked();
     }
 
     /** {@inheritDoc} */
@@ -235,44 +241,7 @@ public final class GridDhtTxPrepareFuture<K, V> extends GridCompoundIdentityFutu
      * @return {@code True} if all locks are owned.
      */
     private boolean checkLocks() {
-        for (IgniteTxEntry<K, V> txEntry : tx.optimisticLockEntries()) {
-            while (true) {
-                GridCacheEntryEx<K, V> cached = txEntry.cached();
-
-                try {
-                    if (txEntry.explicitVersion() == null) {
-                        // Don't compare entry against itself.
-                        if (!cached.lockedLocally(tx.xidVersion())) {
-                            if (log.isDebugEnabled())
-                                log.debug("Transaction entry is not locked by transaction
(will wait) [entry=" +
-                                    cached + ", tx=" + tx + ']');
-
-                            return false;
-                        }
-                    }
-                    else {
-                        if (!cached.lockedBy(txEntry.explicitVersion())) {
-                            if (log.isDebugEnabled())
-                                log.debug("Transaction entry is not locked by explicit version
(will wait) [entry=" +
-                                    cached + ", tx=" + tx + ']');
-
-                            return false;
-                        }
-                    }
-
-                    break; // While.
-                }
-                // Possible if entry cached within transaction is obsolete.
-                catch (GridCacheEntryRemovedException ignored) {
-                    if (log.isDebugEnabled())
-                        log.debug("Got removed entry in future onAllReplies method (will
retry): " + txEntry);
-
-                    txEntry.cached(txEntry.context().cache().entryEx(txEntry.key()), txEntry.keyBytes());
-                }
-            }
-        }
-
-        return true;
+        return locksReady && lockKeys.isEmpty();
     }
 
     /** {@inheritDoc} */
@@ -460,13 +429,26 @@ public final class GridDhtTxPrepareFuture<K, V> extends GridCompoundIdentityFutu
             Collections.singletonList(tx.groupLockEntry()) : writes;
 
         for (IgniteTxEntry<K, V> txEntry : checkEntries) {
-            if (txEntry.cached().isLocal())
+            GridCacheContext<K, V> cacheCtx = txEntry.context();
+
+            if (cacheCtx.isLocal())
                 continue;
 
-            while (true) {
-                GridDistributedCacheEntry<K, V> entry = (GridDistributedCacheEntry<K,
V>)txEntry.cached();
+            GridDistributedCacheEntry<K, V> entry = (GridDistributedCacheEntry<K,
V>)txEntry.cached();
+
+            if (entry == null) {
+                entry = (GridDistributedCacheEntry<K, V>)cacheCtx.cache().entryEx(txEntry.key());
+
+                txEntry.cached(entry, txEntry.keyBytes());
+            }
+
+            if (tx.optimistic() && txEntry.explicitVersion() == null)
+                lockKeys.add(txEntry.txKey());
 
+            while (true) {
                 try {
+                    assert txEntry.explicitVersion() == null || entry.lockedBy(txEntry.explicitVersion());
+
                     GridCacheMvccCandidate<K> c = entry.readyLock(tx.xidVersion());
 
                     if (log.isDebugEnabled())
@@ -479,10 +461,14 @@ public final class GridDhtTxPrepareFuture<K, V> extends GridCompoundIdentityFutu
                     if (log.isDebugEnabled())
                         log.debug("Got removed entry in future onAllReplies method (will
retry): " + txEntry);
 
-                    txEntry.cached(txEntry.context().cache().entryEx(txEntry.key()), txEntry.keyBytes());
+                    entry = (GridDistributedCacheEntry<K, V>)cacheCtx.cache().entryEx(txEntry.key());
+
+                    txEntry.cached(entry, txEntry.keyBytes());
                 }
             }
         }
+
+        locksReady = true;
     }
 
     /**

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/9a995a3a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearLockRequest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearLockRequest.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearLockRequest.java
index e707e85..732703e 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearLockRequest.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearLockRequest.java
@@ -301,11 +301,6 @@ public class GridNearLockRequest<K, V> extends GridDistributedLockRequest<K,
V>
         return dhtVers[idx];
     }
 
-    /** {@inheritDoc} */
-    @Override protected boolean transferExpiryPolicy() {
-        return true;
-    }
-
     /**
      * @return TTL for read operation.
      */

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/9a995a3a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxLocal.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxLocal.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxLocal.java
index f006508..b8bda46 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxLocal.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxLocal.java
@@ -82,7 +82,7 @@ public class GridNearTxLocal<K, V> extends GridDhtTxLocalAdapter<K,
V> {
     private boolean colocatedLocallyMapped;
 
     /** Info for entries accessed locally in optimistic transaction. */
-    private Map<IgniteTxKey, IgniteCacheExpiryPolicy> accessMap;
+    private Map<IgniteTxKey<K>, IgniteCacheExpiryPolicy> accessMap;
 
     /**
      * Empty constructor required for {@link Externalizable}.
@@ -560,8 +560,7 @@ public class GridNearTxLocal<K, V> extends GridDhtTxLocalAdapter<K,
V> {
             while (true) {
                 GridCacheContext<K, V> cacheCtx = txEntry.cached().context();
 
-                if (!cacheCtx.isNear())
-                    break;
+                assert cacheCtx.isNear();
 
                 GridDistributedCacheEntry<K, V> entry = (GridDistributedCacheEntry<K,
V>)txEntry.cached();
 
@@ -1156,7 +1155,7 @@ public class GridNearTxLocal<K, V> extends GridDhtTxLocalAdapter<K,
V> {
 
     /** {@inheritDoc} */
     @Override protected IgniteCacheExpiryPolicy accessPolicy(GridCacheContext ctx,
-        IgniteTxKey key,
+        IgniteTxKey<K> key,
         @Nullable ExpiryPolicy expiryPlc)
     {
         assert optimistic();
@@ -1187,7 +1186,7 @@ public class GridNearTxLocal<K, V> extends GridDhtTxLocalAdapter<K,
V> {
      */
     private IgniteCacheExpiryPolicy accessPolicy(GridCacheContext<K, V> cacheCtx, Collection<?
extends K> keys) {
         if (accessMap != null) {
-            for (Map.Entry<IgniteTxKey, IgniteCacheExpiryPolicy> e : accessMap.entrySet())
{
+            for (Map.Entry<IgniteTxKey<K>, IgniteCacheExpiryPolicy> e : accessMap.entrySet())
{
                 if (e.getKey().cacheId() == cacheCtx.cacheId() && keys.contains(e.getKey().key()))
                     return e.getValue();
             }
@@ -1203,7 +1202,7 @@ public class GridNearTxLocal<K, V> extends GridDhtTxLocalAdapter<K,
V> {
         if (accessMap != null) {
             assert optimistic();
 
-            for (Map.Entry<IgniteTxKey, IgniteCacheExpiryPolicy> e : accessMap.entrySet())
{
+            for (Map.Entry<IgniteTxKey<K>, IgniteCacheExpiryPolicy> e : accessMap.entrySet())
{
                 if (e.getValue().entries() != null) {
                     GridCacheContext cctx0 = cctx.cacheContext(e.getKey().cacheId());
 

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/9a995a3a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxPrepareFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxPrepareFuture.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxPrepareFuture.java
index e7b3601..4d9dadf 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxPrepareFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxPrepareFuture.java
@@ -18,6 +18,7 @@
 package org.apache.ignite.internal.processors.cache.distributed.near;
 
 import org.apache.ignite.*;
+import org.apache.ignite.client.util.*;
 import org.apache.ignite.cluster.*;
 import org.apache.ignite.internal.*;
 import org.apache.ignite.internal.processors.cache.*;
@@ -78,6 +79,9 @@ public final class GridNearTxPrepareFuture<K, V> extends GridCompoundIdentityFut
     /** Full information about transaction nodes mapping. */
     private GridDhtTxMapping<K, V> txMapping;
 
+    /** */
+    private Collection<IgniteTxKey<K>> lockKeys = new GridConcurrentHashSet<>();
+
     /**
      * Empty constructor required for {@link Externalizable}.
      */
@@ -128,6 +132,8 @@ public final class GridNearTxPrepareFuture<K, V> extends GridCompoundIdentityFut
             log.debug("Transaction future received owner changed callback: " + entry);
 
         if (entry.context().isNear() && owner != null && tx.hasWriteKey(entry.txKey()))
{
+            lockKeys.remove(entry.txKey());
+
             // This will check for locks.
             onDone();
 
@@ -213,45 +219,16 @@ public final class GridNearTxPrepareFuture<K, V> extends GridCompoundIdentityFut
      * @return {@code True} if all locks are owned.
      */
     private boolean checkLocks() {
-        Collection<IgniteTxEntry<K, V>> checkEntries = tx.groupLock() ?
-            Collections.singletonList(tx.groupLockEntry()) :
-            tx.writeEntries();
-
-        for (IgniteTxEntry<K, V> txEntry : checkEntries) {
-            // Wait for near locks only.
-            if (!txEntry.context().isNear())
-                continue;
-
-            while (true) {
-                GridCacheEntryEx<K, V> cached = txEntry.cached();
-
-                try {
-                    GridCacheVersion ver = txEntry.explicitVersion() != null ?
-                        txEntry.explicitVersion() : tx.xidVersion();
+        boolean locked = lockKeys.isEmpty();
 
-                    // If locks haven't been acquired yet, keep waiting.
-                    if (!cached.lockedBy(ver)) {
-                        if (log.isDebugEnabled())
-                            log.debug("Transaction entry is not locked by transaction (will
wait) [entry=" + cached +
-                                ", tx=" + tx + ']');
-
-                        return false;
-                    }
-
-                    break; // While.
-                }
-                // Possible if entry cached within transaction is obsolete.
-                catch (GridCacheEntryRemovedException ignored) {
-                    if (log.isDebugEnabled())
-                        log.debug("Got removed entry in future onAllReplies method (will
retry): " + txEntry);
-
-                    txEntry.cached(txEntry.context().cache().entryEx(txEntry.key()), txEntry.keyBytes());
-                }
-            }
+        if (locked) {
+            if (log.isDebugEnabled())
+                log.debug("All locks are acquired for near prepare future: " + this);
+        }
+        else {
+            if (log.isDebugEnabled())
+                log.debug("Still waiting for locks [fut=" + this + ", keys=" + lockKeys +
']');
         }
-
-        if (log.isDebugEnabled())
-            log.debug("All locks are acquired for near prepare future: " + this);
 
         return true;
     }
@@ -564,7 +541,7 @@ public final class GridNearTxPrepareFuture<K, V> extends GridCompoundIdentityFut
      *
      */
     private void preparePessimistic() {
-        Map<ClusterNode, GridDistributedTxMapping<K, V>> mappings = new HashMap<>();
+        Map<IgniteBiTuple<ClusterNode, Boolean>, GridDistributedTxMapping<K,
V>> mappings = new HashMap<>();
 
         long topVer = tx.topologyVersion();
 
@@ -577,12 +554,18 @@ public final class GridNearTxPrepareFuture<K, V> extends GridCompoundIdentityFut
 
             ClusterNode primary = F.first(nodes);
 
-            GridDistributedTxMapping<K, V> nodeMapping = mappings.get(primary);
+            boolean near = cacheCtx.isNear();
+
+            IgniteBiTuple<ClusterNode, Boolean> key = F.t(primary, near);
+
+            GridDistributedTxMapping<K, V> nodeMapping = mappings.get(key);
 
             if (nodeMapping == null) {
                 nodeMapping = new GridDistributedTxMapping<>(primary);
 
-                mappings.put(primary, nodeMapping);
+                nodeMapping.near(cacheCtx.isNear());
+
+                mappings.put(key, nodeMapping);
             }
 
             txEntry.nodeId(primary.id());
@@ -663,13 +646,6 @@ public final class GridNearTxPrepareFuture<K, V> extends GridCompoundIdentityFut
                                 tx.addDhtVersion(m.node().id(), dhtTx.xidVersion());
 
                                 m.dhtVersion(dhtTx.xidVersion());
-
-                                GridCacheVersion min = dhtTx.minVersion();
-
-                                IgniteTxManager<K, V> tm = cctx.tm();
-
-                                tx.readyNearLocks(m, Collections.<GridCacheVersion>emptyList(),
-                                    tm.committedVersions(min), tm.rolledbackVersions(min));
                             }
 
                             tx.implicitSingleResult(dhtTx.implicitSingleResult());
@@ -821,8 +797,9 @@ public final class GridNearTxPrepareFuture<K, V> extends GridCompoundIdentityFut
 
                             IgniteTxManager<K, V> tm = cctx.tm();
 
-                            tx.readyNearLocks(m, Collections.<GridCacheVersion>emptyList(),
-                                tm.committedVersions(min), tm.rolledbackVersions(min));
+                            if (m.near())
+                                tx.readyNearLocks(m, Collections.<GridCacheVersion>emptyList(),
+                                    tm.committedVersions(min), tm.rolledbackVersions(min));
                         }
 
                         // Continue prepare before completing the future.
@@ -902,6 +879,9 @@ public final class GridNearTxPrepareFuture<K, V> extends GridCompoundIdentityFut
         entry.nodeId(primary.id());
 
         if (cacheCtx.isNear()) {
+            if (entry.explicitVersion() == null)
+                lockKeys.add(entry.txKey());
+
             while (true) {
                 try {
                     GridNearCacheEntry<K, V> cached = (GridNearCacheEntry<K, V>)entry.cached();
@@ -1083,7 +1063,8 @@ public final class GridNearTxPrepareFuture<K, V> extends GridCompoundIdentityFut
 
                         m.dhtVersion(res.dhtVersion());
 
-                        tx.readyNearLocks(m, res.pending(), res.committedVersions(), res.rolledbackVersions());
+                        if (m.near())
+                            tx.readyNearLocks(m, res.pending(), res.committedVersions(),
res.rolledbackVersions());
                     }
 
                     // Proceed prepare before finishing mini future.

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/9a995a3a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxLocalAdapter.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxLocalAdapter.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxLocalAdapter.java
index f097833..e481f84 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxLocalAdapter.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxLocalAdapter.java
@@ -1360,9 +1360,11 @@ public abstract class IgniteTxLocalAdapter<K, V> extends IgniteTxAdapter<K,
V>
      * @param expiryPlc Expiry policy.
      * @return Expiry policy wrapper for entries accessed locally in optimistic transaction.
      */
-    protected IgniteCacheExpiryPolicy accessPolicy(GridCacheContext ctx,
-        IgniteTxKey key,
-        @Nullable ExpiryPolicy expiryPlc) {
+    protected IgniteCacheExpiryPolicy accessPolicy(
+        GridCacheContext ctx,
+        IgniteTxKey<K> key,
+        @Nullable ExpiryPolicy expiryPlc
+    ) {
         return null;
     }
 

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/9a995a3a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheAbstractFullApiSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheAbstractFullApiSelfTest.java
b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheAbstractFullApiSelfTest.java
index 237ee2f..5526c7b 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheAbstractFullApiSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheAbstractFullApiSelfTest.java
@@ -4477,10 +4477,13 @@ public abstract class GridCacheAbstractFullApiSelfTest extends GridCacheAbstract
 
         try {
             grid(0).jcache(null).withExpiryPolicy(expiry).put(key, 1);
+
+            if (tx != null)
+                tx.commit();
         }
         finally {
             if (tx != null)
-                tx.commit();
+                tx.close();
         }
 
         long[] expireTimes = new long[gridCount()];


Mime
View raw message