You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by sb...@apache.org on 2014/12/29 09:41:55 UTC
incubator-ignite git commit: ignite-44
Repository: incubator-ignite
Updated Branches:
refs/heads/ignite-44 9fd2f235e -> 0da9afad5
ignite-44
Project: http://git-wip-us.apache.org/repos/asf/incubator-ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-ignite/commit/0da9afad
Tree: http://git-wip-us.apache.org/repos/asf/incubator-ignite/tree/0da9afad
Diff: http://git-wip-us.apache.org/repos/asf/incubator-ignite/diff/0da9afad
Branch: refs/heads/ignite-44
Commit: 0da9afad51430e8a8b0eb8f96456ec1778ab3972
Parents: 9fd2f23
Author: sboikov <sb...@gridgain.com>
Authored: Mon Dec 29 09:43:08 2014 +0300
Committer: sboikov <sb...@gridgain.com>
Committed: Mon Dec 29 11:41:38 2014 +0300
----------------------------------------------------------------------
.../examples/datagrid/CacheApiExample.java | 1 -
.../datagrid/CachePopularNumbersExample.java | 2 +-
.../ScalarCachePopularNumbersExample.scala | 21 +-------
.../dataload/IgniteDataLoadCacheUpdater.java | 4 +-
.../processors/cache/IgniteCacheProxy.java | 22 ++++----
.../processors/cache/CacheInvokeResult.java | 1 -
.../processors/cache/GridCacheEntryEx.java | 5 +-
.../processors/cache/GridCacheMapEntry.java | 24 ++++-----
.../processors/cache/GridCacheProxyImpl.java | 7 ++-
.../GridAtomicCacheQueueImpl.java | 2 +-
.../datastructures/GridCacheQueueAdapter.java | 3 +-
.../dht/atomic/GridDhtAtomicCache.java | 12 +++--
.../dht/atomic/GridDhtAtomicUpdateRequest.java | 6 +--
.../dht/atomic/GridNearAtomicUpdateRequest.java | 2 +-
.../local/atomic/GridLocalAtomicCache.java | 13 +++--
.../transactions/IgniteTxLocalAdapter.java | 6 +--
.../processors/ggfs/GridGgfsDataManager.java | 7 +--
.../processors/ggfs/GridGgfsMetaManager.java | 5 +-
.../cache/IgniteCacheInvokeAbstractTest.java | 56 +++++++++++++++++++-
.../cache/GridCacheAbstractFullApiSelfTest.java | 16 ++++++
.../GridCacheInterceptorAbstractSelfTest.java | 12 ++---
.../processors/cache/GridCacheTestEntryEx.java | 5 +-
.../hadoop/jobtracker/GridHadoopJobTracker.java | 7 ++-
23 files changed, 141 insertions(+), 98 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/0da9afad/examples/src/main/java/org/gridgain/examples/datagrid/CacheApiExample.java
----------------------------------------------------------------------
diff --git a/examples/src/main/java/org/gridgain/examples/datagrid/CacheApiExample.java b/examples/src/main/java/org/gridgain/examples/datagrid/CacheApiExample.java
index f477a89..f05331f 100644
--- a/examples/src/main/java/org/gridgain/examples/datagrid/CacheApiExample.java
+++ b/examples/src/main/java/org/gridgain/examples/datagrid/CacheApiExample.java
@@ -11,7 +11,6 @@ package org.gridgain.examples.datagrid;
import org.apache.ignite.*;
import org.apache.ignite.lang.*;
-import org.gridgain.grid.*;
import org.gridgain.grid.cache.*;
import javax.cache.processor.*;
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/0da9afad/examples/src/main/java/org/gridgain/examples/datagrid/CachePopularNumbersExample.java
----------------------------------------------------------------------
diff --git a/examples/src/main/java/org/gridgain/examples/datagrid/CachePopularNumbersExample.java b/examples/src/main/java/org/gridgain/examples/datagrid/CachePopularNumbersExample.java
index 678a5e2..c9b8f1d 100644
--- a/examples/src/main/java/org/gridgain/examples/datagrid/CachePopularNumbersExample.java
+++ b/examples/src/main/java/org/gridgain/examples/datagrid/CachePopularNumbersExample.java
@@ -156,7 +156,7 @@ public class CachePopularNumbersExample {
@Override public Void process(MutableEntry<Integer, Long> e, Object... args) {
Long val = e.getValue();
- e.setValue(val == null ? 1 : val + 1);
+ e.setValue(val == null ? 1L : val + 1);
return null;
}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/0da9afad/examples/src/main/scala/org/gridgain/scalar/examples/ScalarCachePopularNumbersExample.scala
----------------------------------------------------------------------
diff --git a/examples/src/main/scala/org/gridgain/scalar/examples/ScalarCachePopularNumbersExample.scala b/examples/src/main/scala/org/gridgain/scalar/examples/ScalarCachePopularNumbersExample.scala
index 0d87f19..567cdf0 100644
--- a/examples/src/main/scala/org/gridgain/scalar/examples/ScalarCachePopularNumbersExample.scala
+++ b/examples/src/main/scala/org/gridgain/scalar/examples/ScalarCachePopularNumbersExample.scala
@@ -89,26 +89,7 @@ object ScalarCachePopularNumbersExample extends App {
// Reduce parallel operations since we running the whole grid locally under heavy load.
val ldr = dataLoader$[Int, Long](CACHE_NAME, 2048)
- val f = new EntryProcessor[Int, Long, Void] {
- override def process(e: MutableEntry[Int, Long], arguments: AnyRef*): Void = {
- if (e.exists())
- e.setValue(e.getValue + 1)
- else
- e.setValue(1)
-
- null
- }
- }
-
- // Set custom updater to increment value for each key.
- ldr.updater(new IgniteDataLoadCacheUpdater[Int, Long] {
- def update(cache: IgniteCache[Int, Long], entries: util.Collection[Entry[Int, Long]]) = {
- import scala.collection.JavaConversions._
-
- for (e <- entries)
- cache.invoke(e.getKey, f)
- }
- })
+ // TODO IGNITE-44: restore invoke.
(0 until CNT) foreach (_ => ldr.addData(Random.nextInt(RANGE), 1L))
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/0da9afad/modules/core/src/main/java/org/apache/ignite/dataload/IgniteDataLoadCacheUpdater.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/dataload/IgniteDataLoadCacheUpdater.java b/modules/core/src/main/java/org/apache/ignite/dataload/IgniteDataLoadCacheUpdater.java
index 2e1f92c..9c43db4 100644
--- a/modules/core/src/main/java/org/apache/ignite/dataload/IgniteDataLoadCacheUpdater.java
+++ b/modules/core/src/main/java/org/apache/ignite/dataload/IgniteDataLoadCacheUpdater.java
@@ -15,12 +15,12 @@ import java.io.*;
import java.util.*;
/**
- * Updates cache with batch of entries. Usually it is enough to configure {@link org.apache.ignite.IgniteDataLoader#isolated(boolean)}
+ * Updates cache with batch of entries. Usually it is enough to configure {@link IgniteDataLoader#isolated(boolean)}
* property and appropriate internal cache updater will be chosen automatically. But in some cases to achieve best
* performance custom user-defined implementation may help.
* <p>
* Data loader can be configured to use custom implementation of updater instead of default one using
- * {@link org.apache.ignite.IgniteDataLoader#updater(IgniteDataLoadCacheUpdater)} method.
+ * {@link IgniteDataLoader#updater(IgniteDataLoadCacheUpdater)} method.
*/
public interface IgniteDataLoadCacheUpdater<K, V> extends Serializable {
/**
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/0da9afad/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteCacheProxy.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteCacheProxy.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteCacheProxy.java
index eacae0f..afa2986 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteCacheProxy.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteCacheProxy.java
@@ -826,17 +826,6 @@ public class IgniteCacheProxy<K, V> extends IgniteAsyncSupportAdapter implements
return new IgniteCacheProxy<>(ctx, delegate, prj, true);
}
- /**
- * @param e Checked exception.
- * @return Cache exception.
- */
- private CacheException cacheException(IgniteCheckedException e) {
- if (e instanceof GridCachePartialUpdateException)
- return new CachePartialUpdateException((GridCachePartialUpdateException)e);
-
- return new CacheException(e);
- }
-
/** {@inheritDoc} */
@SuppressWarnings("unchecked")
@Override public <K1, V1> IgniteCache<K1, V1> keepPortable() {
@@ -897,6 +886,17 @@ public class IgniteCacheProxy<K, V> extends IgniteAsyncSupportAdapter implements
}
}
+ /**
+ * @param e Checked exception.
+ * @return Cache exception.
+ */
+ private CacheException cacheException(IgniteCheckedException e) {
+ if (e instanceof GridCachePartialUpdateException)
+ return new CachePartialUpdateException((GridCachePartialUpdateException)e);
+
+ return new CacheException(e);
+ }
+
/** {@inheritDoc} */
@Override public void writeExternal(ObjectOutput out) throws IOException {
out.writeObject(ctx);
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/0da9afad/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/CacheInvokeResult.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/CacheInvokeResult.java b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/CacheInvokeResult.java
index 19b4141..ab0959e 100644
--- a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/CacheInvokeResult.java
+++ b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/CacheInvokeResult.java
@@ -12,7 +12,6 @@ package org.gridgain.grid.kernal.processors.cache;
import org.apache.ignite.marshaller.optimized.*;
import org.gridgain.grid.util.tostring.*;
import org.gridgain.grid.util.typedef.internal.*;
-import org.jetbrains.annotations.*;
import javax.cache.processor.*;
import java.io.*;
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/0da9afad/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/GridCacheEntryEx.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/GridCacheEntryEx.java b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/GridCacheEntryEx.java
index 101427f..0962cd7 100644
--- a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/GridCacheEntryEx.java
+++ b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/GridCacheEntryEx.java
@@ -20,6 +20,7 @@ import org.gridgain.grid.util.lang.*;
import org.jetbrains.annotations.*;
import javax.cache.expiry.*;
+import javax.cache.processor.*;
import java.util.*;
/**
@@ -460,11 +461,11 @@ public interface GridCacheEntryEx<K, V> extends GridMetadataAware {
* @param intercept If {@code true} then calls cache interceptor.
* @param subjId Subject ID initiated this update.
* @param taskName Task name.
- * @return Tuple containing success flag and operation result.
+ * @return Tuple containing success flag, old value and result for invoke operation.
* @throws IgniteCheckedException If update failed.
* @throws GridCacheEntryRemovedException If entry is obsolete.
*/
- public IgniteBiTuple<Boolean, Object> innerUpdateLocal(
+ public GridTuple3<Boolean, V, EntryProcessorResult<Object>> innerUpdateLocal(
GridCacheVersion ver,
GridCacheOperation op,
@Nullable Object writeObj,
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/0da9afad/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/GridCacheMapEntry.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/GridCacheMapEntry.java b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/GridCacheMapEntry.java
index 110d8b1..bb493cc 100644
--- a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/GridCacheMapEntry.java
+++ b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/GridCacheMapEntry.java
@@ -1365,7 +1365,7 @@ public abstract class GridCacheMapEntry<K, V> implements GridCacheEntryEx<K, V>
/** {@inheritDoc} */
@SuppressWarnings("unchecked")
- @Override public IgniteBiTuple<Boolean, Object> innerUpdateLocal(
+ @Override public GridTuple3<Boolean, V, EntryProcessorResult<Object>> innerUpdateLocal(
GridCacheVersion ver,
GridCacheOperation op,
@Nullable Object writeObj,
@@ -1382,12 +1382,14 @@ public abstract class GridCacheMapEntry<K, V> implements GridCacheEntryEx<K, V>
) throws IgniteCheckedException, GridCacheEntryRemovedException {
assert cctx.isLocal() && cctx.atomic();
- Object opRes;
+ V old;
boolean res = true;
IgniteBiTuple<Boolean, ?> interceptorRes = null;
+ EntryProcessorResult<Object> invokeRes = null;
+
synchronized (this) {
boolean needVal = retval || intercept || op == GridCacheOperation.TRANSFORM || !F.isEmpty(filter);
@@ -1398,9 +1400,7 @@ public abstract class GridCacheMapEntry<K, V> implements GridCacheEntryEx<K, V>
unswap(true, retval);
// Possibly get old value form store.
- V old = needVal ? rawGetOrUnmarshalUnlocked(!retval) : val;
-
- opRes = old;
+ old = needVal ? rawGetOrUnmarshalUnlocked(!retval) : val;
GridCacheValueBytes oldBytes = valueBytesUnlocked();
@@ -1431,7 +1431,7 @@ public abstract class GridCacheMapEntry<K, V> implements GridCacheEntryEx<K, V>
updateTtl(ttl);
}
- return new IgniteBiTuple<>(false, (Object)(retval ? old : null));
+ return new T3<>(false, retval ? old : null, null);
}
}
@@ -1458,12 +1458,12 @@ public abstract class GridCacheMapEntry<K, V> implements GridCacheEntryEx<K, V>
updated = cctx.unwrapTemporary(entry.getValue());
- opRes = computed != null ? new CacheInvokeResult<>(cctx.unwrapTemporary(computed)) : null;
+ invokeRes = computed != null ? new CacheInvokeResult<>(cctx.unwrapTemporary(computed)) : null;
}
catch (Exception e) {
updated = old;
- opRes = new CacheInvokeResult<>(e);
+ invokeRes = new CacheInvokeResult<>(e);
}
}
else
@@ -1476,14 +1476,13 @@ public abstract class GridCacheMapEntry<K, V> implements GridCacheEntryEx<K, V>
updated = (V)cctx.config().getInterceptor().onBeforePut(key, old, updated);
if (updated == null)
- return new IgniteBiTuple<>(false, (Object)cctx.<V>unwrapTemporary(old));
+ return new GridTuple3<>(false, cctx.<V>unwrapTemporary(old), invokeRes);
}
else {
interceptorRes = cctx.config().getInterceptor().onBeforeRemove(key, old);
if (cctx.cancelRemove(interceptorRes))
- return new IgniteBiTuple<>(false,
- (Object)cctx.<V>unwrapTemporary(interceptorRes.get2()));
+ return new GridTuple3<>(false, cctx.<V>unwrapTemporary(interceptorRes.get2()), invokeRes);
}
}
@@ -1593,8 +1592,7 @@ public abstract class GridCacheMapEntry<K, V> implements GridCacheEntryEx<K, V>
}
}
- return new IgniteBiTuple<>(res,
- (Object)(cctx.<V>unwrapTemporary(interceptorRes != null ? interceptorRes.get2() : opRes)));
+ return new GridTuple3<>(res, cctx.<V>unwrapTemporary(interceptorRes != null ? interceptorRes.get2() : old), invokeRes);
}
/** {@inheritDoc} */
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/0da9afad/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/GridCacheProxyImpl.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/GridCacheProxyImpl.java b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/GridCacheProxyImpl.java
index 80a776a..dd24388 100644
--- a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/GridCacheProxyImpl.java
+++ b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/GridCacheProxyImpl.java
@@ -797,8 +797,11 @@ public class GridCacheProxyImpl<K, V> implements GridCacheProxy<K, V>, Externali
}
/** {@inheritDoc} */
- @Override public IgniteFuture<Boolean> putxAsync(K key, V val, @Nullable GridCacheEntryEx<K, V> entry,
- long ttl, @Nullable IgnitePredicate<GridCacheEntry<K, V>>... filter) {
+ @Override public IgniteFuture<Boolean> putxAsync(K key,
+ V val,
+ @Nullable GridCacheEntryEx<K, V> entry,
+ long ttl,
+ @Nullable IgnitePredicate<GridCacheEntry<K, V>>... filter) {
GridCacheProjectionImpl<K, V> prev = gate.enter(prj);
try {
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/0da9afad/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/datastructures/GridAtomicCacheQueueImpl.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/datastructures/GridAtomicCacheQueueImpl.java b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/datastructures/GridAtomicCacheQueueImpl.java
index 090fb94..786e863 100644
--- a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/datastructures/GridAtomicCacheQueueImpl.java
+++ b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/datastructures/GridAtomicCacheQueueImpl.java
@@ -223,7 +223,7 @@ public class GridAtomicCacheQueueImpl<T> extends GridCacheQueueAdapter<T> {
/**
* @param c EntryProcessor to be applied for queue header.
- * @return Value computed by the transform closure.
+ * @return Value computed by the entry processor.
* @throws IgniteCheckedException If failed.
*/
@SuppressWarnings("unchecked")
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/0da9afad/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/datastructures/GridCacheQueueAdapter.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/datastructures/GridCacheQueueAdapter.java b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/datastructures/GridCacheQueueAdapter.java
index e10ac90..ed00313 100644
--- a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/datastructures/GridCacheQueueAdapter.java
+++ b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/datastructures/GridCacheQueueAdapter.java
@@ -322,8 +322,7 @@ public abstract class GridCacheQueueAdapter<T> extends AbstractCollection<T> imp
A.ensure(batchSize >= 0, "Batch size cannot be negative: " + batchSize);
try {
- IgniteBiTuple<Long, Long> t = (IgniteBiTuple<Long, Long>)cache.invoke(queueKey,
- new ClearProcessor(id));
+ IgniteBiTuple<Long, Long> t = (IgniteBiTuple<Long, Long>)cache.invoke(queueKey, new ClearProcessor(id));
if (t == null)
return;
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/0da9afad/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java
index e9bb16e..70491e0 100644
--- a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java
+++ b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java
@@ -1276,11 +1276,6 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
CU.<K, V>empty(),
null);
- if (entryProcessorMap == null)
- entryProcessorMap = new HashMap<>();
-
- entryProcessorMap.put(entry.key(), entryProcessor);
-
CacheInvokeEntry<K, V> invokeEntry = new CacheInvokeEntry<>(entry.key(), old);
V updated;
@@ -1335,6 +1330,7 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
firstEntryIdx = i + 1;
putMap = null;
+ entryProcessorMap = null;
filtered = new ArrayList<>();
}
@@ -1376,6 +1372,7 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
firstEntryIdx = i + 1;
rmvKeys = null;
+ entryProcessorMap = null;
filtered = new ArrayList<>();
}
@@ -1385,6 +1382,11 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
putMap.put(entry.key(), ctx.<V>unwrapTemporary(updated));
}
+
+ if (entryProcessorMap == null)
+ entryProcessorMap = new HashMap<>();
+
+ entryProcessorMap.put(entry.key(), entryProcessor);
}
else if (op == UPDATE) {
V updated = req.value(i);
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/0da9afad/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/distributed/dht/atomic/GridDhtAtomicUpdateRequest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/distributed/dht/atomic/GridDhtAtomicUpdateRequest.java b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/distributed/dht/atomic/GridDhtAtomicUpdateRequest.java
index 60e42bc..c34a221 100644
--- a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/distributed/dht/atomic/GridDhtAtomicUpdateRequest.java
+++ b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/distributed/dht/atomic/GridDhtAtomicUpdateRequest.java
@@ -120,11 +120,11 @@ public class GridDhtAtomicUpdateRequest<K, V> extends GridCacheMessage<K, V> imp
@GridDirectVersion(2)
private List<byte[]> entryProcessorsBytes;
- /** Near transform closures. */
+ /** Near entry processors. */
@GridDirectTransient
private List<EntryProcessor<K, V, ?>> nearEntryProcessors;
- /** Near transform closures bytes. */
+ /** Near entry processors bytes. */
@GridDirectCollection(byte[].class)
@GridDirectVersion(2)
private List<byte[]> nearEntryProcessorsBytes;
@@ -133,7 +133,7 @@ public class GridDhtAtomicUpdateRequest<K, V> extends GridCacheMessage<K, V> imp
@GridDirectTransient
private Object[] invokeArgs;
- /** Filter bytes. */
+ /** Entry processor arguments bytes. */
private byte[][] invokeArgsBytes;
/** Subject ID. */
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/0da9afad/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/distributed/dht/atomic/GridNearAtomicUpdateRequest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/distributed/dht/atomic/GridNearAtomicUpdateRequest.java b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/distributed/dht/atomic/GridNearAtomicUpdateRequest.java
index 391a5b2..a210711 100644
--- a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/distributed/dht/atomic/GridNearAtomicUpdateRequest.java
+++ b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/distributed/dht/atomic/GridNearAtomicUpdateRequest.java
@@ -82,7 +82,7 @@ public class GridNearAtomicUpdateRequest<K, V> extends GridCacheMessage<K, V> im
@GridDirectTransient
private Object[] invokeArgs;
- /** Filter bytes. */
+ /** Entry processor arguments bytes. */
private byte[][] invokeArgsBytes;
/** DR versions. */
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/0da9afad/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/local/atomic/GridLocalAtomicCache.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/local/atomic/GridLocalAtomicCache.java b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/local/atomic/GridLocalAtomicCache.java
index a3b4157..639ab38 100644
--- a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/local/atomic/GridLocalAtomicCache.java
+++ b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/local/atomic/GridLocalAtomicCache.java
@@ -19,6 +19,7 @@ import org.gridgain.grid.kernal.processors.cache.local.*;
import org.gridgain.grid.kernal.processors.cache.transactions.*;
import org.gridgain.grid.util.*;
import org.gridgain.grid.util.future.*;
+import org.gridgain.grid.util.lang.*;
import org.gridgain.grid.util.typedef.*;
import org.gridgain.grid.util.typedef.internal.*;
import org.jetbrains.annotations.*;
@@ -882,7 +883,7 @@ public class GridLocalAtomicCache<K, V> extends GridCacheAdapter<K, V> {
try {
entry = entryEx(key);
- IgniteBiTuple<Boolean, Object> t = entry.innerUpdateLocal(
+ GridTuple3<Boolean, V, EntryProcessorResult<Object>> t = entry.innerUpdateLocal(
ver,
val == null ? DELETE : op,
val,
@@ -898,9 +899,7 @@ public class GridLocalAtomicCache<K, V> extends GridCacheAdapter<K, V> {
taskName);
if (op == TRANSFORM) {
- assert t.get2() == null || t.get2() instanceof EntryProcessorResult : t.get2();
-
- if (t.get2() != null) {
+ if (t.get3() != null) {
Map<K, EntryProcessorResult> computedMap;
if (res == null) {
@@ -911,11 +910,11 @@ public class GridLocalAtomicCache<K, V> extends GridCacheAdapter<K, V> {
else
computedMap = (Map<K, EntryProcessorResult>)res.get2();
- computedMap.put(key, (EntryProcessorResult)t.getValue());
+ computedMap.put(key, t.get3());
}
}
else if (res == null)
- res = t;
+ res = new T2(t.get1(), t.get2());
break; // While.
}
@@ -1297,7 +1296,7 @@ public class GridLocalAtomicCache<K, V> extends GridCacheAdapter<K, V> {
assert writeVal != null || op == DELETE : "null write value found.";
- IgniteBiTuple<Boolean, Object> t = entry.innerUpdateLocal(
+ GridTuple3<Boolean, V, EntryProcessorResult<Object>> t = entry.innerUpdateLocal(
ver,
op,
writeVal,
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/0da9afad/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/transactions/IgniteTxLocalAdapter.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/transactions/IgniteTxLocalAdapter.java b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/transactions/IgniteTxLocalAdapter.java
index d349215..6380605 100644
--- a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/transactions/IgniteTxLocalAdapter.java
+++ b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/cache/transactions/IgniteTxLocalAdapter.java
@@ -1850,10 +1850,10 @@ public abstract class IgniteTxLocalAdapter<K, V> extends IgniteTxAdapter<K, V>
Set<K> skipped = null;
- Set<K> missedForInvoke = null;
-
boolean rmv = lookup == null && invokeMap == null;
+ Set<K> missedForInvoke = null;
+
try {
// Set transform flag for transaction.
if (invokeMap != null)
@@ -2285,7 +2285,7 @@ public abstract class IgniteTxLocalAdapter<K, V> extends IgniteTxAdapter<K, V>
}
else {
if (!hasPrevVal)
- v = retval ? cached.rawGetOrUnmarshal(false) : cached.rawGet();
+ v = cached.rawGetOrUnmarshal(false);
}
if (txEntry.op() == TRANSFORM) {
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/0da9afad/modules/core/src/main/java/org/gridgain/grid/kernal/processors/ggfs/GridGgfsDataManager.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/ggfs/GridGgfsDataManager.java b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/ggfs/GridGgfsDataManager.java
index 89bcbe6..1906b95 100644
--- a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/ggfs/GridGgfsDataManager.java
+++ b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/ggfs/GridGgfsDataManager.java
@@ -1118,9 +1118,7 @@ public class GridGgfsDataManager extends GridGgfsManager {
GridGgfsBlockKey key = new GridGgfsBlockKey(colocatedKey.getFileId(), null,
colocatedKey.evictExclude(), colocatedKey.getBlockId());
- IgniteTx tx = dataCachePrj.txStart(PESSIMISTIC, REPEATABLE_READ);
-
- try {
+ try (IgniteTx tx = dataCachePrj.txStart(PESSIMISTIC, REPEATABLE_READ)) {
// Lock keys.
Map<GridGgfsBlockKey, byte[]> vals = dataCachePrj.getAll(F.asList(colocatedKey, key));
@@ -1147,9 +1145,6 @@ public class GridGgfsDataManager extends GridGgfsManager {
tx.commit();
}
- finally {
- tx.close();
- }
}
/**
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/0da9afad/modules/core/src/main/java/org/gridgain/grid/kernal/processors/ggfs/GridGgfsMetaManager.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/ggfs/GridGgfsMetaManager.java b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/ggfs/GridGgfsMetaManager.java
index 87e09b9..722d142 100644
--- a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/ggfs/GridGgfsMetaManager.java
+++ b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/ggfs/GridGgfsMetaManager.java
@@ -2796,11 +2796,8 @@ public class GridGgfsMetaManager extends GridGgfsManager {
GridGgfsListingEntry entry = listing.get(fileName);
- if (entry == null || !entry.fileId().equals(fileId)) {
- e.setValue(fileInfo);
-
+ if (entry == null || !entry.fileId().equals(fileId))
return null;
- }
entry = new GridGgfsListingEntry(entry, entry.length() + lenDelta,
accessTime == -1 ? entry.accessTime() : accessTime,
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/0da9afad/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheInvokeAbstractTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheInvokeAbstractTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheInvokeAbstractTest.java
index 380eced..bda973f 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheInvokeAbstractTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheInvokeAbstractTest.java
@@ -10,6 +10,7 @@
package org.apache.ignite.internal.processors.cache;
import org.apache.ignite.*;
+import org.apache.ignite.lang.*;
import org.apache.ignite.transactions.*;
import org.gridgain.grid.cache.*;
import org.gridgain.grid.util.typedef.internal.*;
@@ -138,9 +139,23 @@ public abstract class IgniteCacheInvokeAbstractTest extends IgniteCacheAbstractT
checkValue(key, 63);
+ IgniteCache<Integer, Integer> asyncCache = cache.enableAsync();
+
+ assertTrue(asyncCache.isAsync());
+
+ assertNull(asyncCache.invoke(key, incProcessor));
+
+ IgniteFuture<Integer> fut = asyncCache.future();
+
+ assertNotNull(fut);
+
+ assertEquals(63, (int)fut.get());
+
+ checkValue(key, 64);
+
tx = startTx(txMode);
- assertNull(cache.invoke(key, new RemoveProcessor(63)));
+ assertNull(cache.invoke(key, new RemoveProcessor(64)));
if (tx != null)
tx.commit();
@@ -374,6 +389,45 @@ public abstract class IgniteCacheInvokeAbstractTest extends IgniteCacheAbstractT
for (Integer key : keys)
checkValue(key, null);
+
+ IgniteCache<Integer, Integer> asyncCache = cache.enableAsync();
+
+ assertTrue(asyncCache.isAsync());
+
+ assertNull(asyncCache.invokeAll(keys, new IncrementProcessor()));
+
+ IgniteFuture<Map<Integer, EntryProcessorResult<Integer>>> fut = asyncCache.future();
+
+ resMap = fut.get();
+
+ exp = new HashMap<>();
+
+ for (Integer key : keys)
+ exp.put(key, -1);
+
+ checkResult(resMap, exp);
+
+ for (Integer key : keys)
+ checkValue(key, 1);
+
+ invokeMap = new HashMap<>();
+
+ for (Integer key : keys)
+ invokeMap.put(key, incProcessor);
+
+ assertNull(asyncCache.invokeAll(invokeMap));
+
+ fut = asyncCache.future();
+
+ resMap = fut.get();
+
+ for (Integer key : keys)
+ exp.put(key, 1);
+
+ checkResult(resMap, exp);
+
+ for (Integer key : keys)
+ checkValue(key, 2);
}
/**
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/0da9afad/modules/core/src/test/java/org/gridgain/grid/kernal/processors/cache/GridCacheAbstractFullApiSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/gridgain/grid/kernal/processors/cache/GridCacheAbstractFullApiSelfTest.java b/modules/core/src/test/java/org/gridgain/grid/kernal/processors/cache/GridCacheAbstractFullApiSelfTest.java
index a6f6e14..54397cc 100644
--- a/modules/core/src/test/java/org/gridgain/grid/kernal/processors/cache/GridCacheAbstractFullApiSelfTest.java
+++ b/modules/core/src/test/java/org/gridgain/grid/kernal/processors/cache/GridCacheAbstractFullApiSelfTest.java
@@ -926,6 +926,22 @@ public abstract class GridCacheAbstractFullApiSelfTest extends GridCacheAbstract
assertEquals("3", res.get("key3").get());
assertEquals(3, res.size());
+
+ cache.remove("key1");
+ cache.put("key2", 1);
+ cache.put("key3", 3);
+
+ res = cache.invokeAll(F.asMap("key1", INCR_PROCESSOR, "key2", INCR_PROCESSOR, "key3", INCR_PROCESSOR));
+
+ assertEquals((Integer)1, cache.get("key1"));
+ assertEquals((Integer)2, cache.get("key2"));
+ assertEquals((Integer)4, cache.get("key3"));
+
+ assertEquals("null", res.get("key1").get());
+ assertEquals("1", res.get("key2").get());
+ assertEquals("3", res.get("key3").get());
+
+ assertEquals(3, res.size());
}
/**
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/0da9afad/modules/core/src/test/java/org/gridgain/grid/kernal/processors/cache/GridCacheInterceptorAbstractSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/gridgain/grid/kernal/processors/cache/GridCacheInterceptorAbstractSelfTest.java b/modules/core/src/test/java/org/gridgain/grid/kernal/processors/cache/GridCacheInterceptorAbstractSelfTest.java
index 66abdc6..3f3d3a6 100644
--- a/modules/core/src/test/java/org/gridgain/grid/kernal/processors/cache/GridCacheInterceptorAbstractSelfTest.java
+++ b/modules/core/src/test/java/org/gridgain/grid/kernal/processors/cache/GridCacheInterceptorAbstractSelfTest.java
@@ -1194,8 +1194,7 @@ public abstract class GridCacheInterceptorAbstractSelfTest extends GridCacheAbst
case UPDATE_FILTER: {
Object old = cache.getAndRemoveIf(key, new IgnitePredicate<GridCacheEntry<String, Integer>>() {
- @Override
- public boolean apply(GridCacheEntry<String, Integer> entry) {
+ @Override public boolean apply(GridCacheEntry<String, Integer> entry) {
return true;
}
});
@@ -1207,8 +1206,7 @@ public abstract class GridCacheInterceptorAbstractSelfTest extends GridCacheAbst
case TRANSFORM: {
cache.invoke(key, new EntryProcessor<String, Integer, Void>() {
- @Override
- public Void process(MutableEntry<String, Integer> e, Object... args) {
+ @Override public Void process(MutableEntry<String, Integer> e, Object... args) {
Integer old = e.getValue();
assertEquals(expOld, old);
@@ -1242,8 +1240,7 @@ public abstract class GridCacheInterceptorAbstractSelfTest extends GridCacheAbst
case UPDATE_FILTER: {
Object old = cache.getAndPutIf(key, val, new P1<GridCacheEntry<String, Integer>>() {
- @Override
- public boolean apply(GridCacheEntry<String, Integer> entry) {
+ @Override public boolean apply(GridCacheEntry<String, Integer> entry) {
return true;
}
});
@@ -1255,8 +1252,7 @@ public abstract class GridCacheInterceptorAbstractSelfTest extends GridCacheAbst
case TRANSFORM: {
cache.invoke(key, new EntryProcessor<String, Integer, Void>() {
- @Override
- public Void process(MutableEntry<String, Integer> e, Object... args) {
+ @Override public Void process(MutableEntry<String, Integer> e, Object... args) {
Integer old = e.getValue();
assertEquals(expOld, old);
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/0da9afad/modules/core/src/test/java/org/gridgain/grid/kernal/processors/cache/GridCacheTestEntryEx.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/gridgain/grid/kernal/processors/cache/GridCacheTestEntryEx.java b/modules/core/src/test/java/org/gridgain/grid/kernal/processors/cache/GridCacheTestEntryEx.java
index af729a1..89c1efa 100644
--- a/modules/core/src/test/java/org/gridgain/grid/kernal/processors/cache/GridCacheTestEntryEx.java
+++ b/modules/core/src/test/java/org/gridgain/grid/kernal/processors/cache/GridCacheTestEntryEx.java
@@ -19,6 +19,7 @@ import org.gridgain.grid.util.typedef.*;
import org.jetbrains.annotations.*;
import javax.cache.expiry.*;
+import javax.cache.processor.*;
import java.util.*;
/**
@@ -423,7 +424,7 @@ public class GridCacheTestEntryEx<K, V> extends GridMetadataAwareAdapter impleme
}
/** {@inheritDoc} */
- @Override public IgniteBiTuple<Boolean, Object> innerUpdateLocal(GridCacheVersion ver,
+ @Override public GridTuple3<Boolean, V, EntryProcessorResult<Object>> innerUpdateLocal(GridCacheVersion ver,
GridCacheOperation op,
@Nullable Object writeObj,
@Nullable Object[] invokeArgs,
@@ -437,7 +438,7 @@ public class GridCacheTestEntryEx<K, V> extends GridMetadataAwareAdapter impleme
UUID subjId,
String taskName)
throws IgniteCheckedException, GridCacheEntryRemovedException {
- return new IgniteBiTuple<>(false, null);
+ return new GridTuple3<>(false, null, null);
}
/** {@inheritDoc} */
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/0da9afad/modules/hadoop/src/main/java/org/gridgain/grid/kernal/processors/hadoop/jobtracker/GridHadoopJobTracker.java
----------------------------------------------------------------------
diff --git a/modules/hadoop/src/main/java/org/gridgain/grid/kernal/processors/hadoop/jobtracker/GridHadoopJobTracker.java b/modules/hadoop/src/main/java/org/gridgain/grid/kernal/processors/hadoop/jobtracker/GridHadoopJobTracker.java
index 23d98c1..6b3260f 100644
--- a/modules/hadoop/src/main/java/org/gridgain/grid/kernal/processors/hadoop/jobtracker/GridHadoopJobTracker.java
+++ b/modules/hadoop/src/main/java/org/gridgain/grid/kernal/processors/hadoop/jobtracker/GridHadoopJobTracker.java
@@ -652,7 +652,8 @@ public class GridHadoopJobTracker extends GridHadoopComponent {
}
/**
- *
+ * @param jobId Job ID.
+ * @param plan Map-reduce plan.
*/
private void printPlan(GridHadoopJobId jobId, GridHadoopMapReducePlan plan) {
log.info("Plan for " + jobId);
@@ -1325,9 +1326,10 @@ public class GridHadoopJobTracker extends GridHadoopComponent {
/**
* @param prev Previous closure.
* @param splits Mapper splits to remove.
+ * @param err Error.
*/
private RemoveMappersProcessor(@Nullable StackedProcessor prev, Collection<GridHadoopInputSplit> splits,
- Throwable err) {
+ Throwable err) {
super(prev);
this.splits = splits;
@@ -1382,6 +1384,7 @@ public class GridHadoopJobTracker extends GridHadoopComponent {
/**
* @param prev Previous closure.
* @param rdc Reducer to remove.
+ * @param err Error.
*/
private RemoveReducerProcessor(@Nullable StackedProcessor prev, int rdc, Throwable err) {
super(prev);