You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by pt...@apache.org on 2020/06/04 06:53:30 UTC
[ignite] branch master updated: IGNITE-13095 .NET: Fix Thin Client
compute task cancellation
This is an automated email from the ASF dual-hosted git repository.
ptupitsyn pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/ignite.git
The following commit(s) were added to refs/heads/master by this push:
new e9aa94b IGNITE-13095 .NET: Fix Thin Client compute task cancellation
e9aa94b is described below
commit e9aa94b0370466fe3488ce29420234a1fd3cf103
Author: Pavel Tupitsyn <pt...@apache.org>
AuthorDate: Thu Jun 4 09:52:55 2020 +0300
IGNITE-13095 .NET: Fix Thin Client compute task cancellation
* Call `OP_RESOURCE_CLOSE` to cancel the task on the server
* Improve query cursors logic to make sure the same connection is used to retrieve batches and close the cursor
---
.../PlatformComputeActiveTaskFuturesTask.java | 70 ++++++
.../Cache/Platform/PlatformCacheTest.cs | 276 ++++++++++-----------
.../Client/ClientConnectionTest.cs | 48 +++-
.../Client/Compute/ComputeClientTests.cs | 132 +++++++---
.../Process/ListDataReader.cs | 5 +
.../Apache.Ignite.Core/Client/ClientStatusCode.cs | 12 +-
.../Impl/Client/Cache/CacheClient.cs | 28 +--
.../Client/Cache/Query/ClientFieldsQueryCursor.cs | 6 +-
.../Impl/Client/Cache/Query/ClientQueryCursor.cs | 6 +-
.../Client/Cache/Query/ClientQueryCursorBase.cs | 19 +-
.../Apache.Ignite.Core/Impl/Client/ClientSocket.cs | 42 ++--
.../Impl/Client/Compute/ComputeClient.cs | 53 ++--
.../Impl/Common/DelegateConverter.cs | 6 +-
13 files changed, 452 insertions(+), 251 deletions(-)
diff --git a/modules/core/src/test/java/org/apache/ignite/platform/PlatformComputeActiveTaskFuturesTask.java b/modules/core/src/test/java/org/apache/ignite/platform/PlatformComputeActiveTaskFuturesTask.java
new file mode 100644
index 0000000..1b93b52
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/platform/PlatformComputeActiveTaskFuturesTask.java
@@ -0,0 +1,70 @@
+/*
+ * 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.platform;
+
+import org.apache.ignite.Ignite;
+import org.apache.ignite.cluster.ClusterNode;
+import org.apache.ignite.compute.ComputeJob;
+import org.apache.ignite.compute.ComputeJobAdapter;
+import org.apache.ignite.compute.ComputeJobResult;
+import org.apache.ignite.compute.ComputeTaskAdapter;
+import org.apache.ignite.lang.IgniteUuid;
+import org.apache.ignite.resources.IgniteInstanceResource;
+import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+/**
+ * Task to get active task futures.
+ */
+@SuppressWarnings({"OptionalGetWithoutIsPresent", "unused"})
+public class PlatformComputeActiveTaskFuturesTask extends ComputeTaskAdapter<Object, IgniteUuid[]> {
+ /** {@inheritDoc} */
+ @NotNull @Override public Map<? extends ComputeJob, ClusterNode> map(List<ClusterNode> subgrid,
+ @Nullable Object arg) {
+ ClusterNode localNode = subgrid.stream().filter(ClusterNode::isLocal).findFirst().get();
+
+ return Collections.singletonMap(new ActiveTaskFuturesJob(), localNode);
+ }
+
+ /** {@inheritDoc} */
+ @Nullable @Override public IgniteUuid[] reduce(List<ComputeJobResult> results) {
+ return results.get(0).getData();
+ }
+
+ /**
+ * Job.
+ */
+ @SuppressWarnings({"ZeroLengthArrayAllocation", "unused"})
+ private static class ActiveTaskFuturesJob extends ComputeJobAdapter {
+ /** Ignite. */
+ @IgniteInstanceResource
+ private Ignite ignite;
+
+ /** {@inheritDoc} */
+ @Nullable @Override public IgniteUuid[] execute() {
+ Set<IgniteUuid> ids = ignite.compute().activeTaskFutures().keySet();
+
+ return ids.toArray(new IgniteUuid[0]);
+ }
+ }
+}
diff --git a/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Cache/Platform/PlatformCacheTest.cs b/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Cache/Platform/PlatformCacheTest.cs
index 079c206..2b79e68 100644
--- a/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Cache/Platform/PlatformCacheTest.cs
+++ b/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Cache/Platform/PlatformCacheTest.cs
@@ -73,7 +73,7 @@ namespace Apache.Ignite.Core.Tests.Cache.Platform
{
EnabledLevels = new[] {LogLevel.Error}
};
-
+
var cfg = new IgniteConfiguration(TestUtils.GetTestConfiguration())
{
CacheConfiguration = new[]
@@ -97,7 +97,7 @@ namespace Apache.Ignite.Core.Tests.Cache.Platform
};
_grid = Ignition.Start(cfg);
-
+
var cfg2 = new IgniteConfiguration(cfg)
{
IgniteInstanceName = "server2"
@@ -113,7 +113,7 @@ namespace Apache.Ignite.Core.Tests.Cache.Platform
};
_client = Ignition.Start(clientCfg);
-
+
WaitForRebalance();
}
@@ -149,14 +149,14 @@ namespace Apache.Ignite.Core.Tests.Cache.Platform
var key = TestUtils.GetKey(_grid, cache.Name, primaryKey: primaryKey);
var obj = new Foo(key);
-
+
cache[key] = obj;
var res1 = cache[key];
var res2 = cache[key];
// Returned object is Equal to the initial.
Assert.AreEqual(obj, res1);
-
+
// But not the same - new instance is stored in platform cache,
// except primary on servers - thread-local optimization avoids extra deserialization there.
if (primaryKey && mode == CacheTestMode.ServerLocal || !primaryKey && mode == CacheTestMode.ServerRemote)
@@ -196,14 +196,14 @@ namespace Apache.Ignite.Core.Tests.Cache.Platform
// Put through remote node: platform cache is updated only on Get.
var remoteCache = GetCache<int, Foo>(
mode == CacheTestMode.Client ? CacheTestMode.ServerRemote : CacheTestMode.Client);
-
+
remoteCache[key] = obj;
}
Assert.AreEqual(3, cache[key].Bar);
Assert.AreSame(cache[key], cache[key]);
}
-
+
/// <summary>
/// Tests that platform cache returns the same object on every get.
/// </summary>
@@ -222,13 +222,13 @@ namespace Apache.Ignite.Core.Tests.Cache.Platform
{
Foo val;
- return localCache.TryGet(key, out val) &&
+ return localCache.TryGet(key, out val) &&
ReferenceEquals(val, localCache.Get(key));
- }, 300);
-
+ });
+
// Invalidate after get.
remoteCache[key] = new Foo(1);
-
+
TestUtils.WaitForTrueCondition(() =>
{
Foo val;
@@ -236,9 +236,9 @@ namespace Apache.Ignite.Core.Tests.Cache.Platform
return localCache.TryGet(key, out val) &&
val.Bar == 1 &&
ReferenceEquals(val, localCache.Get(key));
- }, 300);
+ });
}
-
+
/// <summary>
/// Tests that platform cache is updated from remote node after being populated with local Put call.
/// </summary>
@@ -278,7 +278,7 @@ namespace Apache.Ignite.Core.Tests.Cache.Platform
public void TestPlatformCacheRemoveFromRemoteNodeAfterLocalPut()
{
var localCache = _client.GetOrCreateNearCache<int, int>(CacheName, new NearCacheConfiguration());
-
+
var remoteCache = _grid.GetCache<int, int>(CacheName);
localCache[1] = 1;
@@ -301,7 +301,7 @@ namespace Apache.Ignite.Core.Tests.Cache.Platform
clientCache[key] = 2;
Assert.AreEqual(2, serverCache.LocalPeek(key, CachePeekMode.Platform));
-
+
clientCache[key] = 3;
Assert.AreEqual(3, serverCache.LocalPeek(key, CachePeekMode.Platform));
@@ -320,7 +320,7 @@ namespace Apache.Ignite.Core.Tests.Cache.Platform
NearConfiguration = new NearCacheConfiguration(),
PlatformCacheConfiguration = new PlatformCacheConfiguration()
};
-
+
var cache1 = _grid.CreateCache<int, int>(cfg);
var cache2 = _grid.GetCache<string, string>(cfg.Name);
var cache3 = _grid.GetCache<int, Foo>(cfg.Name);
@@ -346,16 +346,16 @@ namespace Apache.Ignite.Core.Tests.Cache.Platform
NearConfiguration = new NearCacheConfiguration(),
PlatformCacheConfiguration = new PlatformCacheConfiguration()
};
-
+
var cache1 = _grid.CreateCache<int, Foo>(cfg);
cache1[1] = new Foo(42);
cache1[2] = new Foo(43);
-
+
// Use different type parameters.
var cache2 = _grid.GetCache<int, string>(cfg.Name);
cache2[1] = "x";
- // Check that platform cache still works for old entries.
+ // Check that platform cache still works for old entries.
Assert.Throws<InvalidCastException>(() => cache1.Get(1));
Assert.AreEqual(43, cache1[2].Bar);
Assert.AreSame(cache1[2], cache1[2]);
@@ -400,13 +400,13 @@ namespace Apache.Ignite.Core.Tests.Cache.Platform
WriteThrough = true
};
- var cache = GetIgnite(mode).CreateCache<int, Foo>(cfg, new NearCacheConfiguration(),
+ var cache = GetIgnite(mode).CreateCache<int, Foo>(cfg, new NearCacheConfiguration(),
new PlatformCacheConfiguration());
-
+
// First write succeeds.
cache.Put(1, new Foo(1));
Assert.AreEqual(1, cache.LocalPeek(1, CachePeekMode.Platform).Bar);
-
+
// Special value causes write failure. Platform cache value is still correct.
Assert.Throws<CacheStoreException>(() => cache.Put(1, new Foo(FailingCacheStore.FailingValue)));
Assert.AreEqual(1, cache.LocalPeek(1, CachePeekMode.Platform).Bar);
@@ -422,7 +422,7 @@ namespace Apache.Ignite.Core.Tests.Cache.Platform
var value = new Foo(5);
cache[1] = value;
-
+
cache.Query(new SqlFieldsQuery("update Foo set Bar = 7 where Bar = 5"));
var res = cache[1];
@@ -430,19 +430,19 @@ namespace Apache.Ignite.Core.Tests.Cache.Platform
}
/// <summary>
- /// Tests that eviction policy removes platform cache data for the key.
+ /// Tests that eviction policy removes platform cache data for the key.
/// </summary>
[Test]
public void TestFifoEvictionPolicyRemovesPlatformCacheValue(
[Values(CacheTestMode.ServerLocal, CacheTestMode.ServerRemote, CacheTestMode.Client)] CacheTestMode mode)
{
var cache = GetCache<int, Foo>(mode);
-
+
TestEvictionPolicyRemovesPlatformCacheValue(mode, cache);
}
/// <summary>
- /// Tests that eviction policy removes platform cache data for the key.
+ /// Tests that eviction policy removes platform cache data for the key.
/// </summary>
[Test]
public void TestLruEvictionPolicyRemovesPlatformCacheValue(
@@ -464,7 +464,7 @@ namespace Apache.Ignite.Core.Tests.Cache.Platform
var ignite = GetIgnite(mode);
var cache = ignite.CreateCache<int, Foo>(cfg, cfg.NearConfiguration);
-
+
TestEvictionPolicyRemovesPlatformCacheValue(mode, cache);
}
@@ -473,20 +473,20 @@ namespace Apache.Ignite.Core.Tests.Cache.Platform
/// </summary>
[Test]
public void TestEvictionPolicyKeepsLastEntriesInPlatformCache(
- [Values(true, false)] bool lruOrFifo,
+ [Values(true, false)] bool lruOrFifo,
[Values(true, false)] bool getOrCreate)
{
const int maxSize = 30;
-
+
var serverCache = _grid.CreateCache<int, Foo>(TestUtils.TestName);
var nearCfg = new NearCacheConfiguration
{
- EvictionPolicy = lruOrFifo
+ EvictionPolicy = lruOrFifo
? (IEvictionPolicy) new LruEvictionPolicy
{
MaxSize = maxSize
- }
+ }
: new FifoEvictionPolicy
{
MaxSize = maxSize
@@ -495,13 +495,13 @@ namespace Apache.Ignite.Core.Tests.Cache.Platform
var platformCfg = new PlatformCacheConfiguration();
- var clientCache = getOrCreate
+ var clientCache = getOrCreate
? _client.GetOrCreateNearCache<int, Foo>(serverCache.Name, nearCfg, platformCfg)
: _client.CreateNearCache<int, Foo>(serverCache.Name, nearCfg, platformCfg);
var keys = Enumerable.Range(1, maxSize * 5).ToList();
var nearKeys = keys.AsEnumerable().Reverse().Take(maxSize).ToArray();
-
+
keys.ForEach(k => serverCache.Put(k, new Foo(k)));
// Get from client to populate platform cache.
@@ -513,7 +513,7 @@ namespace Apache.Ignite.Core.Tests.Cache.Platform
{
Assert.AreSame(clientCache.LocalPeek(key, CachePeekMode.Platform), clientCache.Get(key));
}
-
+
// Check that GetAll returns instances from platform cache.
var all = clientCache.GetAll(nearKeys);
foreach (var entry in all)
@@ -545,11 +545,11 @@ namespace Apache.Ignite.Core.Tests.Cache.Platform
var serverCache = _grid.CreateCache<int, int>(cfg);
var clientCache = _client.GetOrCreateNearCache<int, int>(cfg.Name, cfg.NearConfiguration);
-
+
clientCache[1] = 1;
clientCache[2] = 2;
serverCache[1] = 11;
-
+
Assert.AreEqual(11, clientCache[1]);
}
@@ -561,7 +561,7 @@ namespace Apache.Ignite.Core.Tests.Cache.Platform
[Values(CacheTestMode.ServerLocal, CacheTestMode.ServerRemote, CacheTestMode.Client)] CacheTestMode mode)
{
var cache = GetCache<int, Foo>(mode);
-
+
const int count = 100;
cache.PutAll(Enumerable.Range(1, count).Select(x => new KeyValuePair<int, Foo>(x, new Foo(x))));
@@ -570,9 +570,9 @@ namespace Apache.Ignite.Core.Tests.Cache.Platform
{
CacheName = cache.Name
};
-
+
var res = cache.Query(new ScanQuery<int, Foo>(filter));
-
+
Assert.AreEqual(count, res.Count());
}
@@ -584,10 +584,10 @@ namespace Apache.Ignite.Core.Tests.Cache.Platform
[Values(CacheTestMode.ServerLocal, CacheTestMode.ServerRemote, CacheTestMode.Client)] CacheTestMode mode)
{
var cache = GetCache<int, Foo>(mode);
-
+
const int count = 100;
var data = Enumerable.Range(1, count).ToDictionary(x=> x, x => new Foo(x));
-
+
cache.PutAll(data);
// Filter will check that value does not come from native platform cache.
@@ -595,7 +595,7 @@ namespace Apache.Ignite.Core.Tests.Cache.Platform
{
CacheName = cache.Name
};
-
+
// Clear platform cache using internal API.
foreach (var ignite in Ignition.GetAll())
{
@@ -606,14 +606,14 @@ namespace Apache.Ignite.Core.Tests.Cache.Platform
platformCache.Clear();
}
}
-
+
var res = cache.Query(new ScanQuery<int, Foo>(filter));
-
+
Assert.AreEqual(count, res.Count());
}
/// <summary>
- /// Tests that local scan query uses platform cache directly, avoiding Java roundtrip.
+ /// Tests that local scan query uses platform cache directly, avoiding Java roundtrip.
/// </summary>
[Test]
public void TestLocalScanQueryUsesKeysAndValuesFromPlatformCache([Values(true, false)] bool withFilter,
@@ -625,23 +625,23 @@ namespace Apache.Ignite.Core.Tests.Cache.Platform
var qry = new ScanQuery<int, Foo>
{
Local = true,
- Filter = withFilter
+ Filter = withFilter
? new ScanQueryPlatformCacheFilter
{
CacheName = cache.Name
}
: null,
- Partition = withPartition
- ? _grid.GetAffinity(cache.Name).GetPartition(TestUtils.GetPrimaryKey(_grid, cache.Name))
+ Partition = withPartition
+ ? _grid.GetAffinity(cache.Name).GetPartition(TestUtils.GetPrimaryKey(_grid, cache.Name))
: (int?) null
};
-
+
var res = cache.Query(qry);
foreach (var entry in res)
{
var localValue = cache.LocalPeek(entry.Key, CachePeekMode.Platform);
-
+
if (withPartition)
{
// Local scan with partition works directly through platform cache.
@@ -661,18 +661,18 @@ namespace Apache.Ignite.Core.Tests.Cache.Platform
}
/// <summary>
- /// Tests that local scan query reserves the partition when <see cref="ScanQuery{TK,TV}.Partition"/> is set.
+ /// Tests that local scan query reserves the partition when <see cref="ScanQuery{TK,TV}.Partition"/> is set.
/// </summary>
[Test]
public void TestLocalScanQueryWithPartitionReservesPartitionAndReleasesItOnDispose()
{
var cache = GetCache<int, Foo>(CacheTestMode.ServerLocal);
-
+
var key = TestUtils.GetPrimaryKey(_grid, cache.Name);
var part = _grid.GetAffinity(cache.Name).GetPartition(key);
cache.PutAll(Enumerable.Range(1, 100).ToDictionary(x => x, x => new Foo(x)));
-
+
var qry = new ScanQuery<int, Foo>
{
Local = true,
@@ -680,7 +680,7 @@ namespace Apache.Ignite.Core.Tests.Cache.Platform
};
Func<bool> isReserved = () => TestUtils.IsPartitionReserved(_grid, cache.Name, part);
-
+
Assert.IsFalse(isReserved());
// Full iteration.
@@ -691,20 +691,20 @@ namespace Apache.Ignite.Core.Tests.Cache.Platform
using (var enumerator = cursor.GetEnumerator())
{
Assert.IsTrue(isReserved());
-
+
while (enumerator.MoveNext())
{
Assert.IsTrue(isReserved());
}
-
+
Assert.IsFalse(isReserved());
}
-
+
Assert.IsFalse(isReserved());
}
-
+
Assert.IsFalse(isReserved());
-
+
// Partial iteration with LINQ.
using (var cursor = cache.Query(qry))
{
@@ -713,10 +713,10 @@ namespace Apache.Ignite.Core.Tests.Cache.Platform
var item = cursor.FirstOrDefault();
Assert.IsNotNull(item);
- // Released because LINQ disposes the iterator.
+ // Released because LINQ disposes the iterator.
Assert.IsFalse(isReserved());
}
-
+
// Partial iteration.
using (var cursor = cache.Query(qry))
{
@@ -727,9 +727,9 @@ namespace Apache.Ignite.Core.Tests.Cache.Platform
Assert.IsTrue(isReserved());
}
-
+
Assert.IsFalse(isReserved());
-
+
// GetAll without using block.
using (var cursor = cache.Query(qry))
{
@@ -740,10 +740,10 @@ namespace Apache.Ignite.Core.Tests.Cache.Platform
Assert.IsFalse(isReserved());
}
-
+
// Exception in filter.
qry.Filter = new ScanQueryPlatformCacheFilter {FailKey = key};
-
+
using (var cursor = cache.Query(qry))
{
Assert.IsTrue(isReserved());
@@ -762,9 +762,9 @@ namespace Apache.Ignite.Core.Tests.Cache.Platform
{
var cache = GetCache<int, Foo>(CacheTestMode.ServerLocal);
var qry = new ScanQuery<int, Foo> {Local = true, Partition = 1024};
-
+
var ex = Assert.Throws<IgniteException>(() => cache.Query(qry));
-
+
Assert.AreEqual("Invalid partition number: 1024", ex.Message);
}
@@ -776,11 +776,11 @@ namespace Apache.Ignite.Core.Tests.Cache.Platform
public void TestLocalScanQueryWithPartitionThrowsOnRemoteKeys()
{
var cache = GetCache<int, Foo>(CacheTestMode.ServerLocal);
-
+
var partition = _grid2.GetAffinity(cache.Name)
.GetPrimaryPartitions(_grid2.GetCluster().GetLocalNode())
.First();
-
+
var qry = new ScanQuery<int, Foo>
{
Local = true,
@@ -788,12 +788,12 @@ namespace Apache.Ignite.Core.Tests.Cache.Platform
};
var ex = Assert.Throws<InvalidOperationException>(() => cache.Query(qry).GetAll());
-
+
Assert.AreEqual(
- string.Format("Failed to reserve partition {0}, it does not belong to the local node.", partition),
+ string.Format("Failed to reserve partition {0}, it does not belong to the local node.", partition),
ex.Message);
}
-
+
/// <summary>
/// Tests local scan query on client node.
/// </summary>
@@ -805,10 +805,10 @@ namespace Apache.Ignite.Core.Tests.Cache.Platform
var clientCache = _client.CreateNearCache<int, Foo>(cache.Name, new NearCacheConfiguration(),
new PlatformCacheConfiguration());
-
+
// Promote key to near cache.
clientCache.Get(2);
-
+
var res = clientCache.Query(new ScanQuery<int, Foo> {Local = true}).GetAll();
// Local scan on client node returns empty collection.
@@ -826,9 +826,9 @@ namespace Apache.Ignite.Core.Tests.Cache.Platform
{
var cache = GetCache<int, Foo>(mode)
.WithExpiryPolicy(new ExpiryPolicy(TimeSpan.FromSeconds(0.2), null, null));
-
+
cache[1] = new Foo(1);
-
+
Assert.AreEqual(1, cache[1].Bar);
Assert.AreEqual(1, cache.LocalPeek(1, CachePeekMode.Platform).Bar);
Assert.AreEqual(1, cache.Count());
@@ -853,14 +853,14 @@ namespace Apache.Ignite.Core.Tests.Cache.Platform
KeepBinary = true
}
};
-
+
var clientCache = _client.CreateCache<int, Foo>(cfg);
var serverCache = _grid2.GetCache<int, object>(cfg.Name);
Assert.IsTrue(serverCache.GetConfiguration().PlatformCacheConfiguration.KeepBinary);
-
+
// Put non-binary from client. There is no platform cache on client.
clientCache[1] = new Foo(2);
-
+
// Read from platform on server.
var res = (IBinaryObject) serverCache.LocalPeek(1, CachePeekMode.Platform);
Assert.AreEqual(2, res.GetField<int>("Bar"));
@@ -895,7 +895,7 @@ namespace Apache.Ignite.Core.Tests.Cache.Platform
remoteCache.Put(key, new Foo(id));
}
});
-
+
var localReader = Task.Factory.StartNew(() =>
{
// ReSharper disable once AccessToModifiedClosure
@@ -913,7 +913,7 @@ namespace Apache.Ignite.Core.Tests.Cache.Platform
// Get actual value with SQL to bypass caches.
// Actual value may not be equal to the latest id because two threads compete in Put calls.
var actualValue = (int) localCache.Query(new SqlFieldsQuery("select Bar from Foo")).GetAll()[0][0];
-
+
Assert.AreEqual(actualValue, localCache[key].Bar, "Local value");
Assert.AreEqual(actualValue, remoteCache[key].Bar, "Remote value");
}
@@ -932,8 +932,8 @@ namespace Apache.Ignite.Core.Tests.Cache.Platform
clientCache[1] = new Foo(1);
serverCache[2] = new Foo(2);
- var res = async
- ? clientCache.GetAllAsync(Enumerable.Range(1, 2)).Result
+ var res = async
+ ? clientCache.GetAllAsync(Enumerable.Range(1, 2)).Result
: clientCache.GetAll(Enumerable.Range(1, 2));
Assert.AreEqual(new[] {1, 2}, res.Select(x => x.Key));
@@ -959,20 +959,20 @@ namespace Apache.Ignite.Core.Tests.Cache.Platform
serverCache[2] = new Foo(2);
Foo foo;
-
+
Assert.IsTrue(clientCache.TryLocalPeek(1, out foo, CachePeekMode.Platform));
Assert.AreEqual(1, foo.Bar);
-
+
Assert.IsTrue(clientCache.TryLocalPeek(1, out foo, CachePeekMode.Platform | CachePeekMode.Near));
Assert.AreEqual(1, foo.Bar);
-
+
Assert.IsFalse(clientCache.TryLocalPeek(2, out foo, CachePeekMode.Platform));
Assert.IsFalse(clientCache.TryLocalPeek(2, out foo, CachePeekMode.Near));
Assert.IsFalse(clientCache.TryLocalPeek(2, out foo, CachePeekMode.All));
-
+
Assert.AreEqual(2, serverCache.LocalPeek(2, CachePeekMode.Platform).Bar);
Assert.AreEqual(2, serverCache.LocalPeek(2, CachePeekMode.Near).Bar);
-
+
Assert.AreSame(serverCache[2], serverCache.LocalPeek(2, CachePeekMode.Platform));
Assert.AreNotSame(serverCache[2], serverCache.LocalPeek(2, CachePeekMode.Near));
}
@@ -984,12 +984,12 @@ namespace Apache.Ignite.Core.Tests.Cache.Platform
{
var cache = GetCache<int, int>(mode);
var cache2 = GetCache<int, int>(CacheTestMode.ServerLocal);
-
+
var data = Enumerable.Range(1, 100).ToDictionary(x => x, x => x);
cache2.PutAll(data);
var act = async
- ? (Func<int, bool>) (k => cache.ContainsKeyAsync(k).Result)
+ ? (Func<int, bool>) (k => cache.ContainsKeyAsync(k).Result)
: k => cache.ContainsKey(k);
foreach (var key in data.Keys)
@@ -1006,12 +1006,12 @@ namespace Apache.Ignite.Core.Tests.Cache.Platform
{
var cache = GetCache<int, int>(mode);
var cache2 = GetCache<int, int>(CacheTestMode.ServerLocal);
-
+
var data = Enumerable.Range(1, 100).ToDictionary(x => x, x => x);
cache2.PutAll(data);
-
+
var act = async
- ? (Func<IEnumerable<int>, bool>) (k => cache.ContainsKeysAsync(k).Result)
+ ? (Func<IEnumerable<int>, bool>) (k => cache.ContainsKeysAsync(k).Result)
: k => cache.ContainsKeys(k);
foreach (var key in data.Keys)
@@ -1019,13 +1019,13 @@ namespace Apache.Ignite.Core.Tests.Cache.Platform
Assert.IsTrue(act(new[] {key}));
Assert.IsFalse(act(new[] {-key}));
}
-
+
Assert.IsTrue(act(data.Keys));
Assert.IsTrue(act(data.Keys.Take(10)));
Assert.IsTrue(act(data.Keys.Skip(10)));
Assert.IsFalse(act(data.Keys.Concat(new[] {-1})));
}
-
+
/// <summary>
/// Tests local size on server node.
/// </summary>
@@ -1057,7 +1057,7 @@ namespace Apache.Ignite.Core.Tests.Cache.Platform
var cache = GetCache<int, int>(CacheTestMode.Client, TestUtils.TestName);
Assert.AreEqual(0, cache.GetLocalSize(CachePeekMode.Platform));
Assert.AreEqual(0, cache.GetLocalSize(CachePeekMode.All));
-
+
cache.PutAll(Enumerable.Range(1, 100).ToDictionary(x => x, x => x));
Assert.AreEqual(0, cache.GetLocalSize());
@@ -1075,14 +1075,14 @@ namespace Apache.Ignite.Core.Tests.Cache.Platform
/// Tests that <see cref="CachePeekMode.Platform"/> works with distributed GetSize overloads.
/// </summary>
[Test]
- public void TestGetSizeWithPlatform([Values(true, false)] bool longMode,
+ public void TestGetSizeWithPlatform([Values(true, false)] bool longMode,
[Values(true, false)] bool async)
{
var cache = GetCache<int, int>(CacheTestMode.Client, TestUtils.TestName);
var primaryAndPlatform = new[] {CachePeekMode.Primary, CachePeekMode.Platform};
var platform = new[] {CachePeekMode.Platform};
var all = new[] {CachePeekMode.All};
-
+
var func =
longMode
? async
@@ -1090,7 +1090,7 @@ namespace Apache.Ignite.Core.Tests.Cache.Platform
: m => cache.GetSizeLong(m)
: async
? (Func<CachePeekMode[], long>) (m => cache.GetSizeAsync(m).Result)
- : m => cache.GetSize(m);
+ : m => cache.GetSize(m);
Assert.AreEqual(0, func(all));
Assert.AreEqual(0, func(platform));
@@ -1132,7 +1132,7 @@ namespace Apache.Ignite.Core.Tests.Cache.Platform
Assert.AreEqual(1, func(part, platform));
Assert.AreEqual(2, func(part, primaryAndPlatform));
Assert.AreEqual(2, func(part, all));
-
+
Assert.AreEqual(0, func(part + 1, platform));
Assert.AreEqual(0, func(part + 1, primaryAndPlatform));
Assert.AreEqual(0, func(part + 1, all));
@@ -1154,14 +1154,14 @@ namespace Apache.Ignite.Core.Tests.Cache.Platform
cache.PutAll(keys.ToDictionary(x => x, x => new Foo(x)));
var localEntries = cache.GetLocalEntries(CachePeekMode.Platform).ToArray();
-
+
// Same set of keys.
CollectionAssert.AreEquivalent(keys, localEntries.Select(e => e.Key));
-
+
// Returns same instances every time.
CollectionAssert.AreEqual(localEntries.Select(e => e.Value),
cache.GetLocalEntries(CachePeekMode.Platform).Select(e => e.Value));
-
+
// Every instance is from platform cache.
foreach (var entry in localEntries)
{
@@ -1187,7 +1187,7 @@ namespace Apache.Ignite.Core.Tests.Cache.Platform
var platform = getKeys(CachePeekMode.Platform);
var all = getKeys(CachePeekMode.All);
var all2 = getKeys(CachePeekMode.Primary | CachePeekMode.Near | CachePeekMode.Platform);
-
+
CollectionAssert.AreEqual(all, all2);
CollectionAssert.AreEquivalent(all, platform.Concat(primary).Concat(near));
CollectionAssert.AreEquivalent(platform, primary.Concat(near));
@@ -1212,10 +1212,10 @@ namespace Apache.Ignite.Core.Tests.Cache.Platform
const int count = 100;
cache1.PutAll(Enumerable.Range(1, count).ToDictionary(x => x, x => x));
-
+
Assert.AreEqual(count, cache1.GetLocalSize(CachePeekMode.Platform));
Assert.AreEqual(count, cache2.GetLocalSize(CachePeekMode.Platform));
-
+
Assert.AreEqual(42, cache1.LocalPeek(42, CachePeekMode.Platform));
Assert.AreEqual(42, cache2.LocalPeek(42, CachePeekMode.Platform));
@@ -1242,10 +1242,10 @@ namespace Apache.Ignite.Core.Tests.Cache.Platform
const int count = 100;
cache1.PutAll(Enumerable.Range(1, count).ToDictionary(x => x, x => x));
-
+
Assert.AreEqual(count, cache1.GetLocalSize(CachePeekMode.Platform));
Assert.AreEqual(count, cache2.GetLocalSize(CachePeekMode.Platform));
-
+
Assert.AreEqual(42, cache1.LocalPeek(42, CachePeekMode.Platform));
Assert.AreEqual(42, cache2.LocalPeek(42, CachePeekMode.Platform));
@@ -1271,7 +1271,7 @@ namespace Apache.Ignite.Core.Tests.Cache.Platform
cache[1] = new Foo(2);
var foo = cache[1];
-
+
Assert.AreEqual(2, foo.Bar);
Assert.AreSame(foo, cache[1]);
@@ -1279,7 +1279,7 @@ namespace Apache.Ignite.Core.Tests.Cache.Platform
{
Assert.AreNotSame(foo, cache[1]);
}
-
+
Assert.AreSame(foo, cache[1]);
using (new TransactionScope())
@@ -1287,9 +1287,9 @@ namespace Apache.Ignite.Core.Tests.Cache.Platform
cache[2] = new Foo(3);
Assert.AreNotSame(foo, cache[1]);
}
-
+
Assert.AreSame(foo, cache[1]);
- }
+ }
/// <summary>
/// Tests platform cache misconfiguration / type mismatch.
@@ -1305,7 +1305,7 @@ namespace Apache.Ignite.Core.Tests.Cache.Platform
ValueTypeName = typeof(Guid).FullName
}
};
-
+
var nearCfg = new NearCacheConfiguration();
var clientCache = _client.CreateCache<int, Foo>(cfg, nearCfg);
@@ -1313,7 +1313,7 @@ namespace Apache.Ignite.Core.Tests.Cache.Platform
// Put Foo, but platform cache expects Guid.
clientCache.GetAndPut(1, new Foo(2));
-
+
// Entry is not in platform cache.
Assert.AreEqual(0, clientCache.GetLocalSize(CachePeekMode.Platform));
@@ -1322,7 +1322,7 @@ namespace Apache.Ignite.Core.Tests.Cache.Platform
Assert.AreEqual(2, clientCache[1].Bar);
// Error is logged.
- Func<ListLogger.Entry> getEntry = () =>
+ Func<ListLogger.Entry> getEntry = () =>
_logger.Entries.FirstOrDefault(e => e.Message.StartsWith("Failure in Java callback"));
var message = string.Join(" | ", _logger.Entries.Select(e => e.Message));
@@ -1331,7 +1331,7 @@ namespace Apache.Ignite.Core.Tests.Cache.Platform
/// <summary>
/// <see cref="ICache{TK,TV}.LoadCache"/> uses same filter mechanism as <see cref="ScanQuery{TK,TV}"/>.
- /// Platform cache should never be used for cache store load filters.
+ /// Platform cache should never be used for cache store load filters.
/// </summary>
[Test]
public void TestCacheStoreLoadFilterDoesNotUseNearCache()
@@ -1349,9 +1349,9 @@ namespace Apache.Ignite.Core.Tests.Cache.Platform
// Put a value to be overwritten from store.
var foo = FailingCacheStore.Foo;
var key = foo.Bar;
-
+
cache[key] = new Foo(1);
-
+
// Filter asserts that values do not come from platform cache.
var filter = new StoreNoPlatformCacheFilter
{
@@ -1369,7 +1369,7 @@ namespace Apache.Ignite.Core.Tests.Cache.Platform
{
const int serverMaxSize = 4;
const int clientMaxSize = serverMaxSize * 3;
-
+
var serverCfg = new CacheConfiguration
{
Name = TestUtils.TestName,
@@ -1381,7 +1381,7 @@ namespace Apache.Ignite.Core.Tests.Cache.Platform
}
}
};
-
+
var clientCfg = new NearCacheConfiguration
{
EvictionPolicy = new FifoEvictionPolicy
@@ -1389,23 +1389,23 @@ namespace Apache.Ignite.Core.Tests.Cache.Platform
MaxSize = clientMaxSize
}
};
-
+
var platformCfg = new PlatformCacheConfiguration();
var serverCache = _grid.CreateCache<int, int>(serverCfg);
var clientCache = _client.CreateNearCache<int, int>(serverCache.Name, clientCfg, platformCfg);
var keys = Enumerable.Range(1, 100).ToList();
-
+
keys.ForEach(k => clientCache.Put(k, k));
-
+
Assert.AreEqual(clientMaxSize, clientCache.GetLocalSize(CachePeekMode.Near));
Assert.AreEqual(clientMaxSize, clientCache.GetLocalSize(CachePeekMode.Platform));
var expectedKeys = keys.AsEnumerable().Reverse().Take(clientMaxSize).ToArray();
- var nearKeys =
+ var nearKeys =
clientCache.GetLocalEntries(CachePeekMode.Platform).Select(e => e.Key).ToArray();
-
+
CollectionAssert.AreEquivalent(expectedKeys, nearKeys);
}
@@ -1416,7 +1416,7 @@ namespace Apache.Ignite.Core.Tests.Cache.Platform
public void TestDataStreamerUpdatesPlatformCache()
{
const int entryCount = 10000;
-
+
var serverCfg = new CacheConfiguration
{
Name = TestUtils.TestName,
@@ -1429,7 +1429,7 @@ namespace Apache.Ignite.Core.Tests.Cache.Platform
},
PlatformCacheConfiguration = new PlatformCacheConfiguration()
};
-
+
var clientCfg = new NearCacheConfiguration
{
EvictionPolicy = new LruEvictionPolicy
@@ -1437,9 +1437,9 @@ namespace Apache.Ignite.Core.Tests.Cache.Platform
MaxSize = entryCount
}
};
-
+
var platformCfg = new PlatformCacheConfiguration();
-
+
var serverCache = _grid.CreateCache<int, int>(serverCfg);
var clientCache = _client.CreateNearCache<int, int>(serverCache.Name, clientCfg, platformCfg);
@@ -1449,7 +1449,7 @@ namespace Apache.Ignite.Core.Tests.Cache.Platform
// Get data on client.
var res = clientCache.GetAll(data.Keys);
Assert.AreEqual(entryCount, res.Count);
-
+
// Check that all entries are in platform cache on client.
Assert.AreEqual(entryCount, clientCache.GetLocalSize(CachePeekMode.Near));
Assert.AreEqual(entryCount, clientCache.GetLocalSize(CachePeekMode.Platform));
@@ -1458,23 +1458,23 @@ namespace Apache.Ignite.Core.Tests.Cache.Platform
using (var streamer = _grid2.GetDataStreamer<int, int>(serverCache.Name))
{
streamer.AllowOverwrite = true;
-
+
foreach (var entry in data)
{
streamer.AddData(entry.Key, entry.Value + 1);
}
}
-
+
// Verify that platform cache contains updated entries.
foreach (var entry in data)
{
var key = entry.Key;
var val = entry.Value + 1;
- TestUtils.WaitForTrueCondition(() => val == clientCache.LocalPeek(key, CachePeekMode.Platform),
+ TestUtils.WaitForTrueCondition(() => val == clientCache.LocalPeek(key, CachePeekMode.Platform),
message: string.Format("{0} = {1}", key, val));
-
- TestUtils.WaitForTrueCondition(() => val == serverCache.LocalPeek(key, CachePeekMode.Platform),
+
+ TestUtils.WaitForTrueCondition(() => val == serverCache.LocalPeek(key, CachePeekMode.Platform),
message: string.Format("{0} = {1}", key, val));
}
}
@@ -1487,7 +1487,7 @@ namespace Apache.Ignite.Core.Tests.Cache.Platform
var nearConfiguration = _grid.GetCache<TK, TV>(CacheName).GetConfiguration().NearConfiguration;
var cacheConfiguration = new CacheConfiguration
{
- NearConfiguration = nearConfiguration,
+ NearConfiguration = nearConfiguration,
Name = name,
PlatformCacheConfiguration = new PlatformCacheConfiguration()
};
@@ -1504,7 +1504,7 @@ namespace Apache.Ignite.Core.Tests.Cache.Platform
}
/// <summary>
- /// Tests that eviction policy removes platform cache data for the key.
+ /// Tests that eviction policy removes platform cache data for the key.
/// </summary>
private void TestEvictionPolicyRemovesPlatformCacheValue(CacheTestMode mode, ICache<int, Foo> cache)
{
@@ -1522,7 +1522,7 @@ namespace Apache.Ignite.Core.Tests.Cache.Platform
cache[item.Bar] = item;
cachedItems.Add(cache[item.Bar]);
}
-
+
// Recent items are in platform cache:
Assert.AreEqual(NearCacheMaxSize, cache.GetLocalSize(CachePeekMode.Platform));
foreach (var item in cachedItems.Skip(items.Length - NearCacheMaxSize))
diff --git a/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Client/ClientConnectionTest.cs b/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Client/ClientConnectionTest.cs
index a37f2e3..46f9770 100644
--- a/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Client/ClientConnectionTest.cs
+++ b/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Client/ClientConnectionTest.cs
@@ -18,10 +18,12 @@
namespace Apache.Ignite.Core.Tests.Client
{
using System;
+ using System.Diagnostics;
using System.IO;
using System.Linq;
using System.Net;
using System.Net.Sockets;
+ using System.Text.RegularExpressions;
using System.Threading;
using System.Threading.Tasks;
using Apache.Ignite.Core.Cache.Configuration;
@@ -31,6 +33,7 @@ namespace Apache.Ignite.Core.Tests.Client
using Apache.Ignite.Core.Configuration;
using Apache.Ignite.Core.Impl.Common;
using Apache.Ignite.Core.Log;
+ using Apache.Ignite.Core.Tests.Client.Cache;
using NUnit.Framework;
/// <summary>
@@ -239,7 +242,7 @@ namespace Apache.Ignite.Core.Tests.Client
{
Assert.AreEqual("foo", client.GetCacheNames().Single());
}
-
+
// Port range.
cfg = new IgniteClientConfiguration("127.0.0.1:10798..10800");
@@ -259,9 +262,9 @@ namespace Apache.Ignite.Core.Tests.Client
var cfg = new IgniteClientConfiguration("127.0.0.1:10800..10700");
var ex = Assert.Throws<IgniteClientException>(() => Ignition.StartClient(cfg));
-
+
Assert.AreEqual(
- "Invalid format of IgniteClientConfiguration.Endpoint, port range is empty: 127.0.0.1:10800..10700",
+ "Invalid format of IgniteClientConfiguration.Endpoint, port range is empty: 127.0.0.1:10800..10700",
ex.Message);
}
@@ -593,6 +596,45 @@ namespace Apache.Ignite.Core.Tests.Client
}
/// <summary>
+ /// Tests that client stops it's receiver thread upon disposal.
+ /// </summary>
+ [Test]
+ public void TestClientDisposalStopsReceiverThread([Values(true, false)] bool async)
+ {
+ Ignition.Start(TestUtils.GetTestConfiguration());
+
+ var logger = new ListLogger {EnabledLevels = new[] {LogLevel.Trace}};
+
+ var cfg = new IgniteClientConfiguration(GetClientConfiguration())
+ {
+ Logger = logger
+ };
+
+ using (var client = Ignition.StartClient(cfg))
+ {
+ var cache = client.GetOrCreateCache<int, int>("c");
+
+ if (async)
+ {
+ cache.PutAsync(1, 1);
+ }
+ else
+ {
+ cache.Put(1, 1);
+ }
+ }
+
+ var threadId = logger.Entries
+ .Select(e => Regex.Match(e.Message, "Receiver thread #([0-9]+) started."))
+ .Where(m => m.Success)
+ .Select(m => int.Parse(m.Groups[1].Value))
+ .First();
+
+ TestUtils.WaitForTrueCondition(() => logger.Entries.Any(
+ e => e.Message == string.Format("Receiver thread #{0} stopped.", threadId)));
+ }
+
+ /// <summary>
/// Starts the client.
/// </summary>
private static IIgniteClient StartClient()
diff --git a/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Client/Compute/ComputeClientTests.cs b/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Client/Compute/ComputeClientTests.cs
index bc060c8..ceffb11 100644
--- a/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Client/Compute/ComputeClientTests.cs
+++ b/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Client/Compute/ComputeClientTests.cs
@@ -25,6 +25,7 @@ namespace Apache.Ignite.Core.Tests.Client.Compute
using Apache.Ignite.Core.Binary;
using Apache.Ignite.Core.Client;
using Apache.Ignite.Core.Client.Compute;
+ using Apache.Ignite.Core.Common;
using Apache.Ignite.Core.Configuration;
using Apache.Ignite.Core.Impl.Binary;
using Apache.Ignite.Core.Impl.Client.Compute;
@@ -40,14 +41,17 @@ namespace Apache.Ignite.Core.Tests.Client.Compute
{
/** */
private const string TestTask = "org.apache.ignite.internal.client.thin.TestTask";
-
+
/** */
private const string TestResultCacheTask = "org.apache.ignite.internal.client.thin.TestResultCacheTask";
-
+
/** */
private const string TestFailoverTask = "org.apache.ignite.internal.client.thin.TestFailoverTask";
/** */
+ private const string ActiveTaskFuturesTask = "org.apache.ignite.platform.PlatformComputeActiveTaskFuturesTask";
+
+ /** */
private const int MaxTasks = 8;
/// <summary>
@@ -66,9 +70,9 @@ namespace Apache.Ignite.Core.Tests.Client.Compute
{
var logger = (ListLogger) Client.GetConfiguration().Logger;
var entries = logger.Entries;
-
+
logger.Clear();
-
+
foreach (var entry in entries)
{
if (entry.Level >= LogLevel.Warn)
@@ -101,7 +105,7 @@ namespace Apache.Ignite.Core.Tests.Client.Compute
var ex = (IgniteClientException) Assert.Throws<AggregateException>(
() => Client.GetCompute().ExecuteJavaTask<int>(ComputeApiTest.EchoTask, -42))
.GetInnermostException();
-
+
StringAssert.EndsWith("Unknown type: -42", ex.Message);
}
@@ -116,16 +120,16 @@ namespace Apache.Ignite.Core.Tests.Client.Compute
{
AutoGenerateIgniteInstanceName = true
};
-
+
var ignite = Ignition.Start(cfg);
var port = ignite.GetCluster().GetLocalNode().GetAttribute<int>("clientListenerPort");
-
+
var clientCfg = new IgniteClientConfiguration(GetClientConfiguration())
{
Endpoints = new[] {"127.0.0.1:" + port}
};
-
+
var client = Ignition.StartClient(clientCfg);
try
@@ -151,7 +155,7 @@ namespace Apache.Ignite.Core.Tests.Client.Compute
{
var res = Client.GetCompute().WithKeepBinary().ExecuteJavaTask<IBinaryObject>(
ComputeApiTest.EchoTask, ComputeApiTest.EchoTypeBinarizableJava);
-
+
Assert.AreEqual(1, res.GetField<int>("field"));
Assert.AreEqual("field", res.GetBinaryType().Fields.Single());
Assert.AreEqual("org.apache.ignite.platform.PlatformComputeJavaBinarizable", res.GetBinaryType().TypeName);
@@ -166,7 +170,7 @@ namespace Apache.Ignite.Core.Tests.Client.Compute
var arg = new PlatformComputeNetBinarizable {Field = 42};
var res = Client.GetCompute().WithKeepBinary().ExecuteJavaTask<int>(ComputeApiTest.BinaryArgTask, arg);
-
+
Assert.AreEqual(arg.Field, res);
}
@@ -183,11 +187,11 @@ namespace Apache.Ignite.Core.Tests.Client.Compute
Client.GetCompute().ExecuteJavaTask<int>(ComputeApiTest.BinaryArgTask, arg));
var clientEx = (IgniteClientException) ex.GetInnermostException();
-
+
var expected = string.Format(
- "Failed to resolve .NET class '{0}' in Java [platformId=0, typeId=-315989221].",
+ "Failed to resolve .NET class '{0}' in Java [platformId=0, typeId=-315989221].",
arg.GetType().FullName);
-
+
Assert.AreEqual(expected, clientEx.Message);
}
@@ -199,7 +203,7 @@ namespace Apache.Ignite.Core.Tests.Client.Compute
{
var computeWithFailover = Client.GetCompute();
var computeWithNoFailover = Client.GetCompute().WithNoFailover();
-
+
Assert.IsTrue(computeWithFailover.ExecuteJavaTask<bool>(TestFailoverTask, null));
Assert.IsFalse(computeWithNoFailover.ExecuteJavaTask<bool>(TestFailoverTask, null));
}
@@ -212,7 +216,7 @@ namespace Apache.Ignite.Core.Tests.Client.Compute
{
var computeWithCache = Client.GetCompute();
var computeWithNoCache = Client.GetCompute().WithNoResultCache();
-
+
Assert.IsTrue(computeWithCache.ExecuteJavaTask<bool>(TestResultCacheTask, null));
Assert.IsFalse(computeWithNoCache.ExecuteJavaTask<bool>(TestResultCacheTask, null));
}
@@ -228,7 +232,7 @@ namespace Apache.Ignite.Core.Tests.Client.Compute
var ex = Assert.Throws<AggregateException>(() => compute.ExecuteJavaTask<object>(TestTask, timeoutMs * 4));
var clientEx = (IgniteClientException) ex.GetInnermostException();
-
+
StringAssert.StartsWith("Task timed out (check logs for error messages):", clientEx.Message);
}
@@ -247,13 +251,50 @@ namespace Apache.Ignite.Core.Tests.Client.Compute
[Test]
public void TestExecuteJavaTaskAsyncCancellation()
{
- const long delayMs = 1000;
-
+ const long delayMs = 10000;
+
+ // GetActiveTaskFutures uses Task internally, so it always returns at least 1 future.
+ var futs1 = GetActiveTaskFutures();
+ Assert.AreEqual(1, futs1.Length);
+
+ // Start a long-running task and verify that 2 futures are active.
var cts = new CancellationTokenSource();
var task = Client.GetCompute().ExecuteJavaTaskAsync<object>(TestTask, delayMs, cts.Token);
-
+
+ var taskFutures = GetActiveTaskFutures();
+ TestUtils.WaitForTrueCondition(() =>
+ {
+ taskFutures = GetActiveTaskFutures();
+ return taskFutures.Length == 2;
+ });
+
+ // Cancel and assert that the future from the step above is no longer present.
cts.Cancel();
Assert.IsTrue(task.IsCanceled);
+
+ TestUtils.WaitForTrueCondition(() =>
+ {
+ var futures = GetActiveTaskFutures();
+
+ return futures.Length == 1 && !taskFutures.Contains(futures[0]);
+ }, message: "Unexpected number of active tasks: " + GetActiveTaskFutures().Length);
+ }
+
+ /// <summary>
+ /// Tests that cancellation does not affect finished tasks.
+ /// </summary>
+ [Test]
+ public void TestExecuteJavaTaskAsyncCancelAfterFinish()
+ {
+ const long delayMs = 1;
+ var cts = new CancellationTokenSource();
+ var task = Client.GetCompute().ExecuteJavaTaskAsync<object>(TestTask, delayMs, cts.Token);
+
+ task.Wait(cts.Token);
+ cts.Cancel();
+
+ Assert.IsTrue(task.IsCompleted);
+ Assert.IsFalse(task.IsFaulted);
}
/// <summary>
@@ -268,16 +309,16 @@ namespace Apache.Ignite.Core.Tests.Client.Compute
return ((ArrayList) res.Item2).OfType<Guid>().ToArray();
};
-
+
// Default: full cluster.
var nodeIds = Client.GetCluster().GetNodes().Select(n => n.Id).ToArray();
CollectionAssert.AreEquivalent(nodeIds, getProjection(Client.GetCompute()));
-
+
// One node.
var nodeId = nodeIds[1];
var proj = Client.GetCluster().ForPredicate(n => n.Id == nodeId);
-
+
Assert.AreEqual(new[]{nodeId}, getProjection(proj.GetCompute()));
// Two nodes.
@@ -293,7 +334,7 @@ namespace Apache.Ignite.Core.Tests.Client.Compute
public void TestExecuteJavaTaskWithMixedModifiers()
{
const long timeoutMs = 200;
-
+
var cluster = Client.GetCluster();
var nodeId = cluster.GetNode().Id;
@@ -306,19 +347,19 @@ namespace Apache.Ignite.Core.Tests.Client.Compute
// KeepBinary.
var res = compute.ExecuteJavaTask<IBinaryObject>(
ComputeApiTest.EchoTask, ComputeApiTest.EchoTypeBinarizableJava);
-
+
Assert.AreEqual(1, res.GetField<int>("field"));
// Timeout.
var ex = Assert.Throws<AggregateException>(() => compute.ExecuteJavaTask<object>(TestTask, timeoutMs * 3));
var clientEx = (IgniteClientException) ex.GetInnermostException();
-
+
StringAssert.StartsWith("Task timed out (check logs for error messages):", clientEx.Message);
-
+
// Flags.
Assert.IsFalse(compute.ExecuteJavaTask<bool>(TestFailoverTask, null));
Assert.IsFalse(compute.WithNoResultCache().ExecuteJavaTask<bool>(TestResultCacheTask, null));
-
+
// Cluster.
var executedNodeIds = (ArrayList) compute.ExecuteJavaTask<IgniteBiTuple>(TestTask, null).Item2;
Assert.AreEqual(nodeId, executedNodeIds.Cast<Guid>().Single());
@@ -334,12 +375,12 @@ namespace Apache.Ignite.Core.Tests.Client.Compute
() => Client.GetCompute().ExecuteJavaTask<int>("bad", null));
var innerEx = ex.GetInnermostException();
-
+
StringAssert.StartsWith(
- "Unknown task name or failed to auto-deploy task (was task (re|un)deployed?) [taskName=bad, ",
+ "Unknown task name or failed to auto-deploy task (was task (re|un)deployed?) [taskName=bad, ",
innerEx.Message);
}
-
+
/// <summary>
/// Tests <see cref="IComputeClient.ExecuteJavaTask{TRes}"/> with exceeded
/// <see cref="ThinClientConfiguration.MaxActiveComputeTasksPerConnection"/>.
@@ -355,9 +396,10 @@ namespace Apache.Ignite.Core.Tests.Client.Compute
.ToArray();
var ex = Assert.Throws<AggregateException>(() => Task.WaitAll(tasks));
- var clientEx = ex.GetInnermostException();
+ var clientEx = (IgniteClientException) ex.GetInnermostException();
StringAssert.StartsWith("Active compute tasks per connection limit (8) exceeded", clientEx.Message);
+ Assert.AreEqual(ClientStatusCode.TooManyComputeTasks, clientEx.StatusCode);
}
/// <summary>
@@ -370,7 +412,7 @@ namespace Apache.Ignite.Core.Tests.Client.Compute
var compute = Client.GetCompute().WithKeepBinary();
var cache = Client.GetOrCreateCache<int, int>(TestUtils.TestName);
cache[1] = 1;
-
+
TestUtils.RunMultiThreaded(() =>
{
while (true)
@@ -378,7 +420,7 @@ namespace Apache.Ignite.Core.Tests.Client.Compute
var arg = new PlatformComputeNetBinarizable { Field = Interlocked.Decrement(ref count) };
var res = compute.ExecuteJavaTask<int>(ComputeApiTest.BinaryArgTask, arg);
-
+
Assert.AreEqual(arg.Field, res);
// Perform other operations to mix notifications and responses.
@@ -403,14 +445,14 @@ namespace Apache.Ignite.Core.Tests.Client.Compute
var task = Client.GetCompute().ExecuteJavaTaskAsync<object>(TestTask, (long) 500);
Client.GetCacheNames();
-
+
Assert.IsFalse(task.IsCompleted);
-
+
task.Wait();
}
/// <summary>
- /// Tests that failed deserialization (caused by missing type) produces correct exception.
+ /// Tests that failed deserialization (caused by missing type) produces correct exception.
/// </summary>
[Test]
public void TestExecuteJavaTaskWithFailedResultDeserializationProducesBinaryObjectException()
@@ -418,9 +460,9 @@ namespace Apache.Ignite.Core.Tests.Client.Compute
var ex = Assert.Throws<AggregateException>(
() => Client.GetCompute().ExecuteJavaTask<object>(
ComputeApiTest.EchoTask, ComputeApiTest.EchoTypeBinarizableJava));
-
+
var clientEx = (IgniteClientException) ex.GetInnermostException();
-
+
Assert.AreEqual("Failed to resolve Java class 'org.apache.ignite.platform.PlatformComputeJavaBinarizable'" +
" in .NET [platformId=1, typeId=-422570294].", clientEx.Message);
}
@@ -432,7 +474,7 @@ namespace Apache.Ignite.Core.Tests.Client.Compute
public void TestClientTimeoutShorterThanTaskDuration()
{
const long timeoutMs = 300;
-
+
var cfg = new IgniteClientConfiguration(GetClientConfiguration())
{
SocketTimeout = TimeSpan.FromMilliseconds(timeoutMs)
@@ -443,7 +485,7 @@ namespace Apache.Ignite.Core.Tests.Client.Compute
var compute = client.GetCompute().WithKeepBinary();
var res = compute.ExecuteJavaTask<IgniteBiTuple>(TestTask, timeoutMs * 3);
-
+
Assert.IsInstanceOf<Guid>(res.Item1);
}
}
@@ -472,5 +514,13 @@ namespace Apache.Ignite.Core.Tests.Client.Compute
EnablePartitionAwareness = false
};
}
+
+ /// <summary>
+ /// Gets active task futures from all server nodes.
+ /// </summary>
+ private IgniteGuid[] GetActiveTaskFutures()
+ {
+ return Client.GetCompute().ExecuteJavaTask<IgniteGuid[]>(ActiveTaskFuturesTask, null);
+ }
}
-}
\ No newline at end of file
+}
diff --git a/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Process/ListDataReader.cs b/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Process/ListDataReader.cs
index e201c96..5660cf9 100644
--- a/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Process/ListDataReader.cs
+++ b/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Process/ListDataReader.cs
@@ -32,6 +32,11 @@ namespace Apache.Ignite.Core.Tests.Process
/** <inheritdoc /> */
public void OnOutput(Process proc, string data, bool err)
{
+ if (string.IsNullOrEmpty(data))
+ {
+ return;
+ }
+
lock (_list)
{
_list.Add(data);
diff --git a/modules/platforms/dotnet/Apache.Ignite.Core/Client/ClientStatusCode.cs b/modules/platforms/dotnet/Apache.Ignite.Core/Client/ClientStatusCode.cs
index 3bdd9e1..b7ce9e7 100644
--- a/modules/platforms/dotnet/Apache.Ignite.Core/Client/ClientStatusCode.cs
+++ b/modules/platforms/dotnet/Apache.Ignite.Core/Client/ClientStatusCode.cs
@@ -50,16 +50,26 @@ namespace Apache.Ignite.Core.Client
CacheExists = 1001,
/// <summary>
- /// The too many cursors (see <see cref="ClientConnectorConfiguration.MaxOpenCursorsPerConnection"/>).
+ /// Too many cursors (see <see cref="ClientConnectorConfiguration.MaxOpenCursorsPerConnection"/>).
/// </summary>
TooManyCursors = 1010,
/// <summary>
+ /// Resource does not exist.
+ /// </summary>
+ ResourceDoesNotExist = 1011,
+
+ /// <summary>
/// Authorization failure.
/// </summary>
SecurityViolation = 1012,
/// <summary>
+ /// Too many compute tasks (see <see cref="ThinClientConfiguration.MaxActiveComputeTasksPerConnection"/>).
+ /// </summary>
+ TooManyComputeTasks = 1030,
+
+ /// <summary>
/// Authentication failed.
/// </summary>
AuthenticationFailed = 2000
diff --git a/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Client/Cache/CacheClient.cs b/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Client/Cache/CacheClient.cs
index 67a3059..b741afe 100644
--- a/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Client/Cache/CacheClient.cs
+++ b/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Client/Cache/CacheClient.cs
@@ -172,7 +172,7 @@ namespace Apache.Ignite.Core.Impl.Client.Cache
{
IgniteArgumentCheck.NotNull(keys, "keys");
- return DoOutInOp(ClientOp.CacheGetAll, ctx => ctx.Writer.WriteEnumerable(keys),
+ return DoOutInOp(ClientOp.CacheGetAll, ctx => ctx.Writer.WriteEnumerable(keys),
s => ReadCacheEntries(s.Stream));
}
@@ -181,7 +181,7 @@ namespace Apache.Ignite.Core.Impl.Client.Cache
{
IgniteArgumentCheck.NotNull(keys, "keys");
- return DoOutInOpAsync(ClientOp.CacheGetAll, ctx => ctx.Writer.WriteEnumerable(keys),
+ return DoOutInOpAsync(ClientOp.CacheGetAll, ctx => ctx.Writer.WriteEnumerable(keys),
s => ReadCacheEntries(s.Stream));
}
@@ -227,7 +227,7 @@ namespace Apache.Ignite.Core.Impl.Client.Cache
{
IgniteArgumentCheck.NotNull(keys, "keys");
- return DoOutInOp(ClientOp.CacheContainsKeys, ctx => ctx.Writer.WriteEnumerable(keys),
+ return DoOutInOp(ClientOp.CacheContainsKeys, ctx => ctx.Writer.WriteEnumerable(keys),
ctx => ctx.Stream.ReadBool());
}
@@ -236,7 +236,7 @@ namespace Apache.Ignite.Core.Impl.Client.Cache
{
IgniteArgumentCheck.NotNull(keys, "keys");
- return DoOutInOpAsync(ClientOp.CacheContainsKeys, ctx => ctx.Writer.WriteEnumerable(keys),
+ return DoOutInOpAsync(ClientOp.CacheContainsKeys, ctx => ctx.Writer.WriteEnumerable(keys),
ctx => ctx.Stream.ReadBool());
}
@@ -249,7 +249,7 @@ namespace Apache.Ignite.Core.Impl.Client.Cache
// For .NET it is a CacheEntryFilterHolder with a predefined id (BinaryTypeId.CacheEntryPredicateHolder).
return DoOutInOp(ClientOp.QueryScan, w => WriteScanQuery(w.Writer, scanQuery),
ctx => new ClientQueryCursor<TK, TV>(
- _ignite, ctx.Stream.ReadLong(), _keepBinary, ctx.Stream, ClientOp.QueryScanCursorGetPage));
+ ctx.Socket, ctx.Stream.ReadLong(), _keepBinary, ctx.Stream, ClientOp.QueryScanCursorGetPage));
}
/** <inheritDoc /> */
@@ -262,7 +262,7 @@ namespace Apache.Ignite.Core.Impl.Client.Cache
return DoOutInOp(ClientOp.QuerySql, w => WriteSqlQuery(w.Writer, sqlQuery),
ctx => new ClientQueryCursor<TK, TV>(
- _ignite, ctx.Stream.ReadLong(), _keepBinary, ctx.Stream, ClientOp.QuerySqlCursorGetPage));
+ ctx.Socket, ctx.Stream.ReadLong(), _keepBinary, ctx.Stream, ClientOp.QuerySqlCursorGetPage));
}
/** <inheritDoc /> */
@@ -281,7 +281,7 @@ namespace Apache.Ignite.Core.Impl.Client.Cache
{
return DoOutInOp(ClientOp.QuerySqlFields,
ctx => WriteSqlFieldsQuery(ctx.Writer, sqlFieldsQuery, false),
- ctx => GetFieldsCursorNoColumnNames(ctx.Stream, readerFunc));
+ ctx => GetFieldsCursorNoColumnNames(ctx, readerFunc));
}
/** <inheritDoc /> */
@@ -545,14 +545,14 @@ namespace Apache.Ignite.Core.Impl.Client.Cache
/** <inheritDoc /> */
public long GetSize(params CachePeekMode[] modes)
{
- return DoOutInOp(ClientOp.CacheGetSize, w => WritePeekModes(modes, w.Stream),
+ return DoOutInOp(ClientOp.CacheGetSize, w => WritePeekModes(modes, w.Stream),
ctx => ctx.Stream.ReadLong());
}
/** <inheritDoc /> */
public Task<long> GetSizeAsync(params CachePeekMode[] modes)
{
- return DoOutInOpAsync(ClientOp.CacheGetSize, w => WritePeekModes(modes, w.Stream),
+ return DoOutInOpAsync(ClientOp.CacheGetSize, w => WritePeekModes(modes, w.Stream),
ctx => ctx.Stream.ReadLong());
}
@@ -892,20 +892,20 @@ namespace Apache.Ignite.Core.Impl.Client.Cache
var cursorId = ctx.Stream.ReadLong();
var columnNames = ClientFieldsQueryCursor.ReadColumns(ctx.Reader);
- return new ClientFieldsQueryCursor(_ignite, cursorId, _keepBinary, ctx.Stream,
+ return new ClientFieldsQueryCursor(ctx.Socket, cursorId, _keepBinary, ctx.Stream,
ClientOp.QuerySqlFieldsCursorGetPage, columnNames);
}
/// <summary>
/// Gets the fields cursor.
/// </summary>
- private ClientQueryCursorBase<T> GetFieldsCursorNoColumnNames<T>(IBinaryStream stream,
+ private ClientQueryCursorBase<T> GetFieldsCursorNoColumnNames<T>(ClientResponseContext ctx,
Func<IBinaryRawReader, int, T> readerFunc)
{
- var cursorId = stream.ReadLong();
- var columnCount = stream.ReadInt();
+ var cursorId = ctx.Stream.ReadLong();
+ var columnCount = ctx.Stream.ReadInt();
- return new ClientQueryCursorBase<T>(_ignite, cursorId, _keepBinary, stream,
+ return new ClientQueryCursorBase<T>(ctx.Socket, cursorId, _keepBinary, ctx.Stream,
ClientOp.QuerySqlFieldsCursorGetPage, r => readerFunc(r, columnCount));
}
diff --git a/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Client/Cache/Query/ClientFieldsQueryCursor.cs b/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Client/Cache/Query/ClientFieldsQueryCursor.cs
index 38d0c3e..4a1e799 100644
--- a/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Client/Cache/Query/ClientFieldsQueryCursor.cs
+++ b/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Client/Cache/Query/ClientFieldsQueryCursor.cs
@@ -33,15 +33,15 @@ namespace Apache.Ignite.Core.Impl.Client.Cache.Query
/// <summary>
/// Initializes a new instance of the <see cref="ClientQueryCursor{TK, TV}" /> class.
/// </summary>
- /// <param name="ignite">The ignite.</param>
+ /// <param name="socket">Connection that holds the cursor.</param>
/// <param name="cursorId">The cursor identifier.</param>
/// <param name="keepBinary">Keep binary flag.</param>
/// <param name="initialBatchStream">Optional stream with initial batch.</param>
/// <param name="getPageOp">The get page op.</param>
/// <param name="columns">The columns.</param>
- public ClientFieldsQueryCursor(IgniteClient ignite, long cursorId, bool keepBinary,
+ public ClientFieldsQueryCursor(ClientSocket socket, long cursorId, bool keepBinary,
IBinaryStream initialBatchStream, ClientOp getPageOp, IList<string> columns)
- : base(ignite, cursorId, keepBinary, initialBatchStream, getPageOp,
+ : base(socket, cursorId, keepBinary, initialBatchStream, getPageOp,
r =>
{
var res = new List<object>(columns.Count);
diff --git a/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Client/Cache/Query/ClientQueryCursor.cs b/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Client/Cache/Query/ClientQueryCursor.cs
index 8e09af7..692c6ce 100644
--- a/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Client/Cache/Query/ClientQueryCursor.cs
+++ b/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Client/Cache/Query/ClientQueryCursor.cs
@@ -30,14 +30,14 @@ namespace Apache.Ignite.Core.Impl.Client.Cache.Query
/// <summary>
/// Initializes a new instance of the <see cref="ClientQueryCursor{TK, TV}" /> class.
/// </summary>
- /// <param name="ignite">The ignite.</param>
+ /// <param name="socket">Connection that holds the cursor.</param>
/// <param name="cursorId">The cursor identifier.</param>
/// <param name="keepBinary">Keep binary flag.</param>
/// <param name="initialBatchStream">Optional stream with initial batch.</param>
/// <param name="getPageOp">The get page op.</param>
- public ClientQueryCursor(IgniteClient ignite, long cursorId, bool keepBinary,
+ public ClientQueryCursor(ClientSocket socket, long cursorId, bool keepBinary,
IBinaryStream initialBatchStream, ClientOp getPageOp)
- : base(ignite, cursorId, keepBinary, initialBatchStream, getPageOp,
+ : base(socket, cursorId, keepBinary, initialBatchStream, getPageOp,
r => new CacheEntry<TK, TV>(r.ReadObject<TK>(), r.ReadObject<TV>()))
{
// No-op.
diff --git a/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Client/Cache/Query/ClientQueryCursorBase.cs b/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Client/Cache/Query/ClientQueryCursorBase.cs
index 5f70126..de49cb6 100644
--- a/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Client/Cache/Query/ClientQueryCursorBase.cs
+++ b/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Client/Cache/Query/ClientQueryCursorBase.cs
@@ -29,7 +29,7 @@ namespace Apache.Ignite.Core.Impl.Client.Cache.Query
internal class ClientQueryCursorBase<T> : QueryCursorBase<T>
{
/** Ignite. */
- private readonly IgniteClient _ignite;
+ private readonly ClientSocket _socket;
/** Cursor ID. */
private readonly long _cursorId;
@@ -40,17 +40,17 @@ namespace Apache.Ignite.Core.Impl.Client.Cache.Query
/// <summary>
/// Initializes a new instance of the <see cref="ClientQueryCursorBase{T}" /> class.
/// </summary>
- /// <param name="ignite">The ignite.</param>
+ /// <param name="socket">Connection that holds the cursor.</param>
/// <param name="cursorId">The cursor identifier.</param>
/// <param name="keepBinary">Keep binary flag.</param>
/// <param name="initialBatchStream">Optional stream with initial batch.</param>
/// <param name="getPageOp">The get page op.</param>
/// <param name="readFunc">Read func.</param>
- public ClientQueryCursorBase(IgniteClient ignite, long cursorId, bool keepBinary,
- IBinaryStream initialBatchStream, ClientOp getPageOp, Func<BinaryReader, T> readFunc)
- : base(ignite.Marshaller, keepBinary, readFunc, initialBatchStream)
+ public ClientQueryCursorBase(ClientSocket socket, long cursorId, bool keepBinary,
+ IBinaryStream initialBatchStream, ClientOp getPageOp, Func<BinaryReader, T> readFunc)
+ : base(socket.Marshaller, keepBinary, readFunc, initialBatchStream)
{
- _ignite = ignite;
+ _socket = socket;
_cursorId = cursorId;
_getPageOp = getPageOp;
}
@@ -70,7 +70,7 @@ namespace Apache.Ignite.Core.Impl.Client.Cache.Query
/** <inheritdoc /> */
protected override T[] GetBatch()
{
- return _ignite.Socket.DoOutInOp(_getPageOp, ctx => ctx.Stream.WriteLong(_cursorId),
+ return _socket.DoOutInOp(_getPageOp, ctx => ctx.Stream.WriteLong(_cursorId),
ctx => ConvertGetBatch(ctx.Stream));
}
@@ -79,7 +79,8 @@ namespace Apache.Ignite.Core.Impl.Client.Cache.Query
{
try
{
- _ignite.Socket.DoOutInOp<object>(ClientOp.ResourceClose, ctx => ctx.Writer.WriteLong(_cursorId), null);
+ _socket.DoOutInOp<object>(ClientOp.ResourceClose,
+ ctx => ctx.Writer.WriteLong(_cursorId), null);
}
finally
{
@@ -87,4 +88,4 @@ namespace Apache.Ignite.Core.Impl.Client.Cache.Query
}
}
}
-}
\ No newline at end of file
+}
diff --git a/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Client/ClientSocket.cs b/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Client/ClientSocket.cs
index 3e1a875..ba13550 100644
--- a/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Client/ClientSocket.cs
+++ b/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Client/ClientSocket.cs
@@ -281,7 +281,7 @@ namespace Apache.Ignite.Core.Impl.Client
{
get { return _features; }
}
-
+
/// <summary>
/// Gets the current remote EndPoint.
/// </summary>
@@ -335,6 +335,8 @@ namespace Apache.Ignite.Core.Impl.Client
[SuppressMessage("Microsoft.Design", "CA1031:DoNotCatchGeneralExceptionTypes")]
private void WaitForMessages()
{
+ _logger.Trace("Receiver thread #{0} started.", Thread.CurrentThread.ManagedThreadId);
+
try
{
// Null exception means active socket.
@@ -364,7 +366,7 @@ namespace Apache.Ignite.Core.Impl.Client
}
var msg = ReceiveMessage();
-
+
HandleResponse(msg);
}
}
@@ -377,6 +379,10 @@ namespace Apache.Ignite.Core.Impl.Client
_exception = ex;
Dispose();
}
+ finally
+ {
+ _logger.Trace("Receiver thread #{0} stopped.", Thread.CurrentThread.ManagedThreadId);
+ }
}
/// <summary>
@@ -419,7 +425,7 @@ namespace Apache.Ignite.Core.Impl.Client
{
return false;
}
-
+
var flags = (ClientFlags) stream.ReadShort();
stream.Seek(-2, SeekOrigin.Current);
@@ -431,12 +437,12 @@ namespace Apache.Ignite.Core.Impl.Client
var count = Interlocked.Decrement(ref _expectedNotifications);
if (count < 0)
{
- throw new IgniteClientException("Unexpected thin client notification: " + requestId);
+ throw new IgniteClientException("Unexpected thin client notification: " + requestId);
}
-
+
_notificationListeners.GetOrAdd(requestId, _ => new ClientNotificationHandler(_logger))
.Handle(stream, null);
-
+
return true;
}
@@ -472,8 +478,8 @@ namespace Apache.Ignite.Core.Impl.Client
if (statusCode == ClientStatusCode.Success)
{
- return readFunc != null
- ? readFunc(new ClientResponseContext(stream, this))
+ return readFunc != null
+ ? readFunc(new ClientResponseContext(stream, this))
: default(T);
}
@@ -513,7 +519,7 @@ namespace Apache.Ignite.Core.Impl.Client
// Writing features.
if (hasFeatures)
{
- BinaryUtils.Marshaller.Marshal(stream,
+ BinaryUtils.Marshaller.Marshal(stream,
w => w.WriteByteArray(ClientFeatures.AllFeatures));
}
@@ -541,7 +547,7 @@ namespace Apache.Ignite.Core.Impl.Client
if (success)
{
BitArray featureBits = null;
-
+
if (hasFeatures)
{
featureBits = new BitArray(BinaryUtils.Marshaller.Unmarshal<byte[]>(stream));
@@ -553,8 +559,8 @@ namespace Apache.Ignite.Core.Impl.Client
}
ServerVersion = version;
-
- _logger.Debug("Handshake completed on {0}, protocol version = {1}",
+
+ _logger.Debug("Handshake completed on {0}, protocol version = {1}",
_socket.RemoteEndPoint, version);
return new ClientFeatures(version, featureBits);
@@ -589,9 +595,9 @@ namespace Apache.Ignite.Core.Impl.Client
if (retry)
{
- _logger.Debug("Retrying handshake on {0} with protocol version {1}",
+ _logger.Debug("Retrying handshake on {0} with protocol version {1}",
_socket.RemoteEndPoint, ServerVersion);
-
+
return Handshake(clientConfiguration, ServerVersion);
}
@@ -755,12 +761,12 @@ namespace Apache.Ignite.Core.Impl.Client
private RequestMessage WriteMessage(Action<ClientRequestContext> writeAction, ClientOp opId)
{
_features.ValidateOp(opId);
-
+
var requestId = Interlocked.Increment(ref _requestId);
-
+
// Potential perf improvements:
// * ArrayPool<T>
- // * Write to socket stream directly (not trivial because of unknown size)
+ // * Write to socket stream directly (not trivial because of unknown size)
var stream = new BinaryHeapStream(256);
stream.WriteInt(0); // Reserve message size.
@@ -840,7 +846,7 @@ namespace Apache.Ignite.Core.Impl.Client
logger.Debug("Socket connection attempt: {0}", endPoint);
socket.Connect(endPoint);
-
+
logger.Debug("Socket connection established: {0} -> {1}", socket.LocalEndPoint, socket.RemoteEndPoint);
return socket;
diff --git a/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Client/Compute/ComputeClient.cs b/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Client/Compute/ComputeClient.cs
index 5ac6fad..4ef2a134 100644
--- a/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Client/Compute/ComputeClient.cs
+++ b/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Client/Compute/ComputeClient.cs
@@ -35,8 +35,8 @@ namespace Apache.Ignite.Core.Impl.Client.Compute
private readonly IgniteClient _ignite;
/** */
- private readonly ComputeClientFlags _flags;
-
+ private readonly ComputeClientFlags _flags;
+
/** */
private readonly TimeSpan _timeout;
@@ -47,8 +47,8 @@ namespace Apache.Ignite.Core.Impl.Client.Compute
/// Initializes a new instance of <see cref="ComputeClient"/>.
/// </summary>
internal ComputeClient(
- IgniteClient ignite,
- ComputeClientFlags flags,
+ IgniteClient ignite,
+ ComputeClientFlags flags,
TimeSpan? timeout,
IClientClusterGroup clusterGroup)
{
@@ -73,20 +73,20 @@ namespace Apache.Ignite.Core.Impl.Client.Compute
}
/** <inheritdoc /> */
- public Task<TRes> ExecuteJavaTaskAsync<TRes>(string taskName, object taskArg,
+ public Task<TRes> ExecuteJavaTaskAsync<TRes>(string taskName, object taskArg,
CancellationToken cancellationToken)
{
IgniteArgumentCheck.NotNullOrEmpty(taskName, "taskName");
-
+
var tcs = new TaskCompletionSource<TRes>();
- cancellationToken.Register(() => tcs.SetCanceled());
+ cancellationToken.Register(() => tcs.TrySetCanceled());
var keepBinary = (_flags & ComputeClientFlags.KeepBinary) == ComputeClientFlags.KeepBinary;
-
+
var task = _ignite.Socket.DoOutInOpAsync(
ClientOp.ComputeTaskExecute,
- ctx => WriteJavaTaskRequest(taskName, taskArg, ctx),
- ctx => ReadJavaTaskResponse(ctx, tcs, keepBinary));
+ ctx => WriteJavaTaskRequest(ctx, taskName, taskArg),
+ ctx => ReadJavaTaskResponse(ctx, tcs, cancellationToken, keepBinary));
// ReSharper disable once AssignNullToNotNullAttribute (t.Exception won't be null).
task.ContinueWith(t => tcs.TrySetException(t.Exception), TaskContinuationOptions.OnlyOnFaulted);
@@ -98,8 +98,8 @@ namespace Apache.Ignite.Core.Impl.Client.Compute
/** <inheritdoc /> */
public IComputeClient WithTimeout(TimeSpan timeout)
{
- return _timeout != timeout
- ? new ComputeClient(_ignite, _flags, timeout, _clusterGroup)
+ return _timeout != timeout
+ ? new ComputeClient(_ignite, _flags, timeout, _clusterGroup)
: this;
}
@@ -136,7 +136,7 @@ namespace Apache.Ignite.Core.Impl.Client.Compute
/// <summary>
/// Writes the java task.
/// </summary>
- private void WriteJavaTaskRequest(string taskName, object taskArg, ClientRequestContext ctx)
+ private void WriteJavaTaskRequest(ClientRequestContext ctx, string taskName, object taskArg)
{
var writer = ctx.Writer;
@@ -159,19 +159,36 @@ namespace Apache.Ignite.Core.Impl.Client.Compute
writer.WriteLong((long) _timeout.TotalMilliseconds);
writer.WriteString(taskName);
writer.WriteObject(taskArg);
-
+
ctx.Socket.ExpectNotifications();
}
-
+
/// <summary>
/// Reads java task execution response.
/// </summary>
[SuppressMessage("Microsoft.Design", "CA1031:DoNotCatchGeneralExceptionTypes")]
private static object ReadJavaTaskResponse<TRes>(ClientResponseContext ctx, TaskCompletionSource<TRes> tcs,
- bool keepBinary)
+ CancellationToken cancellationToken, bool keepBinary)
{
var taskId = ctx.Stream.ReadLong();
+ cancellationToken.Register(() =>
+ {
+ ctx.Socket.DoOutInOpAsync<object>(ClientOp.ResourceClose,
+ c => c.Stream.WriteLong(taskId),
+ _ => null,
+ (status, msg) =>
+ {
+ if (status == ClientStatusCode.ResourceDoesNotExist)
+ {
+ // Task finished before we could cancel it - ignore.
+ return null;
+ }
+
+ throw new IgniteClientException(msg, null, status);
+ });
+ });
+
ctx.Socket.AddNotificationHandler(taskId, (stream, ex) =>
{
if (ex != null)
@@ -182,7 +199,7 @@ namespace Apache.Ignite.Core.Impl.Client.Compute
ctx.Socket.RemoveNotificationHandler(taskId);
- var reader = ctx.Marshaller.StartUnmarshal(stream,
+ var reader = ctx.Marshaller.StartUnmarshal(stream,
keepBinary ? BinaryMode.ForceBinary : BinaryMode.Deserialize);
try
@@ -217,4 +234,4 @@ namespace Apache.Ignite.Core.Impl.Client.Compute
return null;
}
}
-}
\ No newline at end of file
+}
diff --git a/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Common/DelegateConverter.cs b/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Common/DelegateConverter.cs
index 968a5c6..6daea7c 100644
--- a/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Common/DelegateConverter.cs
+++ b/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Common/DelegateConverter.cs
@@ -207,7 +207,7 @@ namespace Apache.Ignite.Core.Impl.Common
/// Compiles a generic ctor with arbitrary number of arguments.
/// </summary>
/// <typeparam name="T">Result func type.</typeparam>
- /// <param name="ctor">Contructor info.</param>
+ /// <param name="ctor">Constructor info.</param>
/// <param name="argTypes">Argument types.</param>
/// <param name="convertResultToObject">
/// Flag that indicates whether ctor return value should be converted to object.</param>
@@ -252,7 +252,7 @@ namespace Apache.Ignite.Core.Impl.Common
/// that takes an uninitialized object as a first arguments.
/// </summary>
/// <typeparam name="T">Result func type.</typeparam>
- /// <param name="ctor">Contructor info.</param>
+ /// <param name="ctor">Constructor info.</param>
/// <param name="argTypes">Argument types.</param>
/// <returns>
/// Compiled generic constructor.
@@ -312,7 +312,7 @@ namespace Apache.Ignite.Core.Impl.Common
}
/// <summary>
- /// Compiles a contructor that reads all arguments from a binary reader.
+ /// Compiles a constructor that reads all arguments from a binary reader.
/// </summary>
/// <typeparam name="T">Result type</typeparam>
/// <param name="ctor">The ctor.</param>