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 2019/09/09 19:52:18 UTC
[ignite] branch master updated: IGNITE-12012 .NET: Add
ICompute.WithExecutor
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 5215e67 IGNITE-12012 .NET: Add ICompute.WithExecutor
5215e67 is described below
commit 5215e67403f1bd7c49e6a609adc200a89c39cbf9
Author: Pavel Tupitsyn <pt...@apache.org>
AuthorDate: Mon Sep 9 22:45:22 2019 +0300
IGNITE-12012 .NET: Add ICompute.WithExecutor
* `ICompute.WithExecutor`
* `IgniteConfiguration.ExecutorConfiguration`
---
.../platform/compute/PlatformCompute.java | 25 +++++
.../platform/utils/PlatformConfigurationUtils.java | 33 ++++++-
.../platform/PlatformComputeGetThreadNameTask.java | 57 ++++++++++++
.../Apache.Ignite.Core.Tests.DotNetCore.csproj | 4 +
.../Apache.Ignite.Core.Tests.csproj | 4 +
.../ApiParity/IgniteConfigurationParityTest.cs | 1 -
.../Compute/ComputeApiTest.JavaTask.cs | 2 +-
.../Compute/ComputeWithExecutorTest.cs | 101 +++++++++++++++++++++
.../Compute/compute-grid-custom-executor.xml | 53 +++++++++++
.../Config/Compute/compute-grid1.xml | 7 ++
.../Config/full-config.xml | 4 +
.../IgniteConfigurationSerializerTest.cs | 16 +++-
.../IgniteConfigurationTest.cs | 16 +++-
.../Apache.Ignite.Core/Apache.Ignite.Core.csproj | 1 +
.../dotnet/Apache.Ignite.Core/Compute/ICompute.cs | 11 +++
.../Configuration/ExecutorConfiguration.cs | 47 ++++++++++
.../Apache.Ignite.Core/IgniteConfiguration.cs | 44 ++++++++-
.../IgniteConfigurationSection.xsd | 26 ++++++
.../Apache.Ignite.Core/Impl/Compute/Compute.cs | 10 +-
.../Apache.Ignite.Core/Impl/Compute/ComputeImpl.cs | 15 +++
20 files changed, 461 insertions(+), 16 deletions(-)
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 88d729c..7b62aa5 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
@@ -76,6 +76,9 @@ public class PlatformCompute extends PlatformAbstractTarget {
/** */
private static final int OP_WITH_NO_RESULT_CACHE = 9;
+ /** */
+ private static final int OP_WITH_EXECUTOR = 10;
+
/** Compute instance. */
private final IgniteComputeImpl compute;
@@ -101,6 +104,20 @@ public class PlatformCompute extends PlatformAbstractTarget {
computeForPlatform = (IgniteComputeImpl)grp.ignite().compute(platformGrp);
}
+ /**
+ * Constructor.
+ *
+ * @param platformCtx Context.
+ * @param compute Compute.
+ * @param computeForPlatform Compute over platform-specific nodes.
+ */
+ private PlatformCompute(PlatformContext platformCtx, IgniteComputeImpl compute,
+ IgniteComputeImpl computeForPlatform) {
+ super(platformCtx);
+ this.compute = compute;
+ this.computeForPlatform = computeForPlatform;
+ }
+
/** {@inheritDoc} */
@Override public PlatformTarget processInStreamOutObject(int type, BinaryRawReaderEx reader)
throws IgniteCheckedException {
@@ -126,6 +143,14 @@ public class PlatformCompute extends PlatformAbstractTarget {
case OP_EXEC_ASYNC:
return wrapListenable((PlatformListenable) executeJavaTask(reader, true));
+ case OP_WITH_EXECUTOR: {
+ String executorName = reader.readString();
+
+ return new PlatformCompute(platformCtx,
+ (IgniteComputeImpl)compute.withExecutor(executorName),
+ (IgniteComputeImpl)computeForPlatform.withExecutor(executorName));
+ }
+
default:
return super.processInStreamOutObject(type, reader);
}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/utils/PlatformConfigurationUtils.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/utils/PlatformConfigurationUtils.java
index 872ec9c..38b3fdf 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/utils/PlatformConfigurationUtils.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/utils/PlatformConfigurationUtils.java
@@ -62,6 +62,7 @@ import org.apache.ignite.configuration.DataPageEvictionMode;
import org.apache.ignite.configuration.DataRegionConfiguration;
import org.apache.ignite.configuration.DataStorageConfiguration;
import org.apache.ignite.configuration.DiskPageCompression;
+import org.apache.ignite.configuration.ExecutorConfiguration;
import org.apache.ignite.configuration.IgniteConfiguration;
import org.apache.ignite.configuration.MemoryConfiguration;
import org.apache.ignite.configuration.MemoryPolicyConfiguration;
@@ -668,9 +669,7 @@ public class PlatformConfigurationUtils {
int sqlSchemasCnt = in.readInt();
- if (sqlSchemasCnt == -1)
- cfg.setSqlSchemas((String[])null);
- else {
+ if (sqlSchemasCnt >= 0) {
String[] sqlSchemas = new String[sqlSchemasCnt];
for (int i = 0; i < sqlSchemasCnt; i++)
@@ -841,6 +840,20 @@ public class PlatformConfigurationUtils {
}
}
+ int execCfgCnt = in.readInt();
+
+ if (execCfgCnt > 0) {
+ ExecutorConfiguration[] execCfgs = new ExecutorConfiguration[execCfgCnt];
+
+ for (int i = 0; i < execCfgCnt; i++) {
+ execCfgs[i] = new ExecutorConfiguration()
+ .setName(in.readString())
+ .setSize(in.readInt());
+ }
+
+ cfg.setExecutorConfiguration(execCfgs);
+ }
+
readPluginConfiguration(cfg, in);
readLocalEventListeners(cfg, in);
@@ -1259,7 +1272,7 @@ public class PlatformConfigurationUtils {
w.writeInt(cfg.getSqlQueryHistorySize());
if (cfg.getSqlSchemas() == null)
- w.writeInt(-1);
+ w.writeInt(0);
else {
w.writeInt(cfg.getSqlSchemas().length);
@@ -1439,6 +1452,18 @@ public class PlatformConfigurationUtils {
} else
w.writeBoolean(false);
+ ExecutorConfiguration[] execCfgs = cfg.getExecutorConfiguration();
+
+ if (execCfgs != null) {
+ w.writeInt(execCfgs.length);
+
+ for (ExecutorConfiguration execCfg : execCfgs) {
+ w.writeString(execCfg.getName());
+ w.writeInt(execCfg.getSize());
+ }
+ } else
+ w.writeInt(0);
+
w.writeString(cfg.getIgniteHome());
w.writeLong(ManagementFactory.getMemoryMXBean().getHeapMemoryUsage().getInit());
diff --git a/modules/core/src/test/java/org/apache/ignite/platform/PlatformComputeGetThreadNameTask.java b/modules/core/src/test/java/org/apache/ignite/platform/PlatformComputeGetThreadNameTask.java
new file mode 100644
index 0000000..17d1200
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/platform/PlatformComputeGetThreadNameTask.java
@@ -0,0 +1,57 @@
+/*
+ * 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.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.internal.util.typedef.F;
+import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Task to get Java thread name.
+ */
+public class PlatformComputeGetThreadNameTask extends ComputeTaskAdapter<Object, String> {
+ /** {@inheritDoc} */
+ @NotNull @Override public Map<? extends ComputeJob, ClusterNode> map(List<ClusterNode> subgrid,
+ @Nullable Object arg) {
+ return Collections.singletonMap(new GetThreadNameTask(), F.first(subgrid));
+ }
+
+ /** {@inheritDoc} */
+ @Nullable @Override public String reduce(List<ComputeJobResult> results) {
+ return results.get(0).getData();
+ }
+
+ /**
+ * Job.
+ */
+ private static class GetThreadNameTask extends ComputeJobAdapter {
+ /** {@inheritDoc} */
+ @Nullable @Override public Object execute() {
+ return Thread.currentThread().getName();
+ }
+ }
+}
diff --git a/modules/platforms/dotnet/Apache.Ignite.Core.Tests.DotNetCore/Apache.Ignite.Core.Tests.DotNetCore.csproj b/modules/platforms/dotnet/Apache.Ignite.Core.Tests.DotNetCore/Apache.Ignite.Core.Tests.DotNetCore.csproj
index 7a9fb3e..370e00b 100644
--- a/modules/platforms/dotnet/Apache.Ignite.Core.Tests.DotNetCore/Apache.Ignite.Core.Tests.DotNetCore.csproj
+++ b/modules/platforms/dotnet/Apache.Ignite.Core.Tests.DotNetCore/Apache.Ignite.Core.Tests.DotNetCore.csproj
@@ -114,6 +114,7 @@
<Compile Include="..\Apache.Ignite.Core.Tests\Common\IgniteProductVersionTests.cs" Link="Common\IgniteProductVersionTests.cs" />
<Compile Include="..\Apache.Ignite.Core.Tests\Compute\ComputeApiTest.cs" Link="Compute\ComputeApiTest.cs" />
<Compile Include="..\Apache.Ignite.Core.Tests\Compute\ComputeApiTest.JavaTask.cs" Link="Compute\ComputeApiTest.JavaTask.cs" />
+ <Compile Include="..\Apache.Ignite.Core.Tests\Compute\ComputeWithExecutorTest.cs" Link="Compute\ComputeWithExecutorTest.cs" />
<Compile Include="..\Apache.Ignite.Core.Tests\DataStructures\AtomicLongTest.cs" Link="DataStructures\AtomicLongTest.cs" />
<Compile Include="..\Apache.Ignite.Core.Tests\DataStructures\AtomicReferenceTest.cs" Link="DataStructures\AtomicReferenceTest.cs" />
<Compile Include="..\Apache.Ignite.Core.Tests\DataStructures\AtomicSequenceTest.cs" Link="DataStructures\AtomicSequenceTest.cs" />
@@ -164,6 +165,9 @@
<Content Include="..\Apache.Ignite.Core.Tests\Config\Compute\compute-grid3.xml" Link="Config\Compute\compute-grid3.xml">
<CopyToOutputDirectory>PreserveNewest</CopyToOutputDirectory>
</Content>
+ <Content Include="..\Apache.Ignite.Core.Tests\Config\Compute\compute-grid-custom-executor.xml" Link="Config\Compute\compute-grid-custom-executor.xml">
+ <CopyToOutputDirectory>PreserveNewest</CopyToOutputDirectory>
+ </Content>
<Content Include="..\Apache.Ignite.Core.Tests\Config\native-client-test-cache.xml" Link="Config\native-client-test-cache.xml">
<CopyToOutputDirectory>PreserveNewest</CopyToOutputDirectory>
</Content>
diff --git a/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Apache.Ignite.Core.Tests.csproj b/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Apache.Ignite.Core.Tests.csproj
index 360355a..2e57bdd 100644
--- a/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Apache.Ignite.Core.Tests.csproj
+++ b/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Apache.Ignite.Core.Tests.csproj
@@ -159,6 +159,7 @@
<Compile Include="Client\IgniteClientConfigurationTest.cs" />
<Compile Include="Common\IgniteProductVersionTests.cs" />
<Compile Include="Compute\ComputeApiTest.JavaTask.cs" />
+ <Compile Include="Compute\ComputeWithExecutorTest.cs" />
<Compile Include="Deployment\CacheGetFunc.cs" />
<Compile Include="Deployment\GetAddressFunc.cs" />
<Compile Include="Deployment\PeerAssemblyLoadingAllApisTest.cs" />
@@ -409,6 +410,9 @@
<Content Include="Config\Client\server-with-ssl.xml">
<CopyToOutputDirectory>PreserveNewest</CopyToOutputDirectory>
</Content>
+ <Content Include="Config\Compute\compute-grid-custom-executor.xml">
+ <CopyToOutputDirectory>PreserveNewest</CopyToOutputDirectory>
+ </Content>
<Content Include="Config\Compute\compute-grid1.xml">
<CopyToOutputDirectory>PreserveNewest</CopyToOutputDirectory>
</Content>
diff --git a/modules/platforms/dotnet/Apache.Ignite.Core.Tests/ApiParity/IgniteConfigurationParityTest.cs b/modules/platforms/dotnet/Apache.Ignite.Core.Tests/ApiParity/IgniteConfigurationParityTest.cs
index 68efe74..50686f9 100644
--- a/modules/platforms/dotnet/Apache.Ignite.Core.Tests/ApiParity/IgniteConfigurationParityTest.cs
+++ b/modules/platforms/dotnet/Apache.Ignite.Core.Tests/ApiParity/IgniteConfigurationParityTest.cs
@@ -64,7 +64,6 @@ namespace Apache.Ignite.Core.Tests.ApiParity
"ClassLoader",
"CacheStoreSessionListenerFactories",
"PlatformConfiguration",
- "ExecutorConfiguration",
"CommunicationFailureResolver",
"EncryptionSpi",
"MetricExporterSpi"
diff --git a/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Compute/ComputeApiTest.JavaTask.cs b/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Compute/ComputeApiTest.JavaTask.cs
index be0a203..b2e50df 100644
--- a/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Compute/ComputeApiTest.JavaTask.cs
+++ b/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Compute/ComputeApiTest.JavaTask.cs
@@ -42,7 +42,7 @@ namespace Apache.Ignite.Core.Tests.Compute
/** Broadcast task name. */
public const string BroadcastTask = "org.apache.ignite.platform.PlatformComputeBroadcastTask";
- /** Broadcast task name. */
+ /** Decimal task name. */
private const string DecimalTask = "org.apache.ignite.platform.PlatformComputeDecimalTask";
/** Echo type: null. */
diff --git a/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Compute/ComputeWithExecutorTest.cs b/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Compute/ComputeWithExecutorTest.cs
new file mode 100644
index 0000000..c8972c9
--- /dev/null
+++ b/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Compute/ComputeWithExecutorTest.cs
@@ -0,0 +1,101 @@
+/*
+ * 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.Tests.Compute
+{
+ using Apache.Ignite.Core.Compute;
+ using Apache.Ignite.Core.Configuration;
+ using NUnit.Framework;
+
+ /// <summary>
+ /// Tests <see cref="ICompute.WithExecutor"/>.
+ /// </summary>
+ public class ComputeWithExecutorTest
+ {
+ /** Thread name task name. */
+ private const string ThreadNameTask = "org.apache.ignite.platform.PlatformComputeGetThreadNameTask";
+
+ /** */
+ private IIgnite _springConfigGrid;
+
+ /** */
+ private IIgnite _codeConfigGrid;
+
+ /// <summary>
+ /// Fixture set up.
+ /// </summary>
+ [TestFixtureSetUp]
+ public void TestFixtureSetUp()
+ {
+ var cfg1 = new IgniteConfiguration(TestUtils.GetTestConfiguration())
+ {
+ SpringConfigUrl = @"Config\Compute\compute-grid-custom-executor.xml"
+ };
+
+ _springConfigGrid = Ignition.Start(cfg1);
+
+ var cfg2 = new IgniteConfiguration(TestUtils.GetTestConfiguration())
+ {
+ ExecutorConfiguration = new[]
+ {
+ new ExecutorConfiguration
+ {
+ Name = "dotNetExecutor",
+ Size = 3
+ },
+ new ExecutorConfiguration
+ {
+ Name = "dotNetExecutor2",
+ Size = 1
+ }
+ }
+ };
+
+ _codeConfigGrid = Ignition.Start(cfg2);
+ }
+
+ /// <summary>
+ /// Fixture tear down.
+ /// </summary>
+ [TestFixtureTearDown]
+ public void TestFixtureTearDown()
+ {
+ Ignition.StopAll(true);
+ }
+
+ /// <summary>
+ /// Tests custom executors.
+ /// </summary>
+ [TestCase("dotNetExecutor", "dotNetExecutor-#", true)]
+ [TestCase("dotNetExecutor2", "pub-#", true)]
+ [TestCase("invalid", "pub-#", true)]
+ [TestCase("dotNetExecutor", "dotNetExecutor-#", false)]
+ [TestCase("dotNetExecutor2", "dotNetExecutor2-#", false)]
+ [TestCase("invalid", "pub-#", false)]
+ public void TestWithExecutor(string executorName, string expectedThreadNamePrefix, bool springConfig)
+ {
+ var grid = springConfig ? _springConfigGrid : _codeConfigGrid;
+ var compute = grid.GetCluster().ForLocal().GetCompute();
+ var computeWithExecutor = compute.WithExecutor(executorName);
+
+ var res = computeWithExecutor.ExecuteJavaTask<string>(ThreadNameTask, null);
+
+ Assert.AreNotSame(compute, computeWithExecutor);
+ Assert.AreEqual(expectedThreadNamePrefix, res.Substring(0, expectedThreadNamePrefix.Length));
+ }
+ }
+}
diff --git a/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Config/Compute/compute-grid-custom-executor.xml b/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Config/Compute/compute-grid-custom-executor.xml
new file mode 100644
index 0000000..ad5a30a
--- /dev/null
+++ b/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Config/Compute/compute-grid-custom-executor.xml
@@ -0,0 +1,53 @@
+<?xml version="1.0" encoding="UTF-8"?>
+
+<!--
+ 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.
+-->
+
+<beans xmlns="http://www.springframework.org/schema/beans"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://www.springframework.org/schema/beans
+ http://www.springframework.org/schema/beans/spring-beans.xsd">
+ <bean id="grid.cfg" class="org.apache.ignite.configuration.IgniteConfiguration">
+ <property name="localHost" value="127.0.0.1"/>
+ <property name="connectorConfiguration"><null/></property>
+
+ <property name="igniteInstanceName" value="grid1"/>
+
+ <property name="discoverySpi">
+ <bean class="org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi">
+ <property name="ipFinder">
+ <bean class="org.apache.ignite.spi.discovery.tcp.ipfinder.vm.TcpDiscoveryVmIpFinder">
+ <property name="addresses">
+ <list>
+ <!-- In distributed environment, replace with actual host IP address. -->
+ <value>127.0.0.1:47500</value>
+ </list>
+ </property>
+ </bean>
+ </property>
+ <property name="socketTimeout" value="300" />
+ </bean>
+ </property>
+
+ <property name="executorConfiguration">
+ <bean class="org.apache.ignite.configuration.ExecutorConfiguration">
+ <property name="name" value="dotNetExecutor" />
+ <property name="size" value="2" />
+ </bean>
+ </property>
+ </bean>
+</beans>
diff --git a/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Config/Compute/compute-grid1.xml b/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Config/Compute/compute-grid1.xml
index c170182..2c08233 100644
--- a/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Config/Compute/compute-grid1.xml
+++ b/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Config/Compute/compute-grid1.xml
@@ -101,5 +101,12 @@
<property name="socketTimeout" value="300" />
</bean>
</property>
+
+ <property name="executorConfiguration">
+ <bean class="org.apache.ignite.configuration.ExecutorConfiguration">
+ <property name="name" value="dotNetExecutor" />
+ <property name="size" value="2" />
+ </bean>
+ </property>
</bean>
</beans>
diff --git a/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Config/full-config.xml b/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Config/full-config.xml
index 7314529..224ad66 100644
--- a/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Config/full-config.xml
+++ b/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Config/full-config.xml
@@ -166,4 +166,8 @@
keyStoreFilePath='KeyStore/server.jks' keyStorePassword='123456'
trustStoreFilePath='KeyStore/trust.jks' trustStorePassword='123456'/>
<failureHandler type="StopNodeOrHaltFailureHandler" tryStop="true" timeout="0:1:0"/>
+ <executorConfiguration>
+ <executorConfiguration name='exec1' size='1' />
+ <executorConfiguration name='exec2' size='2' />
+ </executorConfiguration>
</igniteConfiguration>
diff --git a/modules/platforms/dotnet/Apache.Ignite.Core.Tests/IgniteConfigurationSerializerTest.cs b/modules/platforms/dotnet/Apache.Ignite.Core.Tests/IgniteConfigurationSerializerTest.cs
index 571eb5b..23e81c9 100644
--- a/modules/platforms/dotnet/Apache.Ignite.Core.Tests/IgniteConfigurationSerializerTest.cs
+++ b/modules/platforms/dotnet/Apache.Ignite.Core.Tests/IgniteConfigurationSerializerTest.cs
@@ -359,6 +359,12 @@ namespace Apache.Ignite.Core.Tests
Assert.IsTrue(failureHandler.TryStop);
Assert.AreEqual(TimeSpan.Parse("0:1:0"), failureHandler.Timeout);
+
+ var ec = cfg.ExecutorConfiguration;
+ Assert.NotNull(ec);
+ Assert.AreEqual(2, ec.Count);
+ Assert.AreEqual(new[] {"exec1", "exec2"}, ec.Select(e => e.Name));
+ Assert.AreEqual(new[] {1, 2}, ec.Select(e => e.Size));
}
/// <summary>
@@ -1029,7 +1035,15 @@ namespace Apache.Ignite.Core.Tests
TryStop = false,
Timeout = TimeSpan.FromSeconds(10)
},
- SqlQueryHistorySize = 345
+ SqlQueryHistorySize = 345,
+ ExecutorConfiguration = new[]
+ {
+ new ExecutorConfiguration
+ {
+ Name = "exec-1",
+ Size = 11
+ }
+ }
};
}
diff --git a/modules/platforms/dotnet/Apache.Ignite.Core.Tests/IgniteConfigurationTest.cs b/modules/platforms/dotnet/Apache.Ignite.Core.Tests/IgniteConfigurationTest.cs
index 01fe2c8..3cc1ef4 100644
--- a/modules/platforms/dotnet/Apache.Ignite.Core.Tests/IgniteConfigurationTest.cs
+++ b/modules/platforms/dotnet/Apache.Ignite.Core.Tests/IgniteConfigurationTest.cs
@@ -106,7 +106,7 @@ namespace Apache.Ignite.Core.Tests
}
/// <summary>
- /// Tests all configuration properties.
+ /// Tests all configuration properties roundtrip to Java and back.
/// </summary>
[Test]
public void TestAllConfigurationProperties()
@@ -267,6 +267,9 @@ namespace Apache.Ignite.Core.Tests
Assert.AreEqual(2, resCfg.SqlSchemas.Count);
Assert.IsTrue(resCfg.SqlSchemas.Contains("SCHEMA_3"));
Assert.IsTrue(resCfg.SqlSchemas.Contains("schema_4"));
+
+ Assert.NotNull(cfg.ExecutorConfiguration);
+ AssertExtensions.ReflectionEqual(cfg.ExecutorConfiguration, resCfg.ExecutorConfiguration);
}
}
@@ -877,8 +880,15 @@ namespace Apache.Ignite.Core.Tests
MvccVacuumFrequency = 20000,
MvccVacuumThreadCount = 8,
SqlQueryHistorySize = 99,
-
- SqlSchemas = new List<string> { "SCHEMA_3", "schema_4" }
+ SqlSchemas = new List<string> { "SCHEMA_3", "schema_4" },
+ ExecutorConfiguration = new[]
+ {
+ new ExecutorConfiguration
+ {
+ Name = "ex-1",
+ Size = 11
+ }
+ }
};
}
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 303d6a6..09ba954 100644
--- a/modules/platforms/dotnet/Apache.Ignite.Core/Apache.Ignite.Core.csproj
+++ b/modules/platforms/dotnet/Apache.Ignite.Core/Apache.Ignite.Core.csproj
@@ -75,6 +75,7 @@
<Compile Include="Client\IgniteClientException.cs" />
<Compile Include="Client\IIgniteClient.cs" />
<Compile Include="Common\ExceptionFactory.cs" />
+ <Compile Include="Configuration\ExecutorConfiguration.cs" />
<Compile Include="Failure\IFailureHandler.cs" />
<Compile Include="Failure\NoOpFailureHandler.cs" />
<Compile Include="Failure\StopNodeFailureHandler.cs" />
diff --git a/modules/platforms/dotnet/Apache.Ignite.Core/Compute/ICompute.cs b/modules/platforms/dotnet/Apache.Ignite.Core/Compute/ICompute.cs
index 7d70f97..20f847a 100644
--- a/modules/platforms/dotnet/Apache.Ignite.Core/Compute/ICompute.cs
+++ b/modules/platforms/dotnet/Apache.Ignite.Core/Compute/ICompute.cs
@@ -85,6 +85,17 @@ namespace Apache.Ignite.Core.Compute
ICompute WithKeepBinary();
/// <summary>
+ /// Gets instance of the compute API associated with custom executor. All tasks and closures submitted to
+ /// returned instance will be processed by this executor on both remote and local nodes.
+ /// If an executor with the given name doesn't exist, task will be processed in default ("public") pool.
+ /// <para/>
+ /// Executor should be defined in <see cref="IgniteConfiguration.ExecutorConfiguration"/>.
+ /// </summary>
+ /// <param name="executorName">Executor name.</param>
+ /// <returns>New Compute instance associated with a custom executor.</returns>
+ ICompute WithExecutor(string executorName);
+
+ /// <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>
diff --git a/modules/platforms/dotnet/Apache.Ignite.Core/Configuration/ExecutorConfiguration.cs b/modules/platforms/dotnet/Apache.Ignite.Core/Configuration/ExecutorConfiguration.cs
new file mode 100644
index 0000000..d4a07d0
--- /dev/null
+++ b/modules/platforms/dotnet/Apache.Ignite.Core/Configuration/ExecutorConfiguration.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.Configuration
+{
+ using Apache.Ignite.Core.Compute;
+
+ /// <summary>
+ /// Custom thread pool configuration for compute tasks.
+ /// See <see cref="ICompute.WithExecutor"/>.
+ /// </summary>
+ public class ExecutorConfiguration
+ {
+ /** */
+ private int? _size;
+
+ /// <summary>
+ /// Gets or sets the thread pool name.
+ /// Can not be null and should be unique with respect to other custom executors.
+ /// </summary>
+ public string Name { get; set; }
+
+ /// <summary>
+ /// Gets or sets the thread pool size.
+ /// Defaults to <see cref="IgniteConfiguration.DefaultThreadPoolSize"/>
+ /// </summary>
+ public int Size
+ {
+ get { return _size ?? IgniteConfiguration.DefaultThreadPoolSize; }
+ set { _size = value; }
+ }
+ }
+}
diff --git a/modules/platforms/dotnet/Apache.Ignite.Core/IgniteConfiguration.cs b/modules/platforms/dotnet/Apache.Ignite.Core/IgniteConfiguration.cs
index b5be2be..4c99e29 100644
--- a/modules/platforms/dotnet/Apache.Ignite.Core/IgniteConfiguration.cs
+++ b/modules/platforms/dotnet/Apache.Ignite.Core/IgniteConfiguration.cs
@@ -344,7 +344,7 @@ namespace Apache.Ignite.Core
writer.WriteIntNullable(_sqlQueryHistorySize);
if (SqlSchemas == null)
- writer.WriteInt(-1);
+ writer.WriteInt(0);
else
{
writer.WriteInt(SqlSchemas.Count);
@@ -620,6 +620,20 @@ namespace Apache.Ignite.Core
}
}
+ if (ExecutorConfiguration == null)
+ {
+ writer.WriteInt(0);
+ }
+ else
+ {
+ writer.WriteInt(ExecutorConfiguration.Count);
+ foreach (var exec in ExecutorConfiguration)
+ {
+ writer.WriteString(exec.Name);
+ writer.WriteInt(exec.Size);
+ }
+ }
+
// Plugins (should be last).
if (PluginConfigurations != null)
{
@@ -736,9 +750,7 @@ namespace Apache.Ignite.Core
int sqlSchemasCnt = r.ReadInt();
- if (sqlSchemasCnt == -1)
- SqlSchemas = null;
- else
+ if (sqlSchemasCnt > 0)
{
SqlSchemas = new List<string>(sqlSchemasCnt);
@@ -869,7 +881,7 @@ namespace Apache.Ignite.Core
// SSL context factory.
SslContextFactory = SslFactorySerializer.Read(r);
- //Failure handler.
+ // Failure handler.
if (r.ReadBoolean())
{
switch (r.ReadByte())
@@ -899,6 +911,22 @@ namespace Apache.Ignite.Core
{
FailureHandler = null;
}
+
+ // Executor configuration.
+ var count = r.ReadInt();
+ if (count >= 0)
+ {
+ ExecutorConfiguration = new List<ExecutorConfiguration>(count);
+
+ for (var i = 0; i < count; i++)
+ {
+ ExecutorConfiguration.Add(new ExecutorConfiguration
+ {
+ Name = r.ReadString(),
+ Size = r.ReadInt()
+ });
+ }
+ }
}
/// <summary>
@@ -1678,5 +1706,11 @@ namespace Apache.Ignite.Core
/// </summary>
[SuppressMessage("Microsoft.Usage", "CA2227:CollectionPropertiesShouldBeReadOnly")]
public ICollection<string> SqlSchemas { get; set; }
+
+ /// <summary>
+ /// Gets or sets custom executor configuration for compute tasks.
+ /// </summary>
+ [SuppressMessage("Microsoft.Usage", "CA2227:CollectionPropertiesShouldBeReadOnly")]
+ public ICollection<ExecutorConfiguration> ExecutorConfiguration { get; set; }
}
}
diff --git a/modules/platforms/dotnet/Apache.Ignite.Core/IgniteConfigurationSection.xsd b/modules/platforms/dotnet/Apache.Ignite.Core/IgniteConfigurationSection.xsd
index 60b1276..cd16c6e 100644
--- a/modules/platforms/dotnet/Apache.Ignite.Core/IgniteConfigurationSection.xsd
+++ b/modules/platforms/dotnet/Apache.Ignite.Core/IgniteConfigurationSection.xsd
@@ -2133,6 +2133,32 @@
</xs:sequence>
</xs:complexType>
</xs:element>
+ <xs:element name="executorConfiguration" minOccurs="0">
+ <xs:annotation>
+ <xs:documentation>Custom executor configuration for compute tasks.</xs:documentation>
+ </xs:annotation>
+ <xs:complexType>
+ <xs:sequence>
+ <xs:element name="executorConfiguration" maxOccurs="unbounded" minOccurs="0">
+ <xs:annotation>
+ <xs:documentation>Thread pool configuration for compute tasks.</xs:documentation>
+ </xs:annotation>
+ <xs:complexType>
+ <xs:attribute type="xs:string" name="name" use="required">
+ <xs:annotation>
+ <xs:documentation>Thread pool name.</xs:documentation>
+ </xs:annotation>
+ </xs:attribute>
+ <xs:attribute type="xs:int" name="size" use="required">
+ <xs:annotation>
+ <xs:documentation>Thread pool size.</xs:documentation>
+ </xs:annotation>
+ </xs:attribute>
+ </xs:complexType>
+ </xs:element>
+ </xs:sequence>
+ </xs:complexType>
+ </xs:element>
</xs:all>
<xs:attribute name="igniteInstanceName" type="xs:string">
<xs:annotation>
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 efea1c7..48f3eb7 100644
--- a/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Compute/Compute.cs
+++ b/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Compute/Compute.cs
@@ -84,6 +84,14 @@ namespace Apache.Ignite.Core.Impl.Compute
}
/** <inheritDoc /> */
+ public ICompute WithExecutor(string executorName)
+ {
+ var computeImpl = _compute.WithExecutor(executorName);
+
+ return new Compute(computeImpl);
+ }
+
+ /** <inheritDoc /> */
public TReduceRes ExecuteJavaTask<TReduceRes>(string taskName, object taskArg)
{
return _compute.ExecuteJavaTask<TReduceRes>(taskName, taskArg);
@@ -487,4 +495,4 @@ namespace Apache.Ignite.Core.Impl.Compute
return false;
}
}
-}
\ No newline at end of file
+}
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 05efe3c..b8f892f 100644
--- a/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Compute/ComputeImpl.cs
+++ b/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Compute/ComputeImpl.cs
@@ -68,6 +68,9 @@ namespace Apache.Ignite.Core.Impl.Compute
/** */
private const int OpWithNoResultCache = 9;
+ /** */
+ private const int OpWithExecutor = 10;
+
/** Underlying projection. */
private readonly ClusterGroupImpl _prj;
@@ -139,6 +142,18 @@ namespace Apache.Ignite.Core.Impl.Compute
}
/// <summary>
+ /// Returns a new <see cref="ComputeImpl"/> instance associated with a specified executor.
+ /// </summary>
+ /// <param name="executorName">Executor name.</param>
+ /// <returns>New <see cref="ComputeImpl"/> instance associated with a specified executor.</returns>
+ public ComputeImpl WithExecutor(string executorName)
+ {
+ var target = DoOutOpObject(OpWithExecutor, w => w.WriteString(executorName));
+
+ return new ComputeImpl(target, _prj, _keepBinary.Value);
+ }
+
+ /// <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>