GG-9655 - Fixing tests after merge.

apache · Jan 31, 2015 · 9215a07 · 9215a07
1 parent 4fde8d3
commit 9215a07
Show file tree

Hide file tree

Showing 6 changed files with 77 additions and 13 deletions.
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheReturn.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheReturn.java
@@ -44,6 +44,9 @@ public class GridCacheReturn<V> implements Externalizable, IgniteOptimizedMarsha
     /** Success flag. */
     private volatile boolean success;
 
+    /** */
+    private volatile boolean invokeRes;
+
     /**
      * Empty constructor.
      */
@@ -69,6 +72,19 @@ public GridCacheReturn(V v, boolean success) {
         this.success = success;
     }
 
+    /**
+     *
+     * @param v Value.
+     * @param success Success flag.
+     */
+    public GridCacheReturn(V v, boolean success, boolean invokeRes) {
+        assert !invokeRes || v instanceof Map;
+
+        this.v = v;
+        this.success = success;
+        this.invokeRes = invokeRes;
+    }
+
     /**
      * @return Value.
      */
@@ -85,6 +101,13 @@ public boolean hasValue() {
         return v != null;
     }
 
+    /**
+     * @return If return is invoke result.
+     */
+    public boolean invokeResult() {
+        return invokeRes;
+    }
+
     /**
      * @param v Value.
      * @return This instance for chaining.
@@ -134,6 +157,8 @@ public synchronized void addEntryProcessResult(Object key, EntryProcessorResult<
         assert key != null;
         assert res != null;
 
+        invokeRes = true;
+
         HashMap<Object, EntryProcessorResult> resMap = (HashMap<Object, EntryProcessorResult>)v;
 
         if (resMap == null) {
@@ -145,6 +170,26 @@ public synchronized void addEntryProcessResult(Object key, EntryProcessorResult<
         resMap.put(key, res);
     }
 
+    /**
+     * @param other Other result to merge with.
+     */
+    public synchronized void mergeEntryProcessResults(GridCacheReturn<V> other) {
+        assert invokeRes || v == null : "Invalid state to merge: " + this;
+        assert other.invokeRes;
+
+        invokeRes = true;
+
+        HashMap<Object, EntryProcessorResult> resMap = (HashMap<Object, EntryProcessorResult>)v;
+
+        if (resMap == null) {
+            resMap = new HashMap<>();
+
+            v = (V)resMap;
+        }
+
+        resMap.putAll((Map<Object, EntryProcessorResult>)other.v);
+    }
+
     /** {@inheritDoc} */
     @Override public Object ggClassId() {
         return GG_CLASS_ID;
@@ -154,13 +199,15 @@ public synchronized void addEntryProcessResult(Object key, EntryProcessorResult<
     @Override public void writeExternal(ObjectOutput out) throws IOException {
         out.writeBoolean(success);
         out.writeObject(v);
+        out.writeBoolean(invokeRes);
     }
 
     /** {@inheritDoc} */
     @SuppressWarnings("unchecked")
     @Override public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException {
         success = in.readBoolean();
         v = (V)in.readObject();
+        invokeRes = in.readBoolean();
     }
 
     /** {@inheritDoc} */

diff --git a/...a/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxPrepareFuture.java b/...a/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxPrepareFuture.java
@@ -265,7 +265,7 @@ private boolean checkLocks() {
     /**
      *
      */
-    private void checkFilters() {
+    private void onEntriesLocked() {
         ret = new GridCacheReturn<>(null, true);
 
         for (IgniteTxEntry<K, V> txEntry : tx.optimisticLockEntries()) {
@@ -769,7 +769,7 @@ private void prepare0() {
             // We are holding transaction-level locks for entries here, so we can get next write version.
             tx.writeVersion(cctx.versions().next(tx.topologyVersion()));
 
-            checkFilters();
+            onEntriesLocked();
 
             {
                 Map<UUID, GridDistributedTxMapping<K, V>> futDhtMap = new HashMap<>();

diff --git a/...rg/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java b/...rg/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java
@@ -1805,7 +1805,7 @@ else if (F.contains(readers, node.id())) // Reader became primary or backup.
                         if (retVal == null) {
                             computedMap = U.newHashMap(keys.size());
 
-                            retVal = new GridCacheReturn<>((Object)computedMap, updRes.success());
+                            retVal = new GridCacheReturn<>((Object)computedMap, updRes.success(), true);
                         }
 
                         computedMap.put(k, updRes.computedResult());

diff --git a/...org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxPrepareFuture.java b/...org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxPrepareFuture.java
@@ -230,7 +230,7 @@ private boolean checkLocks() {
                 log.debug("Still waiting for locks [fut=" + this + ", keys=" + lockKeys + ']');
         }
 
-        return true;
+        return locked;
     }
 
     /**
@@ -1046,8 +1046,7 @@ void onResult(UUID nodeId, GridNearTxPrepareResponse<K, V> res) {
                         }
                     }
 
-                    if (tx.implicitSingle())
-                        tx.implicitSingleResult(res.returnValue());
+                    tx.implicitSingleResult(res.returnValue());
 
                     for (IgniteTxKey<K> key : res.filterFailedKeys()) {
                         IgniteTxEntry<K, V> txEntry = tx.entry(key);

diff --git a/...n/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxLocalAdapter.java b/...n/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxLocalAdapter.java
@@ -282,7 +282,10 @@ protected IgniteTxLocalAdapter(
      * @param ret Result.
      */
     public void implicitSingleResult(GridCacheReturn<V> ret) {
-        implicitRes = ret;
+        if (ret.invokeResult())
+            implicitRes.mergeEntryProcessResults(ret);
+        else
+            implicitRes = ret;
     }
 
     /**
@@ -730,17 +733,32 @@ else if (cacheCtx.isNear() && txEntry.locallyMapped())
 
                                     boolean evt = !isNearLocallyMapped(txEntry, false);
 
+                                    if (!F.isEmpty(txEntry.entryProcessors()) || !F.isEmpty(txEntry.filters()))
+                                        txEntry.cached().unswap(true, false);
+
+                                    GridTuple3<GridCacheOperation, V, byte[]> res = applyTransformClosures(txEntry,
+                                        true);
+
                                     // For near local transactions we must record DHT version
                                     // in order to keep near entries on backup nodes until
                                     // backup remote transaction completes.
-                                    if (cacheCtx.isNear())
+                                    if (cacheCtx.isNear()) {
                                         ((GridNearCacheEntry<K, V>)cached).recordDhtVersion(txEntry.dhtVersion());
 
-                                    if (!F.isEmpty(txEntry.entryProcessors()) || !F.isEmpty(txEntry.filters()))
-                                        txEntry.cached().unswap(true, false);
+                                        if (txEntry.op() == CREATE || txEntry.op() == UPDATE && txEntry.drExpireTime() == -1L) {
+                                            ExpiryPolicy expiry = txEntry.expiry();
 
-                                    GridTuple3<GridCacheOperation, V, byte[]> res = applyTransformClosures(txEntry,
-                                        true);
+                                            if (expiry == null)
+                                                expiry = cacheCtx.expiry();
+
+                                            if (expiry != null) {
+                                                Duration duration = cached.hasValue() ?
+                                                    expiry.getExpiryForUpdate() : expiry.getExpiryForCreation();
+
+                                                txEntry.ttl(CU.toTtl(duration));
+                                            }
+                                        }
+                                    }
 
                                     GridCacheOperation op = res.get1();
                                     V val = res.get2();

diff --git a/...st/java/org/apache/ignite/internal/processors/cache/GridCacheAbstractFullApiSelfTest.java b/...st/java/org/apache/ignite/internal/processors/cache/GridCacheAbstractFullApiSelfTest.java
@@ -5392,7 +5392,7 @@ public void testIgniteCacheIterator() throws Exception {
 
         assertFalse(cache.iterator().hasNext());
 
-        final int SIZE = 20000;
+        final int SIZE = 5000;
 
         Map<String, Integer> entries = new HashMap<>();