You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by vo...@apache.org on 2016/01/02 19:52:02 UTC
[1/8] ignite git commit: IGNITE-2235: Replaced ConcurrentLinkedDeque8
with ConcurrentLinkedQueue in GridNioServer.
Repository: ignite
Updated Branches:
refs/heads/ignite-2265 d4d7c24d9 -> 5c1cd29c5
IGNITE-2235: Replaced ConcurrentLinkedDeque8 with ConcurrentLinkedQueue in GridNioServer.
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/2613b958
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/2613b958
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/2613b958
Branch: refs/heads/ignite-2265
Commit: 2613b9581736d8c10eac0ac3250910e76f9600b5
Parents: e298eab
Author: vozerov-gridgain <vo...@gridgain.com>
Authored: Tue Dec 29 15:29:19 2015 +0300
Committer: vozerov-gridgain <vo...@gridgain.com>
Committed: Tue Dec 29 15:29:19 2015 +0300
----------------------------------------------------------------------
.../ignite/internal/util/nio/GridNioServer.java | 16 ++++++++++------
1 file changed, 10 insertions(+), 6 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/2613b958/modules/core/src/main/java/org/apache/ignite/internal/util/nio/GridNioServer.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/util/nio/GridNioServer.java b/modules/core/src/main/java/org/apache/ignite/internal/util/nio/GridNioServer.java
index 17a0b8f..c366fe3 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/util/nio/GridNioServer.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/util/nio/GridNioServer.java
@@ -41,8 +41,9 @@ import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
-import java.util.Queue;
import java.util.Set;
+import java.util.concurrent.ConcurrentLinkedQueue;
+
import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.IgniteException;
import org.apache.ignite.IgniteLogger;
@@ -65,7 +66,6 @@ import org.apache.ignite.plugin.extensions.communication.Message;
import org.apache.ignite.plugin.extensions.communication.MessageWriter;
import org.apache.ignite.thread.IgniteThread;
import org.jetbrains.annotations.Nullable;
-import org.jsr166.ConcurrentLinkedDeque8;
import sun.nio.ch.DirectBuffer;
import static org.apache.ignite.internal.util.nio.GridNioSessionMetaKey.ACK_CLOSURE;
@@ -1105,7 +1105,9 @@ public class GridNioServer<T> {
*/
private void writeSslSystem(GridSelectorNioSessionImpl ses, WritableByteChannel sockCh)
throws IOException {
- ConcurrentLinkedDeque8<ByteBuffer> queue = ses.meta(BUF_SSL_SYSTEM_META_KEY);
+ ConcurrentLinkedQueue<ByteBuffer> queue = ses.meta(BUF_SSL_SYSTEM_META_KEY);
+
+ assert queue != null;
ByteBuffer buf;
@@ -1244,7 +1246,7 @@ public class GridNioServer<T> {
*/
private abstract class AbstractNioClientWorker extends GridWorker {
/** Queue of change requests on this selector. */
- private final Queue<NioOperationFuture> changeReqs = new ConcurrentLinkedDeque8<>();
+ private final ConcurrentLinkedQueue<NioOperationFuture> changeReqs = new ConcurrentLinkedQueue<>();
/** Selector to select read events. */
private Selector selector;
@@ -2105,7 +2107,7 @@ public class GridNioServer<T> {
/** {@inheritDoc} */
@Override public void onSessionOpened(GridNioSession ses) throws IgniteCheckedException {
if (directMode && sslFilter != null)
- ses.addMeta(BUF_SSL_SYSTEM_META_KEY, new ConcurrentLinkedDeque8<>());
+ ses.addMeta(BUF_SSL_SYSTEM_META_KEY, new ConcurrentLinkedQueue<>());
proceedSessionOpened(ses);
}
@@ -2126,7 +2128,9 @@ public class GridNioServer<T> {
boolean sslSys = sslFilter != null && msg instanceof ByteBuffer;
if (sslSys) {
- ConcurrentLinkedDeque8<ByteBuffer> queue = ses.meta(BUF_SSL_SYSTEM_META_KEY);
+ ConcurrentLinkedQueue<ByteBuffer> queue = ses.meta(BUF_SSL_SYSTEM_META_KEY);
+
+ assert queue != null;
queue.offer((ByteBuffer)msg);
[2/8] ignite git commit: Minor fix.
Posted by vo...@apache.org.
Minor fix.
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/ce2a29b0
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/ce2a29b0
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/ce2a29b0
Branch: refs/heads/ignite-2265
Commit: ce2a29b073b7934f2c5a04e6a076f4f07eed4ccc
Parents: 2613b95
Author: vozerov-gridgain <vo...@gridgain.com>
Authored: Tue Dec 29 15:50:39 2015 +0300
Committer: vozerov-gridgain <vo...@gridgain.com>
Committed: Tue Dec 29 15:50:39 2015 +0300
----------------------------------------------------------------------
.../processors/cache/version/GridCachePlainVersionedEntry.java | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/ce2a29b0/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/version/GridCachePlainVersionedEntry.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/version/GridCachePlainVersionedEntry.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/version/GridCachePlainVersionedEntry.java
index c0eb4db..e1da1c4 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/version/GridCachePlainVersionedEntry.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/version/GridCachePlainVersionedEntry.java
@@ -111,7 +111,7 @@ public class GridCachePlainVersionedEntry<K, V> implements GridCacheVersionedEnt
/** {@inheritDoc} */
@Override public long globalTime() {
- return ver.globalTime();
+ return isStartVer ? 0L : ver.globalTime();
}
/** {@inheritDoc} */
[3/8] ignite git commit: PageMemory - WIP.
Posted by vo...@apache.org.
PageMemory - WIP.
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/13e408ef
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/13e408ef
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/13e408ef
Branch: refs/heads/ignite-2265
Commit: 13e408efaeaa6a57061fa4d5f8be4a608495e9c7
Parents: ce2a29b
Author: Alexey Goncharuk <al...@gmail.com>
Authored: Tue Dec 29 17:32:06 2015 +0300
Committer: Alexey Goncharuk <al...@gmail.com>
Committed: Tue Dec 29 17:33:17 2015 +0300
----------------------------------------------------------------------
.../ignite/internal/util/IgniteUtils.java | 40 ++++++++++++++++++++
1 file changed, 40 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/13e408ef/modules/core/src/main/java/org/apache/ignite/internal/util/IgniteUtils.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/util/IgniteUtils.java b/modules/core/src/main/java/org/apache/ignite/internal/util/IgniteUtils.java
index be4851d..480859d 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/util/IgniteUtils.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/util/IgniteUtils.java
@@ -7667,6 +7667,22 @@ public abstract class IgniteUtils {
}
/**
+ * Gets object field offset.
+ *
+ * @param cls Object class.
+ * @param fieldName Field name.
+ * @return Field offset.
+ */
+ public static long fieldOffset(Class<?> cls, String fieldName) {
+ try {
+ return UNSAFE.objectFieldOffset(cls.getDeclaredField(fieldName));
+ }
+ catch (NoSuchFieldException e) {
+ throw new IllegalStateException(e);
+ }
+ }
+
+ /**
* @param cls Class to check.
* @return {@code True} if class is final.
*/
@@ -9174,6 +9190,30 @@ public abstract class IgniteUtils {
}
/**
+ * @param cls The class to search.
+ * @param name Name of a field to get.
+ * @return Field or {@code null}.
+ */
+ @Nullable public static Field findNonPublicField(Class<?> cls, String name) {
+ while (cls != null) {
+ try {
+ Field fld = cls.getDeclaredField(name);
+
+ fld.setAccessible(true);
+
+ return fld;
+ }
+ catch (NoSuchFieldException e) {
+ // No-op.
+ }
+
+ cls = cls.getSuperclass();
+ }
+
+ return null;
+ }
+
+ /**
* @param c Collection.
* @param p Optional filters.
* @return Resulting array list.
[7/8] ignite git commit: Added test for ignite-2328.
Posted by vo...@apache.org.
Added test for ignite-2328.
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/dffcb834
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/dffcb834
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/dffcb834
Branch: refs/heads/ignite-2265
Commit: dffcb834f0f5b8971398e704d520dab9b8901b3a
Parents: 24a78f5
Author: sboikov <sb...@gridgain.com>
Authored: Wed Dec 30 17:22:14 2015 +0300
Committer: sboikov <sb...@gridgain.com>
Committed: Wed Dec 30 17:22:14 2015 +0300
----------------------------------------------------------------------
.../cache/IgniteCacheStoreCollectionTest.java | 163 +++++++++++++++++++
1 file changed, 163 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/dffcb834/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheStoreCollectionTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheStoreCollectionTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheStoreCollectionTest.java
new file mode 100644
index 0000000..57d57ca
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheStoreCollectionTest.java
@@ -0,0 +1,163 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.processors.cache;
+
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Map;
+import java.util.Set;
+import org.apache.ignite.IgniteCache;
+import org.apache.ignite.configuration.CacheConfiguration;
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.internal.util.typedef.F;
+import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
+
+import static org.apache.ignite.cache.CacheAtomicityMode.ATOMIC;
+import static org.apache.ignite.cache.CacheAtomicityMode.TRANSACTIONAL;
+
+/**
+ *
+ */
+public class IgniteCacheStoreCollectionTest extends GridCommonAbstractTest {
+ /** {@inheritDoc} */
+ @Override protected IgniteConfiguration getConfiguration(String gridName) throws Exception {
+ IgniteConfiguration cfg = super.getConfiguration(gridName);
+
+ CacheConfiguration<Object, Object> ccfg1 = new CacheConfiguration<>();
+ ccfg1.setName("cache1");
+ ccfg1.setAtomicityMode(ATOMIC);
+
+ CacheConfiguration<Object, Object> ccfg2 = new CacheConfiguration<>();
+ ccfg2.setName("cache2");
+ ccfg2.setAtomicityMode(TRANSACTIONAL);
+
+ cfg.setCacheConfiguration(ccfg1, ccfg2);
+
+ cfg.setMarshaller(null);
+
+ return cfg;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void beforeTestsStarted() throws Exception {
+ super.beforeTestsStarted();
+
+ startGrid(0);
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void afterTestsStopped() throws Exception {
+ stopAllGrids();
+
+ super.afterTestsStopped();
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testStoreMap() throws Exception {
+ IgniteCache<Object, Object> cache1 = ignite(0).cache("cache1");
+ IgniteCache<Object, Object> cache2 = ignite(0).cache("cache2");
+
+ checkStoreMap(cache1);
+ checkStoreMap(cache2);
+ }
+
+ /**
+ * @param cache Cache.
+ */
+ private void checkStoreMap(IgniteCache<Object, Object> cache) {
+ cache.put(1, new MyMap());
+ cache.put(2, new MyMap());
+
+ MyMap map = (MyMap)cache.get(1);
+
+ assertNotNull(map);
+
+ Map<Integer, MyMap> vals = (Map)cache.getAll(F.asSet(1, 2));
+
+ assertEquals(2, vals.size());
+ assertTrue("Unexpected value: " + vals.get(1), vals.get(1) instanceof MyMap);
+ assertTrue("Unexpected value: " + vals.get(2), vals.get(2) instanceof MyMap);
+ }
+
+ /**
+ *
+ */
+ public static class MyMap implements Map {
+ /** {@inheritDoc} */
+ @Override public int size() {
+ return 0;
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean isEmpty() {
+ return true;
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean containsKey(Object key) {
+ return false;
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean containsValue(Object val) {
+ return false;
+ }
+
+ /** {@inheritDoc} */
+ @Override public Object get(Object key) {
+ return null;
+ }
+
+ /** {@inheritDoc} */
+ @Override public Object put(Object key, Object val) {
+ return null;
+ }
+
+ /** {@inheritDoc} */
+ @Override public Object remove(Object key) {
+ return null;
+ }
+
+ /** {@inheritDoc} */
+ @Override public void putAll(Map m) {
+ // No-op.
+ }
+
+ /** {@inheritDoc} */
+ @Override public void clear() {
+ // No-op.
+ }
+
+ /** {@inheritDoc} */
+ @Override public Set keySet() {
+ return Collections.emptySet();
+ }
+
+ /** {@inheritDoc} */
+ @Override public Collection values() {
+ return Collections.emptySet();
+ }
+
+ /** {@inheritDoc} */
+ @Override public Set<Entry> entrySet() {
+ return Collections.emptySet();
+ }
+ }
+}
[4/8] ignite git commit: # master - added rdf file.
Posted by vo...@apache.org.
# master - added rdf file.
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/e661f172
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/e661f172
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/e661f172
Branch: refs/heads/ignite-2265
Commit: e661f17289c872bf34e40dc4a6001666aec1ca0b
Parents: 13e408e
Author: Dmitiry Setrakyan <ds...@gridgain.com>
Authored: Tue Dec 29 13:49:08 2015 -0800
Committer: Dmitiry Setrakyan <ds...@gridgain.com>
Committed: Tue Dec 29 13:49:08 2015 -0800
----------------------------------------------------------------------
doap_Ignite.rdf | 67 ++++++++++++++++++++++++++++++++++++++++++++++++++++
1 file changed, 67 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/e661f172/doap_Ignite.rdf
----------------------------------------------------------------------
diff --git a/doap_Ignite.rdf b/doap_Ignite.rdf
new file mode 100644
index 0000000..fb51a15
--- /dev/null
+++ b/doap_Ignite.rdf
@@ -0,0 +1,67 @@
+<?xml version="1.0"?>
+<?xml-stylesheet type="text/xsl"?>
+<rdf:RDF xml:lang="en"
+ xmlns="http://usefulinc.com/ns/doap#"
+ xmlns:rdf="http://www.w3.org/1999/02/22-rdf-syntax-ns#"
+ xmlns:asfext="http://projects.apache.org/ns/asfext#"
+ xmlns:foaf="http://xmlns.com/foaf/0.1/">
+<!--
+ Licensed to the Apache Software Foundation (ASF) under one or more
+ contributor license agreements. See the NOTICE file distributed with
+ this work for additional information regarding copyright ownership.
+ The ASF licenses this file to You under the Apache License, Version 2.0
+ (the "License"); you may not use this file except in compliance with
+ the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+-->
+ <Project rdf:about="http://ignite.apache.org">
+ <created>2014-10-01</created>
+ <license rdf:resource="http://spdx.org/licenses/Apache-2.0" />
+ <name>Apache Ignite</name>
+ <homepage rdf:resource="http://ignite.apache.org" />
+ <asfext:pmc rdf:resource="http://ignite.apache.org" />
+ <shortdesc>Apache Ignite is an In-Memory Data Fabric providing in-memory data caching, partitioning, processing, and querying components.</shortdesc>
+ <description>Apache Ignite In-Memory Data Fabric is designed to deliver uncompromised performance for a wide set of in-memory computing use cases from high performance computing, to the industry most advanced data grid, in-memory SQL, in-memory file system, streaming, and more.</description>
+ <bug-database rdf:resource="https://issues.apache.org/jira/browse/IGNITE" />
+ <mailing-list rdf:resource="https://ignite.apache.org/community/resources.html#mail-lists" />
+ <download-page rdf:resource="https://ignite.apache.org/download" />
+
+ <programming-language>Java</programming-language>
+ <programming-language>C#</programming-language>
+ <programming-language>C++</programming-language>
+ <programming-language>Node.JS</programming-language>
+ <programming-language>SQL</programming-language>
+ <programming-language>JDBC</programming-language>
+ <programming-language>ODBC</programming-language>
+
+ <category rdf:resource="http://projects.apache.org/category/big-data" />
+ <category rdf:resource="http://projects.apache.org/category/network-server" />
+ <category rdf:resource="http://projects.apache.org/category/network-client" />
+
+ <repository>
+ <GitRepository>
+ <location rdf:resource="https://git-wip-us.apache.org/repos/asf/ignite"/>
+ <browse rdf:resource="https://git-wip-us.apache.org/repos/asf/ignite"/>
+ </GitRepository>
+ </repository>
+ <maintainer>
+ <foaf:Person>
+ <foaf:name>Dmitriy Setrakyan</foaf:name>
+ <foaf:mbox rdf:resource="mailto:dsetrakyan@apache.org"/>
+ </foaf:Person>
+ </maintainer>
+ <asfext:implements><asfext:Standard>
+ <asfext:title>JCACHE - Java Temporary Caching API</asfext:title>
+ <asfext:body>JCP</asfext:body>
+ <asfext:id>107</asfext:id>
+ <asfext:url rdf:resource="https://jcp.org/en/jsr/detail?id=107"/>
+ </asfext:Standard></asfext:implements>
+ </Project>
+</rdf:RDF>
[6/8] ignite git commit: IGNITE-2228: .NET: Compute futures could be
cancelled.
Posted by vo...@apache.org.
IGNITE-2228: .NET: Compute futures could be cancelled.
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/24a78f5d
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/24a78f5d
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/24a78f5d
Branch: refs/heads/ignite-2265
Commit: 24a78f5de27908f608cf53915064164d25c0f633
Parents: e661f17
Author: Pavel Tupitsyn <pt...@gridgain.com>
Authored: Wed Dec 30 13:51:32 2015 +0300
Committer: vozerov-gridgain <vo...@gridgain.com>
Committed: Wed Dec 30 13:51:32 2015 +0300
----------------------------------------------------------------------
.../platform/PlatformAbstractTarget.java | 17 +-
.../processors/platform/PlatformTarget.java | 22 ++
.../platform/compute/PlatformCompute.java | 34 ++-
.../platform/utils/PlatformFutureUtils.java | 119 +++++++---
.../platform/utils/PlatformListenable.java | 47 ++++
.../cpp/common/include/ignite/common/exports.h | 7 +-
.../cpp/common/include/ignite/common/java.h | 13 +-
.../platforms/cpp/common/project/vs/module.def | 6 +-
modules/platforms/cpp/common/src/exports.cpp | 20 +-
modules/platforms/cpp/common/src/java.cpp | 70 +++++-
.../Compute/ComputeApiTest.cs | 20 ++
.../Apache.Ignite.Core.csproj | 2 +
.../Common/IgniteFutureCancelledException.cs | 65 ++++++
.../Apache.Ignite.Core/Compute/ICompute.cs | 233 +++++++++++++++++++
.../Impl/Common/CancelledTask.cs | 47 ++++
.../Apache.Ignite.Core/Impl/Common/Future.cs | 74 +++++-
.../Apache.Ignite.Core/Impl/Compute/Compute.cs | 157 ++++++++++++-
.../Impl/Compute/ComputeImpl.cs | 15 +-
.../Apache.Ignite.Core/Impl/ExceptionUtils.cs | 4 +
.../Apache.Ignite.Core/Impl/PlatformTarget.cs | 52 +++++
.../Impl/Unmanaged/IgniteJniNativeMethods.cs | 16 +-
.../Impl/Unmanaged/UnmanagedUtils.cs | 31 ++-
22 files changed, 995 insertions(+), 76 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/24a78f5d/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/PlatformAbstractTarget.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/PlatformAbstractTarget.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/PlatformAbstractTarget.java
index 34a2cca..7ffceef 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/PlatformAbstractTarget.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/PlatformAbstractTarget.java
@@ -24,7 +24,7 @@ import org.apache.ignite.internal.binary.BinaryRawReaderEx;
import org.apache.ignite.internal.binary.BinaryRawWriterEx;
import org.apache.ignite.internal.processors.platform.memory.PlatformMemory;
import org.apache.ignite.internal.processors.platform.memory.PlatformOutputStream;
-import org.apache.ignite.internal.processors.platform.utils.PlatformFutureUtils;
+import org.apache.ignite.internal.processors.platform.utils.*;
import org.apache.ignite.internal.util.future.IgniteFutureImpl;
import org.apache.ignite.lang.IgniteFuture;
import org.jetbrains.annotations.Nullable;
@@ -184,12 +184,23 @@ public abstract class PlatformAbstractTarget implements PlatformTarget {
/** {@inheritDoc} */
@Override public void listenFuture(final long futId, int typ) throws Exception {
- PlatformFutureUtils.listen(platformCtx, currentFutureWrapped(), futId, typ, null, this);
+ listenFutureAndGet(futId, typ);
}
/** {@inheritDoc} */
@Override public void listenFutureForOperation(final long futId, int typ, int opId) throws Exception {
- PlatformFutureUtils.listen(platformCtx, currentFutureWrapped(), futId, typ, futureWriter(opId), this);
+ listenFutureForOperationAndGet(futId, typ, opId);
+ }
+
+ /** {@inheritDoc} */
+ @Override public PlatformListenable listenFutureAndGet(final long futId, int typ) throws Exception {
+ return PlatformFutureUtils.listen(platformCtx, currentFutureWrapped(), futId, typ, null, this);
+ }
+
+ /** {@inheritDoc} */
+ @Override public PlatformListenable listenFutureForOperationAndGet(final long futId, int typ, int opId)
+ throws Exception {
+ return PlatformFutureUtils.listen(platformCtx, currentFutureWrapped(), futId, typ, futureWriter(opId), this);
}
/**
http://git-wip-us.apache.org/repos/asf/ignite/blob/24a78f5d/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/PlatformTarget.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/PlatformTarget.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/PlatformTarget.java
index bf657d1..1ebf700 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/PlatformTarget.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/PlatformTarget.java
@@ -18,6 +18,7 @@
package org.apache.ignite.internal.processors.platform;
import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.internal.processors.platform.utils.*;
import org.jetbrains.annotations.Nullable;
/**
@@ -113,4 +114,25 @@ public interface PlatformTarget {
*/
@SuppressWarnings("UnusedDeclaration")
public void listenFutureForOperation(final long futId, int typ, int opId) throws Exception;
+
+ /**
+ * Start listening for the future.
+ *
+ * @param futId Future ID.
+ * @param typ Result type.
+ * @throws IgniteCheckedException In case of failure.
+ */
+ @SuppressWarnings("UnusedDeclaration")
+ public PlatformListenable listenFutureAndGet(final long futId, int typ) throws Exception;
+
+ /**
+ * Start listening for the future for specific operation type.
+ *
+ * @param futId Future ID.
+ * @param typ Result type.
+ * @param opId Operation ID required to pick correct result writer.
+ * @throws IgniteCheckedException In case of failure.
+ */
+ @SuppressWarnings("UnusedDeclaration")
+ public PlatformListenable listenFutureForOperationAndGet(final long futId, int typ, int opId) throws Exception;
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/ignite/blob/24a78f5d/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/compute/PlatformCompute.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/compute/PlatformCompute.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/compute/PlatformCompute.java
index 9ef6b5e..1dad126 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/compute/PlatformCompute.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/compute/PlatformCompute.java
@@ -30,6 +30,7 @@ import org.apache.ignite.internal.binary.BinaryRawReaderEx;
import org.apache.ignite.internal.binary.BinaryRawWriterEx;
import org.apache.ignite.internal.processors.platform.PlatformAbstractTarget;
import org.apache.ignite.internal.processors.platform.PlatformContext;
+import org.apache.ignite.internal.processors.platform.utils.*;
import org.apache.ignite.internal.util.typedef.C1;
import org.apache.ignite.lang.IgniteFuture;
import org.apache.ignite.lang.IgniteInClosure;
@@ -75,36 +76,31 @@ public class PlatformCompute extends PlatformAbstractTarget {
}
/** {@inheritDoc} */
- @Override protected long processInStreamOutLong(int type, BinaryRawReaderEx reader) throws IgniteCheckedException {
+ @Override protected Object processInStreamOutObject(int type, BinaryRawReaderEx reader)
+ throws IgniteCheckedException {
switch (type) {
case OP_UNICAST:
- processClosures(reader.readLong(), reader, false, false);
-
- return TRUE;
+ return processClosures(reader.readLong(), reader, false, false);
case OP_BROADCAST:
- processClosures(reader.readLong(), reader, true, false);
-
- return TRUE;
+ return processClosures(reader.readLong(), reader, true, false);
case OP_AFFINITY:
- processClosures(reader.readLong(), reader, false, true);
-
- return TRUE;
+ return processClosures(reader.readLong(), reader, false, true);
default:
- return super.processInStreamOutLong(type, reader);
+ return super.processInStreamOutObject(type, reader);
}
}
/**
* Process closure execution request.
- *
- * @param taskPtr Task pointer.
+ * @param taskPtr Task pointer.
* @param reader Reader.
* @param broadcast broadcast flag.
*/
- private void processClosures(long taskPtr, BinaryRawReaderEx reader, boolean broadcast, boolean affinity) {
+ private PlatformListenable processClosures(long taskPtr, BinaryRawReaderEx reader, boolean broadcast,
+ boolean affinity) {
PlatformAbstractTask task;
int size = reader.readInt();
@@ -155,7 +151,7 @@ public class PlatformCompute extends PlatformAbstractTarget {
platformCtx.kernalContext().task().setThreadContext(TC_SUBGRID, compute.clusterGroup().nodes());
- executeNative0(task);
+ return executeNative0(task);
}
/**
@@ -194,10 +190,10 @@ public class PlatformCompute extends PlatformAbstractTarget {
* @param taskPtr Pointer to the task.
* @param topVer Topology version.
*/
- public void executeNative(long taskPtr, long topVer) {
+ public PlatformListenable executeNative(long taskPtr, long topVer) {
final PlatformFullTask task = new PlatformFullTask(platformCtx, compute, taskPtr, topVer);
- executeNative0(task);
+ return executeNative0(task);
}
/**
@@ -231,7 +227,7 @@ public class PlatformCompute extends PlatformAbstractTarget {
*
* @param task Task.
*/
- private void executeNative0(final PlatformAbstractTask task) {
+ private PlatformListenable executeNative0(final PlatformAbstractTask task) {
IgniteInternalFuture fut = compute.executeAsync(task, null);
fut.listen(new IgniteInClosure<IgniteInternalFuture>() {
@@ -248,6 +244,8 @@ public class PlatformCompute extends PlatformAbstractTarget {
}
}
});
+
+ return PlatformFutureUtils.getListenable(fut);
}
/**
http://git-wip-us.apache.org/repos/asf/ignite/blob/24a78f5d/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/utils/PlatformFutureUtils.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/utils/PlatformFutureUtils.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/utils/PlatformFutureUtils.java
index e6f28c9..7a86201 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/utils/PlatformFutureUtils.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/utils/PlatformFutureUtils.java
@@ -17,6 +17,7 @@
package org.apache.ignite.internal.processors.platform.utils;
+import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.internal.IgniteInternalFuture;
import org.apache.ignite.internal.binary.BinaryRawWriterEx;
import org.apache.ignite.internal.processors.platform.PlatformAbstractTarget;
@@ -67,10 +68,15 @@ public class PlatformFutureUtils {
* @param fut Java future.
* @param futPtr Native future pointer.
* @param typ Expected return type.
+ * @return Resulting listenable.
*/
- public static void listen(final PlatformContext ctx, IgniteInternalFuture fut, final long futPtr, final int typ,
- PlatformAbstractTarget target) {
- listen(ctx, new InternalFutureListenable(fut), futPtr, typ, null, target);
+ public static PlatformListenable listen(final PlatformContext ctx, IgniteInternalFuture fut, final long futPtr,
+ final int typ, PlatformAbstractTarget target) {
+ PlatformListenable listenable = getListenable(fut);
+
+ listen(ctx, listenable, futPtr, typ, null, target);
+
+ return listenable;
}
/**
* Listen future.
@@ -79,10 +85,15 @@ public class PlatformFutureUtils {
* @param fut Java future.
* @param futPtr Native future pointer.
* @param typ Expected return type.
+ * @return Resulting listenable.
*/
- public static void listen(final PlatformContext ctx, IgniteFuture fut, final long futPtr, final int typ,
- PlatformAbstractTarget target) {
- listen(ctx, new FutureListenable(fut), futPtr, typ, null, target);
+ public static PlatformListenable listen(final PlatformContext ctx, IgniteFuture fut, final long futPtr,
+ final int typ, PlatformAbstractTarget target) {
+ PlatformListenable listenable = getListenable(fut);
+
+ listen(ctx, listenable, futPtr, typ, null, target);
+
+ return listenable;
}
/**
@@ -93,10 +104,15 @@ public class PlatformFutureUtils {
* @param futPtr Native future pointer.
* @param typ Expected return type.
* @param writer Writer.
+ * @return Resulting listenable.
*/
- public static void listen(final PlatformContext ctx, IgniteInternalFuture fut, final long futPtr, final int typ,
- Writer writer, PlatformAbstractTarget target) {
- listen(ctx, new InternalFutureListenable(fut), futPtr, typ, writer, target);
+ public static PlatformListenable listen(final PlatformContext ctx, IgniteInternalFuture fut, final long futPtr,
+ final int typ, Writer writer, PlatformAbstractTarget target) {
+ PlatformListenable listenable = getListenable(fut);
+
+ listen(ctx, listenable, futPtr, typ, writer, target);
+
+ return listenable;
}
/**
@@ -107,10 +123,15 @@ public class PlatformFutureUtils {
* @param futPtr Native future pointer.
* @param typ Expected return type.
* @param writer Writer.
+ * @return Resulting listenable.
*/
- public static void listen(final PlatformContext ctx, IgniteFuture fut, final long futPtr, final int typ,
- Writer writer, PlatformAbstractTarget target) {
- listen(ctx, new FutureListenable(fut), futPtr, typ, writer, target);
+ public static PlatformListenable listen(final PlatformContext ctx, IgniteFuture fut, final long futPtr,
+ final int typ, Writer writer, PlatformAbstractTarget target) {
+ PlatformListenable listenable = getListenable(fut);
+
+ listen(ctx, listenable, futPtr, typ, writer, target);
+
+ return listenable;
}
/**
@@ -120,10 +141,35 @@ public class PlatformFutureUtils {
* @param fut Java future.
* @param futPtr Native future pointer.
* @param writer Writer.
+ * @return Resulting listenable.
+ */
+ public static PlatformListenable listen(final PlatformContext ctx, IgniteInternalFuture fut, final long futPtr,
+ Writer writer, PlatformAbstractTarget target) {
+ PlatformListenable listenable = getListenable(fut);
+
+ listen(ctx, listenable, futPtr, TYP_OBJ, writer, target);
+
+ return listenable;
+ }
+
+ /**
+ * Gets the listenable.
+ *
+ * @param fut Future.
+ * @return Platform listenable.
*/
- public static void listen(final PlatformContext ctx, IgniteInternalFuture fut, final long futPtr, Writer writer,
- PlatformAbstractTarget target) {
- listen(ctx, new InternalFutureListenable(fut), futPtr, TYP_OBJ, writer, target);
+ public static PlatformListenable getListenable(IgniteInternalFuture fut) {
+ return new InternalFutureListenable(fut);
+ }
+
+ /**
+ * Gets the listenable.
+ *
+ * @param fut Future.
+ * @return Platform listenable.
+ */
+ public static PlatformListenable getListenable(IgniteFuture fut) {
+ return new FutureListenable(fut);
}
/**
@@ -136,8 +182,8 @@ public class PlatformFutureUtils {
* @param writer Optional writer.
*/
@SuppressWarnings("unchecked")
- private static void listen(final PlatformContext ctx, Listenable listenable, final long futPtr, final int typ,
- @Nullable final Writer writer, final PlatformAbstractTarget target) {
+ private static void listen(final PlatformContext ctx, PlatformListenable listenable, final long futPtr, final
+ int typ, @Nullable final Writer writer, final PlatformAbstractTarget target) {
final PlatformCallbackGateway gate = ctx.gateway();
listenable.listen(new IgniteBiInClosure<Object, Throwable>() {
@@ -312,21 +358,9 @@ public class PlatformFutureUtils {
}
/**
- * Listenable entry.
- */
- private static interface Listenable {
- /**
- * Listen.
- *
- * @param lsnr Listener.
- */
- public void listen(IgniteBiInClosure<Object, Throwable> lsnr);
- }
-
- /**
* Listenable around Ignite future.
*/
- private static class FutureListenable implements Listenable {
+ private static class FutureListenable implements PlatformListenable {
/** Future. */
private final IgniteFuture fut;
@@ -358,12 +392,22 @@ public class PlatformFutureUtils {
}
});
}
+
+ /** {@inheritDoc} */
+ @Override public boolean cancel() {
+ return fut.cancel();
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean isCancelled() {
+ return fut.isCancelled();
+ }
}
/**
* Listenable around Ignite future.
*/
- private static class InternalFutureListenable implements Listenable {
+ private static class InternalFutureListenable implements PlatformListenable {
/** Future. */
private final IgniteInternalFuture fut;
@@ -392,6 +436,15 @@ public class PlatformFutureUtils {
}
});
}
- }
-}
+ /** {@inheritDoc} */
+ @Override public boolean cancel() throws IgniteCheckedException {
+ return fut.cancel();
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean isCancelled() {
+ return fut.isCancelled();
+ }
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/ignite/blob/24a78f5d/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/utils/PlatformListenable.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/utils/PlatformListenable.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/utils/PlatformListenable.java
new file mode 100644
index 0000000..223590d
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/utils/PlatformListenable.java
@@ -0,0 +1,47 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.processors.platform.utils;
+
+import org.apache.ignite.*;
+import org.apache.ignite.lang.*;
+
+/**
+ * Platform listenable.
+ */
+public interface PlatformListenable {
+ /**
+ * Listen.
+ *
+ * @param lsnr Listener.
+ */
+ public void listen(IgniteBiInClosure<Object, Throwable> lsnr);
+
+ /**
+ * Cancel this instance.
+ *
+ * @return True if canceled.
+ */
+ public boolean cancel() throws IgniteCheckedException;
+
+ /**
+ * Returns true if this listenable was canceled before completion.
+ *
+ * @return True if this listenable was canceled before completion.
+ */
+ public boolean isCancelled();
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/24a78f5d/modules/platforms/cpp/common/include/ignite/common/exports.h
----------------------------------------------------------------------
diff --git a/modules/platforms/cpp/common/include/ignite/common/exports.h b/modules/platforms/cpp/common/include/ignite/common/exports.h
index 23b9665..3eb775d 100644
--- a/modules/platforms/cpp/common/include/ignite/common/exports.h
+++ b/modules/platforms/cpp/common/include/ignite/common/exports.h
@@ -55,6 +55,8 @@ extern "C" {
void* IGNITE_CALL IgniteTargetOutObject(gcj::JniContext* ctx, void* obj, int opType);
void IGNITE_CALL IgniteTargetListenFuture(gcj::JniContext* ctx, void* obj, long long futId, int typ);
void IGNITE_CALL IgniteTargetListenFutureForOperation(gcj::JniContext* ctx, void* obj, long long futId, int typ, int opId);
+ void* IGNITE_CALL IgniteTargetListenFutureAndGet(gcj::JniContext* ctx, void* obj, long long futId, int typ);
+ void* IGNITE_CALL IgniteTargetListenFutureForOperationAndGet(gcj::JniContext* ctx, void* obj, long long futId, int typ, int opId);
int IGNITE_CALL IgniteAffinityPartitions(gcj::JniContext* ctx, void* obj);
@@ -80,7 +82,7 @@ extern "C" {
void IGNITE_CALL IgniteComputeWithNoFailover(gcj::JniContext* ctx, void* obj);
void IGNITE_CALL IgniteComputeWithTimeout(gcj::JniContext* ctx, void* obj, long long timeout);
- void IGNITE_CALL IgniteComputeExecuteNative(gcj::JniContext* ctx, void* obj, long long taskPtr, long long topVer);
+ void* IGNITE_CALL IgniteComputeExecuteNative(gcj::JniContext* ctx, void* obj, long long taskPtr, long long topVer);
void IGNITE_CALL IgniteContinuousQueryClose(gcj::JniContext* ctx, void* obj);
void* IGNITE_CALL IgniteContinuousQueryGetInitialQueryCursor(gcj::JniContext* ctx, void* obj);
@@ -153,6 +155,9 @@ extern "C" {
long long IGNITE_CALL IgniteAtomicLongCompareAndSetAndGet(gcj::JniContext* ctx, void* obj, long long expVal, long long newVal);
bool IGNITE_CALL IgniteAtomicLongIsClosed(gcj::JniContext* ctx, void* obj);
void IGNITE_CALL IgniteAtomicLongClose(gcj::JniContext* ctx, void* obj);
+
+ bool IGNITE_CALL IgniteListenableCancel(gcj::JniContext* ctx, void* obj);
+ bool IGNITE_CALL IgniteListenableIsCancelled(gcj::JniContext* ctx, void* obj);
}
#endif
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/ignite/blob/24a78f5d/modules/platforms/cpp/common/include/ignite/common/java.h
----------------------------------------------------------------------
diff --git a/modules/platforms/cpp/common/include/ignite/common/java.h b/modules/platforms/cpp/common/include/ignite/common/java.h
index 7a2165c..e629c77 100644
--- a/modules/platforms/cpp/common/include/ignite/common/java.h
+++ b/modules/platforms/cpp/common/include/ignite/common/java.h
@@ -318,6 +318,8 @@ namespace ignite
jmethodID m_PlatformTarget_inObjectStreamOutStream;
jmethodID m_PlatformTarget_listenFuture;
jmethodID m_PlatformTarget_listenFutureForOperation;
+ jmethodID m_PlatformTarget_listenFutureAndGet;
+ jmethodID m_PlatformTarget_listenFutureForOperationAndGet;
jclass c_PlatformTransactions;
jmethodID m_PlatformTransactions_txStart;
@@ -347,6 +349,10 @@ namespace ignite
jmethodID m_PlatformAtomicLong_isClosed;
jmethodID m_PlatformAtomicLong_close;
+ jclass c_PlatformListenable;
+ jmethodID m_PlatformListenable_cancel;
+ jmethodID m_PlatformListenable_isCancelled;
+
/**
* Constructor.
*/
@@ -501,6 +507,8 @@ namespace ignite
jobject TargetOutObject(jobject obj, int opType, JniErrorInfo* errInfo = NULL);
void TargetListenFuture(jobject obj, long long futId, int typ);
void TargetListenFutureForOperation(jobject obj, long long futId, int typ, int opId);
+ void* TargetListenFutureAndGet(jobject obj, long long futId, int typ);
+ void* TargetListenFutureForOperationAndGet(jobject obj, long long futId, int typ, int opId);
int AffinityPartitions(jobject obj);
@@ -526,7 +534,7 @@ namespace ignite
void ComputeWithNoFailover(jobject obj);
void ComputeWithTimeout(jobject obj, long long timeout);
- void ComputeExecuteNative(jobject obj, long long taskPtr, long long topVer);
+ void* ComputeExecuteNative(jobject obj, long long taskPtr, long long topVer);
void ContinuousQueryClose(jobject obj);
jobject ContinuousQueryGetInitialQueryCursor(jobject obj);
@@ -589,6 +597,9 @@ namespace ignite
bool AtomicLongIsClosed(jobject obj);
void AtomicLongClose(jobject obj);
+ bool ListenableCancel(jobject obj);
+ bool ListenableIsCancelled(jobject obj);
+
jobject Acquire(jobject obj);
void DestroyJvm();
http://git-wip-us.apache.org/repos/asf/ignite/blob/24a78f5d/modules/platforms/cpp/common/project/vs/module.def
----------------------------------------------------------------------
diff --git a/modules/platforms/cpp/common/project/vs/module.def b/modules/platforms/cpp/common/project/vs/module.def
index 99cec2d..3d166bd 100644
--- a/modules/platforms/cpp/common/project/vs/module.def
+++ b/modules/platforms/cpp/common/project/vs/module.def
@@ -108,4 +108,8 @@ IgniteAtomicLongGetAndDecrement @105
IgniteAtomicLongGetAndSet @106
IgniteAtomicLongCompareAndSetAndGet @107
IgniteAtomicLongIsClosed @108
-IgniteAtomicLongClose @109
\ No newline at end of file
+IgniteAtomicLongClose @109
+IgniteListenableCancel @110
+IgniteListenableIsCancelled @111
+IgniteTargetListenFutureAndGet @112
+IgniteTargetListenFutureForOperationAndGet @113
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/ignite/blob/24a78f5d/modules/platforms/cpp/common/src/exports.cpp
----------------------------------------------------------------------
diff --git a/modules/platforms/cpp/common/src/exports.cpp b/modules/platforms/cpp/common/src/exports.cpp
index 327719e..08425a4 100644
--- a/modules/platforms/cpp/common/src/exports.cpp
+++ b/modules/platforms/cpp/common/src/exports.cpp
@@ -138,6 +138,14 @@ extern "C" {
ctx->TargetListenFutureForOperation(static_cast<jobject>(obj), futId, typ, opId);
}
+ void* IGNITE_CALL IgniteTargetListenFutureAndGet(gcj::JniContext* ctx, void* obj, long long futId, int typ) {
+ return ctx->TargetListenFutureAndGet(static_cast<jobject>(obj), futId, typ);
+ }
+
+ void* IGNITE_CALL IgniteTargetListenFutureForOperationAndGet(gcj::JniContext* ctx, void* obj, long long futId, int typ, int opId) {
+ return ctx->TargetListenFutureForOperationAndGet(static_cast<jobject>(obj), futId, typ, opId);
+ }
+
int IGNITE_CALL IgniteAffinityPartitions(gcj::JniContext* ctx, void* obj) {
return ctx->AffinityPartitions(static_cast<jobject>(obj));
}
@@ -219,8 +227,8 @@ extern "C" {
ctx->ComputeWithTimeout(static_cast<jobject>(obj), timeout);
}
- void IGNITE_CALL IgniteComputeExecuteNative(gcj::JniContext* ctx, void* obj, long long taskPtr, long long topVer) {
- ctx->ComputeExecuteNative(static_cast<jobject>(obj), taskPtr, topVer);
+ void* IGNITE_CALL IgniteComputeExecuteNative(gcj::JniContext* ctx, void* obj, long long taskPtr, long long topVer) {
+ return ctx->ComputeExecuteNative(static_cast<jobject>(obj), taskPtr, topVer);
}
void IGNITE_CALL IgniteContinuousQueryClose(gcj::JniContext* ctx, void* obj) {
@@ -458,4 +466,12 @@ extern "C" {
void IGNITE_CALL IgniteAtomicLongClose(gcj::JniContext* ctx, void* obj) {
return ctx->AtomicLongClose(static_cast<jobject>(obj));
}
+
+ bool IGNITE_CALL IgniteListenableCancel(gcj::JniContext* ctx, void* obj) {
+ return ctx->ListenableCancel(static_cast<jobject>(obj));
+ }
+
+ bool IGNITE_CALL IgniteListenableIsCancelled(gcj::JniContext* ctx, void* obj) {
+ return ctx->ListenableIsCancelled(static_cast<jobject>(obj));
+ }
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/ignite/blob/24a78f5d/modules/platforms/cpp/common/src/java.cpp
----------------------------------------------------------------------
diff --git a/modules/platforms/cpp/common/src/java.cpp b/modules/platforms/cpp/common/src/java.cpp
index 64f5d9c..63deba5 100644
--- a/modules/platforms/cpp/common/src/java.cpp
+++ b/modules/platforms/cpp/common/src/java.cpp
@@ -211,6 +211,8 @@ namespace ignite
JniMethod M_PLATFORM_TARGET_OUT_OBJECT = JniMethod("outObject", "(I)Ljava/lang/Object;", false);
JniMethod M_PLATFORM_TARGET_LISTEN_FUTURE = JniMethod("listenFuture", "(JI)V", false);
JniMethod M_PLATFORM_TARGET_LISTEN_FOR_OPERATION = JniMethod("listenFutureForOperation", "(JII)V", false);
+ JniMethod M_PLATFORM_TARGET_LISTEN_FUTURE_AND_GET = JniMethod("listenFutureAndGet", "(JI)Lorg/apache/ignite/internal/processors/platform/utils/PlatformListenable;", false);
+ JniMethod M_PLATFORM_TARGET_LISTEN_FOR_OPERATION_AND_GET = JniMethod("listenFutureForOperationAndGet", "(JII)Lorg/apache/ignite/internal/processors/platform/utils/PlatformListenable;", false);
const char* C_PLATFORM_CLUSTER_GRP = "org/apache/ignite/internal/processors/platform/cluster/PlatformClusterGroup";
JniMethod M_PLATFORM_CLUSTER_GRP_FOR_OTHERS = JniMethod("forOthers", "(Lorg/apache/ignite/internal/processors/platform/cluster/PlatformClusterGroup;)Lorg/apache/ignite/internal/processors/platform/cluster/PlatformClusterGroup;", false);
@@ -227,7 +229,7 @@ namespace ignite
const char* C_PLATFORM_COMPUTE = "org/apache/ignite/internal/processors/platform/compute/PlatformCompute";
JniMethod M_PLATFORM_COMPUTE_WITH_NO_FAILOVER = JniMethod("withNoFailover", "()V", false);
JniMethod M_PLATFORM_COMPUTE_WITH_TIMEOUT = JniMethod("withTimeout", "(J)V", false);
- JniMethod M_PLATFORM_COMPUTE_EXECUTE_NATIVE = JniMethod("executeNative", "(JJ)V", false);
+ JniMethod M_PLATFORM_COMPUTE_EXECUTE_NATIVE = JniMethod("executeNative", "(JJ)Lorg/apache/ignite/internal/processors/platform/utils/PlatformListenable;", false);
const char* C_PLATFORM_CACHE = "org/apache/ignite/internal/processors/platform/cache/PlatformCache";
JniMethod M_PLATFORM_CACHE_WITH_SKIP_STORE = JniMethod("withSkipStore", "()Lorg/apache/ignite/internal/processors/platform/cache/PlatformCache;", false);
@@ -390,6 +392,10 @@ namespace ignite
JniMethod M_PLATFORM_ATOMIC_LONG_IS_CLOSED = JniMethod("isClosed", "()Z", false);
JniMethod M_PLATFORM_ATOMIC_LONG_CLOSE = JniMethod("close", "()V", false);
+ const char* C_PLATFORM_LISTENABLE = "org/apache/ignite/internal/processors/platform/utils/PlatformListenable";
+ JniMethod M_PLATFORM_LISTENABLE_CANCEL = JniMethod("cancel", "()Z", false);
+ JniMethod M_PLATFORM_LISTENABLE_IS_CANCELED = JniMethod("isCancelled", "()Z", false);
+
/* STATIC STATE. */
gcc::CriticalSection JVM_LOCK;
JniJvm JVM;
@@ -650,6 +656,8 @@ namespace ignite
m_PlatformTarget_inObjectStreamOutStream = FindMethod(env, c_PlatformTarget, M_PLATFORM_TARGET_IN_OBJECT_STREAM_OUT_STREAM);
m_PlatformTarget_listenFuture = FindMethod(env, c_PlatformTarget, M_PLATFORM_TARGET_LISTEN_FUTURE);
m_PlatformTarget_listenFutureForOperation = FindMethod(env, c_PlatformTarget, M_PLATFORM_TARGET_LISTEN_FOR_OPERATION);
+ m_PlatformTarget_listenFutureAndGet = FindMethod(env, c_PlatformTarget, M_PLATFORM_TARGET_LISTEN_FUTURE_AND_GET);
+ m_PlatformTarget_listenFutureForOperationAndGet = FindMethod(env, c_PlatformTarget, M_PLATFORM_TARGET_LISTEN_FOR_OPERATION_AND_GET);
c_PlatformTransactions = FindClass(env, C_PLATFORM_TRANSACTIONS);
m_PlatformTransactions_txStart = FindMethod(env, c_PlatformTransactions, M_PLATFORM_TRANSACTIONS_TX_START);
@@ -666,7 +674,7 @@ namespace ignite
m_PlatformUtils_reallocate = FindMethod(env, c_PlatformUtils, M_PLATFORM_UTILS_REALLOC);
m_PlatformUtils_errData = FindMethod(env, c_PlatformUtils, M_PLATFORM_UTILS_ERR_DATA);
- jclass c_PlatformAtomicLong = FindClass(env, C_PLATFORM_ATOMIC_LONG);
+ c_PlatformAtomicLong = FindClass(env, C_PLATFORM_ATOMIC_LONG);
m_PlatformAtomicLong_get = FindMethod(env, c_PlatformAtomicLong, M_PLATFORM_ATOMIC_LONG_GET);
m_PlatformAtomicLong_incrementAndGet = FindMethod(env, c_PlatformAtomicLong, M_PLATFORM_ATOMIC_LONG_INCREMENT_AND_GET);
m_PlatformAtomicLong_getAndIncrement = FindMethod(env, c_PlatformAtomicLong, M_PLATFORM_ATOMIC_LONG_GET_AND_INCREMENT);
@@ -679,6 +687,10 @@ namespace ignite
m_PlatformAtomicLong_isClosed = FindMethod(env, c_PlatformAtomicLong, M_PLATFORM_ATOMIC_LONG_IS_CLOSED);
m_PlatformAtomicLong_close = FindMethod(env, c_PlatformAtomicLong, M_PLATFORM_ATOMIC_LONG_CLOSE);
+ c_PlatformListenable = FindClass(env, C_PLATFORM_LISTENABLE);
+ m_PlatformListenable_cancel = FindMethod(env, c_PlatformListenable, M_PLATFORM_LISTENABLE_CANCEL);
+ m_PlatformListenable_isCancelled = FindMethod(env, c_PlatformListenable, M_PLATFORM_LISTENABLE_IS_CANCELED);
+
// Find utility classes which are not used from context, but are still required in other places.
CheckClass(env, C_PLATFORM_NO_CALLBACK_EXCEPTION);
}
@@ -1322,6 +1334,28 @@ namespace ignite
ExceptionCheck(env);
}
+ void* JniContext::TargetListenFutureAndGet(jobject obj, long long futId, int typ) {
+ JNIEnv* env = Attach();
+
+ jobject res = env->CallObjectMethod(obj,
+ jvm->GetMembers().m_PlatformTarget_listenFutureAndGet, futId, typ);
+
+ ExceptionCheck(env);
+
+ return LocalToGlobal(env, res);
+ }
+
+ void* JniContext::TargetListenFutureForOperationAndGet(jobject obj, long long futId, int typ, int opId) {
+ JNIEnv* env = Attach();
+
+ jobject res = env->CallObjectMethod(obj,
+ jvm->GetMembers().m_PlatformTarget_listenFutureForOperationAndGet, futId, typ, opId);
+
+ ExceptionCheck(env);
+
+ return LocalToGlobal(env, res);
+ }
+
int JniContext::AffinityPartitions(jobject obj) {
JNIEnv* env = Attach();
@@ -1517,12 +1551,15 @@ namespace ignite
ExceptionCheck(env);
}
- void JniContext::ComputeExecuteNative(jobject obj, long long taskPtr, long long topVer) {
+ void* JniContext::ComputeExecuteNative(jobject obj, long long taskPtr, long long topVer) {
JNIEnv* env = Attach();
- env->CallVoidMethod(obj, jvm->GetMembers().m_PlatformCompute_executeNative, taskPtr, topVer);
+ jobject res = env->CallObjectMethod(obj,
+ jvm->GetMembers().m_PlatformCompute_executeNative, taskPtr, topVer);
ExceptionCheck(env);
+
+ return LocalToGlobal(env, res);
}
void JniContext::ContinuousQueryClose(jobject obj) {
@@ -1536,7 +1573,8 @@ namespace ignite
jobject JniContext::ContinuousQueryGetInitialQueryCursor(jobject obj) {
JNIEnv* env = Attach();
- jobject res = env->CallObjectMethod(obj, jvm->GetMembers().m_PlatformContinuousQuery_getInitialQueryCursor);
+ jobject res = env->CallObjectMethod(obj,
+ jvm->GetMembers().m_PlatformContinuousQuery_getInitialQueryCursor);
ExceptionCheck(env);
@@ -2037,6 +2075,28 @@ namespace ignite
ExceptionCheck(env);
}
+ bool JniContext::ListenableCancel(jobject obj)
+ {
+ JNIEnv* env = Attach();
+
+ jboolean res = env->CallBooleanMethod(obj, jvm->GetMembers().m_PlatformListenable_cancel);
+
+ ExceptionCheck(env);
+
+ return res != 0;;
+ }
+
+ bool JniContext::ListenableIsCancelled(jobject obj)
+ {
+ JNIEnv* env = Attach();
+
+ jboolean res = env->CallBooleanMethod(obj, jvm->GetMembers().m_PlatformListenable_isCancelled);
+
+ ExceptionCheck(env);
+
+ return res != 0;;
+ }
+
jobject JniContext::Acquire(jobject obj)
{
if (obj) {
http://git-wip-us.apache.org/repos/asf/ignite/blob/24a78f5d/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Compute/ComputeApiTest.cs
----------------------------------------------------------------------
diff --git a/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Compute/ComputeApiTest.cs b/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Compute/ComputeApiTest.cs
index 87b7f9d..fe7d78f 100644
--- a/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Compute/ComputeApiTest.cs
+++ b/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Compute/ComputeApiTest.cs
@@ -1005,6 +1005,25 @@ namespace Apache.Ignite.Core.Tests.Compute
}
/// <summary>
+ /// Tests single action run.
+ /// </summary>
+ [Test]
+ public void TestRunActionAsyncCancel()
+ {
+ using (var cts = new CancellationTokenSource())
+ {
+ // Cancel while executing
+ var task = _grid1.GetCompute().RunAsync(new ComputeAction(), cts.Token);
+ cts.Cancel();
+ Assert.IsTrue(task.IsCanceled);
+
+ // Use cancelled token
+ task = _grid1.GetCompute().RunAsync(new ComputeAction(), cts.Token);
+ Assert.IsTrue(task.IsCanceled);
+ }
+ }
+
+ /// <summary>
/// Tests multiple actions run.
/// </summary>
[Test]
@@ -1274,6 +1293,7 @@ namespace Apache.Ignite.Core.Tests.Compute
public void Invoke()
{
+ Thread.Sleep(10);
Interlocked.Increment(ref InvokeCount);
LastNodeId = _grid.GetCluster().GetLocalNode().Id;
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/24a78f5d/modules/platforms/dotnet/Apache.Ignite.Core/Apache.Ignite.Core.csproj
----------------------------------------------------------------------
diff --git a/modules/platforms/dotnet/Apache.Ignite.Core/Apache.Ignite.Core.csproj b/modules/platforms/dotnet/Apache.Ignite.Core/Apache.Ignite.Core.csproj
index f758863..12404be 100644
--- a/modules/platforms/dotnet/Apache.Ignite.Core/Apache.Ignite.Core.csproj
+++ b/modules/platforms/dotnet/Apache.Ignite.Core/Apache.Ignite.Core.csproj
@@ -104,6 +104,7 @@
<Compile Include="Cluster\IClusterNodeFilter.cs" />
<Compile Include="Cluster\Package-Info.cs" />
<Compile Include="Common\IgniteException.cs" />
+ <Compile Include="Common\IgniteFutureCancelledException.cs" />
<Compile Include="Common\IgniteGuid.cs" />
<Compile Include="Common\Package-Info.cs" />
<Compile Include="Compute\ComputeExecutionRejectedException.cs" />
@@ -187,6 +188,7 @@
<Compile Include="Impl\Collections\MultiValueDictionary.cs" />
<Compile Include="Impl\Collections\ReadOnlyCollection.cs" />
<Compile Include="Impl\Collections\ReadOnlyDictionary.cs" />
+ <Compile Include="Impl\Common\CancelledTask.cs" />
<Compile Include="Impl\Common\Classpath.cs" />
<Compile Include="Impl\Common\CopyOnWriteConcurrentDictionary.cs" />
<Compile Include="Impl\Common\DelegateConverter.cs" />
http://git-wip-us.apache.org/repos/asf/ignite/blob/24a78f5d/modules/platforms/dotnet/Apache.Ignite.Core/Common/IgniteFutureCancelledException.cs
----------------------------------------------------------------------
diff --git a/modules/platforms/dotnet/Apache.Ignite.Core/Common/IgniteFutureCancelledException.cs b/modules/platforms/dotnet/Apache.Ignite.Core/Common/IgniteFutureCancelledException.cs
new file mode 100644
index 0000000..02433ce
--- /dev/null
+++ b/modules/platforms/dotnet/Apache.Ignite.Core/Common/IgniteFutureCancelledException.cs
@@ -0,0 +1,65 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+namespace Apache.Ignite.Core.Common
+{
+ using System;
+ using System.Runtime.Serialization;
+
+ /// <summary>
+ /// Indicates future cancellation within Ignite.
+ /// </summary>
+ public class IgniteFutureCancelledException : IgniteException
+ {
+ /// <summary>
+ /// Initializes a new instance of the <see cref="IgniteFutureCancelledException"/> class.
+ /// </summary>
+ public IgniteFutureCancelledException()
+ {
+ // No-op.
+ }
+
+ /// <summary>
+ /// Initializes a new instance of the <see cref="IgniteFutureCancelledException"/> class.
+ /// </summary>
+ /// <param name="message">The message that describes the error.</param>
+ public IgniteFutureCancelledException(string message) : base(message)
+ {
+ // No-op.
+ }
+
+ /// <summary>
+ /// Initializes a new instance of the <see cref="IgniteFutureCancelledException"/> class.
+ /// </summary>
+ /// <param name="message">The message.</param>
+ /// <param name="cause">The cause.</param>
+ public IgniteFutureCancelledException(string message, Exception cause) : base(message, cause)
+ {
+ // No-op.
+ }
+
+ /// <summary>
+ /// Initializes a new instance of the <see cref="IgniteFutureCancelledException"/> class.
+ /// </summary>
+ /// <param name="info">Serialization information.</param>
+ /// <param name="ctx">Streaming context.</param>
+ protected IgniteFutureCancelledException(SerializationInfo info, StreamingContext ctx) : base(info, ctx)
+ {
+ // No-op.
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/24a78f5d/modules/platforms/dotnet/Apache.Ignite.Core/Compute/ICompute.cs
----------------------------------------------------------------------
diff --git a/modules/platforms/dotnet/Apache.Ignite.Core/Compute/ICompute.cs b/modules/platforms/dotnet/Apache.Ignite.Core/Compute/ICompute.cs
index d818153..a677f39 100644
--- a/modules/platforms/dotnet/Apache.Ignite.Core/Compute/ICompute.cs
+++ b/modules/platforms/dotnet/Apache.Ignite.Core/Compute/ICompute.cs
@@ -19,6 +19,7 @@ namespace Apache.Ignite.Core.Compute
{
using System;
using System.Collections.Generic;
+ using System.Threading;
using System.Threading.Tasks;
using Apache.Ignite.Core.Cluster;
@@ -98,6 +99,19 @@ namespace Apache.Ignite.Core.Compute
Task<TRes> ExecuteJavaTaskAsync<TRes>(string taskName, object taskArg);
/// <summary>
+ /// Executes given Java task on the grid projection. If task for given name has not been deployed yet,
+ /// then 'taskName' will be used as task class name to auto-deploy the task.
+ /// </summary>
+ /// <typeparam name="TRes">Type of task result.</typeparam>
+ /// <param name="taskName">Java task name</param>
+ /// <param name="taskArg">Optional argument of task execution, can be null.</param>
+ /// <param name="cancellationToken">The cancellation token.</param>
+ /// <returns>
+ /// Task result.
+ /// </returns>
+ Task<TRes> ExecuteJavaTaskAsync<TRes>(string taskName, object taskArg, CancellationToken cancellationToken);
+
+ /// <summary>
/// Executes given task on the grid projection. For step-by-step explanation of task execution process
/// refer to <see cref="IComputeTask{A,T,R}"/> documentation.
/// </summary>
@@ -123,6 +137,22 @@ namespace Apache.Ignite.Core.Compute
/// <summary>
/// Executes given task on the grid projection. For step-by-step explanation of task execution process
+ /// refer to <see cref="IComputeTask{A,T,R}" /> documentation.
+ /// </summary>
+ /// <typeparam name="TArg">Argument type.</typeparam>
+ /// <typeparam name="TJobRes">Type of job result.</typeparam>
+ /// <typeparam name="TRes">Type of final task result.</typeparam>
+ /// <param name="task">Task to execute.</param>
+ /// <param name="taskArg">Optional task argument.</param>
+ /// <param name="cancellationToken">The cancellation token.</param>
+ /// <returns>
+ /// Task result.
+ /// </returns>
+ Task<TRes> ExecuteAsync<TArg, TJobRes, TRes>(IComputeTask<TArg, TJobRes, TRes> task, TArg taskArg,
+ CancellationToken cancellationToken);
+
+ /// <summary>
+ /// Executes given task on the grid projection. For step-by-step explanation of task execution process
/// refer to <see cref="IComputeTask{A,T,R}"/> documentation.
/// </summary>
/// <param name="task">Task to execute.</param>
@@ -143,6 +173,19 @@ namespace Apache.Ignite.Core.Compute
/// <summary>
/// Executes given task on the grid projection. For step-by-step explanation of task execution process
+ /// refer to <see cref="IComputeTask{A,T,R}" /> documentation.
+ /// </summary>
+ /// <typeparam name="TJobRes">Type of job result.</typeparam>
+ /// <typeparam name="TRes">Type of reduce result.</typeparam>
+ /// <param name="task">Task to execute.</param>
+ /// <param name="cancellationToken">The cancellation token.</param>
+ /// <returns>
+ /// Task result.
+ /// </returns>
+ Task<TRes> ExecuteAsync<TJobRes, TRes>(IComputeTask<TJobRes, TRes> task, CancellationToken cancellationToken);
+
+ /// <summary>
+ /// Executes given task on the grid projection. For step-by-step explanation of task execution process
/// refer to <see cref="IComputeTask{A,T,R}"/> documentation.
/// </summary>
/// <param name="taskType">Task type.</param>
@@ -167,6 +210,21 @@ namespace Apache.Ignite.Core.Compute
/// <summary>
/// Executes given task on the grid projection. For step-by-step explanation of task execution process
+ /// refer to <see cref="IComputeTask{A,T,R}" /> documentation.
+ /// </summary>
+ /// <typeparam name="TArg">Argument type.</typeparam>
+ /// <typeparam name="TJobRes">Type of job result.</typeparam>
+ /// <typeparam name="TRes">Type of reduce result.</typeparam>
+ /// <param name="taskType">Task type.</param>
+ /// <param name="taskArg">Optional task argument.</param>
+ /// <param name="cancellationToken">The cancellation token.</param>
+ /// <returns>
+ /// Task result.
+ /// </returns>
+ Task<TRes> ExecuteAsync<TArg, TJobRes, TRes>(Type taskType, TArg taskArg, CancellationToken cancellationToken);
+
+ /// <summary>
+ /// Executes given task on the grid projection. For step-by-step explanation of task execution process
/// refer to <see cref="IComputeTask{A,T,R}"/> documentation.
/// </summary>
/// <param name="taskType">Task type.</param>
@@ -186,6 +244,19 @@ namespace Apache.Ignite.Core.Compute
Task<TRes> ExecuteAsync<TJobRes, TRes>(Type taskType);
/// <summary>
+ /// Executes given task on the grid projection. For step-by-step explanation of task execution process
+ /// refer to <see cref="IComputeTask{A,T,R}" /> documentation.
+ /// </summary>
+ /// <typeparam name="TJobRes">Type of job result.</typeparam>
+ /// <typeparam name="TRes">Type of reduce result.</typeparam>
+ /// <param name="taskType">Task type.</param>
+ /// <param name="cancellationToken">The cancellation token.</param>
+ /// <returns>
+ /// Task result.
+ /// </returns>
+ Task<TRes> ExecuteAsync<TJobRes, TRes>(Type taskType, CancellationToken cancellationToken);
+
+ /// <summary>
/// Executes provided job on a node in this grid projection. The result of the
/// job execution is returned from the result closure.
/// </summary>
@@ -204,6 +275,18 @@ namespace Apache.Ignite.Core.Compute
Task<TRes> CallAsync<TRes>(IComputeFunc<TRes> clo);
/// <summary>
+ /// Executes provided job on a node in this grid projection. The result of the
+ /// job execution is returned from the result closure.
+ /// </summary>
+ /// <typeparam name="TRes">Type of job result.</typeparam>
+ /// <param name="clo">Job to execute.</param>
+ /// <param name="cancellationToken">The cancellation token.</param>
+ /// <returns>
+ /// Job result for this execution.
+ /// </returns>
+ Task<TRes> CallAsync<TRes>(IComputeFunc<TRes> clo, CancellationToken cancellationToken);
+
+ /// <summary>
/// Executes given job on the node where data for provided affinity key is located
/// (a.k.a. affinity co-location).
/// </summary>
@@ -226,6 +309,21 @@ namespace Apache.Ignite.Core.Compute
Task<TRes> AffinityCallAsync<TRes>(string cacheName, object affinityKey, IComputeFunc<TRes> clo);
/// <summary>
+ /// Executes given job on the node where data for provided affinity key is located
+ /// (a.k.a. affinity co-location).
+ /// </summary>
+ /// <typeparam name="TRes">Type of job result.</typeparam>
+ /// <param name="cacheName">Name of the cache to use for affinity co-location.</param>
+ /// <param name="affinityKey">Affinity key.</param>
+ /// <param name="clo">Job to execute.</param>
+ /// <param name="cancellationToken">The cancellation token.</param>
+ /// <returns>
+ /// Job result for this execution.
+ /// </returns>
+ Task<TRes> AffinityCallAsync<TRes>(string cacheName, object affinityKey, IComputeFunc<TRes> clo,
+ CancellationToken cancellationToken);
+
+ /// <summary>
/// Executes collection of jobs on nodes within this grid projection.
/// </summary>
/// <param name="clos">Collection of jobs to execute.</param>
@@ -249,6 +347,20 @@ namespace Apache.Ignite.Core.Compute
/// <summary>
/// Executes collection of jobs on nodes within this grid projection.
/// </summary>
+ /// <typeparam name="TFuncRes">Type of function result.</typeparam>
+ /// <typeparam name="TRes">Type of result after reduce.</typeparam>
+ /// <param name="clos">Collection of jobs to execute.</param>
+ /// <param name="reducer">Reducer to reduce all job results into one individual return value.</param>
+ /// <param name="cancellationToken">The cancellation token.</param>
+ /// <returns>
+ /// Reduced job result for this execution.
+ /// </returns>
+ Task<TRes> CallAsync<TFuncRes, TRes>(IEnumerable<IComputeFunc<TFuncRes>> clos,
+ IComputeReducer<TFuncRes, TRes> reducer, CancellationToken cancellationToken);
+
+ /// <summary>
+ /// Executes collection of jobs on nodes within this grid projection.
+ /// </summary>
/// <param name="clos">Collection of jobs to execute.</param>
/// <returns>Collection of job results for this execution.</returns>
/// <typeparam name="TRes">Type of job result.</typeparam>
@@ -263,6 +375,18 @@ namespace Apache.Ignite.Core.Compute
Task<ICollection<TRes>> CallAsync<TRes>(IEnumerable<IComputeFunc<TRes>> clos);
/// <summary>
+ /// Executes collection of jobs on nodes within this grid projection.
+ /// </summary>
+ /// <typeparam name="TRes">Type of job result.</typeparam>
+ /// <param name="clos">Collection of jobs to execute.</param>
+ /// <param name="cancellationToken">The cancellation token.</param>
+ /// <returns>
+ /// Collection of job results for this execution.
+ /// </returns>
+ Task<ICollection<TRes>> CallAsync<TRes>(IEnumerable<IComputeFunc<TRes>> clos,
+ CancellationToken cancellationToken);
+
+ /// <summary>
/// Broadcasts given job to all nodes in grid projection. Every participating node will return a job result.
/// </summary>
/// <param name="clo">Job to broadcast to all projection nodes.</param>
@@ -277,6 +401,17 @@ namespace Apache.Ignite.Core.Compute
Task<ICollection<TRes>> BroadcastAsync<TRes>(IComputeFunc<TRes> clo);
/// <summary>
+ /// Broadcasts given job to all nodes in grid projection. Every participating node will return a job result.
+ /// </summary>
+ /// <typeparam name="TRes">The type of the resource.</typeparam>
+ /// <param name="clo">Job to broadcast to all projection nodes.</param>
+ /// <param name="cancellationToken">The cancellation token.</param>
+ /// <returns>
+ /// Collection of results for this execution.
+ /// </returns>
+ Task<ICollection<TRes>> BroadcastAsync<TRes>(IComputeFunc<TRes> clo, CancellationToken cancellationToken);
+
+ /// <summary>
/// Broadcasts given closure job with passed in argument to all nodes in grid projection.
/// Every participating node will return a job result.
/// </summary>
@@ -299,6 +434,21 @@ namespace Apache.Ignite.Core.Compute
Task<ICollection<TRes>> BroadcastAsync<TArg, TRes>(IComputeFunc<TArg, TRes> clo, TArg arg);
/// <summary>
+ /// Broadcasts given closure job with passed in argument to all nodes in grid projection.
+ /// Every participating node will return a job result.
+ /// </summary>
+ /// <typeparam name="TArg">Type of argument.</typeparam>
+ /// <typeparam name="TRes">Type of job result.</typeparam>
+ /// <param name="clo">Job to broadcast to all projection nodes.</param>
+ /// <param name="arg">Job closure argument.</param>
+ /// <param name="cancellationToken">The cancellation token.</param>
+ /// <returns>
+ /// Collection of results for this execution.
+ /// </returns>
+ Task<ICollection<TRes>> BroadcastAsync<TArg, TRes>(IComputeFunc<TArg, TRes> clo, TArg arg,
+ CancellationToken cancellationToken);
+
+ /// <summary>
/// Broadcasts given job to all nodes in grid projection.
/// </summary>
/// <param name="action">Job to broadcast to all projection nodes.</param>
@@ -311,6 +461,14 @@ namespace Apache.Ignite.Core.Compute
Task BroadcastAsync(IComputeAction action);
/// <summary>
+ /// Broadcasts given job to all nodes in grid projection.
+ /// </summary>
+ /// <param name="action">Job to broadcast to all projection nodes.</param>
+ /// <param name="cancellationToken">The cancellation token.</param>
+ /// <returns>Task.</returns>
+ Task BroadcastAsync(IComputeAction action, CancellationToken cancellationToken);
+
+ /// <summary>
/// Executes provided job on a node in this grid projection.
/// </summary>
/// <param name="action">Job to execute.</param>
@@ -323,6 +481,13 @@ namespace Apache.Ignite.Core.Compute
Task RunAsync(IComputeAction action);
/// <summary>
+ /// Executes provided job on a node in this grid projection.
+ /// </summary>
+ /// <param name="action">Job to execute.</param>
+ /// <param name="cancellationToken">The cancellation token.</param>
+ Task RunAsync(IComputeAction action, CancellationToken cancellationToken);
+
+ /// <summary>
/// Executes given job on the node where data for provided affinity key is located
/// (a.k.a. affinity co-location).
/// </summary>
@@ -341,6 +506,18 @@ namespace Apache.Ignite.Core.Compute
Task AffinityRunAsync(string cacheName, object affinityKey, IComputeAction action);
/// <summary>
+ /// Executes given job on the node where data for provided affinity key is located
+ /// (a.k.a. affinity co-location).
+ /// </summary>
+ /// <param name="cacheName">Name of the cache to use for affinity co-location.</param>
+ /// <param name="affinityKey">Affinity key.</param>
+ /// <param name="action">Job to execute.</param>
+ /// <param name="cancellationToken">The cancellation token.</param>
+ /// <returns>Task.</returns>
+ Task AffinityRunAsync(string cacheName, object affinityKey, IComputeAction action,
+ CancellationToken cancellationToken);
+
+ /// <summary>
/// Executes collection of jobs on Ignite nodes within this grid projection.
/// </summary>
/// <param name="actions">Jobs to execute.</param>
@@ -353,6 +530,14 @@ namespace Apache.Ignite.Core.Compute
Task RunAsync(IEnumerable<IComputeAction> actions);
/// <summary>
+ /// Executes collection of jobs on Ignite nodes within this grid projection.
+ /// </summary>
+ /// <param name="actions">Jobs to execute.</param>
+ /// <param name="cancellationToken">The cancellation token.</param>
+ /// <returns>Task.</returns>
+ Task RunAsync(IEnumerable<IComputeAction> actions, CancellationToken cancellationToken);
+
+ /// <summary>
/// Executes provided closure job on a node in this grid projection.
/// </summary>
/// <param name="clo">Job to run.</param>
@@ -373,6 +558,19 @@ namespace Apache.Ignite.Core.Compute
Task<TRes> ApplyAsync<TArg, TRes>(IComputeFunc<TArg, TRes> clo, TArg arg);
/// <summary>
+ /// Executes provided closure job on a node in this grid projection.
+ /// </summary>
+ /// <typeparam name="TArg">Type of argument.</typeparam>
+ /// <typeparam name="TRes">Type of job result.</typeparam>
+ /// <param name="clo">Job to run.</param>
+ /// <param name="arg">Job argument.</param>
+ /// <param name="cancellationToken">The cancellation token.</param>
+ /// <returns>
+ /// Job result for this execution.
+ /// </returns>
+ Task<TRes> ApplyAsync<TArg, TRes>(IComputeFunc<TArg, TRes> clo, TArg arg, CancellationToken cancellationToken);
+
+ /// <summary>
/// Executes provided closure job on nodes within this grid projection. A new job is executed for
/// every argument in the passed in collection. The number of actual job executions will be
/// equal to size of the job arguments collection.
@@ -399,6 +597,22 @@ namespace Apache.Ignite.Core.Compute
/// <summary>
/// Executes provided closure job on nodes within this grid projection. A new job is executed for
/// every argument in the passed in collection. The number of actual job executions will be
+ /// equal to size of the job arguments collection.
+ /// </summary>
+ /// <typeparam name="TArg">Type of argument.</typeparam>
+ /// <typeparam name="TRes">Type of job result.</typeparam>
+ /// <param name="clo">Job to run.</param>
+ /// <param name="args">Job arguments.</param>
+ /// <param name="cancellationToken">The cancellation token.</param>
+ /// <returns>
+ /// Сollection of job results.
+ /// </returns>
+ Task<ICollection<TRes>> ApplyAsync<TArg, TRes>(IComputeFunc<TArg, TRes> clo, IEnumerable<TArg> args,
+ CancellationToken cancellationToken);
+
+ /// <summary>
+ /// Executes provided closure job on nodes within this grid projection. A new job is executed for
+ /// every argument in the passed in collection. The number of actual job executions will be
/// equal to size of the job arguments collection. The returned job results will be reduced
/// into an individual result by provided reducer.
/// </summary>
@@ -427,5 +641,24 @@ namespace Apache.Ignite.Core.Compute
/// <typeparam name="TRes">Type of result after reduce.</typeparam>
Task<TRes> ApplyAsync<TArg, TFuncRes, TRes>(IComputeFunc<TArg, TFuncRes> clo, IEnumerable<TArg> args,
IComputeReducer<TFuncRes, TRes> rdc);
+
+ /// <summary>
+ /// Executes provided closure job on nodes within this grid projection. A new job is executed for
+ /// every argument in the passed in collection. The number of actual job executions will be
+ /// equal to size of the job arguments collection. The returned job results will be reduced
+ /// into an individual result by provided reducer.
+ /// </summary>
+ /// <typeparam name="TArg">Type of argument.</typeparam>
+ /// <typeparam name="TFuncRes">Type of function result.</typeparam>
+ /// <typeparam name="TRes">Type of result after reduce.</typeparam>
+ /// <param name="clo">Job to run.</param>
+ /// <param name="args">Job arguments.</param>
+ /// <param name="rdc">Reducer to reduce all job results into one individual return value.</param>
+ /// <param name="cancellationToken">The cancellation token.</param>
+ /// <returns>
+ /// Reduced job result for this execution.
+ /// </returns>
+ Task<TRes> ApplyAsync<TArg, TFuncRes, TRes>(IComputeFunc<TArg, TFuncRes> clo, IEnumerable<TArg> args,
+ IComputeReducer<TFuncRes, TRes> rdc, CancellationToken cancellationToken);
}
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/24a78f5d/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Common/CancelledTask.cs
----------------------------------------------------------------------
diff --git a/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Common/CancelledTask.cs b/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Common/CancelledTask.cs
new file mode 100644
index 0000000..0a84d81
--- /dev/null
+++ b/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Common/CancelledTask.cs
@@ -0,0 +1,47 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+namespace Apache.Ignite.Core.Impl.Common
+{
+ using System.Threading.Tasks;
+
+ /// <summary>
+ /// Provides cancelled tasks of given type.
+ /// </summary>
+ internal static class CancelledTask<T>
+ {
+ /** Task source. */
+ private static readonly TaskCompletionSource<T> TaskCompletionSource;
+
+ /// <summary>
+ /// Initializes the <see cref="CancelledTask{T}"/> class.
+ /// </summary>
+ static CancelledTask()
+ {
+ TaskCompletionSource = new TaskCompletionSource<T>();
+ TaskCompletionSource.SetCanceled();
+ }
+
+ /// <summary>
+ /// Gets the cancelled task.
+ /// </summary>
+ public static Task<T> Instance
+ {
+ get { return TaskCompletionSource.Task; }
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/24a78f5d/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Common/Future.cs
----------------------------------------------------------------------
diff --git a/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Common/Future.cs b/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Common/Future.cs
index 9460be6..0325b71 100644
--- a/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Common/Future.cs
+++ b/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Common/Future.cs
@@ -18,9 +18,13 @@
namespace Apache.Ignite.Core.Impl.Common
{
using System;
+ using System.Diagnostics;
using System.Diagnostics.CodeAnalysis;
+ using System.Threading;
using System.Threading.Tasks;
+ using Apache.Ignite.Core.Common;
using Apache.Ignite.Core.Impl.Binary.IO;
+ using Apache.Ignite.Core.Impl.Unmanaged;
/// <summary>
/// Grid future implementation.
@@ -35,6 +39,9 @@ namespace Apache.Ignite.Core.Impl.Common
/** Task completion source. */
private readonly TaskCompletionSource<T> _taskCompletionSource = new TaskCompletionSource<T>();
+ /** */
+ private volatile IUnmanagedTarget _unmanagedTarget;
+
/// <summary>
/// Constructor.
/// </summary>
@@ -44,7 +51,9 @@ namespace Apache.Ignite.Core.Impl.Common
_converter = converter;
}
- /** <inheritdoc/> */
+ /// <summary>
+ /// Gets the result.
+ /// </summary>
public T Get()
{
try
@@ -57,12 +66,28 @@ namespace Apache.Ignite.Core.Impl.Common
}
}
- /** <inheritdoc/> */
+ /// <summary>
+ /// Gets the task.
+ /// </summary>
public Task<T> Task
{
get { return _taskCompletionSource.Task; }
}
+ /// <summary>
+ /// Gets the task with cancellation.
+ /// </summary>
+ /// <param name="cancellationToken">The cancellation token.</param>
+ public Task<T> GetTask(CancellationToken cancellationToken)
+ {
+ Debug.Assert(_unmanagedTarget != null);
+
+ // OnTokenCancel will fire even if cancellationToken is already cancelled.
+ cancellationToken.Register(OnTokenCancel);
+
+ return Task;
+ }
+
/** <inheritdoc /> */
[SuppressMessage("Microsoft.Design", "CA1031:DoNotCatchGeneralExceptionTypes")]
public void OnResult(IBinaryStream stream)
@@ -80,7 +105,10 @@ namespace Apache.Ignite.Core.Impl.Common
/** <inheritdoc /> */
public void OnError(Exception err)
{
- _taskCompletionSource.TrySetException(err);
+ if (err is IgniteFutureCancelledException)
+ _taskCompletionSource.TrySetCanceled();
+ else
+ _taskCompletionSource.TrySetException(err);
}
/** <inheritdoc /> */
@@ -124,5 +152,45 @@ namespace Apache.Ignite.Core.Impl.Common
else
OnResult(res);
}
+
+ /// <summary>
+ /// Sets unmanaged future target for cancellation.
+ /// </summary>
+ internal void SetTarget(IUnmanagedTarget target)
+ {
+ Debug.Assert(target != null);
+
+ _unmanagedTarget = target;
+ }
+
+ /// <summary>
+ /// Cancels this instance.
+ /// </summary>
+ internal bool Cancel()
+ {
+ if (_unmanagedTarget == null)
+ return false;
+
+ return UnmanagedUtils.ListenableCancel(_unmanagedTarget);
+ }
+
+ /// <summary>
+ /// Determines whether this instance is cancelled.
+ /// </summary>
+ internal bool IsCancelled()
+ {
+ if (_unmanagedTarget == null)
+ return false;
+
+ return UnmanagedUtils.ListenableIsCancelled(_unmanagedTarget);
+ }
+
+ /// <summary>
+ /// Called when token cancellation occurs.
+ /// </summary>
+ private void OnTokenCancel()
+ {
+ Cancel();
+ }
}
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/24a78f5d/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Compute/Compute.cs
----------------------------------------------------------------------
diff --git a/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Compute/Compute.cs b/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Compute/Compute.cs
index 0f8fd33..300e944 100644
--- a/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Compute/Compute.cs
+++ b/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Compute/Compute.cs
@@ -20,9 +20,11 @@ namespace Apache.Ignite.Core.Impl.Compute
using System;
using System.Collections.Generic;
using System.Diagnostics;
+ using System.Threading;
using System.Threading.Tasks;
using Apache.Ignite.Core.Cluster;
using Apache.Ignite.Core.Compute;
+ using Apache.Ignite.Core.Impl.Common;
/// <summary>
/// Synchronous Compute facade.
@@ -86,6 +88,14 @@ namespace Apache.Ignite.Core.Impl.Compute
}
/** <inheritDoc /> */
+ public Task<TRes> ExecuteJavaTaskAsync<TRes>(string taskName, object taskArg,
+ CancellationToken cancellationToken)
+ {
+ return GetTaskIfAlreadyCancelled<TRes>(cancellationToken) ??
+ _compute.ExecuteJavaTaskAsync<TRes>(taskName, taskArg).GetTask(cancellationToken);
+ }
+
+ /** <inheritDoc /> */
public TReduceRes Execute<TArg, TJobRes, TReduceRes>(IComputeTask<TArg, TJobRes, TReduceRes> task, TArg taskArg)
{
return _compute.Execute(task, taskArg).Get();
@@ -98,6 +108,14 @@ namespace Apache.Ignite.Core.Impl.Compute
}
/** <inheritDoc /> */
+ public Task<TRes> ExecuteAsync<TArg, TJobRes, TRes>(IComputeTask<TArg, TJobRes, TRes> task, TArg taskArg,
+ CancellationToken cancellationToken)
+ {
+ return GetTaskIfAlreadyCancelled<TRes>(cancellationToken) ??
+ _compute.Execute(task, taskArg).GetTask(cancellationToken);
+ }
+
+ /** <inheritDoc /> */
public TJobRes Execute<TArg, TJobRes>(IComputeTask<TArg, TJobRes> task)
{
return _compute.Execute(task, null).Get();
@@ -110,6 +128,14 @@ namespace Apache.Ignite.Core.Impl.Compute
}
/** <inheritDoc /> */
+ public Task<TRes> ExecuteAsync<TJobRes, TRes>(IComputeTask<TJobRes, TRes> task,
+ CancellationToken cancellationToken)
+ {
+ return GetTaskIfAlreadyCancelled<TRes>(cancellationToken) ??
+ _compute.Execute(task, null).GetTask(cancellationToken);
+ }
+
+ /** <inheritDoc /> */
public TReduceRes Execute<TArg, TJobRes, TReduceRes>(Type taskType, TArg taskArg)
{
return _compute.Execute<TArg, TJobRes, TReduceRes>(taskType, taskArg).Get();
@@ -122,6 +148,14 @@ namespace Apache.Ignite.Core.Impl.Compute
}
/** <inheritDoc /> */
+ public Task<TReduceRes> ExecuteAsync<TArg, TJobRes, TReduceRes>(Type taskType, TArg taskArg,
+ CancellationToken cancellationToken)
+ {
+ return GetTaskIfAlreadyCancelled<TReduceRes>(cancellationToken) ??
+ _compute.Execute<TArg, TJobRes, TReduceRes>(taskType, taskArg).GetTask(cancellationToken);
+ }
+
+ /** <inheritDoc /> */
public TReduceRes Execute<TArg, TReduceRes>(Type taskType)
{
return _compute.Execute<object, TArg, TReduceRes>(taskType, null).Get();
@@ -134,6 +168,13 @@ namespace Apache.Ignite.Core.Impl.Compute
}
/** <inheritDoc /> */
+ public Task<TReduceRes> ExecuteAsync<TArg, TReduceRes>(Type taskType, CancellationToken cancellationToken)
+ {
+ return GetTaskIfAlreadyCancelled<TReduceRes>(cancellationToken) ??
+ _compute.Execute<object, TArg, TReduceRes>(taskType, null).GetTask(cancellationToken);
+ }
+
+ /** <inheritDoc /> */
public TJobRes Call<TJobRes>(IComputeFunc<TJobRes> clo)
{
return _compute.Execute(clo).Get();
@@ -146,6 +187,13 @@ namespace Apache.Ignite.Core.Impl.Compute
}
/** <inheritDoc /> */
+ public Task<TRes> CallAsync<TRes>(IComputeFunc<TRes> clo, CancellationToken cancellationToken)
+ {
+ return GetTaskIfAlreadyCancelled<TRes>(cancellationToken) ??
+ _compute.Execute(clo).GetTask(cancellationToken);
+ }
+
+ /** <inheritDoc /> */
public TJobRes AffinityCall<TJobRes>(string cacheName, object affinityKey, IComputeFunc<TJobRes> clo)
{
return _compute.AffinityCall(cacheName, affinityKey, clo).Get();
@@ -158,18 +206,35 @@ namespace Apache.Ignite.Core.Impl.Compute
}
/** <inheritDoc /> */
+ public Task<TRes> AffinityCallAsync<TRes>(string cacheName, object affinityKey, IComputeFunc<TRes> clo,
+ CancellationToken cancellationToken)
+ {
+ return GetTaskIfAlreadyCancelled<TRes>(cancellationToken) ??
+ _compute.AffinityCall(cacheName, affinityKey, clo).GetTask(cancellationToken);
+ }
+
+ /** <inheritDoc /> */
public TJobRes Call<TJobRes>(Func<TJobRes> func)
{
return _compute.Execute(func).Get();
}
/** <inheritDoc /> */
- public Task<TRes> CallAsync<TFuncRes, TRes>(IEnumerable<IComputeFunc<TFuncRes>> clos, IComputeReducer<TFuncRes, TRes> reducer)
+ public Task<TRes> CallAsync<TFuncRes, TRes>(IEnumerable<IComputeFunc<TFuncRes>> clos,
+ IComputeReducer<TFuncRes, TRes> reducer)
{
return _compute.Execute(clos, reducer).Task;
}
/** <inheritDoc /> */
+ public Task<TRes> CallAsync<TFuncRes, TRes>(IEnumerable<IComputeFunc<TFuncRes>> clos,
+ IComputeReducer<TFuncRes, TRes> reducer, CancellationToken cancellationToken)
+ {
+ return GetTaskIfAlreadyCancelled<TRes>(cancellationToken) ??
+ _compute.Execute(clos, reducer).GetTask(cancellationToken);
+ }
+
+ /** <inheritDoc /> */
public ICollection<TJobRes> Call<TJobRes>(IEnumerable<IComputeFunc<TJobRes>> clos)
{
return _compute.Execute(clos).Get();
@@ -182,6 +247,14 @@ namespace Apache.Ignite.Core.Impl.Compute
}
/** <inheritDoc /> */
+ public Task<ICollection<TRes>> CallAsync<TRes>(IEnumerable<IComputeFunc<TRes>> clos,
+ CancellationToken cancellationToken)
+ {
+ return GetTaskIfAlreadyCancelled<ICollection<TRes>>(cancellationToken) ??
+ _compute.Execute(clos).GetTask(cancellationToken);
+ }
+
+ /** <inheritDoc /> */
public TReduceRes Call<TJobRes, TReduceRes>(IEnumerable<IComputeFunc<TJobRes>> clos,
IComputeReducer<TJobRes, TReduceRes> reducer)
{
@@ -201,6 +274,13 @@ namespace Apache.Ignite.Core.Impl.Compute
}
/** <inheritDoc /> */
+ public Task<ICollection<TRes>> BroadcastAsync<TRes>(IComputeFunc<TRes> clo, CancellationToken cancellationToken)
+ {
+ return GetTaskIfAlreadyCancelled<ICollection<TRes>>(cancellationToken) ??
+ _compute.Broadcast(clo).GetTask(cancellationToken);
+ }
+
+ /** <inheritDoc /> */
public ICollection<TJobRes> Broadcast<T, TJobRes>(IComputeFunc<T, TJobRes> clo, T arg)
{
return _compute.Broadcast(clo, arg).Get();
@@ -213,6 +293,14 @@ namespace Apache.Ignite.Core.Impl.Compute
}
/** <inheritDoc /> */
+ public Task<ICollection<TRes>> BroadcastAsync<TArg, TRes>(IComputeFunc<TArg, TRes> clo, TArg arg,
+ CancellationToken cancellationToken)
+ {
+ return GetTaskIfAlreadyCancelled<ICollection<TRes>>(cancellationToken) ??
+ _compute.Broadcast(clo, arg).GetTask(cancellationToken);
+ }
+
+ /** <inheritDoc /> */
public void Broadcast(IComputeAction action)
{
_compute.Broadcast(action).Get();
@@ -225,6 +313,13 @@ namespace Apache.Ignite.Core.Impl.Compute
}
/** <inheritDoc /> */
+ public Task BroadcastAsync(IComputeAction action, CancellationToken cancellationToken)
+ {
+ return GetTaskIfAlreadyCancelled<object>(cancellationToken) ??
+ _compute.Broadcast(action).GetTask(cancellationToken);
+ }
+
+ /** <inheritDoc /> */
public void Run(IComputeAction action)
{
_compute.Run(action).Get();
@@ -237,6 +332,13 @@ namespace Apache.Ignite.Core.Impl.Compute
}
/** <inheritDoc /> */
+ public Task RunAsync(IComputeAction action, CancellationToken cancellationToken)
+ {
+ return GetTaskIfAlreadyCancelled<object>(cancellationToken) ??
+ _compute.Run(action).GetTask(cancellationToken);
+ }
+
+ /** <inheritDoc /> */
public void AffinityRun(string cacheName, object affinityKey, IComputeAction action)
{
_compute.AffinityRun(cacheName, affinityKey, action).Get();
@@ -249,6 +351,14 @@ namespace Apache.Ignite.Core.Impl.Compute
}
/** <inheritDoc /> */
+ public Task AffinityRunAsync(string cacheName, object affinityKey, IComputeAction action,
+ CancellationToken cancellationToken)
+ {
+ return GetTaskIfAlreadyCancelled<object>(cancellationToken) ??
+ _compute.AffinityRun(cacheName, affinityKey, action).GetTask(cancellationToken);
+ }
+
+ /** <inheritDoc /> */
public void Run(IEnumerable<IComputeAction> actions)
{
_compute.Run(actions).Get();
@@ -261,6 +371,13 @@ namespace Apache.Ignite.Core.Impl.Compute
}
/** <inheritDoc /> */
+ public Task RunAsync(IEnumerable<IComputeAction> actions, CancellationToken cancellationToken)
+ {
+ return GetTaskIfAlreadyCancelled<object>(cancellationToken) ??
+ _compute.Run(actions).GetTask(cancellationToken);
+ }
+
+ /** <inheritDoc /> */
public TJobRes Apply<TArg, TJobRes>(IComputeFunc<TArg, TJobRes> clo, TArg arg)
{
return _compute.Apply(clo, arg).Get();
@@ -273,6 +390,14 @@ namespace Apache.Ignite.Core.Impl.Compute
}
/** <inheritDoc /> */
+ public Task<TRes> ApplyAsync<TArg, TRes>(IComputeFunc<TArg, TRes> clo, TArg arg,
+ CancellationToken cancellationToken)
+ {
+ return GetTaskIfAlreadyCancelled<TRes>(cancellationToken) ??
+ _compute.Apply(clo, arg).GetTask(cancellationToken);
+ }
+
+ /** <inheritDoc /> */
public ICollection<TJobRes> Apply<TArg, TJobRes>(IComputeFunc<TArg, TJobRes> clo, IEnumerable<TArg> args)
{
return _compute.Apply(clo, args).Get();
@@ -285,6 +410,14 @@ namespace Apache.Ignite.Core.Impl.Compute
}
/** <inheritDoc /> */
+ public Task<ICollection<TRes>> ApplyAsync<TArg, TRes>(IComputeFunc<TArg, TRes> clo, IEnumerable<TArg> args,
+ CancellationToken cancellationToken)
+ {
+ return GetTaskIfAlreadyCancelled<ICollection<TRes>>(cancellationToken) ??
+ _compute.Apply(clo, args).GetTask(cancellationToken);
+ }
+
+ /** <inheritDoc /> */
public TReduceRes Apply<TArg, TJobRes, TReduceRes>(IComputeFunc<TArg, TJobRes> clo,
IEnumerable<TArg> args, IComputeReducer<TJobRes, TReduceRes> rdc)
{
@@ -292,9 +425,29 @@ namespace Apache.Ignite.Core.Impl.Compute
}
/** <inheritDoc /> */
- public Task<TRes> ApplyAsync<TArg, TFuncRes, TRes>(IComputeFunc<TArg, TFuncRes> clo, IEnumerable<TArg> args, IComputeReducer<TFuncRes, TRes> rdc)
+ public Task<TRes> ApplyAsync<TArg, TFuncRes, TRes>(IComputeFunc<TArg, TFuncRes> clo, IEnumerable<TArg> args,
+ IComputeReducer<TFuncRes, TRes> rdc)
{
return _compute.Apply(clo, args, rdc).Task;
}
+
+ /** <inheritDoc /> */
+ public Task<TRes> ApplyAsync<TArg, TFuncRes, TRes>(IComputeFunc<TArg, TFuncRes> clo, IEnumerable<TArg> args,
+ IComputeReducer<TFuncRes, TRes> rdc, CancellationToken cancellationToken)
+ {
+ return GetTaskIfAlreadyCancelled<TRes>(cancellationToken) ??
+ _compute.Apply(clo, args, rdc).GetTask(cancellationToken);
+ }
+
+ /// <summary>
+ /// Gets the cancelled task if specified token is cancelled.
+ /// </summary>
+ private static Task<T> GetTaskIfAlreadyCancelled<T>(CancellationToken cancellationToken)
+ {
+ if (cancellationToken.IsCancellationRequested)
+ return CancelledTask<T>.Instance;
+
+ return null;
+ }
}
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/ignite/blob/24a78f5d/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Compute/ComputeImpl.cs
----------------------------------------------------------------------
diff --git a/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Compute/ComputeImpl.cs b/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Compute/ComputeImpl.cs
index b44b2ee..86dee30 100644
--- a/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Compute/ComputeImpl.cs
+++ b/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Compute/ComputeImpl.cs
@@ -165,7 +165,8 @@ namespace Apache.Ignite.Core.Impl.Compute
WriteTask(writer, taskName, taskArg, nodes);
}, input =>
{
- fut = GetFuture<TReduceRes>((futId, futTyp) => UU.TargetListenFuture(Target, futId, futTyp), _keepBinary.Value);
+ fut = GetFuture<TReduceRes>((futId, futTyp) =>
+ UU.TargetListenFutureAndGet(Target, futId, futTyp), _keepBinary.Value);
});
return fut;
@@ -192,9 +193,13 @@ namespace Apache.Ignite.Core.Impl.Compute
long ptr = Marshaller.Ignite.HandleRegistry.Allocate(holder);
- UU.ComputeExecuteNative(Target, ptr, _prj.TopologyVersion);
+ var futTarget = UU.ComputeExecuteNative(Target, ptr, _prj.TopologyVersion);
- return holder.Future;
+ var future = holder.Future;
+
+ future.SetTarget(futTarget);
+
+ return future;
}
/// <summary>
@@ -522,7 +527,7 @@ namespace Apache.Ignite.Core.Impl.Compute
try
{
- DoOutOp(opId, writer =>
+ var futTarget = DoOutOpObject(opId, writer =>
{
writer.WriteLong(taskHandle);
@@ -546,6 +551,8 @@ namespace Apache.Ignite.Core.Impl.Compute
if (writeAction != null)
writeAction(writer);
});
+
+ holder.Future.SetTarget(futTarget);
}
catch (Exception e)
{
http://git-wip-us.apache.org/repos/asf/ignite/blob/24a78f5d/modules/platforms/dotnet/Apache.Ignite.Core/Impl/ExceptionUtils.cs
----------------------------------------------------------------------
diff --git a/modules/platforms/dotnet/Apache.Ignite.Core/Impl/ExceptionUtils.cs b/modules/platforms/dotnet/Apache.Ignite.Core/Impl/ExceptionUtils.cs
index e0735e1..665d37e 100644
--- a/modules/platforms/dotnet/Apache.Ignite.Core/Impl/ExceptionUtils.cs
+++ b/modules/platforms/dotnet/Apache.Ignite.Core/Impl/ExceptionUtils.cs
@@ -95,6 +95,10 @@ namespace Apache.Ignite.Core.Impl
// Security exceptions.
EXS["org.apache.ignite.IgniteAuthenticationException"] = m => new SecurityException(m);
EXS["org.apache.ignite.plugin.security.GridSecurityException"] = m => new SecurityException(m);
+
+ // Future exceptions
+ EXS["org.apache.ignite.lang.IgniteFutureCancelledException"] = m => new IgniteFutureCancelledException(m);
+ EXS["org.apache.ignite.internal.IgniteFutureCancelledCheckedException"] = m => new IgniteFutureCancelledException(m);
}
/// <summary>
[8/8] ignite git commit: Merge branch 'master' into ignite-2265
Posted by vo...@apache.org.
Merge branch 'master' into ignite-2265
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/5c1cd29c
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/5c1cd29c
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/5c1cd29c
Branch: refs/heads/ignite-2265
Commit: 5c1cd29c5bedba24972ab242d177e26e682e4592
Parents: d4d7c24 dffcb83
Author: vozerov-gridgain <vo...@gridgain.com>
Authored: Sat Jan 2 21:52:57 2016 +0300
Committer: vozerov-gridgain <vo...@gridgain.com>
Committed: Sat Jan 2 21:52:57 2016 +0300
----------------------------------------------------------------------
doap_Ignite.rdf | 67 ++++++
.../version/GridCachePlainVersionedEntry.java | 2 +-
.../platform/PlatformAbstractTarget.java | 17 +-
.../processors/platform/PlatformTarget.java | 22 ++
.../platform/compute/PlatformCompute.java | 34 ++-
.../platform/utils/PlatformFutureUtils.java | 119 +++++++---
.../platform/utils/PlatformListenable.java | 47 ++++
.../ignite/internal/util/IgniteUtils.java | 40 ++++
.../ignite/internal/util/nio/GridNioServer.java | 16 +-
.../cache/IgniteCacheStoreCollectionTest.java | 163 +++++++++++++
.../cpp/common/include/ignite/common/exports.h | 7 +-
.../cpp/common/include/ignite/common/java.h | 13 +-
.../platforms/cpp/common/project/vs/module.def | 6 +-
modules/platforms/cpp/common/src/exports.cpp | 20 +-
modules/platforms/cpp/common/src/java.cpp | 70 +++++-
.../Compute/ComputeApiTest.cs | 20 ++
.../Apache.Ignite.Core.csproj | 2 +
.../Common/IgniteFutureCancelledException.cs | 65 ++++++
.../Apache.Ignite.Core/Compute/ICompute.cs | 233 +++++++++++++++++++
.../Impl/Common/CancelledTask.cs | 47 ++++
.../Apache.Ignite.Core/Impl/Common/Future.cs | 74 +++++-
.../Apache.Ignite.Core/Impl/Compute/Compute.cs | 157 ++++++++++++-
.../Impl/Compute/ComputeImpl.cs | 15 +-
.../Apache.Ignite.Core/Impl/ExceptionUtils.cs | 4 +
.../Apache.Ignite.Core/Impl/PlatformTarget.cs | 52 +++++
.../Impl/Unmanaged/IgniteJniNativeMethods.cs | 16 +-
.../Impl/Unmanaged/UnmanagedUtils.cs | 31 ++-
27 files changed, 1276 insertions(+), 83 deletions(-)
----------------------------------------------------------------------
[5/8] ignite git commit: IGNITE-2228: .NET: Compute futures could be
cancelled.
Posted by vo...@apache.org.
http://git-wip-us.apache.org/repos/asf/ignite/blob/24a78f5d/modules/platforms/dotnet/Apache.Ignite.Core/Impl/PlatformTarget.cs
----------------------------------------------------------------------
diff --git a/modules/platforms/dotnet/Apache.Ignite.Core/Impl/PlatformTarget.cs b/modules/platforms/dotnet/Apache.Ignite.Core/Impl/PlatformTarget.cs
index 4a4f93b..0472ce4 100644
--- a/modules/platforms/dotnet/Apache.Ignite.Core/Impl/PlatformTarget.cs
+++ b/modules/platforms/dotnet/Apache.Ignite.Core/Impl/PlatformTarget.cs
@@ -22,6 +22,7 @@ namespace Apache.Ignite.Core.Impl
using System.Diagnostics;
using System.Diagnostics.CodeAnalysis;
using System.IO;
+ using System.Threading;
using System.Threading.Tasks;
using Apache.Ignite.Core.Impl.Binary;
using Apache.Ignite.Core.Impl.Binary.IO;
@@ -289,6 +290,26 @@ namespace Apache.Ignite.Core.Impl
}
/// <summary>
+ /// Perform out operation.
+ /// </summary>
+ /// <param name="type">Operation type.</param>
+ /// <param name="action">Action to be performed on the stream.</param>
+ /// <returns></returns>
+ protected IUnmanagedTarget DoOutOpObject(int type, Action<BinaryWriter> action)
+ {
+ using (var stream = IgniteManager.Memory.Allocate().GetStream())
+ {
+ var writer = _marsh.StartMarshal(stream);
+
+ action(writer);
+
+ FinishMarshal(writer);
+
+ return UU.TargetInStreamOutObject(_target, type, stream.SynchronizeOutput());
+ }
+ }
+
+ /// <summary>
/// Perform simple output operation accepting single argument.
/// </summary>
/// <param name="type">Operation type.</param>
@@ -633,6 +654,37 @@ namespace Apache.Ignite.Core.Impl
/// <param name="keepBinary">Keep binary flag, only applicable to object futures. False by default.</param>
/// <param name="convertFunc">The function to read future result from stream.</param>
/// <returns>Created future.</returns>
+ protected Future<T> GetFuture<T>(Func<long, int, IUnmanagedTarget> listenAction, bool keepBinary = false,
+ Func<BinaryReader, T> convertFunc = null)
+ {
+ var futType = FutureType.Object;
+
+ var type = typeof(T);
+
+ if (type.IsPrimitive)
+ IgniteFutureTypeMap.TryGetValue(type, out futType);
+
+ var fut = convertFunc == null && futType != FutureType.Object
+ ? new Future<T>()
+ : new Future<T>(new FutureConverter<T>(_marsh, keepBinary, convertFunc));
+
+ var futHnd = _marsh.Ignite.HandleRegistry.Allocate(fut);
+
+ var futTarget = listenAction(futHnd, (int) futType);
+
+ fut.SetTarget(futTarget);
+
+ return fut;
+ }
+
+ /// <summary>
+ /// Creates a future and starts listening.
+ /// </summary>
+ /// <typeparam name="T">Future result type</typeparam>
+ /// <param name="listenAction">The listen action.</param>
+ /// <param name="keepBinary">Keep binary flag, only applicable to object futures. False by default.</param>
+ /// <param name="convertFunc">The function to read future result from stream.</param>
+ /// <returns>Created future.</returns>
protected Future<T> GetFuture<T>(Action<long, int> listenAction, bool keepBinary = false,
Func<BinaryReader, T> convertFunc = null)
{
http://git-wip-us.apache.org/repos/asf/ignite/blob/24a78f5d/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Unmanaged/IgniteJniNativeMethods.cs
----------------------------------------------------------------------
diff --git a/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Unmanaged/IgniteJniNativeMethods.cs b/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Unmanaged/IgniteJniNativeMethods.cs
index 860e703..5e54a4c 100644
--- a/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Unmanaged/IgniteJniNativeMethods.cs
+++ b/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Unmanaged/IgniteJniNativeMethods.cs
@@ -113,6 +113,12 @@ namespace Apache.Ignite.Core.Impl.Unmanaged
[DllImport(IgniteUtils.FileIgniteJniDll, EntryPoint = "IgniteTargetListenFutureForOperation")]
public static extern void TargetListenFutForOp(void* ctx, void* target, long futId, int typ, int opId);
+ [DllImport(IgniteUtils.FileIgniteJniDll, EntryPoint = "IgniteTargetListenFutureAndGet")]
+ public static extern void* TargetListenFutAndGet(void* ctx, void* target, long futId, int typ);
+
+ [DllImport(IgniteUtils.FileIgniteJniDll, EntryPoint = "IgniteTargetListenFutureForOperationAndGet")]
+ public static extern void* TargetListenFutForOpAndGet(void* ctx, void* target, long futId, int typ, int opId);
+
[DllImport(IgniteUtils.FileIgniteJniDll, EntryPoint = "IgniteAffinityPartitions")]
public static extern int AffinityParts(void* ctx, void* target);
@@ -178,7 +184,7 @@ namespace Apache.Ignite.Core.Impl.Unmanaged
public static extern void ComputeWithTimeout(void* ctx, void* target, long timeout);
[DllImport(IgniteUtils.FileIgniteJniDll, EntryPoint = "IgniteComputeExecuteNative")]
- public static extern void ComputeExecuteNative(void* ctx, void* target, long taskPtr, long topVer);
+ public static extern void* ComputeExecuteNative(void* ctx, void* target, long taskPtr, long topVer);
[DllImport(IgniteUtils.FileIgniteJniDll, EntryPoint = "IgniteContinuousQueryClose")]
public static extern void ContinuousQryClose(void* ctx, void* target);
@@ -354,5 +360,13 @@ namespace Apache.Ignite.Core.Impl.Unmanaged
[DllImport(IgniteUtils.FileIgniteJniDll, EntryPoint = "IgniteAtomicLongClose")]
public static extern void AtomicLongClose(void* ctx, void* target);
+
+ [DllImport(IgniteUtils.FileIgniteJniDll, EntryPoint = "IgniteListenableCancel")]
+ [return: MarshalAs(UnmanagedType.U1)]
+ public static extern bool ListenableCancel(void* ctx, void* target);
+
+ [DllImport(IgniteUtils.FileIgniteJniDll, EntryPoint = "IgniteListenableIsCancelled")]
+ [return: MarshalAs(UnmanagedType.U1)]
+ public static extern bool ListenableIsCancelled(void* ctx, void* target);
}
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/ignite/blob/24a78f5d/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Unmanaged/UnmanagedUtils.cs
----------------------------------------------------------------------
diff --git a/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Unmanaged/UnmanagedUtils.cs b/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Unmanaged/UnmanagedUtils.cs
index 56a184d..4c8f1dc 100644
--- a/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Unmanaged/UnmanagedUtils.cs
+++ b/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Unmanaged/UnmanagedUtils.cs
@@ -305,6 +305,21 @@ namespace Apache.Ignite.Core.Impl.Unmanaged
JNI.TargetListenFutForOp(target.Context, target.Target, futId, typ, opId);
}
+ internal static IUnmanagedTarget TargetListenFutureAndGet(IUnmanagedTarget target, long futId, int typ)
+ {
+ var res = JNI.TargetListenFutAndGet(target.Context, target.Target, futId, typ);
+
+ return target.ChangeTarget(res);
+ }
+
+ internal static IUnmanagedTarget TargetListenFutureForOperationAndGet(IUnmanagedTarget target, long futId,
+ int typ, int opId)
+ {
+ var res = JNI.TargetListenFutForOpAndGet(target.Context, target.Target, futId, typ, opId);
+
+ return target.ChangeTarget(res);
+ }
+
#endregion
#region NATIVE METHODS: AFFINITY
@@ -440,9 +455,11 @@ namespace Apache.Ignite.Core.Impl.Unmanaged
JNI.ComputeWithTimeout(target.Context, target.Target, timeout);
}
- internal static void ComputeExecuteNative(IUnmanagedTarget target, long taskPtr, long topVer)
+ internal static IUnmanagedTarget ComputeExecuteNative(IUnmanagedTarget target, long taskPtr, long topVer)
{
- JNI.ComputeExecuteNative(target.Context, target.Target, taskPtr, topVer);
+ void* res = JNI.ComputeExecuteNative(target.Context, target.Target, taskPtr, topVer);
+
+ return target.ChangeTarget(res);
}
#endregion
@@ -816,6 +833,16 @@ namespace Apache.Ignite.Core.Impl.Unmanaged
JNI.AtomicLongClose(target.Context, target.Target);
}
+ internal static bool ListenableCancel(IUnmanagedTarget target)
+ {
+ return JNI.ListenableCancel(target.Context, target.Target);
+ }
+
+ internal static bool ListenableIsCancelled(IUnmanagedTarget target)
+ {
+ return JNI.ListenableIsCancelled(target.Context, target.Target);
+ }
+
#endregion
}
}