You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@reef.apache.org by tm...@apache.org on 2015/02/11 23:58:06 UTC
[1/4] incubator-reef git commit: [REEF-150] Adding group
communication to REEF .Net
Repository: incubator-reef
Updated Branches:
refs/heads/master 7edb8570e -> 0292caf14
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/0292caf1/lang/cs/Org.Apache.REEF.Network/Group/Topology/TaskNode.cs
----------------------------------------------------------------------
diff --git a/lang/cs/Org.Apache.REEF.Network/Group/Topology/TaskNode.cs b/lang/cs/Org.Apache.REEF.Network/Group/Topology/TaskNode.cs
new file mode 100644
index 0000000..54a46a1
--- /dev/null
+++ b/lang/cs/Org.Apache.REEF.Network/Group/Topology/TaskNode.cs
@@ -0,0 +1,69 @@
+/**
+ * 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.
+ */
+
+using Org.Apache.REEF.Tang.Annotations;
+using Org.Apache.REEF.Tang.Util;
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.Threading.Tasks;
+
+namespace Org.Apache.REEF.Network.Group.Topology
+{
+ /// <summary>
+ /// Represents a node in the operator topology graph.
+ /// </summary>
+ internal class TaskNode
+ {
+ private string _groupName;
+ private string _operatorName;
+ private string _taskId;
+ private string _driverId;
+
+ private TaskNode _parent;
+ private List<TaskNode> _children;
+
+ public TaskNode(
+ string groupName,
+ string operatorName,
+ string taskId,
+ string driverId)
+ {
+ _groupName = groupName;
+ _operatorName = operatorName;
+ _taskId = taskId;
+ _driverId = driverId;
+
+ _children = new List<TaskNode>();
+ }
+
+ public string TaskId { get; private set; }
+
+ public void AddChild(TaskNode child)
+ {
+ _children.Add(child);
+ }
+
+ public void SetParent(TaskNode parent)
+ {
+ _parent = parent;
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/0292caf1/lang/cs/Org.Apache.REEF.Network/Org.Apache.REEF.Network.csproj
----------------------------------------------------------------------
diff --git a/lang/cs/Org.Apache.REEF.Network/Org.Apache.REEF.Network.csproj b/lang/cs/Org.Apache.REEF.Network/Org.Apache.REEF.Network.csproj
index b759c24..fb6e183 100644
--- a/lang/cs/Org.Apache.REEF.Network/Org.Apache.REEF.Network.csproj
+++ b/lang/cs/Org.Apache.REEF.Network/Org.Apache.REEF.Network.csproj
@@ -95,6 +95,49 @@ under the License.
<Reference Include="System.Xml" />
</ItemGroup>
<ItemGroup>
+ <Compile Include="Group\Codec\GcmMessageProto.cs" />
+ <Compile Include="Group\Codec\GroupCommunicationMessageCodec.cs" />
+ <Compile Include="Group\Config\MpiConfigurationOptions.cs" />
+ <Compile Include="Group\Driver\ICommunicationGroupDriver.cs" />
+ <Compile Include="Group\Driver\IMpiDriver.cs" />
+ <Compile Include="Group\Driver\Impl\CommunicationGroupDriver.cs" />
+ <Compile Include="Group\Driver\Impl\GroupCommunicationMessage.cs" />
+ <Compile Include="Group\Driver\Impl\MessageType.cs" />
+ <Compile Include="Group\Driver\Impl\MpiDriver.cs" />
+ <Compile Include="Group\Driver\Impl\TaskStarter.cs" />
+ <Compile Include="Group\Operators\IBroadcastReceiver.cs" />
+ <Compile Include="Group\Operators\IBroadcastSender.cs" />
+ <Compile Include="Group\Operators\IMpiOperator.cs" />
+ <Compile Include="Group\Operators\Impl\BroadcastOperatorSpec.cs" />
+ <Compile Include="Group\Operators\Impl\BroadcastReceiver.cs" />
+ <Compile Include="Group\Operators\Impl\BroadcastSender.cs" />
+ <Compile Include="Group\Operators\Impl\ReduceFunction.cs" />
+ <Compile Include="Group\Operators\Impl\ReduceOperatorSpec.cs" />
+ <Compile Include="Group\Operators\Impl\ReduceReceiver.cs" />
+ <Compile Include="Group\Operators\Impl\ReduceSender.cs" />
+ <Compile Include="Group\Operators\Impl\ScatterOperatorSpec.cs" />
+ <Compile Include="Group\Operators\Impl\ScatterReceiver.cs" />
+ <Compile Include="Group\Operators\Impl\ScatterSender.cs" />
+ <Compile Include="Group\Operators\Impl\Sender.cs" />
+ <Compile Include="Group\Operators\IOperatorSpec.cs" />
+ <Compile Include="Group\Operators\IReduceFunction.cs" />
+ <Compile Include="Group\Operators\IReduceReceiver.cs" />
+ <Compile Include="Group\Operators\IReduceSender.cs" />
+ <Compile Include="Group\Operators\IScatterReceiver.cs" />
+ <Compile Include="Group\Operators\IScatterSender.cs" />
+ <Compile Include="Group\Task\ICommunicationGroupClient.cs" />
+ <Compile Include="Group\Task\ICommunicationGroupNetworkObserver.cs" />
+ <Compile Include="Group\Task\IMpiClient.cs" />
+ <Compile Include="Group\Task\IMpiNetworkObserver.cs" />
+ <Compile Include="Group\Task\Impl\CommunicationGroupClient.cs" />
+ <Compile Include="Group\Task\Impl\CommunicationGroupNetworkObserver.cs" />
+ <Compile Include="Group\Task\Impl\MpiClient.cs" />
+ <Compile Include="Group\Task\Impl\MpiNetworkObserver.cs" />
+ <Compile Include="Group\Task\Impl\NodeStruct.cs" />
+ <Compile Include="Group\Task\Impl\OperatorTopology.cs" />
+ <Compile Include="Group\Topology\FlatTopology.cs" />
+ <Compile Include="Group\Topology\ITopology.cs" />
+ <Compile Include="Group\Topology\TaskNode.cs" />
<Compile Include="Naming\Codec\NamingLookupRequestCodec.cs" />
<Compile Include="Naming\Codec\NamingLookupResponseCodec.cs" />
<Compile Include="Naming\Codec\NamingRegisterRequestCodec.cs" />
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/0292caf1/lang/cs/Org.Apache.REEF.Tests/Functional/ML/KMeans/TestKMeans.cs
----------------------------------------------------------------------
diff --git a/lang/cs/Org.Apache.REEF.Tests/Functional/ML/KMeans/TestKMeans.cs b/lang/cs/Org.Apache.REEF.Tests/Functional/ML/KMeans/TestKMeans.cs
new file mode 100644
index 0000000..8d00e69
--- /dev/null
+++ b/lang/cs/Org.Apache.REEF.Tests/Functional/ML/KMeans/TestKMeans.cs
@@ -0,0 +1,163 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+using System;
+using System.Collections.Generic;
+using System.Globalization;
+using System.IO;
+using Microsoft.VisualStudio.TestTools.UnitTesting;
+using Org.Apache.REEF.Common.Io;
+using Org.Apache.REEF.Driver.Bridge;
+using Org.Apache.REEF.Examples.MachineLearning.KMeans;
+using Org.Apache.REEF.Network.NetworkService;
+using Org.Apache.REEF.Tang.Interface;
+using Org.Apache.REEF.Tang.Util;
+using Org.Apache.REEF.Utilities.Logging;
+
+namespace Org.Apache.REEF.Tests.Functional.ML.KMeans
+{
+ [TestClass]
+ public class TestKMeans : ReefFunctionalTest
+ {
+ private const int K = 3;
+ private const int Partitions = 2;
+ private const string SmallMouseDataFile = @"mouseData_small.csv";
+ private const string MouseDataFile = @"mouseData.csv";
+
+ private bool _useSmallDataSet = false;
+ private string _dataFile = MouseDataFile;
+
+ [TestInitialize()]
+ public void TestSetup()
+ {
+ if (_useSmallDataSet)
+ {
+ _dataFile = SmallMouseDataFile;
+ }
+
+ CleanUp();
+ Init();
+ }
+
+ [TestCleanup]
+ public void TestCleanup()
+ {
+ Console.WriteLine("Post test check and clean up");
+ CleanUp();
+ }
+
+ [TestMethod, Priority(1), TestCategory("FunctionalGated")]
+ [Description("Test KMeans clustering with things directly run without reef")]
+ [DeploymentItem(@".")]
+ [DeploymentItem(@"Data", ".")]
+ [Ignore]
+ [Timeout(180 * 1000)]
+ public void TestKMeansOnDirectRunViaFileSystem()
+ {
+ int iteration = 0;
+ string executionDirectory = Path.Combine(Directory.GetCurrentDirectory(), Constants.KMeansExecutionBaseDirectory, Guid.NewGuid().ToString("N").Substring(0, 4));
+ List<DataVector> centroids = DataVector.ShuffleDataAndGetInitialCentriods(_dataFile, Partitions, K, executionDirectory);
+
+ // initialize all tasks
+ List<LegacyKMeansTask> tasks = new List<LegacyKMeansTask>();
+ List<DataVector> labeledData = new List<DataVector>();
+ for (int i = 0; i < Partitions; i++)
+ {
+ DataPartitionCache p = new DataPartitionCache(i, executionDirectory);
+ tasks.Add(new LegacyKMeansTask(p, K, executionDirectory));
+ labeledData.AddRange(p.DataVectors);
+ }
+
+ float loss = float.MaxValue;
+ while (true)
+ {
+ for (int i = 0; i < Partitions; i++)
+ {
+ tasks[i].CallWithWritingToFileSystem(null);
+ }
+ List<DataVector> newCentroids = PartialMean.AggregateTrueMeansToFileSystem(Partitions, K, executionDirectory);
+ DataVector.WriteToCentroidFile(newCentroids, executionDirectory);
+ centroids = newCentroids;
+ float newLoss = LegacyKMeansTask.ComputeLossFunction(centroids, labeledData);
+ if (newLoss > loss)
+ {
+ throw new InvalidOperationException(
+ string.Format(CultureInfo.InvariantCulture, "The new loss {0} is larger than previous loss {1}, while loss function must be monotonically decreasing across iterations", newLoss, loss));
+ }
+ else if (newLoss.Equals(loss))
+ {
+ Console.WriteLine(string.Format(CultureInfo.InvariantCulture, "KMeans clustering has converged with a loss value of {0} at iteration {1} ", newLoss, iteration));
+ break;
+ }
+ else
+ {
+ loss = newLoss;
+ }
+ iteration++;
+ }
+ }
+
+ [TestMethod, Priority(1), TestCategory("FunctionalGated")]
+ [Description("Test KMeans clustering on reef local runtime with group communications")]
+ [DeploymentItem(@".")]
+ [DeploymentItem(@"Data", ".")]
+ [Ignore]
+ [Timeout(180 * 1000)]
+ public void TestKMeansOnLocalRuntimeWithGroupCommunications()
+ {
+ IsOnLocalRuntiime = true;
+ TestRun(AssembliesToCopy(), DriverConfiguration());
+ ValidateSuccessForLocalRuntime(Partitions + 1);
+ }
+
+ [TestMethod, Priority(1), TestCategory("FunctionalGated")]
+ [Description("Test KMeans clustering on reef YARN runtime - one box")]
+ [DeploymentItem(@".")]
+ [DeploymentItem(@"Data", ".")]
+ [Timeout(180 * 1000)]
+ [Ignore] // ignored by default
+ public void TestKMeansOnYarnOneBoxWithGroupCommunications()
+ {
+ TestRun(AssembliesToCopy(), DriverConfiguration(), runOnYarn: true);
+ Assert.IsNotNull("BreakPointChecker");
+ }
+
+ private IConfiguration DriverConfiguration()
+ {
+ return DriverBridgeConfiguration.ConfigurationModule
+ .Set(DriverBridgeConfiguration.OnDriverStarted, GenericType<KMeansDriverHandlers>.Class)
+ .Set(DriverBridgeConfiguration.OnEvaluatorAllocated, GenericType<KMeansDriverHandlers>.Class)
+ .Set(DriverBridgeConfiguration.OnEvaluatorRequested, GenericType<KMeansDriverHandlers>.Class)
+ .Set(DriverBridgeConfiguration.OnContextActive, GenericType<KMeansDriverHandlers>.Class)
+ .Set(DriverBridgeConfiguration.CommandLineArguments, "DataFile:" + _dataFile)
+ .Set(DriverBridgeConfiguration.CustomTraceLevel, Level.Info.ToString())
+ .Build();
+ }
+
+ private HashSet<string> AssembliesToCopy()
+ {
+ HashSet<string> appDlls = new HashSet<string>();
+ appDlls.Add(typeof(KMeansDriverHandlers).Assembly.GetName().Name);
+ appDlls.Add(typeof(LegacyKMeansTask).Assembly.GetName().Name);
+ appDlls.Add(typeof(INameClient).Assembly.GetName().Name);
+ appDlls.Add(typeof(INetworkService<>).Assembly.GetName().Name);
+ return appDlls;
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/0292caf1/lang/cs/Org.Apache.REEF.Tests/Functional/MPI/BroadcastReduceTest/BroadcastReduceDriver.cs
----------------------------------------------------------------------
diff --git a/lang/cs/Org.Apache.REEF.Tests/Functional/MPI/BroadcastReduceTest/BroadcastReduceDriver.cs b/lang/cs/Org.Apache.REEF.Tests/Functional/MPI/BroadcastReduceTest/BroadcastReduceDriver.cs
new file mode 100644
index 0000000..7472d56
--- /dev/null
+++ b/lang/cs/Org.Apache.REEF.Tests/Functional/MPI/BroadcastReduceTest/BroadcastReduceDriver.cs
@@ -0,0 +1,187 @@
+/**
+ * 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.
+ */
+
+using System;
+using System.Collections.Generic;
+using System.Globalization;
+using System.Linq;
+using Org.Apache.REEF.Common.Io;
+using Org.Apache.REEF.Common.Tasks;
+using Org.Apache.REEF.Driver;
+using Org.Apache.REEF.Driver.Bridge;
+using Org.Apache.REEF.Driver.Context;
+using Org.Apache.REEF.Driver.Evaluator;
+using Org.Apache.REEF.Network.Group.Driver;
+using Org.Apache.REEF.Network.Group.Driver.Impl;
+using Org.Apache.REEF.Network.Group.Operators;
+using Org.Apache.REEF.Network.Group.Operators.Impl;
+using Org.Apache.REEF.Network.NetworkService;
+using Org.Apache.REEF.Tang.Annotations;
+using Org.Apache.REEF.Tang.Formats;
+using Org.Apache.REEF.Tang.Implementations.Tang;
+using Org.Apache.REEF.Tang.Interface;
+using Org.Apache.REEF.Tang.Util;
+using Org.Apache.REEF.Utilities.Logging;
+using Org.Apache.REEF.Wake.Remote.Impl;
+
+namespace Org.Apache.REEF.Tests.Functional.MPI.BroadcastReduceTest
+{
+ public class BroadcastReduceDriver : IStartHandler, IObserver<IEvaluatorRequestor>, IObserver<IAllocatedEvaluator>, IObserver<IActiveContext>, IObserver<IFailedEvaluator>
+ {
+ private static readonly Logger LOGGER = Logger.GetLogger(typeof(BroadcastReduceDriver));
+
+ private int _numEvaluators;
+ private int _numIterations;
+
+ private IMpiDriver _mpiDriver;
+ private ICommunicationGroupDriver _commGroup;
+ private TaskStarter _mpiTaskStarter;
+
+ [Inject]
+ public BroadcastReduceDriver(
+ [Parameter(typeof(MpiTestConfig.NumEvaluators))] int numEvaluators,
+ [Parameter(typeof(MpiTestConfig.NumIterations))] int numIterations,
+ AvroConfigurationSerializer confSerializer)
+ {
+ Identifier = "BroadcastStartHandler";
+ _numEvaluators = numEvaluators;
+ _numIterations = numIterations;
+
+ _mpiDriver = new MpiDriver(
+ MpiTestConstants.DriverId,
+ MpiTestConstants.MasterTaskId,
+ confSerializer);
+
+ _commGroup = _mpiDriver.NewCommunicationGroup(
+ MpiTestConstants.GroupName,
+ numEvaluators)
+ .AddBroadcast(
+ MpiTestConstants.BroadcastOperatorName,
+ new BroadcastOperatorSpec<int>(
+ MpiTestConstants.MasterTaskId,
+ new IntCodec()))
+ .AddReduce(
+ MpiTestConstants.ReduceOperatorName,
+ new ReduceOperatorSpec<int>(
+ MpiTestConstants.MasterTaskId,
+ new IntCodec(),
+ new SumFunction()))
+ .Build();
+
+ _mpiTaskStarter = new TaskStarter(_mpiDriver, numEvaluators);
+
+ CreateClassHierarchy();
+ }
+
+ public string Identifier { get; set; }
+
+ public void OnNext(IEvaluatorRequestor evaluatorRequestor)
+ {
+ EvaluatorRequest request = new EvaluatorRequest(_numEvaluators, 512, 2, "WonderlandRack", "BroadcastEvaluator");
+ evaluatorRequestor.Submit(request);
+ }
+
+ public void OnNext(IAllocatedEvaluator allocatedEvaluator)
+ {
+ IConfiguration contextConf = _mpiDriver.GetContextConfiguration();
+ IConfiguration serviceConf = _mpiDriver.GetServiceConfiguration();
+ allocatedEvaluator.SubmitContextAndService(contextConf, serviceConf);
+ }
+
+ public void OnNext(IActiveContext activeContext)
+ {
+ if (_mpiDriver.IsMasterTaskContext(activeContext))
+ {
+ // Configure Master Task
+ IConfiguration partialTaskConf = TangFactory.GetTang().NewConfigurationBuilder(
+ TaskConfiguration.ConfigurationModule
+ .Set(TaskConfiguration.Identifier, MpiTestConstants.MasterTaskId)
+ .Set(TaskConfiguration.Task, GenericType<MasterTask>.Class)
+ .Build())
+ .BindNamedParameter<MpiTestConfig.NumEvaluators, int>(
+ GenericType<MpiTestConfig.NumEvaluators>.Class,
+ _numEvaluators.ToString(CultureInfo.InvariantCulture))
+ .BindNamedParameter<MpiTestConfig.NumIterations, int>(
+ GenericType<MpiTestConfig.NumIterations>.Class,
+ _numIterations.ToString(CultureInfo.InvariantCulture))
+ .Build();
+
+ _commGroup.AddTask(MpiTestConstants.MasterTaskId);
+ _mpiTaskStarter.QueueTask(partialTaskConf, activeContext);
+ }
+ else
+ {
+ // Configure Slave Task
+ string slaveTaskId = "SlaveTask-" + activeContext.Id;
+ IConfiguration partialTaskConf = TangFactory.GetTang().NewConfigurationBuilder(
+ TaskConfiguration.ConfigurationModule
+ .Set(TaskConfiguration.Identifier, slaveTaskId)
+ .Set(TaskConfiguration.Task, GenericType<SlaveTask>.Class)
+ .Build())
+ .BindNamedParameter<MpiTestConfig.NumEvaluators, int>(
+ GenericType<MpiTestConfig.NumEvaluators>.Class,
+ _numEvaluators.ToString(CultureInfo.InvariantCulture))
+ .BindNamedParameter<MpiTestConfig.NumIterations, int>(
+ GenericType<MpiTestConfig.NumIterations>.Class,
+ _numIterations.ToString(CultureInfo.InvariantCulture))
+ .Build();
+
+ _commGroup.AddTask(slaveTaskId);
+ _mpiTaskStarter.QueueTask(partialTaskConf, activeContext);
+ }
+ }
+
+ public void OnNext(IFailedEvaluator value)
+ {
+ }
+
+ public void OnError(Exception error)
+ {
+ }
+
+ public void OnCompleted()
+ {
+ }
+
+ private void CreateClassHierarchy()
+ {
+ HashSet<string> clrDlls = new HashSet<string>();
+ clrDlls.Add(typeof(IDriver).Assembly.GetName().Name);
+ clrDlls.Add(typeof(ITask).Assembly.GetName().Name);
+ clrDlls.Add(typeof(BroadcastReduceDriver).Assembly.GetName().Name);
+ clrDlls.Add(typeof(INameClient).Assembly.GetName().Name);
+ clrDlls.Add(typeof(INetworkService<>).Assembly.GetName().Name);
+
+ ClrHandlerHelper.GenerateClassHierarchy(clrDlls);
+ }
+
+ private class SumFunction : IReduceFunction<int>
+ {
+ [Inject]
+ public SumFunction()
+ {
+ }
+
+ public int Reduce(IEnumerable<int> elements)
+ {
+ return elements.Sum();
+ }
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/0292caf1/lang/cs/Org.Apache.REEF.Tests/Functional/MPI/BroadcastReduceTest/BroadcastReduceTest.cs
----------------------------------------------------------------------
diff --git a/lang/cs/Org.Apache.REEF.Tests/Functional/MPI/BroadcastReduceTest/BroadcastReduceTest.cs b/lang/cs/Org.Apache.REEF.Tests/Functional/MPI/BroadcastReduceTest/BroadcastReduceTest.cs
new file mode 100644
index 0000000..c74c923
--- /dev/null
+++ b/lang/cs/Org.Apache.REEF.Tests/Functional/MPI/BroadcastReduceTest/BroadcastReduceTest.cs
@@ -0,0 +1,83 @@
+/**
+ * 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.
+ */
+
+using System.Collections.Generic;
+using System.Globalization;
+using Microsoft.VisualStudio.TestTools.UnitTesting;
+using Org.Apache.REEF.Common.Io;
+using Org.Apache.REEF.Common.Tasks;
+using Org.Apache.REEF.Driver;
+using Org.Apache.REEF.Driver.Bridge;
+using Org.Apache.REEF.Network.NetworkService;
+using Org.Apache.REEF.Tang.Implementations.Tang;
+using Org.Apache.REEF.Tang.Interface;
+using Org.Apache.REEF.Tang.Util;
+using Org.Apache.REEF.Utilities.Logging;
+
+namespace Org.Apache.REEF.Tests.Functional.MPI.BroadcastReduceTest
+{
+ [TestClass]
+ public class BroadcastReduceTest : ReefFunctionalTest
+ {
+ [TestInitialize]
+ public void TestSetup()
+ {
+ CleanUp();
+ }
+
+ [TestCleanup]
+ public void TestCleanup()
+ {
+ CleanUp();
+ }
+
+ [TestMethod]
+ public void TestBroadcastAndReduce()
+ {
+ int numTasks = 4;
+
+ IConfiguration driverConfig = TangFactory.GetTang().NewConfigurationBuilder(
+ DriverBridgeConfiguration.ConfigurationModule
+ .Set(DriverBridgeConfiguration.OnDriverStarted, GenericType<BroadcastReduceDriver>.Class)
+ .Set(DriverBridgeConfiguration.OnEvaluatorAllocated, GenericType<BroadcastReduceDriver>.Class)
+ .Set(DriverBridgeConfiguration.OnEvaluatorRequested, GenericType<BroadcastReduceDriver>.Class)
+ .Set(DriverBridgeConfiguration.OnEvaluatorFailed, GenericType<BroadcastReduceDriver>.Class)
+ .Set(DriverBridgeConfiguration.OnContextActive, GenericType<BroadcastReduceDriver>.Class)
+ .Set(DriverBridgeConfiguration.CustomTraceLevel, Level.Info.ToString())
+ .Build())
+ .BindNamedParameter<MpiTestConfig.NumIterations, int>(
+ GenericType<MpiTestConfig.NumIterations>.Class,
+ MpiTestConstants.NumIterations.ToString(CultureInfo.InvariantCulture))
+ .BindNamedParameter<MpiTestConfig.NumEvaluators, int>(
+ GenericType<MpiTestConfig.NumEvaluators>.Class,
+ numTasks.ToString(CultureInfo.InvariantCulture))
+ .Build();
+
+ HashSet<string> appDlls = new HashSet<string>();
+ appDlls.Add(typeof(IDriver).Assembly.GetName().Name);
+ appDlls.Add(typeof(ITask).Assembly.GetName().Name);
+ appDlls.Add(typeof(BroadcastReduceDriver).Assembly.GetName().Name);
+ appDlls.Add(typeof(INameClient).Assembly.GetName().Name);
+ appDlls.Add(typeof(INetworkService<>).Assembly.GetName().Name);
+
+ TestRun(appDlls, driverConfig, false, JavaLoggingSetting.VERBOSE);
+ ValidateSuccessForLocalRuntime(numTasks);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/0292caf1/lang/cs/Org.Apache.REEF.Tests/Functional/MPI/BroadcastReduceTest/MasterTask.cs
----------------------------------------------------------------------
diff --git a/lang/cs/Org.Apache.REEF.Tests/Functional/MPI/BroadcastReduceTest/MasterTask.cs b/lang/cs/Org.Apache.REEF.Tests/Functional/MPI/BroadcastReduceTest/MasterTask.cs
new file mode 100644
index 0000000..78f8840
--- /dev/null
+++ b/lang/cs/Org.Apache.REEF.Tests/Functional/MPI/BroadcastReduceTest/MasterTask.cs
@@ -0,0 +1,89 @@
+/**
+ * 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.
+ */
+
+using System;
+using System.Linq;
+using Org.Apache.REEF.Common.Tasks;
+using Org.Apache.REEF.Network.Group.Operators;
+using Org.Apache.REEF.Network.Group.Task;
+using Org.Apache.REEF.Tang.Annotations;
+using Org.Apache.REEF.Utilities.Logging;
+
+namespace Org.Apache.REEF.Tests.Functional.MPI.BroadcastReduceTest
+{
+ public class MasterTask : ITask
+ {
+ private static Logger _logger = Logger.GetLogger(typeof(MasterTask));
+
+ private int _numIters;
+ private int _numReduceSenders;
+
+ private IMpiClient _mpiClient;
+ private ICommunicationGroupClient _commGroup;
+ private IBroadcastSender<int> _broadcastSender;
+ private IReduceReceiver<int> _sumReducer;
+
+ [Inject]
+ public MasterTask(
+ [Parameter(typeof(MpiTestConfig.NumIterations))] int numIters,
+ [Parameter(typeof(MpiTestConfig.NumEvaluators))] int numEvaluators,
+ IMpiClient mpiClient)
+ {
+ _logger.Log(Level.Info, "Hello from master task");
+ _numIters = numIters;
+ _numReduceSenders = numEvaluators - 1;
+ _mpiClient = mpiClient;
+
+ _commGroup = mpiClient.GetCommunicationGroup(MpiTestConstants.GroupName);
+ _broadcastSender = _commGroup.GetBroadcastSender<int>(MpiTestConstants.BroadcastOperatorName);
+ _sumReducer = _commGroup.GetReduceReceiver<int>(MpiTestConstants.ReduceOperatorName);
+ }
+
+ public byte[] Call(byte[] memento)
+ {
+ for (int i = 1; i <= _numIters; i++)
+ {
+ // Each slave task calculates the nth triangle number
+ _broadcastSender.Send(i);
+
+ // Sum up all of the calculated triangle numbers
+ int sum = _sumReducer.Reduce();
+ _logger.Log(Level.Info, "Received sum: {0} on iteration: {1}", sum, i);
+
+ int expected = TriangleNumber(i) * _numReduceSenders;
+ if (sum != TriangleNumber(i) * _numReduceSenders)
+ {
+ throw new Exception("Expected " + expected + " but got " + sum);
+ }
+ }
+
+ return null;
+ }
+
+ public void Dispose()
+ {
+ _mpiClient.Dispose();
+ }
+
+ private int TriangleNumber(int n)
+ {
+ return Enumerable.Range(1, n).Sum();
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/0292caf1/lang/cs/Org.Apache.REEF.Tests/Functional/MPI/BroadcastReduceTest/SlaveTask.cs
----------------------------------------------------------------------
diff --git a/lang/cs/Org.Apache.REEF.Tests/Functional/MPI/BroadcastReduceTest/SlaveTask.cs b/lang/cs/Org.Apache.REEF.Tests/Functional/MPI/BroadcastReduceTest/SlaveTask.cs
new file mode 100644
index 0000000..20b9351
--- /dev/null
+++ b/lang/cs/Org.Apache.REEF.Tests/Functional/MPI/BroadcastReduceTest/SlaveTask.cs
@@ -0,0 +1,80 @@
+/**
+ * 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.
+ */
+
+using System.Linq;
+using Org.Apache.REEF.Common.Tasks;
+using Org.Apache.REEF.Network.Group.Operators;
+using Org.Apache.REEF.Network.Group.Task;
+using Org.Apache.REEF.Tang.Annotations;
+using Org.Apache.REEF.Utilities.Logging;
+
+namespace Org.Apache.REEF.Tests.Functional.MPI.BroadcastReduceTest
+{
+ public class SlaveTask : ITask
+ {
+ private static Logger _logger = Logger.GetLogger(typeof(SlaveTask));
+
+ private int _numIterations;
+ private IMpiClient _mpiClient;
+ private ICommunicationGroupClient _commGroup;
+ private IBroadcastReceiver<int> _broadcastReceiver;
+ private IReduceSender<int> _triangleNumberSender;
+
+ [Inject]
+ public SlaveTask(
+ [Parameter(typeof(MpiTestConfig.NumIterations))] int numIters,
+ IMpiClient mpiClient)
+ {
+ _logger.Log(Level.Info, "Hello from slave task");
+
+ _numIterations = numIters;
+ _mpiClient = mpiClient;
+ _commGroup = _mpiClient.GetCommunicationGroup(MpiTestConstants.GroupName);
+ _broadcastReceiver = _commGroup.GetBroadcastReceiver<int>(MpiTestConstants.BroadcastOperatorName);
+ _triangleNumberSender = _commGroup.GetReduceSender<int>(MpiTestConstants.ReduceOperatorName);
+ }
+
+ public byte[] Call(byte[] memento)
+ {
+ for (int i = 0; i < _numIterations; i++)
+ {
+ // Receive n from Master Task
+ int n = _broadcastReceiver.Receive();
+ _logger.Log(Level.Info, "Calculating TriangleNumber({0}) on slave task...", n);
+
+ // Calculate the nth Triangle number and send it back to driver
+ int triangleNum = TriangleNumber(n);
+ _logger.Log(Level.Info, "Sending sum: {0} on iteration {1}.", triangleNum, i);
+ _triangleNumberSender.Send(triangleNum);
+ }
+
+ return null;
+ }
+
+ public void Dispose()
+ {
+ _mpiClient.Dispose();
+ }
+
+ private int TriangleNumber(int n)
+ {
+ return Enumerable.Range(1, n).Sum();
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/0292caf1/lang/cs/Org.Apache.REEF.Tests/Functional/MPI/MpiTestConfig.cs
----------------------------------------------------------------------
diff --git a/lang/cs/Org.Apache.REEF.Tests/Functional/MPI/MpiTestConfig.cs b/lang/cs/Org.Apache.REEF.Tests/Functional/MPI/MpiTestConfig.cs
new file mode 100644
index 0000000..9450e20
--- /dev/null
+++ b/lang/cs/Org.Apache.REEF.Tests/Functional/MPI/MpiTestConfig.cs
@@ -0,0 +1,36 @@
+/**
+ * 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.
+ */
+
+using Org.Apache.REEF.Tang.Annotations;
+
+namespace Org.Apache.REEF.Tests.Functional.MPI
+{
+ public class MpiTestConfig
+ {
+ [NamedParameter("Number of iterations of messages to send")]
+ public class NumIterations : Name<int>
+ {
+ }
+
+ [NamedParameter("Number of Evaluators")]
+ public class NumEvaluators : Name<int>
+ {
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/0292caf1/lang/cs/Org.Apache.REEF.Tests/Functional/MPI/MpiTestConstants.cs
----------------------------------------------------------------------
diff --git a/lang/cs/Org.Apache.REEF.Tests/Functional/MPI/MpiTestConstants.cs b/lang/cs/Org.Apache.REEF.Tests/Functional/MPI/MpiTestConstants.cs
new file mode 100644
index 0000000..5cff899
--- /dev/null
+++ b/lang/cs/Org.Apache.REEF.Tests/Functional/MPI/MpiTestConstants.cs
@@ -0,0 +1,33 @@
+/**
+ * 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 Org.Apache.REEF.Tests.Functional.MPI
+{
+ internal class MpiTestConstants
+ {
+ public const string DriverId = "BroadcastReduceDriver";
+ public const string GroupName = "BroadcastReduceGroup";
+ public const string BroadcastOperatorName = "Broadcast";
+ public const string ReduceOperatorName = "Reduce";
+ public const string ScatterOperatorName = "Scatter";
+ public const string MasterTaskId = "MasterTask";
+ public const string SlaveTaskId = "SlaveTask-";
+ public const int NumIterations = 10;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/0292caf1/lang/cs/Org.Apache.REEF.Tests/Functional/MPI/ScatterReduceTest/MasterTask.cs
----------------------------------------------------------------------
diff --git a/lang/cs/Org.Apache.REEF.Tests/Functional/MPI/ScatterReduceTest/MasterTask.cs b/lang/cs/Org.Apache.REEF.Tests/Functional/MPI/ScatterReduceTest/MasterTask.cs
new file mode 100644
index 0000000..07168c2
--- /dev/null
+++ b/lang/cs/Org.Apache.REEF.Tests/Functional/MPI/ScatterReduceTest/MasterTask.cs
@@ -0,0 +1,72 @@
+/**
+ * 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.
+ */
+
+using System.Collections.Generic;
+using System.Linq;
+using Org.Apache.REEF.Common.Tasks;
+using Org.Apache.REEF.Network.Group.Operators;
+using Org.Apache.REEF.Network.Group.Task;
+using Org.Apache.REEF.Tang.Annotations;
+using Org.Apache.REEF.Utilities.Logging;
+
+namespace Org.Apache.REEF.Tests.Functional.MPI.ScatterReduceTest
+{
+ public class MasterTask : ITask
+ {
+ private static Logger _logger = Logger.GetLogger(typeof(MasterTask));
+
+ private IMpiClient _mpiClient;
+ private ICommunicationGroupClient _commGroup;
+ private IScatterSender<int> _scatterSender;
+ private IReduceReceiver<int> _sumReducer;
+
+ [Inject]
+ public MasterTask(IMpiClient mpiClient)
+ {
+ _logger.Log(Level.Info, "Hello from master task");
+ _mpiClient = mpiClient;
+
+ _commGroup = mpiClient.GetCommunicationGroup(MpiTestConstants.GroupName);
+ _scatterSender = _commGroup.GetScatterSender<int>(MpiTestConstants.ScatterOperatorName);
+ _sumReducer = _commGroup.GetReduceReceiver<int>(MpiTestConstants.ReduceOperatorName);
+ }
+
+ public byte[] Call(byte[] memento)
+ {
+ List<int> data = Enumerable.Range(1, 100).ToList();
+ List<string> order = GetScatterOrder();
+ _scatterSender.Send(data, order);
+
+ int sum = _sumReducer.Reduce();
+ _logger.Log(Level.Info, "Received sum: {0}", sum);
+
+ return null;
+ }
+
+ public void Dispose()
+ {
+ _mpiClient.Dispose();
+ }
+
+ private List<string> GetScatterOrder()
+ {
+ return new List<string> { "SlaveTask-4", "SlaveTask-3", "SlaveTask-2", "SlaveTask-1" };
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/0292caf1/lang/cs/Org.Apache.REEF.Tests/Functional/MPI/ScatterReduceTest/ScatterReduceDriver.cs
----------------------------------------------------------------------
diff --git a/lang/cs/Org.Apache.REEF.Tests/Functional/MPI/ScatterReduceTest/ScatterReduceDriver.cs b/lang/cs/Org.Apache.REEF.Tests/Functional/MPI/ScatterReduceTest/ScatterReduceDriver.cs
new file mode 100644
index 0000000..18ff22b
--- /dev/null
+++ b/lang/cs/Org.Apache.REEF.Tests/Functional/MPI/ScatterReduceTest/ScatterReduceDriver.cs
@@ -0,0 +1,168 @@
+/**
+ * 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.
+ */
+
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using Org.Apache.REEF.Common.Io;
+using Org.Apache.REEF.Common.Tasks;
+using Org.Apache.REEF.Driver;
+using Org.Apache.REEF.Driver.Bridge;
+using Org.Apache.REEF.Driver.Context;
+using Org.Apache.REEF.Driver.Evaluator;
+using Org.Apache.REEF.Network.Group.Driver;
+using Org.Apache.REEF.Network.Group.Driver.Impl;
+using Org.Apache.REEF.Network.Group.Operators;
+using Org.Apache.REEF.Network.Group.Operators.Impl;
+using Org.Apache.REEF.Network.NetworkService;
+using Org.Apache.REEF.Tang.Annotations;
+using Org.Apache.REEF.Tang.Formats;
+using Org.Apache.REEF.Tang.Interface;
+using Org.Apache.REEF.Tang.Util;
+using Org.Apache.REEF.Utilities.Logging;
+using Org.Apache.REEF.Wake.Remote.Impl;
+
+namespace Org.Apache.REEF.Tests.Functional.MPI.ScatterReduceTest
+{
+ public class ScatterReduceDriver : IStartHandler, IObserver<IEvaluatorRequestor>, IObserver<IAllocatedEvaluator>, IObserver<IActiveContext>, IObserver<IFailedEvaluator>
+ {
+ private static readonly Logger LOGGER = Logger.GetLogger(typeof(ScatterReduceDriver));
+
+ private int _numEvaluators;
+
+ private IMpiDriver _mpiDriver;
+ private ICommunicationGroupDriver _commGroup;
+ private TaskStarter _mpiTaskStarter;
+
+ [Inject]
+ public ScatterReduceDriver(
+ [Parameter(typeof(MpiTestConfig.NumEvaluators))] int numEvaluators,
+ AvroConfigurationSerializer confSerializer)
+ {
+ Identifier = "BroadcastStartHandler";
+ _numEvaluators = numEvaluators;
+
+ _mpiDriver = new MpiDriver(
+ MpiTestConstants.DriverId,
+ MpiTestConstants.MasterTaskId,
+ confSerializer);
+
+ _commGroup = _mpiDriver.NewCommunicationGroup(
+ MpiTestConstants.GroupName,
+ numEvaluators)
+ .AddScatter(
+ MpiTestConstants.ScatterOperatorName,
+ new ScatterOperatorSpec<int>(
+ MpiTestConstants.MasterTaskId,
+ new IntCodec()))
+ .AddReduce(
+ MpiTestConstants.ReduceOperatorName,
+ new ReduceOperatorSpec<int>(
+ MpiTestConstants.MasterTaskId,
+ new IntCodec(),
+ new SumFunction()))
+ .Build();
+
+ _mpiTaskStarter = new TaskStarter(_mpiDriver, numEvaluators);
+
+ CreateClassHierarchy();
+ }
+
+ public string Identifier { get; set; }
+
+ public void OnNext(IEvaluatorRequestor evaluatorRequestor)
+ {
+ EvaluatorRequest request = new EvaluatorRequest(_numEvaluators, 512, 2, "WonderlandRack", "BroadcastEvaluator");
+ evaluatorRequestor.Submit(request);
+ }
+
+ public void OnNext(IAllocatedEvaluator allocatedEvaluator)
+ {
+ IConfiguration contextConf = _mpiDriver.GetContextConfiguration();
+ IConfiguration serviceConf = _mpiDriver.GetServiceConfiguration();
+ allocatedEvaluator.SubmitContextAndService(contextConf, serviceConf);
+ }
+
+ public void OnNext(IActiveContext activeContext)
+ {
+ if (_mpiDriver.IsMasterTaskContext(activeContext))
+ {
+ // Configure Master Task
+ IConfiguration partialTaskConf = TaskConfiguration.ConfigurationModule
+ .Set(TaskConfiguration.Identifier, MpiTestConstants.MasterTaskId)
+ .Set(TaskConfiguration.Task, GenericType<MasterTask>.Class)
+ .Build();
+
+ _commGroup.AddTask(MpiTestConstants.MasterTaskId);
+ _mpiTaskStarter.QueueTask(partialTaskConf, activeContext);
+ }
+ else
+ {
+ // Configure Slave Task
+ string slaveTaskId = MpiTestConstants.SlaveTaskId +
+ _mpiDriver.GetContextNum(activeContext);
+
+ IConfiguration partialTaskConf = TaskConfiguration.ConfigurationModule
+ .Set(TaskConfiguration.Identifier, slaveTaskId)
+ .Set(TaskConfiguration.Task, GenericType<SlaveTask>.Class)
+ .Build();
+
+ _commGroup.AddTask(slaveTaskId);
+ _mpiTaskStarter.QueueTask(partialTaskConf, activeContext);
+ }
+ }
+
+ public void OnNext(IFailedEvaluator value)
+ {
+ }
+
+ public void OnError(Exception error)
+ {
+ }
+
+ public void OnCompleted()
+ {
+ }
+
+ private void CreateClassHierarchy()
+ {
+ HashSet<string> clrDlls = new HashSet<string>();
+ clrDlls.Add(typeof(IDriver).Assembly.GetName().Name);
+ clrDlls.Add(typeof(ITask).Assembly.GetName().Name);
+ clrDlls.Add(typeof(ScatterReduceDriver).Assembly.GetName().Name);
+ clrDlls.Add(typeof(INameClient).Assembly.GetName().Name);
+ clrDlls.Add(typeof(INetworkService<>).Assembly.GetName().Name);
+
+ ClrHandlerHelper.GenerateClassHierarchy(clrDlls);
+ }
+
+ private class SumFunction : IReduceFunction<int>
+ {
+ [Inject]
+ public SumFunction()
+ {
+ }
+
+ public int Reduce(IEnumerable<int> elements)
+ {
+ return elements.Sum();
+ }
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/0292caf1/lang/cs/Org.Apache.REEF.Tests/Functional/MPI/ScatterReduceTest/ScatterReduceTest.cs
----------------------------------------------------------------------
diff --git a/lang/cs/Org.Apache.REEF.Tests/Functional/MPI/ScatterReduceTest/ScatterReduceTest.cs b/lang/cs/Org.Apache.REEF.Tests/Functional/MPI/ScatterReduceTest/ScatterReduceTest.cs
new file mode 100644
index 0000000..1a0eef5
--- /dev/null
+++ b/lang/cs/Org.Apache.REEF.Tests/Functional/MPI/ScatterReduceTest/ScatterReduceTest.cs
@@ -0,0 +1,80 @@
+/**
+ * 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.
+ */
+
+using System.Collections.Generic;
+using System.Globalization;
+using Microsoft.VisualStudio.TestTools.UnitTesting;
+using Org.Apache.REEF.Common.Io;
+using Org.Apache.REEF.Common.Tasks;
+using Org.Apache.REEF.Driver;
+using Org.Apache.REEF.Driver.Bridge;
+using Org.Apache.REEF.Network.NetworkService;
+using Org.Apache.REEF.Tang.Implementations.Tang;
+using Org.Apache.REEF.Tang.Interface;
+using Org.Apache.REEF.Tang.Util;
+using Org.Apache.REEF.Utilities.Logging;
+
+namespace Org.Apache.REEF.Tests.Functional.MPI.ScatterReduceTest
+{
+ [TestClass]
+ public class ScatterReduceTest : ReefFunctionalTest
+ {
+ [TestInitialize]
+ public void TestSetup()
+ {
+ CleanUp();
+ }
+
+ [TestCleanup]
+ public void TestCleanup()
+ {
+ CleanUp();
+ }
+
+ [TestMethod]
+ public void TestScatterAndReduce()
+ {
+ int numTasks = 5;
+
+ IConfiguration driverConfig = TangFactory.GetTang().NewConfigurationBuilder(
+ DriverBridgeConfiguration.ConfigurationModule
+ .Set(DriverBridgeConfiguration.OnDriverStarted, GenericType<ScatterReduceDriver>.Class)
+ .Set(DriverBridgeConfiguration.OnEvaluatorAllocated, GenericType<ScatterReduceDriver>.Class)
+ .Set(DriverBridgeConfiguration.OnEvaluatorRequested, GenericType<ScatterReduceDriver>.Class)
+ .Set(DriverBridgeConfiguration.OnEvaluatorFailed, GenericType<ScatterReduceDriver>.Class)
+ .Set(DriverBridgeConfiguration.OnContextActive, GenericType<ScatterReduceDriver>.Class)
+ .Set(DriverBridgeConfiguration.CustomTraceLevel, Level.Info.ToString())
+ .Build())
+ .BindNamedParameter<MpiTestConfig.NumEvaluators, int>(
+ GenericType<MpiTestConfig.NumEvaluators>.Class,
+ numTasks.ToString(CultureInfo.InvariantCulture))
+ .Build();
+
+ HashSet<string> appDlls = new HashSet<string>();
+ appDlls.Add(typeof(IDriver).Assembly.GetName().Name);
+ appDlls.Add(typeof(ITask).Assembly.GetName().Name);
+ appDlls.Add(typeof(ScatterReduceDriver).Assembly.GetName().Name);
+ appDlls.Add(typeof(INameClient).Assembly.GetName().Name);
+ appDlls.Add(typeof(INetworkService<>).Assembly.GetName().Name);
+
+ TestRun(appDlls, driverConfig, false, JavaLoggingSetting.VERBOSE);
+ ValidateSuccessForLocalRuntime(numTasks);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/0292caf1/lang/cs/Org.Apache.REEF.Tests/Functional/MPI/ScatterReduceTest/SlaveTask.cs
----------------------------------------------------------------------
diff --git a/lang/cs/Org.Apache.REEF.Tests/Functional/MPI/ScatterReduceTest/SlaveTask.cs b/lang/cs/Org.Apache.REEF.Tests/Functional/MPI/ScatterReduceTest/SlaveTask.cs
new file mode 100644
index 0000000..327b991
--- /dev/null
+++ b/lang/cs/Org.Apache.REEF.Tests/Functional/MPI/ScatterReduceTest/SlaveTask.cs
@@ -0,0 +1,67 @@
+/**
+ * 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.
+ */
+
+using System.Collections.Generic;
+using System.Linq;
+using Org.Apache.REEF.Common.Tasks;
+using Org.Apache.REEF.Network.Group.Operators;
+using Org.Apache.REEF.Network.Group.Task;
+using Org.Apache.REEF.Tang.Annotations;
+using Org.Apache.REEF.Utilities.Logging;
+
+namespace Org.Apache.REEF.Tests.Functional.MPI.ScatterReduceTest
+{
+ public class SlaveTask : ITask
+ {
+ private static Logger _logger = Logger.GetLogger(typeof(SlaveTask));
+
+ private IMpiClient _mpiClient;
+ private ICommunicationGroupClient _commGroup;
+ private IScatterReceiver<int> _scatterReceiver;
+ private IReduceSender<int> _sumSender;
+
+ [Inject]
+ public SlaveTask(IMpiClient mpiClient)
+ {
+ _logger.Log(Level.Info, "Hello from slave task");
+
+ _mpiClient = mpiClient;
+ _commGroup = _mpiClient.GetCommunicationGroup(MpiTestConstants.GroupName);
+ _scatterReceiver = _commGroup.GetScatterReceiver<int>(MpiTestConstants.ScatterOperatorName);
+ _sumSender = _commGroup.GetReduceSender<int>(MpiTestConstants.ReduceOperatorName);
+ }
+
+ public byte[] Call(byte[] memento)
+ {
+ List<int> data = _scatterReceiver.Receive();
+ _logger.Log(Level.Info, "Received data: {0}", string.Join(" ", data));
+
+ int sum = data.Sum();
+ _logger.Log(Level.Info, "Sending back sum: {0}", sum);
+ _sumSender.Send(sum);
+
+ return null;
+ }
+
+ public void Dispose()
+ {
+ _mpiClient.Dispose();
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/0292caf1/lang/cs/Org.Apache.REEF.Tests/Org.Apache.REEF.Tests.csproj
----------------------------------------------------------------------
diff --git a/lang/cs/Org.Apache.REEF.Tests/Org.Apache.REEF.Tests.csproj b/lang/cs/Org.Apache.REEF.Tests/Org.Apache.REEF.Tests.csproj
index f8b4385..45c39c5 100644
--- a/lang/cs/Org.Apache.REEF.Tests/Org.Apache.REEF.Tests.csproj
+++ b/lang/cs/Org.Apache.REEF.Tests/Org.Apache.REEF.Tests.csproj
@@ -126,6 +126,17 @@ under the License.
<Compile Include="Functional\Messaging\MessageDriver.cs" />
<Compile Include="Functional\Messaging\MessageTask.cs" />
<Compile Include="Functional\Messaging\TestTaskMessage.cs" />
+ <Compile Include="Functional\ML\KMeans\TestKMeans.cs" />
+ <Compile Include="Functional\MPI\BroadcastReduceTest\BroadcastReduceDriver.cs" />
+ <Compile Include="Functional\MPI\BroadcastReduceTest\BroadcastReduceTest.cs" />
+ <Compile Include="Functional\MPI\BroadcastReduceTest\MasterTask.cs" />
+ <Compile Include="Functional\MPI\BroadcastReduceTest\SlaveTask.cs" />
+ <Compile Include="Functional\MPI\MpiTestConfig.cs" />
+ <Compile Include="Functional\MPI\MpiTestConstants.cs" />
+ <Compile Include="Functional\MPI\ScatterReduceTest\MasterTask.cs" />
+ <Compile Include="Functional\MPI\ScatterReduceTest\ScatterReduceDriver.cs" />
+ <Compile Include="Functional\MPI\ScatterReduceTest\ScatterReduceTest.cs" />
+ <Compile Include="Functional\MPI\ScatterReduceTest\SlaveTask.cs" />
<Compile Include="Functional\ReefFunctionalTest.cs" />
<Compile Include="Network\BlockingCollectionExtensionTests.cs" />
<Compile Include="Network\NameServerTests.cs" />
[2/4] incubator-reef git commit: [REEF-150] Adding group
communication to REEF .Net
Posted by tm...@apache.org.
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/0292caf1/lang/cs/Org.Apache.REEF.Network/Group/Operators/Impl/ReduceSender.cs
----------------------------------------------------------------------
diff --git a/lang/cs/Org.Apache.REEF.Network/Group/Operators/Impl/ReduceSender.cs b/lang/cs/Org.Apache.REEF.Network/Group/Operators/Impl/ReduceSender.cs
new file mode 100644
index 0000000..ffc5165
--- /dev/null
+++ b/lang/cs/Org.Apache.REEF.Network/Group/Operators/Impl/ReduceSender.cs
@@ -0,0 +1,97 @@
+/**
+ * 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.
+ */
+
+using Org.Apache.REEF.Network.Group.Config;
+using Org.Apache.REEF.Network.Group.Driver;
+using Org.Apache.REEF.Network.Group.Driver.Impl;
+using Org.Apache.REEF.Network.Group.Task;
+using Org.Apache.REEF.Network.Group.Task.Impl;
+using Org.Apache.REEF.Tang.Annotations;
+using System;
+using System.Reactive;
+
+namespace Org.Apache.REEF.Network.Group.Operators.Impl
+{
+ /// <summary>
+ /// MPI Operator used to send messages to be reduced by the ReduceReceiver.
+ /// </summary>
+ /// <typeparam name="T">The message type</typeparam>
+ public class ReduceSender<T> : IReduceSender<T>
+ {
+ private const int DefaultVersion = 1;
+
+ private ICommunicationGroupNetworkObserver _networkHandler;
+ private OperatorTopology<T> _topology;
+
+ /// <summary>
+ /// Creates a new ReduceSender.
+ /// </summary>
+ /// <param name="operatorName">The name of the reduce operator</param>
+ /// <param name="groupName">The name of the reduce operator's CommunicationGroup</param>
+ /// <param name="topology">The Task's operator topology graph</param>
+ /// <param name="networkHandler">The handler used to handle incoming messages</param>
+ [Inject]
+ public ReduceSender(
+ [Parameter(typeof(MpiConfigurationOptions.OperatorName))] string operatorName,
+ [Parameter(typeof(MpiConfigurationOptions.CommunicationGroupName))] string groupName,
+ OperatorTopology<T> topology,
+ ICommunicationGroupNetworkObserver networkHandler)
+ {
+ OperatorName = operatorName;
+ GroupName = groupName;
+ Version = DefaultVersion;
+
+ _networkHandler = networkHandler;
+ _topology = topology;
+ _topology.Initialize();
+
+ var msgHandler = Observer.Create<GroupCommunicationMessage>(message => _topology.OnNext(message));
+ _networkHandler.Register(operatorName, msgHandler);
+ }
+
+ /// <summary>
+ /// Returns the name of the reduce operator.
+ /// </summary>
+ public string OperatorName { get; private set; }
+
+ /// <summary>
+ /// Returns the name of the operator's CommunicationGroup.
+ /// </summary>
+ public string GroupName { get; private set; }
+
+ /// <summary>
+ /// Returns the operator version.
+ /// </summary>
+ public int Version { get; private set; }
+
+ /// <summary>
+ /// Sends the data to the operator's ReduceReceiver to be aggregated.
+ /// </summary>
+ /// <param name="data">The data to send</param>
+ public void Send(T data)
+ {
+ if (data == null)
+ {
+ throw new ArgumentNullException("data");
+ }
+
+ _topology.SendToParent(data, MessageType.Data);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/0292caf1/lang/cs/Org.Apache.REEF.Network/Group/Operators/Impl/ScatterOperatorSpec.cs
----------------------------------------------------------------------
diff --git a/lang/cs/Org.Apache.REEF.Network/Group/Operators/Impl/ScatterOperatorSpec.cs b/lang/cs/Org.Apache.REEF.Network/Group/Operators/Impl/ScatterOperatorSpec.cs
new file mode 100644
index 0000000..8219eb6
--- /dev/null
+++ b/lang/cs/Org.Apache.REEF.Network/Group/Operators/Impl/ScatterOperatorSpec.cs
@@ -0,0 +1,58 @@
+/**
+ * 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.
+ */
+
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.Threading.Tasks;
+using Org.Apache.REEF.Wake.Remote;
+
+namespace Org.Apache.REEF.Network.Group.Operators.Impl
+{
+ /// <summary>
+ /// The specification used to define Scatter MPI Operators.
+ /// </summary>
+ public class ScatterOperatorSpec<T> : IOperatorSpec<T>
+ {
+ /// <summary>
+ /// Creates a new ScatterOperatorSpec.
+ /// </summary>
+ /// <param name="senderId">The identifier of the task that will
+ /// be sending messages</param>
+ /// <param name="codec">The codec used to serialize and
+ /// deserialize messages</param>
+ public ScatterOperatorSpec(string senderId, ICodec<T> codec)
+ {
+ SenderId = senderId;
+ Codec = codec;
+ }
+
+ /// <summary>
+ /// Returns the identifier for the task that splits and scatters a list
+ /// of messages to other tasks.
+ /// </summary>
+ public string SenderId { get; private set; }
+
+ /// <summary>
+ /// The codec used to serialize and deserialize messages.
+ /// </summary>
+ public ICodec<T> Codec { get; private set; }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/0292caf1/lang/cs/Org.Apache.REEF.Network/Group/Operators/Impl/ScatterReceiver.cs
----------------------------------------------------------------------
diff --git a/lang/cs/Org.Apache.REEF.Network/Group/Operators/Impl/ScatterReceiver.cs b/lang/cs/Org.Apache.REEF.Network/Group/Operators/Impl/ScatterReceiver.cs
new file mode 100644
index 0000000..85b5c13
--- /dev/null
+++ b/lang/cs/Org.Apache.REEF.Network/Group/Operators/Impl/ScatterReceiver.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.
+ */
+
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Reactive;
+using System.Text;
+using System.Threading.Tasks;
+using Org.Apache.REEF.Network.Group.Config;
+using Org.Apache.REEF.Network.Group.Driver.Impl;
+using Org.Apache.REEF.Network.Group.Task;
+using Org.Apache.REEF.Network.Group.Task.Impl;
+using Org.Apache.REEF.Tang.Annotations;
+
+namespace Org.Apache.REEF.Network.Group.Operators.Impl
+{
+ /// <summary>
+ /// MPI operator used to receive a sublist of messages sent
+ /// from the IScatterSender.
+ /// </summary>
+ /// <typeparam name="T">The message type</typeparam>
+ public class ScatterReceiver<T> : IScatterReceiver<T>
+ {
+ private const int DefaultVersion = 1;
+
+ private ICommunicationGroupNetworkObserver _networkHandler;
+ private OperatorTopology<T> _topology;
+
+ /// <summary>
+ /// Creates a new ScatterReceiver.
+ /// </summary>
+ /// <param name="operatorName">The name of the scatter operator</param>
+ /// <param name="groupName">The name of the operator's CommunicationGroup</param>
+ /// <param name="topology">The task's operator topology graph</param>
+ /// <param name="networkHandler">Handles incoming messages from other tasks</param>
+ [Inject]
+ public ScatterReceiver(
+ [Parameter(typeof(MpiConfigurationOptions.OperatorName))] string operatorName,
+ [Parameter(typeof(MpiConfigurationOptions.CommunicationGroupName))] string groupName,
+ OperatorTopology<T> topology,
+ ICommunicationGroupNetworkObserver networkHandler)
+ {
+ OperatorName = operatorName;
+ GroupName = groupName;
+ Version = DefaultVersion;
+
+ _networkHandler = networkHandler;
+ _topology = topology;
+ _topology.Initialize();
+
+ var msgHandler = Observer.Create<GroupCommunicationMessage>(message => _topology.OnNext(message));
+ _networkHandler.Register(operatorName, msgHandler);
+ }
+
+ /// <summary>
+ /// Returns the name of the reduce operator
+ /// </summary>
+ public string OperatorName { get; private set; }
+
+ /// <summary>
+ /// Returns the name of the operator's CommunicationGroup.
+ /// </summary>
+ public string GroupName { get; private set; }
+
+ /// <summary>
+ /// Returns the operator version.
+ /// </summary>
+ public int Version { get; private set; }
+
+ /// <summary>
+ /// Returns the class used to reduce incoming messages sent by ReduceSenders.
+ /// </summary>
+ public IReduceFunction<T> ReduceFunction { get; private set; }
+
+ /// <summary>
+ /// Receive a sublist of messages sent from the IScatterSender.
+ /// </summary>
+ /// <returns>The sublist of messages</returns>
+ public List<T> Receive()
+ {
+ return _topology.ReceiveListFromParent();
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/0292caf1/lang/cs/Org.Apache.REEF.Network/Group/Operators/Impl/ScatterSender.cs
----------------------------------------------------------------------
diff --git a/lang/cs/Org.Apache.REEF.Network/Group/Operators/Impl/ScatterSender.cs b/lang/cs/Org.Apache.REEF.Network/Group/Operators/Impl/ScatterSender.cs
new file mode 100644
index 0000000..ee9e683
--- /dev/null
+++ b/lang/cs/Org.Apache.REEF.Network/Group/Operators/Impl/ScatterSender.cs
@@ -0,0 +1,112 @@
+/**
+ * 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.
+ */
+
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Reactive;
+using System.Text;
+using System.Threading.Tasks;
+using Org.Apache.REEF.Network.Group.Config;
+using Org.Apache.REEF.Network.Group.Driver.Impl;
+using Org.Apache.REEF.Network.Group.Task;
+using Org.Apache.REEF.Network.Group.Task.Impl;
+using Org.Apache.REEF.Tang.Annotations;
+
+namespace Org.Apache.REEF.Network.Group.Operators.Impl
+{
+ /// <summary>
+ /// MPI operator used to scatter a list of elements to all
+ /// of the IScatterReceivers.
+ /// </summary>
+ /// <typeparam name="T">The message type</typeparam>
+ public class ScatterSender<T> : IScatterSender<T>
+ {
+ private const int DefaultVersion = 1;
+
+ private ICommunicationGroupNetworkObserver _networkHandler;
+ private OperatorTopology<T> _topology;
+
+ /// <summary>
+ /// Creates a new ScatterSender.
+ /// </summary>
+ /// <param name="operatorName">The name of the scatter operator</param>
+ /// <param name="groupName">The name of the operator's Communication Group</param>
+ /// <param name="topology">The operator topology</param>
+ /// <param name="networkHandler">The network handler</param>
+ [Inject]
+ public ScatterSender(
+ [Parameter(typeof(MpiConfigurationOptions.OperatorName))] string operatorName,
+ [Parameter(typeof(MpiConfigurationOptions.CommunicationGroupName))] string groupName,
+ OperatorTopology<T> topology,
+ ICommunicationGroupNetworkObserver networkHandler)
+ {
+ OperatorName = operatorName;
+ GroupName = groupName;
+ Version = DefaultVersion;
+
+ _networkHandler = networkHandler;
+ _topology = topology;
+ _topology.Initialize();
+
+ var msgHandler = Observer.Create<GroupCommunicationMessage>(message => _topology.OnNext(message));
+ _networkHandler.Register(operatorName, msgHandler);
+ }
+
+ public string OperatorName { get; private set; }
+
+ public string GroupName { get; private set; }
+
+ public int Version { get; private set; }
+
+ /// <summary>
+ /// Split up the list of elements evenly and scatter each chunk
+ /// to the IScatterReceivers.
+ /// </summary>
+ /// <param name="elements">The list of elements to send.</param>
+ public void Send(List<T> elements)
+ {
+ _topology.ScatterToChildren(elements, MessageType.Data);
+ }
+
+ /// <summary>
+ /// Split up the list of elements and scatter each chunk
+ /// to the IScatterReceivers. Each receiver will receive
+ /// a sublist of the specified size.
+ /// </summary>
+ /// <param name="elements">The list of elements to send.</param>
+ /// <param name="count">The size of each sublist</param>
+ public void Send(List<T> elements, int count)
+ {
+ _topology.ScatterToChildren(elements, count, MessageType.Data);
+ }
+
+ /// <summary>
+ /// Split up the list of elements and scatter each chunk
+ /// to the IScatterReceivers in the specified task order.
+ /// </summary>
+ /// <param name="elements">The list of elements to send.</param>
+ /// <param name="order">The list of task identifiers representing
+ /// the order in which to scatter each sublist</param>
+ public void Send(List<T> elements, List<string> order)
+ {
+ _topology.ScatterToChildren(elements, order, MessageType.Data);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/0292caf1/lang/cs/Org.Apache.REEF.Network/Group/Operators/Impl/Sender.cs
----------------------------------------------------------------------
diff --git a/lang/cs/Org.Apache.REEF.Network/Group/Operators/Impl/Sender.cs b/lang/cs/Org.Apache.REEF.Network/Group/Operators/Impl/Sender.cs
new file mode 100644
index 0000000..c5ca60f
--- /dev/null
+++ b/lang/cs/Org.Apache.REEF.Network/Group/Operators/Impl/Sender.cs
@@ -0,0 +1,74 @@
+/**
+ * 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.
+ */
+
+using System;
+using Org.Apache.REEF.Network.Group.Driver;
+using Org.Apache.REEF.Network.Group.Driver.Impl;
+using Org.Apache.REEF.Network.NetworkService;
+using Org.Apache.REEF.Tang.Annotations;
+using Org.Apache.REEF.Wake;
+using Org.Apache.REEF.Wake.Remote.Impl;
+
+namespace Org.Apache.REEF.Network.Group.Operators.Impl
+{
+ /// <summary>
+ /// MPI operator used to do point-to-point communication between named Tasks.
+ /// </summary>
+ public class Sender
+ {
+ private INetworkService<GroupCommunicationMessage> _networkService;
+ private IIdentifierFactory _idFactory;
+
+ /// <summary>
+ /// Creates a new Sender.
+ /// </summary>
+ /// <param name="networkService">The network services used to send messages.</param>
+ /// <param name="idFactory">Used to create IIdentifier for GroupCommunicationMessages.</param>
+ [Inject]
+ public Sender(
+ NetworkService<GroupCommunicationMessage> networkService,
+ IIdentifierFactory idFactory)
+ {
+ _networkService = networkService;
+ _idFactory = idFactory;
+ }
+
+ /// <summary>
+ /// Send the GroupCommunicationMessage to the Task whose name is
+ /// included in the message.
+ /// </summary>
+ /// <param name="message">The message to send.</param>
+ public void Send(GroupCommunicationMessage message)
+ {
+ if (message == null)
+ {
+ throw new ArgumentNullException("message");
+ }
+ if (string.IsNullOrEmpty(message.Destination))
+ {
+ throw new ArgumentException("Message destination cannot be null or empty");
+ }
+
+ IIdentifier destId = _idFactory.Create(message.Destination);
+ var conn = _networkService.NewConnection(destId);
+ conn.Open();
+ conn.Write(message);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/0292caf1/lang/cs/Org.Apache.REEF.Network/Group/Task/ICommunicationGroupClient.cs
----------------------------------------------------------------------
diff --git a/lang/cs/Org.Apache.REEF.Network/Group/Task/ICommunicationGroupClient.cs b/lang/cs/Org.Apache.REEF.Network/Group/Task/ICommunicationGroupClient.cs
new file mode 100644
index 0000000..9b96a9a
--- /dev/null
+++ b/lang/cs/Org.Apache.REEF.Network/Group/Task/ICommunicationGroupClient.cs
@@ -0,0 +1,90 @@
+/**
+ * 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.
+ */
+
+using Org.Apache.REEF.Network.Group.Operators;
+using Org.Apache.REEF.Network.Group.Task.Impl;
+using Org.Apache.REEF.Tang.Annotations;
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.Threading.Tasks;
+
+namespace Org.Apache.REEF.Network.Group.Task
+{
+ /// <summary>
+ /// Used by Tasks to fetch MPI Operators in the group configured by the driver.
+ /// </summary>
+ [DefaultImplementation(typeof(CommunicationGroupClient))]
+ public interface ICommunicationGroupClient
+ {
+ /// <summary>
+ /// Returns the Communication Group name
+ /// </summary>
+ string GroupName { get; }
+
+ /// <summary>
+ /// Gets the BroadcastSender with the given name and message type.
+ /// </summary>
+ /// <typeparam name="T">The message type</typeparam>
+ /// <param name="operatorName">The name of the Broadcast operator</param>
+ /// <returns>The BroadcastSender</returns>
+ IBroadcastSender<T> GetBroadcastSender<T>(string operatorName);
+
+ /// <summary>
+ /// Gets the BroadcastReceiver with the given name and message type.
+ /// </summary>
+ /// <typeparam name="T">The message type</typeparam>
+ /// <param name="operatorName">The name of the Broadcast operator</param>
+ /// <returns>The BroadcastReceiver</returns>
+ IBroadcastReceiver<T> GetBroadcastReceiver<T>(string operatorName);
+
+ /// <summary>
+ /// Gets the ReduceSender with the given name and message type.
+ /// </summary>
+ /// <typeparam name="T">The message type</typeparam>
+ /// <param name="operatorName">The name of the Reduce operator</param>
+ /// <returns>The ReduceSender</returns>
+ IReduceSender<T> GetReduceSender<T>(string operatorName);
+
+ /// <summary>
+ /// Gets the ReduceReceiver with the given name and message type.
+ /// </summary>
+ /// <typeparam name="T">The message type</typeparam>
+ /// <param name="operatorName">The name of the Reduce operator</param>
+ /// <returns>The ReduceReceiver</returns>
+ IReduceReceiver<T> GetReduceReceiver<T>(string operatorName);
+
+ /// <summary>
+ /// Gets the ScatterSender with the given name and message type.
+ /// </summary>
+ /// <typeparam name="T">The message type</typeparam>
+ /// <param name="operatorName">The name of the Scatter operator</param>
+ /// <returns>The ScatterSender</returns>
+ IScatterSender<T> GetScatterSender<T>(string operatorName);
+
+ /// <summary>
+ /// Gets the ScatterReceiver with the given name and message type.
+ /// </summary>
+ /// <typeparam name="T">The message type</typeparam>
+ /// <param name="operatorName">The name of the Scatter operator</param>
+ /// <returns>The ScatterReceiver</returns>
+ IScatterReceiver<T> GetScatterReceiver<T>(string operatorName);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/0292caf1/lang/cs/Org.Apache.REEF.Network/Group/Task/ICommunicationGroupNetworkObserver.cs
----------------------------------------------------------------------
diff --git a/lang/cs/Org.Apache.REEF.Network/Group/Task/ICommunicationGroupNetworkObserver.cs b/lang/cs/Org.Apache.REEF.Network/Group/Task/ICommunicationGroupNetworkObserver.cs
new file mode 100644
index 0000000..d3034fc
--- /dev/null
+++ b/lang/cs/Org.Apache.REEF.Network/Group/Task/ICommunicationGroupNetworkObserver.cs
@@ -0,0 +1,49 @@
+/**
+ * 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.
+ */
+
+using Org.Apache.REEF.Network.Group.Driver;
+using Org.Apache.REEF.Network.Group.Driver.Impl;
+using Org.Apache.REEF.Network.Group.Task.Impl;
+using Org.Apache.REEF.Tang.Annotations;
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.Threading.Tasks;
+
+namespace Org.Apache.REEF.Network.Group.Task
+{
+ /// <summary>
+ /// Handles incoming messages sent to this Communication Group.
+ /// </summary>
+ [DefaultImplementation(typeof(CommunicationGroupNetworkObserver))]
+ public interface ICommunicationGroupNetworkObserver : IObserver<GroupCommunicationMessage>
+ {
+ /// <summary>
+ /// Registers the handler with the CommunicationGroupNetworkObserver.
+ /// Messages that are to be sent to the operator specified by operatorName
+ /// are handled by the given observer.
+ /// </summary>
+ /// <param name="operatorName">The name of the operator whose handler
+ /// will be invoked</param>
+ /// <param name="observer">The handler to invoke when messages are sent
+ /// to the operator specified by operatorName</param>
+ void Register(string operatorName, IObserver<GroupCommunicationMessage> observer);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/0292caf1/lang/cs/Org.Apache.REEF.Network/Group/Task/IMpiClient.cs
----------------------------------------------------------------------
diff --git a/lang/cs/Org.Apache.REEF.Network/Group/Task/IMpiClient.cs b/lang/cs/Org.Apache.REEF.Network/Group/Task/IMpiClient.cs
new file mode 100644
index 0000000..2592b04
--- /dev/null
+++ b/lang/cs/Org.Apache.REEF.Network/Group/Task/IMpiClient.cs
@@ -0,0 +1,44 @@
+/**
+ * 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.
+ */
+
+using Org.Apache.REEF.Network.Group.Operators;
+using Org.Apache.REEF.Network.Group.Task.Impl;
+using Org.Apache.REEF.Tang.Annotations;
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.Threading.Tasks;
+
+namespace Org.Apache.REEF.Network.Group.Task
+{
+ /// <summary>
+ /// Used by Tasks to fetch CommunicationGroupClients.
+ /// </summary>
+ [DefaultImplementation(typeof(MpiClient))]
+ public interface IMpiClient : IDisposable
+ {
+ /// <summary>
+ /// Gets the CommunicationGroupClient with the given group name.
+ /// </summary>
+ /// <param name="groupName">The name of the CommunicationGroupClient</param>
+ /// <returns>The configured CommunicationGroupClient</returns>
+ ICommunicationGroupClient GetCommunicationGroup(string groupName);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/0292caf1/lang/cs/Org.Apache.REEF.Network/Group/Task/IMpiNetworkObserver.cs
----------------------------------------------------------------------
diff --git a/lang/cs/Org.Apache.REEF.Network/Group/Task/IMpiNetworkObserver.cs b/lang/cs/Org.Apache.REEF.Network/Group/Task/IMpiNetworkObserver.cs
new file mode 100644
index 0000000..5fe948c
--- /dev/null
+++ b/lang/cs/Org.Apache.REEF.Network/Group/Task/IMpiNetworkObserver.cs
@@ -0,0 +1,50 @@
+/**
+ * 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.
+ */
+
+using Org.Apache.REEF.Network.Group.Driver;
+using Org.Apache.REEF.Network.Group.Driver.Impl;
+using Org.Apache.REEF.Network.Group.Task.Impl;
+using Org.Apache.REEF.Network.NetworkService;
+using Org.Apache.REEF.Tang.Annotations;
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.Threading.Tasks;
+using Org.Apache.REEF.Network.Group.Codec;
+
+namespace Org.Apache.REEF.Network.Group.Task
+{
+ /// <summary>
+ /// Handles all incoming messages for this Task.
+ /// </summary>
+ [DefaultImplementation(typeof(MpiNetworkObserver))]
+ public interface IMpiNetworkObserver : IObserver<NsMessage<GroupCommunicationMessage>>
+ {
+ /// <summary>
+ /// Registers the network handler for the given CommunicationGroup.
+ /// When messages are sent to the specified group name, the given handler
+ /// will be invoked with that message.
+ /// </summary>
+ /// <param name="groupName">The group name for the network handler</param>
+ /// <param name="commGroupHandler">The network handler to invoke when
+ /// messages are sent to the given group.</param>
+ void Register(string groupName, IObserver<GroupCommunicationMessage> commGroupHandler);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/0292caf1/lang/cs/Org.Apache.REEF.Network/Group/Task/Impl/CommunicationGroupClient.cs
----------------------------------------------------------------------
diff --git a/lang/cs/Org.Apache.REEF.Network/Group/Task/Impl/CommunicationGroupClient.cs b/lang/cs/Org.Apache.REEF.Network/Group/Task/Impl/CommunicationGroupClient.cs
new file mode 100644
index 0000000..e6d653d
--- /dev/null
+++ b/lang/cs/Org.Apache.REEF.Network/Group/Task/Impl/CommunicationGroupClient.cs
@@ -0,0 +1,219 @@
+/**
+ * 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.
+ */
+
+using Org.Apache.REEF.Network.Group.Config;
+using Org.Apache.REEF.Network.Group.Driver;
+using Org.Apache.REEF.Network.Group.Driver.Impl;
+using Org.Apache.REEF.Network.Group.Operators;
+using Org.Apache.REEF.Network.Group.Operators.Impl;
+using Org.Apache.REEF.Network.NetworkService;
+using Org.Apache.REEF.Utilities.Logging;
+using Org.Apache.REEF.Tang.Annotations;
+using Org.Apache.REEF.Tang.Exceptions;
+using Org.Apache.REEF.Tang.Formats;
+using Org.Apache.REEF.Tang.Interface;
+using Org.Apache.REEF.Tang.Util;
+using System;
+using System.Collections.Generic;
+using Org.Apache.REEF.Common.Tasks;
+using Org.Apache.REEF.Tang.Implementations.Tang;
+
+namespace Org.Apache.REEF.Network.Group.Task.Impl
+{
+ /// <summary>
+ /// Used by Tasks to fetch MPI Operators in the group configured by the driver.
+ /// </summary>
+ public class CommunicationGroupClient : ICommunicationGroupClient
+ {
+ private readonly Logger LOGGER = Logger.GetLogger(typeof(CommunicationGroupClient));
+
+ private string _taskId;
+ private string _driverId;
+
+ private Dictionary<string, IInjector> _operatorInjectors;
+ private Dictionary<string, object> _operators;
+ private NetworkService<GroupCommunicationMessage> _networkService;
+ private IMpiNetworkObserver _mpiNetworkHandler;
+ private ICommunicationGroupNetworkObserver _commGroupNetworkHandler;
+
+ /// <summary>
+ /// Creates a new CommunicationGroupClient.
+ /// </summary>
+ /// <param name="taskId">The identifier for this Task.</param>
+ /// <param name="groupName">The name of the CommunicationGroup</param>
+ /// <param name="driverId">The identifier for the driver</param>
+ /// <param name="operatorConfigs">The serialized operator configurations</param>
+ /// <param name="mpiNetworkObserver">The handler for all incoming messages
+ /// across all Communication Groups</param>
+ /// <param name="networkService">The network service used to send messages.</param>
+ /// <param name="configSerializer">Used to deserialize operator configuration.</param>
+ [Inject]
+ public CommunicationGroupClient(
+ [Parameter(typeof(TaskConfigurationOptions.Identifier))] string taskId,
+ [Parameter(typeof(MpiConfigurationOptions.CommunicationGroupName))] string groupName,
+ [Parameter(typeof(MpiConfigurationOptions.DriverId))] string driverId,
+ [Parameter(typeof(MpiConfigurationOptions.SerializedOperatorConfigs))] ISet<string> operatorConfigs,
+ IMpiNetworkObserver mpiNetworkObserver,
+ NetworkService<GroupCommunicationMessage> networkService,
+ AvroConfigurationSerializer configSerializer)
+ {
+ _taskId = taskId;
+ _driverId = driverId;
+ GroupName = groupName;
+
+ _operators = new Dictionary<string, object>();
+ _operatorInjectors = new Dictionary<string, IInjector>();
+
+ _networkService = networkService;
+ _mpiNetworkHandler = mpiNetworkObserver;
+ _commGroupNetworkHandler = new CommunicationGroupNetworkObserver();
+ _mpiNetworkHandler.Register(groupName, _commGroupNetworkHandler);
+
+ // Deserialize operator configuration and store each injector.
+ // When user requests the MPI Operator, use type information to
+ // create the instance.
+ foreach (string operatorConfigStr in operatorConfigs)
+ {
+ IConfiguration operatorConfig = configSerializer.FromString(operatorConfigStr);
+
+ IInjector injector = TangFactory.GetTang().NewInjector(operatorConfig);
+ string operatorName = injector.GetNamedInstance<MpiConfigurationOptions.OperatorName, string>(
+ GenericType<MpiConfigurationOptions.OperatorName>.Class);
+ _operatorInjectors[operatorName] = injector;
+ }
+ }
+
+ /// <summary>
+ /// Returns the Communication Group name
+ /// </summary>
+ public string GroupName { get; private set; }
+
+ /// <summary>
+ /// Gets the BroadcastSender with the given name and message type.
+ /// </summary>
+ /// <typeparam name="T">The message type</typeparam>
+ /// <param name="operatorName">The name of the Broadcast operator</param>
+ /// <returns>The BroadcastSender</returns>
+ public IBroadcastSender<T> GetBroadcastSender<T>(string operatorName)
+ {
+ return GetOperatorInstance<BroadcastSender<T>>(operatorName);
+ }
+
+ /// <summary>
+ /// Gets the BroadcastReceiver with the given name and message type.
+ /// </summary>
+ /// <typeparam name="T">The message type</typeparam>
+ /// <param name="operatorName">The name of the Broadcast operator</param>
+ /// <returns>The BroadcastReceiver</returns>
+ public IBroadcastReceiver<T> GetBroadcastReceiver<T>(string operatorName)
+ {
+ return GetOperatorInstance<BroadcastReceiver<T>>(operatorName);
+ }
+
+ /// <summary>
+ /// Gets the ReduceSender with the given name and message type.
+ /// </summary>
+ /// <typeparam name="T">The message type</typeparam>
+ /// <param name="operatorName">The name of the Reduce operator</param>
+ /// <returns>The ReduceSender</returns>
+ public IReduceSender<T> GetReduceSender<T>(string operatorName)
+ {
+ return GetOperatorInstance<ReduceSender<T>>(operatorName);
+ }
+
+ /// <summary>
+ /// Gets the ReduceReceiver with the given name and message type.
+ /// </summary>
+ /// <typeparam name="T">The message type</typeparam>
+ /// <param name="operatorName">The name of the Reduce operator</param>
+ /// <returns>The ReduceReceiver</returns>
+ public IReduceReceiver<T> GetReduceReceiver<T>(string operatorName)
+ {
+ return GetOperatorInstance<ReduceReceiver<T>>(operatorName);
+ }
+
+ /// <summary>
+ /// Gets the ScatterSender with the given name and message type.
+ /// </summary>
+ /// <typeparam name="T">The message type</typeparam>
+ /// <param name="operatorName">The name of the Scatter operator</param>
+ /// <returns>The ScatterSender</returns>
+ public IScatterSender<T> GetScatterSender<T>(string operatorName)
+ {
+ return GetOperatorInstance<ScatterSender<T>>(operatorName);
+ }
+
+ /// <summary>
+ /// Gets the ScatterReceiver with the given name and message type.
+ /// </summary>
+ /// <typeparam name="T">The message type</typeparam>
+ /// <param name="operatorName">The name of the Scatter operator</param>
+ /// <returns>The ScatterReceiver</returns>
+ public IScatterReceiver<T> GetScatterReceiver<T>(string operatorName)
+ {
+ return GetOperatorInstance<ScatterReceiver<T>>(operatorName);
+ }
+
+ /// <summary>
+ /// Gets the MPI operator with the specified name and type.
+ /// If the operator hasn't been instanciated yet, find the injector
+ /// associated with the given operator name and use the type information
+ /// to create a new operator of that type.
+ /// </summary>
+ /// <typeparam name="T">The type of operator to create</typeparam>
+ /// <param name="operatorName">The name of the operator</param>
+ /// <returns>The newly created MPI Operator</returns>
+ private T GetOperatorInstance<T>(string operatorName) where T : class
+ {
+ if (string.IsNullOrEmpty(operatorName))
+ {
+ throw new ArgumentNullException("operatorName");
+ }
+ if (!_operatorInjectors.ContainsKey(operatorName))
+ {
+ throw new ArgumentException("Invalid operator name, cannot create CommunicationGroupClient");
+ }
+
+ object op;
+ if (!_operators.TryGetValue(operatorName, out op))
+ {
+ IInjector injector = _operatorInjectors[operatorName];
+
+ injector.BindVolatileParameter(GenericType<TaskConfigurationOptions.Identifier>.Class, _taskId);
+ injector.BindVolatileParameter(GenericType<MpiConfigurationOptions.CommunicationGroupName>.Class, GroupName);
+ injector.BindVolatileInstance(GenericType<ICommunicationGroupNetworkObserver>.Class, _commGroupNetworkHandler);
+ injector.BindVolatileInstance(GenericType<NetworkService<GroupCommunicationMessage>>.Class, _networkService);
+ injector.BindVolatileInstance(GenericType<ICommunicationGroupClient>.Class, this);
+
+ try
+ {
+ op = injector.GetInstance<T>();
+ _operators[operatorName] = op;
+ }
+ catch (InjectionException)
+ {
+ LOGGER.Log(Level.Error, "Cannot inject MPI operator: No known operator of type: {0}", typeof(T));
+ throw;
+ }
+ }
+
+ return (T) op;
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/0292caf1/lang/cs/Org.Apache.REEF.Network/Group/Task/Impl/CommunicationGroupNetworkObserver.cs
----------------------------------------------------------------------
diff --git a/lang/cs/Org.Apache.REEF.Network/Group/Task/Impl/CommunicationGroupNetworkObserver.cs b/lang/cs/Org.Apache.REEF.Network/Group/Task/Impl/CommunicationGroupNetworkObserver.cs
new file mode 100644
index 0000000..97ab082
--- /dev/null
+++ b/lang/cs/Org.Apache.REEF.Network/Group/Task/Impl/CommunicationGroupNetworkObserver.cs
@@ -0,0 +1,93 @@
+/**
+ * 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.
+ */
+
+using Org.Apache.REEF.Network.Group.Driver;
+using Org.Apache.REEF.Network.Group.Driver.Impl;
+using Org.Apache.REEF.Tang.Annotations;
+using System;
+using System.Collections.Generic;
+
+namespace Org.Apache.REEF.Network.Group.Task.Impl
+{
+ /// <summary>
+ /// Handles incoming messages sent to this Communication Group.
+ /// </summary>
+ public class CommunicationGroupNetworkObserver : ICommunicationGroupNetworkObserver
+ {
+ private Dictionary<string, IObserver<GroupCommunicationMessage>> _handlers;
+
+ /// <summary>
+ /// Creates a new CommunicationGroupNetworkObserver.
+ /// </summary>
+ [Inject]
+ public CommunicationGroupNetworkObserver()
+ {
+ _handlers = new Dictionary<string, IObserver<GroupCommunicationMessage>>();
+ }
+
+ /// <summary>
+ /// Registers the handler with the CommunicationGroupNetworkObserver.
+ /// Messages that are to be sent to the operator specified by operatorName
+ /// are handled by the given observer.
+ /// </summary>
+ /// <param name="operatorName">The name of the operator whose handler
+ /// will be invoked</param>
+ /// <param name="observer">The handler to invoke when messages are sent
+ /// to the operator specified by operatorName</param>
+ public void Register(string operatorName, IObserver<GroupCommunicationMessage> observer)
+ {
+ if (string.IsNullOrEmpty(operatorName))
+ {
+ throw new ArgumentNullException("operatorName");
+ }
+ if (observer == null)
+ {
+ throw new ArgumentNullException("observer");
+ }
+
+ _handlers[operatorName] = observer;
+ }
+
+ /// <summary>
+ /// Handles the incoming GroupCommunicationMessage sent to this Communication Group.
+ /// Looks for the operator that the message is being sent to and invoke its handler.
+ /// </summary>
+ /// <param name="message">The incoming message</param>
+ public void OnNext(GroupCommunicationMessage message)
+ {
+ string operatorName = message.OperatorName;
+
+ IObserver<GroupCommunicationMessage> handler;
+ if (!_handlers.TryGetValue(operatorName, out handler))
+ {
+ throw new ArgumentException("No handler registered with the operator name: " + operatorName);
+ }
+
+ handler.OnNext(message);
+ }
+
+ public void OnError(Exception error)
+ {
+ }
+
+ public void OnCompleted()
+ {
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/0292caf1/lang/cs/Org.Apache.REEF.Network/Group/Task/Impl/MpiClient.cs
----------------------------------------------------------------------
diff --git a/lang/cs/Org.Apache.REEF.Network/Group/Task/Impl/MpiClient.cs b/lang/cs/Org.Apache.REEF.Network/Group/Task/Impl/MpiClient.cs
new file mode 100644
index 0000000..7cec022
--- /dev/null
+++ b/lang/cs/Org.Apache.REEF.Network/Group/Task/Impl/MpiClient.cs
@@ -0,0 +1,108 @@
+/**
+ * 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.
+ */
+
+using Org.Apache.REEF.Network.Group.Config;
+using Org.Apache.REEF.Network.Group.Driver;
+using Org.Apache.REEF.Network.Group.Driver.Impl;
+using Org.Apache.REEF.Network.NetworkService;
+using Org.Apache.REEF.Tang.Annotations;
+using Org.Apache.REEF.Tang.Formats;
+using Org.Apache.REEF.Tang.Interface;
+using Org.Apache.REEF.Tang.Util;
+using Org.Apache.REEF.Wake.Remote.Impl;
+using System;
+using System.Collections.Generic;
+using Org.Apache.REEF.Common.Tasks;
+using Org.Apache.REEF.Tang.Implementations.Tang;
+
+namespace Org.Apache.REEF.Network.Group.Task.Impl
+{
+ /// <summary>
+ /// Used by Tasks to fetch CommunicationGroupClients.
+ /// </summary>
+ public class MpiClient : IMpiClient
+ {
+ private Dictionary<string, ICommunicationGroupClient> _commGroups;
+
+ private INetworkService<GroupCommunicationMessage> _networkService;
+
+ /// <summary>
+ /// Creates a new MpiClient and registers the task ID with the Name Server.
+ /// </summary>
+ /// <param name="groupConfigs">The set of serialized Group Communication configurations</param>
+ /// <param name="taskId">The identifier for this task</param>
+ /// <param name="mpiNetworkObserver">The network handler to receive incoming messages
+ /// for this task</param>
+ /// <param name="networkService">The network service used to send messages</param>
+ /// <param name="configSerializer">Used to deserialize Group Communication configuration</param>
+ [Inject]
+ public MpiClient(
+ [Parameter(typeof(MpiConfigurationOptions.SerializedGroupConfigs))] ISet<string> groupConfigs,
+ [Parameter(typeof(TaskConfigurationOptions.Identifier))] string taskId,
+ IMpiNetworkObserver mpiNetworkObserver,
+ NetworkService<GroupCommunicationMessage> networkService,
+ AvroConfigurationSerializer configSerializer)
+ {
+ _commGroups = new Dictionary<string, ICommunicationGroupClient>();
+ _networkService = networkService;
+ networkService.Register(new StringIdentifier(taskId));
+
+ foreach (string serializedGroupConfig in groupConfigs)
+ {
+ IConfiguration groupConfig = configSerializer.FromString(serializedGroupConfig);
+
+ IInjector injector = TangFactory.GetTang().NewInjector(groupConfig);
+ injector.BindVolatileParameter(GenericType<TaskConfigurationOptions.Identifier>.Class, taskId);
+ injector.BindVolatileInstance(GenericType<IMpiNetworkObserver>.Class, mpiNetworkObserver);
+ injector.BindVolatileInstance(GenericType<NetworkService<GroupCommunicationMessage>>.Class, networkService);
+
+ ICommunicationGroupClient commGroup = injector.GetInstance<ICommunicationGroupClient>();
+ _commGroups[commGroup.GroupName] = commGroup;
+ }
+ }
+
+ /// <summary>
+ /// Gets the CommunicationGroupClient for the given group name.
+ /// </summary>
+ /// <param name="groupName">The name of the CommunicationGroupClient</param>
+ /// <returns>The CommunicationGroupClient</returns>
+ public ICommunicationGroupClient GetCommunicationGroup(string groupName)
+ {
+ if (string.IsNullOrEmpty(groupName))
+ {
+ throw new ArgumentNullException("groupName");
+ }
+ if (!_commGroups.ContainsKey(groupName))
+ {
+ throw new ArgumentException("No CommunicationGroupClient with name: " + groupName);
+ }
+
+ return _commGroups[groupName];
+ }
+
+ /// <summary>
+ /// Disposes of the MpiClient's services.
+ /// </summary>
+ public void Dispose()
+ {
+ _networkService.Unregister();
+ _networkService.Dispose();
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/0292caf1/lang/cs/Org.Apache.REEF.Network/Group/Task/Impl/MpiNetworkObserver.cs
----------------------------------------------------------------------
diff --git a/lang/cs/Org.Apache.REEF.Network/Group/Task/Impl/MpiNetworkObserver.cs b/lang/cs/Org.Apache.REEF.Network/Group/Task/Impl/MpiNetworkObserver.cs
new file mode 100644
index 0000000..baa2e5e
--- /dev/null
+++ b/lang/cs/Org.Apache.REEF.Network/Group/Task/Impl/MpiNetworkObserver.cs
@@ -0,0 +1,109 @@
+/**
+ * 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.
+ */
+
+using Org.Apache.REEF.Network.Group.Driver;
+using Org.Apache.REEF.Network.Group.Driver.Impl;
+using Org.Apache.REEF.Network.NetworkService;
+using Org.Apache.REEF.Utilities.Logging;
+using Org.Apache.REEF.Tang.Annotations;
+using System;
+using System.Collections.Generic;
+using System.Linq;
+
+namespace Org.Apache.REEF.Network.Group.Task.Impl
+{
+ /// <summary>
+ /// Handles all incoming messages for this Task.
+ /// </summary>
+ public class MpiNetworkObserver : IMpiNetworkObserver
+ {
+ private static Logger LOGGER = Logger.GetLogger(typeof(MpiNetworkObserver));
+
+ private Dictionary<string, IObserver<GroupCommunicationMessage>> _commGroupHandlers;
+
+ /// <summary>
+ /// Creates a new MpiNetworkObserver.
+ /// </summary>
+ [Inject]
+ public MpiNetworkObserver()
+ {
+ _commGroupHandlers = new Dictionary<string, IObserver<GroupCommunicationMessage>>();
+ }
+
+ /// <summary>
+ /// Handles the incoming NsMessage for this Task.
+ /// Delegates the GroupCommunicationMessage to the correct
+ /// CommunicationGroupNetworkObserver.
+ /// </summary>
+ /// <param name="nsMessage"></param>
+ public void OnNext(NsMessage<GroupCommunicationMessage> nsMessage)
+ {
+ if (nsMessage == null)
+ {
+ throw new ArgumentNullException("nsMessage");
+ }
+
+ try
+ {
+ GroupCommunicationMessage gcm = nsMessage.Data.First();
+ _commGroupHandlers[gcm.GroupName].OnNext(gcm);
+ }
+ catch (InvalidOperationException)
+ {
+ LOGGER.Log(Level.Error, "Mpi Network Handler received message with no data");
+ throw;
+ }
+ catch (KeyNotFoundException)
+ {
+ LOGGER.Log(Level.Error, "Mpi Network Handler received message for nonexistant group");
+ throw;
+ }
+ }
+
+ /// <summary>
+ /// Registers the network handler for the given CommunicationGroup.
+ /// When messages are sent to the specified group name, the given handler
+ /// will be invoked with that message.
+ /// </summary>
+ /// <param name="groupName">The group name for the network handler</param>
+ /// <param name="commGroupHandler">The network handler to invoke when
+ /// messages are sent to the given group.</param>
+ public void Register(string groupName, IObserver<GroupCommunicationMessage> commGroupHandler)
+ {
+ if (string.IsNullOrEmpty(groupName))
+ {
+ throw new ArgumentNullException("groupName");
+ }
+ if (commGroupHandler == null)
+ {
+ throw new ArgumentNullException("commGroupHandler");
+ }
+
+ _commGroupHandlers[groupName] = commGroupHandler;
+ }
+
+ public void OnError(Exception error)
+ {
+ }
+
+ public void OnCompleted()
+ {
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/0292caf1/lang/cs/Org.Apache.REEF.Network/Group/Task/Impl/NodeStruct.cs
----------------------------------------------------------------------
diff --git a/lang/cs/Org.Apache.REEF.Network/Group/Task/Impl/NodeStruct.cs b/lang/cs/Org.Apache.REEF.Network/Group/Task/Impl/NodeStruct.cs
new file mode 100644
index 0000000..f4c7a60
--- /dev/null
+++ b/lang/cs/Org.Apache.REEF.Network/Group/Task/Impl/NodeStruct.cs
@@ -0,0 +1,67 @@
+/**
+ * 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.
+ */
+
+using Org.Apache.REEF.Network.Group.Driver;
+using System.Collections.Concurrent;
+using Org.Apache.REEF.Network.Group.Driver.Impl;
+
+namespace Org.Apache.REEF.Network.Group.Task.Impl
+{
+ /// <summary>
+ /// Stores all incoming messages sent by a particular Task.
+ /// </summary>
+ internal class NodeStruct
+ {
+ private BlockingCollection<GroupCommunicationMessage> _messageQueue;
+
+ /// <summary>
+ /// Creates a new NodeStruct.
+ /// </summary>
+ /// <param name="id">The Task identifier</param>
+ public NodeStruct(string id)
+ {
+ Identifier = id;
+ _messageQueue = new BlockingCollection<GroupCommunicationMessage>();
+ }
+
+ /// <summary>
+ /// Returns the identifier for the Task that sent all
+ /// messages in the message queue.
+ /// </summary>
+ public string Identifier { get; private set; }
+
+ /// <summary>
+ /// Gets the first message in the message queue.
+ /// </summary>
+ /// <returns>The first available message.</returns>
+ public byte[][] GetData()
+ {
+ return _messageQueue.Take().Data;
+ }
+
+ /// <summary>
+ /// Adds an incoming message to the message queue.
+ /// </summary>
+ /// <param name="gcm">The incoming message</param>
+ public void AddData(GroupCommunicationMessage gcm)
+ {
+ _messageQueue.Add(gcm);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/0292caf1/lang/cs/Org.Apache.REEF.Network/Group/Task/Impl/OperatorTopology.cs
----------------------------------------------------------------------
diff --git a/lang/cs/Org.Apache.REEF.Network/Group/Task/Impl/OperatorTopology.cs b/lang/cs/Org.Apache.REEF.Network/Group/Task/Impl/OperatorTopology.cs
new file mode 100644
index 0000000..8752203
--- /dev/null
+++ b/lang/cs/Org.Apache.REEF.Network/Group/Task/Impl/OperatorTopology.cs
@@ -0,0 +1,484 @@
+/**
+ * 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.
+ */
+
+using Org.Apache.REEF.Network.Group.Config;
+using Org.Apache.REEF.Network.Group.Driver;
+using Org.Apache.REEF.Network.Group.Driver.Impl;
+using Org.Apache.REEF.Network.Group.Operators;
+using Org.Apache.REEF.Network.Group.Operators.Impl;
+using Org.Apache.REEF.Network.NetworkService;
+using Org.Apache.REEF.Network.Utilities;
+using Org.Apache.REEF.Utilities.Logging;
+using Org.Apache.REEF.Tang.Annotations;
+using Org.Apache.REEF.Tang.Exceptions;
+using Org.Apache.REEF.Wake.Remote;
+using System;
+using System.Collections.Concurrent;
+using System.Collections.Generic;
+using System.Linq;
+using System.Threading;
+using Org.Apache.REEF.Common.Io;
+using Org.Apache.REEF.Common.Tasks;
+
+namespace Org.Apache.REEF.Network.Group.Task.Impl
+{
+ /// <summary>
+ /// Contains the Operator's topology graph.
+ /// Used to send or receive messages to/from operators in the same
+ /// Communication Group.
+ /// </summary>
+ /// <typeparam name="T">The message type</typeparam>
+ public class OperatorTopology<T> : IObserver<GroupCommunicationMessage>
+ {
+ private const int DefaultTimeout = 10000;
+ private const int RetryCount = 5;
+
+ private static Logger LOGGER = Logger.GetLogger(typeof(OperatorTopology<>));
+
+ private string _groupName;
+ private string _operatorName;
+ private string _selfId;
+ private string _driverId;
+
+ private NodeStruct _parent;
+ private List<NodeStruct> _children;
+ private Dictionary<string, NodeStruct> _idToNodeMap;
+ private ICodec<T> _codec;
+ private INameClient _nameClient;
+ private Sender _sender;
+ private BlockingCollection<NodeStruct> _nodesWithData;
+
+ /// <summary>
+ /// Creates a new OperatorTopology object.
+ /// </summary>
+ /// <param name="operatorName">The name of the MPI Operator</param>
+ /// <param name="groupName">The name of the operator's Communication Group</param>
+ /// <param name="taskId">The operator's Task identifier</param>
+ /// <param name="driverId">The identifer for the driver</param>
+ /// <param name="rootId">The identifier for the root Task in the topology graph</param>
+ /// <param name="childIds">The set of child Task identifiers in the topology graph</param>
+ /// <param name="networkService">The network service</param>
+ /// <param name="codec">The codec used to serialize and deserialize messages</param>
+ /// <param name="sender">The Sender used to do point to point communication</param>
+ [Inject]
+ public OperatorTopology(
+ [Parameter(typeof(MpiConfigurationOptions.OperatorName))] string operatorName,
+ [Parameter(typeof(MpiConfigurationOptions.CommunicationGroupName))] string groupName,
+ [Parameter(typeof(TaskConfigurationOptions.Identifier))] string taskId,
+ [Parameter(typeof(MpiConfigurationOptions.DriverId))] string driverId,
+ [Parameter(typeof(MpiConfigurationOptions.TopologyRootTaskId))] string rootId,
+ [Parameter(typeof(MpiConfigurationOptions.TopologyChildTaskIds))] ISet<string> childIds,
+ NetworkService<GroupCommunicationMessage> networkService,
+ ICodec<T> codec,
+ Sender sender)
+ {
+ _operatorName = operatorName;
+ _groupName = groupName;
+ _selfId = taskId;
+ _driverId = driverId;
+ _codec = codec;
+ _nameClient = networkService.NamingClient;
+ _sender = sender;
+ _nodesWithData = new BlockingCollection<NodeStruct>();
+ _children = new List<NodeStruct>();
+ _idToNodeMap = new Dictionary<string, NodeStruct>();
+
+ if (_selfId.Equals(rootId))
+ {
+ _parent = null;
+ foreach (string childId in childIds)
+ {
+ NodeStruct node = new NodeStruct(childId);
+ _children.Add(node);
+ _idToNodeMap[childId] = node;
+ }
+ }
+ else
+ {
+ _parent = new NodeStruct(rootId);
+ _idToNodeMap[rootId] = _parent;
+ }
+ }
+
+ /// <summary>
+ /// Initializes operator topology.
+ /// Waits until all Tasks in the CommunicationGroup have registered themselves
+ /// with the Name Service.
+ /// </summary>
+ public void Initialize()
+ {
+ using (LOGGER.LogFunction("OperatorTopology::Initialize"))
+ {
+ if (_parent != null)
+ {
+ WaitForTaskRegistration(_parent.Identifier, RetryCount);
+ }
+
+ if (_children.Count > 0)
+ {
+ foreach (NodeStruct child in _children)
+ {
+ WaitForTaskRegistration(child.Identifier, RetryCount);
+ }
+ }
+ }
+ }
+
+ /// <summary>
+ /// Handles the incoming GroupCommunicationMessage.
+ /// Updates the sending node's message queue.
+ /// </summary>
+ /// <param name="gcm">The incoming message</param>
+ public void OnNext(GroupCommunicationMessage gcm)
+ {
+ if (gcm == null)
+ {
+ throw new ArgumentNullException("gcm");
+ }
+ if (gcm.Source == null)
+ {
+ throw new ArgumentException("Message must have a source");
+ }
+
+ NodeStruct sourceNode = FindNode(gcm.Source);
+ if (sourceNode == null)
+ {
+ throw new IllegalStateException("Received message from invalid task id: " + gcm.Source);
+ }
+
+ _nodesWithData.Add(sourceNode);
+ sourceNode.AddData(gcm);
+ }
+
+ /// <summary>
+ /// Sends the message to the parent Task.
+ /// </summary>
+ /// <param name="message">The message to send</param>
+ /// <param name="type">The message type</param>
+ public void SendToParent(T message, MessageType type)
+ {
+ if (_parent == null)
+ {
+ throw new ArgumentException("No parent for node");
+ }
+
+ SendToNode(message, MessageType.Data, _parent);
+ }
+
+ /// <summary>
+ /// Sends the message to all child Tasks.
+ /// </summary>
+ /// <param name="message">The message to send</param>
+ /// <param name="type">The message type</param>
+ public void SendToChildren(T message, MessageType type)
+ {
+ if (message == null)
+ {
+ throw new ArgumentNullException("message");
+ }
+
+ foreach (NodeStruct child in _children)
+ {
+ SendToNode(message, MessageType.Data, child);
+ }
+ }
+
+ /// <summary>
+ /// Splits the list of messages up evenly and sends each sublist
+ /// to the child Tasks.
+ /// </summary>
+ /// <param name="messages">The list of messages to scatter</param>
+ /// <param name="type">The message type</param>
+ public void ScatterToChildren(List<T> messages, MessageType type)
+ {
+ if (messages == null)
+ {
+ throw new ArgumentNullException("messages");
+ }
+ if (_children.Count <= 0)
+ {
+ throw new ArgumentException("Cannot scatter, no children available");
+ }
+
+ int count = (int) Math.Ceiling(((double) messages.Count) / _children.Count);
+ ScatterHelper(messages, _children, count);
+ }
+
+ /// <summary>
+ /// Splits the list of messages up into chunks of the specified size
+ /// and sends each sublist to the child Tasks.
+ /// </summary>
+ /// <param name="messages">The list of messages to scatter</param>
+ /// <param name="count">The size of each sublist</param>
+ /// <param name="type">The message type</param>
+ public void ScatterToChildren(List<T> messages, int count, MessageType type)
+ {
+ if (messages == null)
+ {
+ throw new ArgumentNullException("messages");
+ }
+ if (count <= 0)
+ {
+ throw new ArgumentException("Count must be positive");
+ }
+
+ ScatterHelper(messages, _children, count);
+ }
+
+ /// <summary>
+ /// Splits the list of messages up into chunks of the specified size
+ /// and sends each sublist to the child Tasks in the specified order.
+ /// </summary>
+ /// <param name="messages">The list of messages to scatter</param>
+ /// <param name="order">The order to send messages</param>
+ /// <param name="type">The message type</param>
+ public void ScatterToChildren(List<T> messages, List<string> order, MessageType type)
+ {
+ if (messages == null)
+ {
+ throw new ArgumentNullException("messages");
+ }
+ if (order == null || order.Count != _children.Count)
+ {
+ throw new ArgumentException("order cannot be null and must have the same number of elements as child tasks");
+ }
+
+ List<NodeStruct> nodes = new List<NodeStruct>();
+ foreach (string taskId in order)
+ {
+ NodeStruct node = FindNode(taskId);
+ if (node == null)
+ {
+ throw new IllegalStateException("Received message from invalid task id: " + taskId);
+ }
+
+ nodes.Add(node);
+ }
+
+ int count = (int) Math.Ceiling(((double) messages.Count) / _children.Count);
+ ScatterHelper(messages, nodes, count);
+ }
+
+ /// <summary>
+ /// Receive an incoming message from the parent Task.
+ /// </summary>
+ /// <returns>The parent Task's message</returns>
+ public T ReceiveFromParent()
+ {
+ byte[][] data = ReceiveFromNode(_parent, true);
+ if (data == null || data.Length != 1)
+ {
+ throw new InvalidOperationException("Cannot receive data from parent node");
+ }
+
+ return _codec.Decode(data[0]);
+ }
+
+ /// <summary>
+ /// Receive a list of incoming messages from the parent Task.
+ /// </summary>
+ /// <returns>The parent Task's list of messages</returns>
+ public List<T> ReceiveListFromParent()
+ {
+ byte[][] data = ReceiveFromNode(_parent, true);
+ if (data == null || data.Length == 0)
+ {
+ throw new InvalidOperationException("Cannot receive data from parent node");
+ }
+
+ return data.Select(b => _codec.Decode(b)).ToList();
+ }
+
+ /// <summary>
+ /// Receives all messages from child Tasks and reduces them with the
+ /// given IReduceFunction.
+ /// </summary>
+ /// <param name="reduceFunction">The class used to reduce messages</param>
+ /// <returns>The result of reducing messages</returns>
+ public T ReceiveFromChildren(IReduceFunction<T> reduceFunction)
+ {
+ if (reduceFunction == null)
+ {
+ throw new ArgumentNullException("reduceFunction");
+ }
+
+ var receivedData = new List<T>();
+ var childrenToReceiveFrom = new HashSet<string>(_children.Select(node => node.Identifier));
+
+ while (childrenToReceiveFrom.Count > 0)
+ {
+ NodeStruct childWithData = GetNodeWithData();
+ byte[][] data = ReceiveFromNode(childWithData, false);
+ if (data == null || data.Length != 1)
+ {
+ throw new InvalidOperationException("Received invalid data from child with id: " + childWithData.Identifier);
+ }
+
+ receivedData.Add(_codec.Decode(data[0]));
+ childrenToReceiveFrom.Remove(childWithData.Identifier);
+ }
+
+ return reduceFunction.Reduce(receivedData);
+ }
+
+ public void OnError(Exception error)
+ {
+ }
+
+ public void OnCompleted()
+ {
+ }
+
+ /// <summary>
+ /// Get a node containing an incoming message.
+ /// </summary>
+ /// <returns>A NodeStruct with incoming data.</returns>
+ private NodeStruct GetNodeWithData()
+ {
+ CancellationTokenSource timeoutSource = new CancellationTokenSource(DefaultTimeout);
+
+ try
+ {
+ return _nodesWithData.Take(timeoutSource.Token);
+ }
+ catch (OperationCanceledException)
+ {
+ LOGGER.Log(Level.Error, "No data to read from child");
+ throw;
+ }
+ catch (ObjectDisposedException)
+ {
+ LOGGER.Log(Level.Error, "No data to read from child");
+ throw;
+ }
+ catch (InvalidOperationException)
+ {
+ LOGGER.Log(Level.Error, "No data to read from child");
+ throw;
+ }
+ }
+
+ /// <summary>
+ /// Sends the message to the Task represented by the given NodeStruct.
+ /// </summary>
+ /// <param name="message">The message to send</param>
+ /// <param name="msgType">The message type</param>
+ /// <param name="node">The NodeStruct representing the Task to send to</param>
+ private void SendToNode(T message, MessageType msgType, NodeStruct node)
+ {
+ GroupCommunicationMessage gcm = new GroupCommunicationMessage(_groupName, _operatorName,
+ _selfId, node.Identifier, _codec.Encode(message), msgType);
+
+ _sender.Send(gcm);
+ }
+
+ /// <summary>
+ /// Sends the list of messages to the Task represented by the given NodeStruct.
+ /// </summary>
+ /// <param name="messages">The list of messages to send</param>
+ /// <param name="msgType">The message type</param>
+ /// <param name="node">The NodeStruct representing the Task to send to</param>
+ private void SendToNode(List<T> messages, MessageType msgType, NodeStruct node)
+ {
+ byte[][] encodedMessages = messages.Select(message => _codec.Encode(message)).ToArray();
+ GroupCommunicationMessage gcm = new GroupCommunicationMessage(_groupName, _operatorName,
+ _selfId, node.Identifier, encodedMessages, msgType);
+
+ _sender.Send(gcm);
+ }
+
+ private void ScatterHelper(List<T> messages, List<NodeStruct> order, int count)
+ {
+ if (count <= 0)
+ {
+ throw new ArgumentException("Count must be positive");
+ }
+
+ int i = 0;
+ foreach (NodeStruct nodeStruct in order)
+ {
+ // The last sublist might be smaller than count if the number of
+ // child tasks is not evenly divisible by count
+ int left = messages.Count - i;
+ int size = (left < count) ? left : count;
+ if (size <= 0)
+ {
+ throw new ArgumentException("Scatter count must be positive");
+ }
+
+ List<T> sublist = messages.GetRange(i, size);
+ SendToNode(sublist, MessageType.Data, nodeStruct);
+
+ i += size;
+ }
+ }
+
+ /// <summary>
+ /// Receive a message from the Task represented by the given NodeStruct.
+ /// Removes the NodeStruct from the nodesWithData queue if requested.
+ /// </summary>
+ /// <param name="node">The node to receive from</param>
+ /// <param name="removeFromQueue">Whether or not to remove the NodeStruct
+ /// from the nodesWithData queue</param>
+ /// <returns>The byte array message from the node</returns>
+ private byte[][] ReceiveFromNode(NodeStruct node, bool removeFromQueue)
+ {
+ byte[][] data = node.GetData();
+ if (removeFromQueue)
+ {
+ _nodesWithData.Take(node);
+ }
+
+ return data;
+ }
+
+ /// <summary>
+ /// Find the NodeStruct with the given Task identifier.
+ /// </summary>
+ /// <param name="identifier">The identifier of the Task</param>
+ /// <returns>The NodeStruct</returns>
+ private NodeStruct FindNode(string identifier)
+ {
+ NodeStruct node;
+ return _idToNodeMap.TryGetValue(identifier, out node) ? node : null;
+ }
+
+ /// <summary>
+ /// Checks if the identifier is registered with the Name Server.
+ /// Throws exception if the operation fails more than the retry count.
+ /// </summary>
+ /// <param name="identifier">The identifier to look up</param>
+ /// <param name="retries">The number of times to retry the lookup operation</param>
+ private void WaitForTaskRegistration(string identifier, int retries)
+ {
+ for (int i = 0; i < retries; i++)
+ {
+ if (_nameClient.Lookup(identifier) != null)
+ {
+ return;
+ }
+
+ Thread.Sleep(500);
+ LOGGER.Log(Level.Verbose, "Retry {0}: retrying lookup for node: {1}", i + 1, identifier);
+ }
+
+ throw new IllegalStateException("Failed to initialize operator topology for node: " + identifier);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/0292caf1/lang/cs/Org.Apache.REEF.Network/Group/Topology/FlatTopology.cs
----------------------------------------------------------------------
diff --git a/lang/cs/Org.Apache.REEF.Network/Group/Topology/FlatTopology.cs b/lang/cs/Org.Apache.REEF.Network/Group/Topology/FlatTopology.cs
new file mode 100644
index 0000000..5342410
--- /dev/null
+++ b/lang/cs/Org.Apache.REEF.Network/Group/Topology/FlatTopology.cs
@@ -0,0 +1,201 @@
+/**
+ * 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.
+ */
+
+using Org.Apache.REEF.Network.Group.Config;
+using Org.Apache.REEF.Network.Group.Operators;
+using Org.Apache.REEF.Network.Group.Operators.Impl;
+using Org.Apache.REEF.Tang.Annotations;
+using Org.Apache.REEF.Tang.Implementations;
+using Org.Apache.REEF.Tang.Interface;
+using Org.Apache.REEF.Tang.Util;
+using Org.Apache.REEF.Wake.Remote;
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.Threading.Tasks;
+using Org.Apache.REEF.Tang.Implementations.Tang;
+
+namespace Org.Apache.REEF.Network.Group.Topology
+{
+ /// <summary>
+ /// Represents a graph of MPI Operators where there are only two levels of
+ /// nodes: the root and all children extending from the root.
+ /// </summary>
+ /// <typeparam name="T">The message type</typeparam>
+ public class FlatTopology<T> : ITopology<T>
+ {
+ private string _groupName;
+ private string _operatorName;
+
+ private string _rootId;
+ private string _driverId;
+
+ private Dictionary<string, TaskNode> _nodes;
+ private TaskNode _root;
+
+ /// <summary>
+ /// Creates a new FlatTopology.
+ /// </summary>
+ /// <param name="operatorName">The operator name</param>
+ /// <param name="groupName">The name of the topology's CommunicationGroup</param>
+ /// <param name="rootId">The root Task identifier</param>
+ /// <param name="driverId">The driver identifier</param>
+ /// <param name="operatorSpec">The operator specification</param>
+ public FlatTopology(
+ string operatorName,
+ string groupName,
+ string rootId,
+ string driverId,
+ IOperatorSpec<T> operatorSpec)
+ {
+ _groupName = groupName;
+ _operatorName = operatorName;
+ _rootId = rootId;
+ _driverId = driverId;
+
+ OperatorSpec = operatorSpec;
+
+ _nodes = new Dictionary<string, TaskNode>();
+ }
+
+ /// <summary>
+ /// Gets the Operator specification
+ /// </summary>
+ public IOperatorSpec<T> OperatorSpec { get; set; }
+
+ /// <summary>
+ /// Gets the task configuration for the operator topology.
+ /// </summary>
+ /// <param name="taskId">The task identifier</param>
+ /// <returns>The task configuration</returns>
+ public IConfiguration GetTaskConfiguration(string taskId)
+ {
+ var confBuilder = TangFactory.GetTang().NewConfigurationBuilder()
+ .BindImplementation(typeof(ICodec<T>), OperatorSpec.Codec.GetType())
+ .BindNamedParameter<MpiConfigurationOptions.TopologyRootTaskId, string>(
+ GenericType<MpiConfigurationOptions.TopologyRootTaskId>.Class,
+ _rootId);
+
+ foreach (string tId in _nodes.Keys)
+ {
+ if (!tId.Equals(_rootId))
+ {
+ confBuilder.BindSetEntry<MpiConfigurationOptions.TopologyChildTaskIds, string>(
+ GenericType<MpiConfigurationOptions.TopologyChildTaskIds>.Class,
+ tId);
+ }
+ }
+
+ if (OperatorSpec is BroadcastOperatorSpec<T>)
+ {
+ BroadcastOperatorSpec<T> broadcastSpec = OperatorSpec as BroadcastOperatorSpec<T>;
+ if (taskId.Equals(broadcastSpec.SenderId))
+ {
+ confBuilder.BindImplementation(GenericType<IMpiOperator<T>>.Class, GenericType<BroadcastSender<T>>.Class);
+ }
+ else
+ {
+ confBuilder.BindImplementation(GenericType<IMpiOperator<T>>.Class, GenericType<BroadcastReceiver<T>>.Class);
+ }
+ }
+ else if (OperatorSpec is ReduceOperatorSpec<T>)
+ {
+ ReduceOperatorSpec<T> reduceSpec = OperatorSpec as ReduceOperatorSpec<T>;
+ confBuilder.BindImplementation(typeof(IReduceFunction<T>), reduceSpec.ReduceFunction.GetType());
+
+ if (taskId.Equals(reduceSpec.ReceiverId))
+ {
+ confBuilder.BindImplementation(GenericType<IMpiOperator<T>>.Class, GenericType<ReduceReceiver<T>>.Class);
+ }
+ else
+ {
+ confBuilder.BindImplementation(GenericType<IMpiOperator<T>>.Class, GenericType<ReduceSender<T>>.Class);
+ }
+ }
+ else if (OperatorSpec is ScatterOperatorSpec<T>)
+ {
+ ScatterOperatorSpec<T> scatterSpec = OperatorSpec as ScatterOperatorSpec<T>;
+ if (taskId.Equals(scatterSpec.SenderId))
+ {
+ confBuilder.BindImplementation(GenericType<IMpiOperator<T>>.Class, GenericType<ScatterSender<T>>.Class);
+ }
+ else
+ {
+ confBuilder.BindImplementation(GenericType<IMpiOperator<T>>.Class, GenericType<ScatterReceiver<T>>.Class);
+ }
+ }
+ else
+ {
+ throw new NotSupportedException("Spec type not supported");
+ }
+
+ return confBuilder.Build();
+ }
+
+ /// <summary>
+ /// Adds a task to the topology graph.
+ /// </summary>
+ /// <param name="taskId">The identifier of the task to add</param>
+ public void AddTask(string taskId)
+ {
+ if (string.IsNullOrEmpty(taskId))
+ {
+ throw new ArgumentNullException("taskId");
+ }
+ else if (_nodes.ContainsKey(taskId))
+ {
+ throw new ArgumentException("Task has already been added to the topology");
+ }
+
+ if (taskId.Equals(_rootId))
+ {
+ SetRootNode(_rootId);
+ }
+ else
+ {
+ AddChild(taskId);
+ }
+ }
+
+ private void SetRootNode(string rootId)
+ {
+ TaskNode rootNode = new TaskNode(_groupName, _operatorName, rootId, _driverId);
+ _root = rootNode;
+
+ foreach (TaskNode childNode in _nodes.Values)
+ {
+ rootNode.AddChild(childNode);
+ childNode.SetParent(rootNode);
+ }
+ }
+
+ private void AddChild(string childId)
+ {
+ TaskNode childNode = new TaskNode(_groupName, _operatorName, childId, _driverId);
+ _nodes[childId] = childNode;
+
+ if (_root != null)
+ {
+ _root.AddChild(childNode);
+ childNode.SetParent(_root);
+ }
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/0292caf1/lang/cs/Org.Apache.REEF.Network/Group/Topology/ITopology.cs
----------------------------------------------------------------------
diff --git a/lang/cs/Org.Apache.REEF.Network/Group/Topology/ITopology.cs b/lang/cs/Org.Apache.REEF.Network/Group/Topology/ITopology.cs
new file mode 100644
index 0000000..c4dc9e6
--- /dev/null
+++ b/lang/cs/Org.Apache.REEF.Network/Group/Topology/ITopology.cs
@@ -0,0 +1,41 @@
+/**
+ * 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.
+ */
+
+using Org.Apache.REEF.Network.Group.Operators;
+using Org.Apache.REEF.Tang.Interface;
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.Threading.Tasks;
+
+namespace Org.Apache.REEF.Network.Group.Topology
+{
+ /// <summary>
+ /// Represents a topology graph for IMpiOperators.
+ /// </summary>
+ public interface ITopology<T>
+ {
+ IOperatorSpec<T> OperatorSpec { get; }
+
+ IConfiguration GetTaskConfiguration(string taskId);
+
+ void AddTask(string taskId);
+ }
+}
[3/4] incubator-reef git commit: [REEF-150] Adding group
communication to REEF .Net
Posted by tm...@apache.org.
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/0292caf1/lang/cs/Org.Apache.REEF.Network/Group/Config/MpiConfigurationOptions.cs
----------------------------------------------------------------------
diff --git a/lang/cs/Org.Apache.REEF.Network/Group/Config/MpiConfigurationOptions.cs b/lang/cs/Org.Apache.REEF.Network/Group/Config/MpiConfigurationOptions.cs
new file mode 100644
index 0000000..423e916
--- /dev/null
+++ b/lang/cs/Org.Apache.REEF.Network/Group/Config/MpiConfigurationOptions.cs
@@ -0,0 +1,71 @@
+/**
+ * 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.
+ */
+
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.Threading.Tasks;
+using Org.Apache.REEF.Tang.Annotations;
+
+namespace Org.Apache.REEF.Network.Group.Config
+{
+ public class MpiConfigurationOptions
+ {
+ [NamedParameter("Name of the communication group")]
+ public class CommunicationGroupName : Name<string>
+ {
+ }
+
+ [NamedParameter("Name of the MPI operator")]
+ public class OperatorName : Name<string>
+ {
+ }
+
+ [NamedParameter("Driver identifier")]
+ public class DriverId : Name<string>
+ {
+ }
+
+ [NamedParameter("Master task identifier")]
+ public class MasterTaskId : Name<string>
+ {
+ }
+
+ [NamedParameter("Serialized communication group configuration")]
+ public class SerializedGroupConfigs : Name<ISet<string>>
+ {
+ }
+
+ [NamedParameter("Serialized operator configuration")]
+ public class SerializedOperatorConfigs : Name<ISet<string>>
+ {
+ }
+
+ [NamedParameter("Id of root task in operator topology")]
+ public class TopologyRootTaskId : Name<string>
+ {
+ }
+
+ [NamedParameter("Ids of child tasks in operator topology")]
+ public class TopologyChildTaskIds : Name<ISet<string>>
+ {
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/0292caf1/lang/cs/Org.Apache.REEF.Network/Group/Driver/ICommunicationGroupDriver.cs
----------------------------------------------------------------------
diff --git a/lang/cs/Org.Apache.REEF.Network/Group/Driver/ICommunicationGroupDriver.cs b/lang/cs/Org.Apache.REEF.Network/Group/Driver/ICommunicationGroupDriver.cs
new file mode 100644
index 0000000..a2a249d
--- /dev/null
+++ b/lang/cs/Org.Apache.REEF.Network/Group/Driver/ICommunicationGroupDriver.cs
@@ -0,0 +1,89 @@
+/**
+ * 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.
+ */
+
+using Org.Apache.REEF.Network.Group.Operators;
+using Org.Apache.REEF.Network.Group.Operators.Impl;
+using Org.Apache.REEF.Tang.Interface;
+using System.Collections.Generic;
+
+namespace Org.Apache.REEF.Network.Group.Driver
+{
+ /// <summary>
+ /// Used to configure MPI operators in Reef driver.
+ /// All operators in the same Communication Group run on the the
+ /// same set of tasks.
+ /// </summary>
+ public interface ICommunicationGroupDriver
+ {
+ /// <summary>
+ /// Returns the list of task ids that belong to this Communication Group
+ /// </summary>
+ List<string> TaskIds { get; }
+
+ /// <summary>
+ /// Adds the Broadcast MPI operator to the communication group.
+ /// </summary>
+ /// <typeparam name="T">The type of messages that operators will send</typeparam>
+ /// <param name="operatorName">The name of the broadcast operator</param>
+ /// <param name="spec">The specification that defines the Broadcast operator</param>
+ /// <returns>The same CommunicationGroupDriver with the added Broadcast operator info</returns>
+ ICommunicationGroupDriver AddBroadcast<T>(string operatorName, BroadcastOperatorSpec<T> spec);
+
+ /// <summary>
+ /// Adds the Reduce MPI operator to the communication group.
+ /// </summary>
+ /// <typeparam name="T">The type of messages that operators will send</typeparam>
+ /// <param name="operatorName">The name of the reduce operator</param>
+ /// <param name="spec">The specification that defines the Reduce operator</param>
+ /// <returns>The same CommunicationGroupDriver with the added Reduce operator info</returns>
+ ICommunicationGroupDriver AddReduce<T>(string operatorName, ReduceOperatorSpec<T> spec);
+
+ /// <summary>
+ /// Adds the Scatter MPI operator to the communication group.
+ /// </summary>
+ /// <typeparam name="T">The type of messages that operators will send</typeparam>
+ /// <param name="operatorName">The name of the scatter operator</param>
+ /// <param name="spec">The specification that defines the Scatter operator</param>
+ /// <returns>The same CommunicationGroupDriver with the added Scatter operator info</returns>
+ ICommunicationGroupDriver AddScatter<T>(string operatorName, ScatterOperatorSpec<T> spec);
+
+ /// <summary>
+ /// Finalizes the CommunicationGroupDriver.
+ /// After the CommunicationGroupDriver has been finalized, no more operators may
+ /// be added to the group.
+ /// </summary>
+ /// <returns>The same finalized CommunicationGroupDriver</returns>
+ ICommunicationGroupDriver Build();
+
+ /// <summary>
+ /// Add a task to the communication group.
+ /// The CommunicationGroupDriver must have called Build() before adding tasks to the group.
+ /// </summary>
+ /// <param name="taskId">The id of the task to add</param>
+ void AddTask(string taskId);
+
+ /// <summary>
+ /// Get the Task Configuration for this communication group.
+ /// Must be called only after all tasks have been added to the CommunicationGroupDriver.
+ /// </summary>
+ /// <param name="taskId">The task id of the task that belongs to this Communication Group</param>
+ /// <returns>The Task Configuration for this communication group</returns>
+ IConfiguration GetGroupTaskConfiguration(string taskId);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/0292caf1/lang/cs/Org.Apache.REEF.Network/Group/Driver/IMpiDriver.cs
----------------------------------------------------------------------
diff --git a/lang/cs/Org.Apache.REEF.Network/Group/Driver/IMpiDriver.cs b/lang/cs/Org.Apache.REEF.Network/Group/Driver/IMpiDriver.cs
new file mode 100644
index 0000000..422d63e
--- /dev/null
+++ b/lang/cs/Org.Apache.REEF.Network/Group/Driver/IMpiDriver.cs
@@ -0,0 +1,93 @@
+/**
+ * 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.
+ */
+
+using Org.Apache.REEF.Driver.Context;
+using Org.Apache.REEF.Tang.Interface;
+
+namespace Org.Apache.REEF.Network.Group.Driver
+{
+ /// <summary>
+ /// Used to create Communication Groups for MPI Operators.
+ /// Also manages configuration for MPI tasks/services.
+ /// </summary>
+ public interface IMpiDriver
+ {
+ /// <summary>
+ /// Returns the identifier for the master task
+ /// </summary>
+ string MasterTaskId { get; }
+
+ /// <summary>
+ /// Create a new CommunicationGroup with the given name and number of tasks/operators.
+ /// </summary>
+ /// <param name="groupName">The new group name</param>
+ /// <param name="numTasks">The number of tasks/operators in the group.</param>
+ /// <returns>The new Communication Group</returns>
+ ICommunicationGroupDriver NewCommunicationGroup(string groupName, int numTasks);
+
+ /// <summary>
+ /// Generates context configuration with a unique identifier.
+ /// </summary>
+ /// <returns>The configured context configuration</returns>
+ IConfiguration GetContextConfiguration();
+
+ /// <summary>
+ /// Get the service configuration required for running MPI on Reef tasks.
+ /// </summary>
+ /// <returns>The service configuration for the Reef tasks</returns>
+ IConfiguration GetServiceConfiguration();
+
+ /// <summary>
+ /// Checks whether this active context can be used to run the Master Task.
+ /// </summary>
+ /// <param name="activeContext">The active context to check</param>
+ /// <returns>True if the active context can run the Master task,
+ /// otherwise false.</returns>
+ bool IsMasterTaskContext(IActiveContext activeContext);
+
+ /// <summary>
+ /// Checks whether this context configuration is used to configure the Master Task.
+ /// </summary>
+ /// <param name="contextConfiguration">The context configuration to check</param>
+ /// <returns>True if the context configuration is used to configure the Master
+ /// Task, otherwise false.</returns>
+ bool IsMasterContextConfiguration(IConfiguration contextConfiguration);
+
+ /// <summary>
+ /// Gets the context number associated with the Active Context id.
+ /// </summary>
+ /// <param name="activeContext">The active context to check</param>
+ /// <returns>The context number associated with the active context id</returns>
+ int GetContextNum(IActiveContext activeContext);
+
+ /// <summary>
+ /// Get the configuration for a particular task.
+ ///
+ /// The task may belong to many Communication Groups, so each one is serialized
+ /// in the configuration as a SerializedGroupConfig.
+ ///
+ /// The user must merge their part of task configuration (task id, task class)
+ /// with this returned MPI task configuration.
+ /// </summary>
+ /// <param name="taskId">The id of the task Configuration to generate</param>
+ /// <returns>The MPI task configuration with communication group and
+ /// operator configuration set.</returns>
+ IConfiguration GetMpiTaskConfiguration(string taskId);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/0292caf1/lang/cs/Org.Apache.REEF.Network/Group/Driver/Impl/CommunicationGroupDriver.cs
----------------------------------------------------------------------
diff --git a/lang/cs/Org.Apache.REEF.Network/Group/Driver/Impl/CommunicationGroupDriver.cs b/lang/cs/Org.Apache.REEF.Network/Group/Driver/Impl/CommunicationGroupDriver.cs
new file mode 100644
index 0000000..d29c504
--- /dev/null
+++ b/lang/cs/Org.Apache.REEF.Network/Group/Driver/Impl/CommunicationGroupDriver.cs
@@ -0,0 +1,260 @@
+/**
+ * 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.
+ */
+
+using Org.Apache.REEF.Network.Group.Config;
+using Org.Apache.REEF.Network.Group.Operators.Impl;
+using Org.Apache.REEF.Network.Group.Topology;
+using Org.Apache.REEF.Network.Utilities;
+using Org.Apache.REEF.Utilities.Logging;
+using Org.Apache.REEF.Tang.Exceptions;
+using Org.Apache.REEF.Tang.Formats;
+using Org.Apache.REEF.Tang.Implementations;
+using Org.Apache.REEF.Tang.Interface;
+using Org.Apache.REEF.Tang.Util;
+using System.Collections.Generic;
+using System.Reflection;
+using Org.Apache.REEF.Tang.Implementations.Tang;
+
+namespace Org.Apache.REEF.Network.Group.Driver.Impl
+{
+ /// <summary>
+ /// Used to configure MPI operators in Reef driver.
+ /// All operators in the same Communication Group run on the the
+ /// same set of tasks.
+ /// </summary>
+ public class CommunicationGroupDriver : ICommunicationGroupDriver
+ {
+ private static readonly Logger LOGGER = Logger.GetLogger(typeof(CommunicationGroupDriver));
+
+ private string _groupName;
+ private string _driverId;
+ private int _numTasks;
+ private int _tasksAdded;
+ private bool _finalized;
+
+ private AvroConfigurationSerializer _confSerializer;
+
+ private object _topologyLock;
+ private Dictionary<string, object> _operatorSpecs;
+ private Dictionary<string, object> _topologies;
+
+ /// <summary>
+ /// Create a new CommunicationGroupDriver.
+ /// </summary>
+ /// <param name="groupName">The communication group name</param>
+ /// <param name="driverId">Identifier of the Reef driver</param>
+ /// <param name="numTasks">The number of tasks each operator will use</param>
+ /// <param name="confSerializer">Used to serialize task configuration</param>
+ public CommunicationGroupDriver(
+ string groupName,
+ string driverId,
+ int numTasks,
+ AvroConfigurationSerializer confSerializer)
+ {
+ _confSerializer = confSerializer;
+ _groupName = groupName;
+ _driverId = driverId;
+ _numTasks = numTasks;
+ _tasksAdded = 0;
+ _finalized = false;
+
+ _topologyLock = new object();
+
+ _operatorSpecs = new Dictionary<string, object>();
+ _topologies = new Dictionary<string, object>();
+ TaskIds = new List<string>();
+ }
+
+ /// <summary>
+ /// Returns the list of task ids that belong to this Communication Group
+ /// </summary>
+ public List<string> TaskIds { get; private set; }
+
+ /// <summary>
+ /// Adds the Broadcast MPI operator to the communication group.
+ /// </summary>
+ /// <typeparam name="T">The type of messages that operators will send</typeparam>
+ /// <param name="operatorName">The name of the broadcast operator</param>
+ /// <param name="spec">The specification that defines the Broadcast operator</param>
+ /// <returns>The same CommunicationGroupDriver with the added Broadcast operator info</returns>
+ public ICommunicationGroupDriver AddBroadcast<T>(
+ string operatorName,
+ BroadcastOperatorSpec<T> spec)
+ {
+ if (_finalized)
+ {
+ throw new IllegalStateException("Can't add operators once the spec has been built.");
+ }
+
+ ITopology<T> topology = new FlatTopology<T>(operatorName, _groupName, spec.SenderId, _driverId, spec);
+ _topologies[operatorName] = topology;
+ _operatorSpecs[operatorName] = spec;
+
+ return this;
+ }
+
+ /// <summary>
+ /// Adds the Reduce MPI operator to the communication group.
+ /// </summary>
+ /// <typeparam name="T">The type of messages that operators will send</typeparam>
+ /// <param name="operatorName">The name of the reduce operator</param>
+ /// <param name="spec">The specification that defines the Reduce operator</param>
+ /// <returns>The same CommunicationGroupDriver with the added Reduce operator info</returns>
+ public ICommunicationGroupDriver AddReduce<T>(
+ string operatorName,
+ ReduceOperatorSpec<T> spec)
+ {
+ if (_finalized)
+ {
+ throw new IllegalStateException("Can't add operators once the spec has been built.");
+ }
+
+ ITopology<T> topology = new FlatTopology<T>(operatorName, _groupName, spec.ReceiverId, _driverId, spec);
+ _topologies[operatorName] = topology;
+ _operatorSpecs[operatorName] = spec;
+
+ return this;
+ }
+
+ /// <summary>
+ /// Adds the Scatter MPI operator to the communication group.
+ /// </summary>
+ /// <typeparam name="T">The type of messages that operators will send</typeparam>
+ /// <param name="operatorName">The name of the scatter operator</param>
+ /// <param name="spec">The specification that defines the Scatter operator</param>
+ /// <returns>The same CommunicationGroupDriver with the added Scatter operator info</returns>
+ public ICommunicationGroupDriver AddScatter<T>(string operatorName, ScatterOperatorSpec<T> spec)
+ {
+ if (_finalized)
+ {
+ throw new IllegalStateException("Can't add operators once the spec has been built.");
+ }
+
+ ITopology<T> topology = new FlatTopology<T>(operatorName, _groupName, spec.SenderId, _driverId, spec);
+ _topologies[operatorName] = topology;
+ _operatorSpecs[operatorName] = spec;
+
+ return this;
+ }
+
+ /// <summary>
+ /// Finalizes the CommunicationGroupDriver.
+ /// After the CommunicationGroupDriver has been finalized, no more operators may
+ /// be added to the group.
+ /// </summary>
+ /// <returns>The same finalized CommunicationGroupDriver</returns>
+ public ICommunicationGroupDriver Build()
+ {
+ _finalized = true;
+ return this;
+ }
+
+ /// <summary>
+ /// Add a task to the communication group.
+ /// The CommunicationGroupDriver must have called Build() before adding tasks to the group.
+ /// </summary>
+ /// <param name="taskId">The id of the task to add</param>
+ public void AddTask(string taskId)
+ {
+ if (!_finalized)
+ {
+ throw new IllegalStateException("CommunicationGroupDriver must call Build() before adding tasks to the group.");
+ }
+
+ lock (_topologyLock)
+ {
+ _tasksAdded++;
+ if (_tasksAdded > _numTasks)
+ {
+ throw new IllegalStateException("Added too many tasks to Communication Group, expected: " + _numTasks);
+ }
+
+ TaskIds.Add(taskId);
+ foreach (string operatorName in _operatorSpecs.Keys)
+ {
+ AddTask(operatorName, taskId);
+ }
+ }
+ }
+
+ /// <summary>
+ /// Get the Task Configuration for this communication group.
+ /// Must be called only after all tasks have been added to the CommunicationGroupDriver.
+ /// </summary>
+ /// <param name="taskId">The task id of the task that belongs to this Communication Group</param>
+ /// <returns>The Task Configuration for this communication group</returns>
+ public IConfiguration GetGroupTaskConfiguration(string taskId)
+ {
+ if (!TaskIds.Contains(taskId))
+ {
+ return null;
+ }
+
+ // Make sure all tasks have been added to communication group before generating config
+ lock (_topologyLock)
+ {
+ if (_tasksAdded != _numTasks)
+ {
+ throw new IllegalStateException(
+ "Must add all tasks to communication group before fetching configuration");
+ }
+ }
+
+ var confBuilder = TangFactory.GetTang().NewConfigurationBuilder()
+ .BindNamedParameter<MpiConfigurationOptions.DriverId, string>(
+ GenericType<MpiConfigurationOptions.DriverId>.Class,
+ _driverId)
+ .BindNamedParameter<MpiConfigurationOptions.CommunicationGroupName, string>(
+ GenericType<MpiConfigurationOptions.CommunicationGroupName>.Class,
+ _groupName);
+
+ foreach (var operatorName in _topologies.Keys)
+ {
+ var innerConf = TangFactory.GetTang().NewConfigurationBuilder(GetOperatorConfiguration(operatorName, taskId))
+ .BindNamedParameter<MpiConfigurationOptions.DriverId, string>(
+ GenericType<MpiConfigurationOptions.DriverId>.Class,
+ _driverId)
+ .BindNamedParameter<MpiConfigurationOptions.OperatorName, string>(
+ GenericType<MpiConfigurationOptions.OperatorName>.Class,
+ operatorName)
+ .Build();
+
+ confBuilder.BindSetEntry<MpiConfigurationOptions.SerializedOperatorConfigs, string>(
+ GenericType<MpiConfigurationOptions.SerializedOperatorConfigs>.Class,
+ _confSerializer.ToString(innerConf));
+ }
+
+ return confBuilder.Build();
+ }
+
+ private void AddTask(string operatorName, string taskId)
+ {
+ var topology = _topologies[operatorName];
+ MethodInfo info = topology.GetType().GetMethod("AddTask");
+ info.Invoke(topology, new[] { (object) taskId });
+ }
+
+ private IConfiguration GetOperatorConfiguration(string operatorName, string taskId)
+ {
+ var topology = _topologies[operatorName];
+ MethodInfo info = topology.GetType().GetMethod("GetTaskConfiguration");
+ return (IConfiguration) info.Invoke(topology, new[] { (object) taskId });
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/0292caf1/lang/cs/Org.Apache.REEF.Network/Group/Driver/Impl/GroupCommunicationMessage.cs
----------------------------------------------------------------------
diff --git a/lang/cs/Org.Apache.REEF.Network/Group/Driver/Impl/GroupCommunicationMessage.cs b/lang/cs/Org.Apache.REEF.Network/Group/Driver/Impl/GroupCommunicationMessage.cs
new file mode 100644
index 0000000..1439a36
--- /dev/null
+++ b/lang/cs/Org.Apache.REEF.Network/Group/Driver/Impl/GroupCommunicationMessage.cs
@@ -0,0 +1,107 @@
+/**
+ * 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 Org.Apache.REEF.Network.Group.Driver.Impl
+{
+ /// <summary>
+ /// Messages sent by MPI Operators
+ /// </summary>
+ public class GroupCommunicationMessage
+ {
+ /// <summary>
+ /// Create new CommunicationGroupMessage.
+ /// </summary>
+ /// <param name="groupName">The name of the communication group</param>
+ /// <param name="operatorName">The name of the MPI operator</param>
+ /// <param name="source">The message source</param>
+ /// <param name="destination">The message destination</param>
+ /// <param name="data">The actual byte array of data</param>
+ /// <param name="messageType">The type of message to send</param>
+ public GroupCommunicationMessage(
+ string groupName,
+ string operatorName,
+ string source,
+ string destination,
+ byte[] data,
+ MessageType messageType)
+ {
+ GroupName = groupName;
+ OperatorName = operatorName;
+ Source = source;
+ Destination = destination;
+ Data = new[] { data };
+ MsgType = messageType;
+ }
+
+ /// <summary>
+ /// Create new CommunicationGroupMessage.
+ /// </summary>
+ /// <param name="groupName">The name of the communication group</param>
+ /// <param name="operatorName">The name of the MPI operator</param>
+ /// <param name="source">The message source</param>
+ /// <param name="destination">The message destination</param>
+ /// <param name="data">The actual byte array of data</param>
+ /// <param name="messageType">The type of message to send</param>
+ public GroupCommunicationMessage(
+ string groupName,
+ string operatorName,
+ string source,
+ string destination,
+ byte[][] data,
+ MessageType messageType)
+ {
+ GroupName = groupName;
+ OperatorName = operatorName;
+ Source = source;
+ Destination = destination;
+ Data = data;
+ MsgType = messageType;
+ }
+
+ /// <summary>
+ /// Returns the Communication Group name.
+ /// </summary>
+ public string GroupName { get; private set; }
+
+ /// <summary>
+ /// Returns the MPI Operator name.
+ /// </summary>
+ public string OperatorName { get; private set; }
+
+ /// <summary>
+ /// Returns the source of the message.
+ /// </summary>
+ public string Source { get; private set; }
+
+ /// <summary>
+ /// Returns the destination of the message.
+ /// </summary>
+ public string Destination { get; private set; }
+
+ /// <summary>
+ /// Returns the message data.
+ /// </summary>
+ public byte[][] Data { get; private set; }
+
+ /// <summary>
+ /// Returns the type of message being sent.
+ /// </summary>
+ public MessageType MsgType { get; private set; }
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/0292caf1/lang/cs/Org.Apache.REEF.Network/Group/Driver/Impl/MessageType.cs
----------------------------------------------------------------------
diff --git a/lang/cs/Org.Apache.REEF.Network/Group/Driver/Impl/MessageType.cs b/lang/cs/Org.Apache.REEF.Network/Group/Driver/Impl/MessageType.cs
new file mode 100644
index 0000000..cd8ace2
--- /dev/null
+++ b/lang/cs/Org.Apache.REEF.Network/Group/Driver/Impl/MessageType.cs
@@ -0,0 +1,30 @@
+/**
+ * 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 Org.Apache.REEF.Network.Group.Driver.Impl
+{
+ /// <summary>
+ /// Represents the different types of messages that Mpi Tasks can
+ /// send to each other.
+ /// </summary>
+ public enum MessageType
+ {
+ Data = 0
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/0292caf1/lang/cs/Org.Apache.REEF.Network/Group/Driver/Impl/MpiDriver.cs
----------------------------------------------------------------------
diff --git a/lang/cs/Org.Apache.REEF.Network/Group/Driver/Impl/MpiDriver.cs b/lang/cs/Org.Apache.REEF.Network/Group/Driver/Impl/MpiDriver.cs
new file mode 100644
index 0000000..a373ef3
--- /dev/null
+++ b/lang/cs/Org.Apache.REEF.Network/Group/Driver/Impl/MpiDriver.cs
@@ -0,0 +1,239 @@
+/**
+ * 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.
+ */
+
+using System;
+using System.Collections.Generic;
+using System.Globalization;
+using System.Net;
+using System.Threading;
+using Org.Apache.REEF.Common.Io;
+using Org.Apache.REEF.Common.Services;
+using Org.Apache.REEF.Driver.Context;
+using Org.Apache.REEF.Network.Group.Codec;
+using Org.Apache.REEF.Network.Group.Config;
+using Org.Apache.REEF.Network.Group.Task.Impl;
+using Org.Apache.REEF.Network.Naming;
+using Org.Apache.REEF.Network.NetworkService;
+using Org.Apache.REEF.Utilities.Logging;
+using Org.Apache.REEF.Tang.Annotations;
+using Org.Apache.REEF.Tang.Formats;
+using Org.Apache.REEF.Tang.Implementations.Tang;
+using Org.Apache.REEF.Tang.Interface;
+using Org.Apache.REEF.Tang.Util;
+using Org.Apache.REEF.Wake.Remote;
+
+namespace Org.Apache.REEF.Network.Group.Driver.Impl
+{
+ /// <summary>
+ /// Used to create Communication Groups for MPI Operators on the Reef driver.
+ /// Also manages configuration for MPI tasks/services.
+ /// </summary>
+ public class MpiDriver : IMpiDriver
+ {
+ private const string MasterTaskContextName = "MasterTaskContext";
+ private const string SlaveTaskContextName = "SlaveTaskContext";
+
+ private static Logger LOGGER = Logger.GetLogger(typeof(MpiDriver));
+
+ private string _driverId;
+ private string _nameServerAddr;
+ private int _nameServerPort;
+ private int _contextIds;
+
+ private Dictionary<string, ICommunicationGroupDriver> _commGroups;
+ private AvroConfigurationSerializer _configSerializer;
+ private NameServer _nameServer;
+
+ /// <summary>
+ /// Create a new MpiDriver object.
+ /// </summary>
+ /// <param name="driverId">Identifer for the REEF driver</param>
+ /// <param name="masterTaskId">Identifer for MPI master task</param>
+ /// <param name="configSerializer">Used to serialize task configuration</param>
+ [Inject]
+ public MpiDriver(
+ [Parameter(typeof(MpiConfigurationOptions.DriverId))] string driverId,
+ [Parameter(typeof(MpiConfigurationOptions.MasterTaskId))] string masterTaskId,
+ AvroConfigurationSerializer configSerializer)
+ {
+ _driverId = driverId;
+ _contextIds = -1;
+ MasterTaskId = masterTaskId;
+
+ _configSerializer = configSerializer;
+ _commGroups = new Dictionary<string, ICommunicationGroupDriver>();
+ _nameServer = new NameServer(0);
+
+ IPEndPoint localEndpoint = _nameServer.LocalEndpoint;
+ _nameServerAddr = localEndpoint.Address.ToString();
+ _nameServerPort = localEndpoint.Port;
+ }
+
+ /// <summary>
+ /// Returns the identifier for the master task
+ /// </summary>
+ public string MasterTaskId { get; private set; }
+
+ /// <summary>
+ /// Create a new CommunicationGroup with the given name and number of tasks/operators.
+ /// </summary>
+ /// <param name="groupName">The new group name</param>
+ /// <param name="numTasks">The number of tasks/operators in the group.</param>
+ /// <returns>The new Communication Group</returns>
+ public ICommunicationGroupDriver NewCommunicationGroup(string groupName, int numTasks)
+ {
+ if (string.IsNullOrEmpty(groupName))
+ {
+ throw new ArgumentNullException("groupName");
+ }
+ else if (numTasks < 1)
+ {
+ throw new ArgumentException("NumTasks must be greater than 0");
+ }
+ else if (_commGroups.ContainsKey(groupName))
+ {
+ throw new ArgumentException("Group Name already registered with MpiDriver");
+ }
+
+ var commGroup = new CommunicationGroupDriver(groupName, _driverId, numTasks, _configSerializer);
+ _commGroups[groupName] = commGroup;
+ return commGroup;
+ }
+
+ /// <summary>
+ /// Generates context configuration with a unique identifier.
+ /// </summary>
+ /// <returns>The configured context configuration</returns>
+ public IConfiguration GetContextConfiguration()
+ {
+ int contextNum = Interlocked.Increment(ref _contextIds);
+ string id = (contextNum == 0)
+ ? MasterTaskContextName
+ : GetSlaveTaskContextName(contextNum);
+
+ return ContextConfiguration.ConfigurationModule
+ .Set(ContextConfiguration.Identifier, id)
+ .Build();
+ }
+
+ /// <summary>
+ /// Get the service configuration required for running MPI on Reef tasks.
+ /// </summary>
+ /// <returns>The service configuration for the Reef tasks</returns>
+ public IConfiguration GetServiceConfiguration()
+ {
+ IConfiguration serviceConfig = ServiceConfiguration.ConfigurationModule
+ .Set(ServiceConfiguration.Services, GenericType<NetworkService<GroupCommunicationMessage>>.Class)
+ .Build();
+
+ return TangFactory.GetTang().NewConfigurationBuilder(serviceConfig)
+ .BindImplementation(
+ GenericType<IObserver<NsMessage<GroupCommunicationMessage>>>.Class,
+ GenericType<MpiNetworkObserver>.Class)
+ .BindImplementation(
+ GenericType<ICodec<GroupCommunicationMessage>>.Class,
+ GenericType<GroupCommunicationMessageCodec>.Class)
+ .BindNamedParameter<NamingConfigurationOptions.NameServerAddress, string>(
+ GenericType<NamingConfigurationOptions.NameServerAddress>.Class,
+ _nameServerAddr)
+ .BindNamedParameter<NamingConfigurationOptions.NameServerPort, int>(
+ GenericType<NamingConfigurationOptions.NameServerPort>.Class,
+ _nameServerPort.ToString(CultureInfo.InvariantCulture))
+ .Build();
+ }
+
+ /// <summary>
+ /// Get the configuration for a particular task.
+ ///
+ /// The task may belong to many Communication Groups, so each one is serialized
+ /// in the configuration as a SerializedGroupConfig.
+ ///
+ /// The user must merge their part of task configuration (task id, task class)
+ /// with this returned MPI task configuration.
+ /// </summary>
+ /// <param name="taskId">The id of the task Configuration to generate</param>
+ /// <returns>The MPI task configuration with communication group and
+ /// operator configuration set.</returns>
+ public IConfiguration GetMpiTaskConfiguration(string taskId)
+ {
+ var confBuilder = TangFactory.GetTang().NewConfigurationBuilder();
+
+ foreach (ICommunicationGroupDriver commGroup in _commGroups.Values)
+ {
+ var taskConf = commGroup.GetGroupTaskConfiguration(taskId);
+ if (taskConf != null)
+ {
+ confBuilder.BindSetEntry<MpiConfigurationOptions.SerializedGroupConfigs, string>(
+ GenericType<MpiConfigurationOptions.SerializedGroupConfigs>.Class,
+ _configSerializer.ToString(taskConf));
+ }
+ }
+
+ return confBuilder.Build();
+ }
+
+ /// <summary>
+ /// Checks whether this active context can be used to run the Master Task.
+ /// </summary>
+ /// <param name="activeContext">The active context to check</param>
+ /// <returns>True if the active context can run the Master task,
+ /// otherwise false.</returns>
+ public bool IsMasterTaskContext(IActiveContext activeContext)
+ {
+ return activeContext.Id.Equals(MasterTaskContextName);
+ }
+
+ /// <summary>
+ /// Checks whether this context configuration is used to configure the Master Task.
+ /// </summary>
+ /// <param name="contextConfiguration">The context configuration to check</param>
+ /// <returns>True if the context configuration is used to configure the Master
+ /// Task, otherwise false.</returns>
+ public bool IsMasterContextConfiguration(IConfiguration contextConfiguration)
+ {
+ return Utilities.Utils.GetContextId(contextConfiguration).Equals(MasterTaskContextName);
+ }
+
+ /// <summary>
+ /// Gets the context number associated with the Active Context id.
+ /// </summary>
+ /// <param name="activeContext">The active context to check</param>
+ /// <returns>The context number associated with the active context id</returns>
+ public int GetContextNum(IActiveContext activeContext)
+ {
+ if (activeContext.Id.Equals(MasterTaskContextName))
+ {
+ return 0;
+ }
+
+ string[] parts = activeContext.Id.Split('-');
+ if (parts.Length != 2)
+ {
+ throw new ArgumentException("Invalid id in active context");
+ }
+
+ return int.Parse(parts[1], CultureInfo.InvariantCulture);
+ }
+
+ private string GetSlaveTaskContextName(int contextNum)
+ {
+ return string.Format(CultureInfo.InvariantCulture, "{0}-{1}", SlaveTaskContextName, contextNum);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/0292caf1/lang/cs/Org.Apache.REEF.Network/Group/Driver/Impl/TaskStarter.cs
----------------------------------------------------------------------
diff --git a/lang/cs/Org.Apache.REEF.Network/Group/Driver/Impl/TaskStarter.cs b/lang/cs/Org.Apache.REEF.Network/Group/Driver/Impl/TaskStarter.cs
new file mode 100644
index 0000000..48463dc
--- /dev/null
+++ b/lang/cs/Org.Apache.REEF.Network/Group/Driver/Impl/TaskStarter.cs
@@ -0,0 +1,135 @@
+/**
+ * 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.
+ */
+
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Threading;
+using Org.Apache.REEF.Driver.Context;
+using Org.Apache.REEF.Network.Utilities;
+using Org.Apache.REEF.Utilities.Logging;
+using Org.Apache.REEF.Tang.Implementations.Configuration;
+using Org.Apache.REEF.Tang.Interface;
+
+namespace Org.Apache.REEF.Network.Group.Driver.Impl
+{
+ /// <summary>
+ /// Helper class to start MPI tasks.
+ /// </summary>
+ public class TaskStarter
+ {
+ private static Logger LOGGER = Logger.GetLogger(typeof(TaskStarter));
+
+ private object _lock;
+ private int _numTasks;
+ private int _tasksAdded;
+ private string _masterTaskId;
+
+ private IMpiDriver _mpiDriver;
+ private List<Tuple<string, IConfiguration, IActiveContext>> _taskTuples;
+
+ /// <summary>
+ /// Create new TaskStarter.
+ /// After adding the correct number of tasks to the TaskStarter, the
+ /// Tasks will be started on their given active context.
+ /// </summary>
+ /// <param name="mpiDriver">The IMpiDriver for the MPI tasks</param>
+ /// <param name="numTasks">The number of Tasks that need to be added before
+ /// the Tasks will be started. </param>
+ public TaskStarter(IMpiDriver mpiDriver, int numTasks)
+ {
+ LOGGER.Log(Level.Verbose, "Creating TaskStarter");
+ _masterTaskId = mpiDriver.MasterTaskId;
+ _numTasks = numTasks;
+ _tasksAdded = 0;
+ _lock = new object();
+
+ _mpiDriver = mpiDriver;
+ _taskTuples = new List<Tuple<string, IConfiguration, IActiveContext>>();
+ }
+
+ /// <summary>
+ /// Queues the task into the TaskStarter.
+ ///
+ /// Once the correct number of tasks have been queued, the final Configuration
+ /// will be generated and run on the given Active Context.
+ /// </summary>
+ /// <param name="partialTaskConfig">The partial task configuration containing Task
+ /// identifier and Task class</param>
+ /// <param name="activeContext">The Active Context to run the Task on</param>
+ public void QueueTask(IConfiguration partialTaskConfig, IActiveContext activeContext)
+ {
+ string taskId = Utils.GetTaskId(partialTaskConfig);
+ LOGGER.Log(Level.Verbose, "Adding context with identifier: " + taskId);
+
+ lock (_lock)
+ {
+ _taskTuples.Add(
+ new Tuple<string, IConfiguration, IActiveContext>(taskId, partialTaskConfig, activeContext));
+
+ if (Interlocked.Increment(ref _tasksAdded) == _numTasks)
+ {
+ StartTasks();
+ }
+ }
+ }
+
+ /// <summary>
+ /// Starts the Master Task followed by the Slave Tasks.
+ /// </summary>
+ private void StartTasks()
+ {
+ Tuple<string, IConfiguration, IActiveContext> masterTaskTuple;
+ try
+ {
+ masterTaskTuple = _taskTuples.Single(tuple => tuple.Item1.Equals(_masterTaskId));
+ }
+ catch (InvalidOperationException)
+ {
+ LOGGER.Log(Level.Error, "There must be exactly one master task. The driver has been misconfigured.");
+ throw;
+ }
+
+ LOGGER.Log(Level.Verbose, "Starting master task on context id: {0}.", masterTaskTuple.Item3.Id);
+ StartTask(masterTaskTuple.Item1, masterTaskTuple.Item2, masterTaskTuple.Item3);
+
+ LOGGER.Log(Level.Verbose, "Starting slave tasks.");
+ foreach (Tuple<string, IConfiguration, IActiveContext> taskTuple in _taskTuples)
+ {
+ string taskId = taskTuple.Item1;
+ if (taskId.Equals(_masterTaskId))
+ {
+ continue;
+ }
+
+ StartTask(taskId, taskTuple.Item2, taskTuple.Item3);
+ }
+ }
+
+ private void StartTask(
+ string taskId,
+ IConfiguration userPartialTaskConf,
+ IActiveContext activeContext)
+ {
+ IConfiguration mpiTaskConfiguration = _mpiDriver.GetMpiTaskConfiguration(taskId);
+ IConfiguration mergedTaskConf = Configurations.Merge(userPartialTaskConf, mpiTaskConfiguration);
+ activeContext.SubmitTask(mergedTaskConf);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/0292caf1/lang/cs/Org.Apache.REEF.Network/Group/Operators/IBroadcastReceiver.cs
----------------------------------------------------------------------
diff --git a/lang/cs/Org.Apache.REEF.Network/Group/Operators/IBroadcastReceiver.cs b/lang/cs/Org.Apache.REEF.Network/Group/Operators/IBroadcastReceiver.cs
new file mode 100644
index 0000000..09c8c30
--- /dev/null
+++ b/lang/cs/Org.Apache.REEF.Network/Group/Operators/IBroadcastReceiver.cs
@@ -0,0 +1,40 @@
+/**
+ * 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.
+ */
+
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.Threading.Tasks;
+
+namespace Org.Apache.REEF.Network.Group.Operators
+{
+ /// <summary>
+ /// MPI Operator used to receive broadcast messages.
+ /// </summary>
+ /// <typeparam name="T">The type of message being sent.</typeparam>
+ public interface IBroadcastReceiver<T> : IMpiOperator<T>
+ {
+ /// <summary>
+ /// Receive a message from parent BroadcastSender.
+ /// </summary>
+ /// <returns>The incoming message</returns>
+ T Receive();
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/0292caf1/lang/cs/Org.Apache.REEF.Network/Group/Operators/IBroadcastSender.cs
----------------------------------------------------------------------
diff --git a/lang/cs/Org.Apache.REEF.Network/Group/Operators/IBroadcastSender.cs b/lang/cs/Org.Apache.REEF.Network/Group/Operators/IBroadcastSender.cs
new file mode 100644
index 0000000..534fa9f
--- /dev/null
+++ b/lang/cs/Org.Apache.REEF.Network/Group/Operators/IBroadcastSender.cs
@@ -0,0 +1,40 @@
+/**
+ * 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.
+ */
+
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.Threading.Tasks;
+
+namespace Org.Apache.REEF.Network.Group.Operators
+{
+ /// <summary>
+ /// MPI Operator used to send messages to child Tasks.
+ /// </summary>
+ /// <typeparam name="T">The message type</typeparam>
+ public interface IBroadcastSender<T> : IMpiOperator<T>
+ {
+ /// <summary>
+ /// Send the data to all BroadcastReceivers.
+ /// </summary>
+ /// <param name="data">The data to send.</param>
+ void Send(T data);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/0292caf1/lang/cs/Org.Apache.REEF.Network/Group/Operators/IMpiOperator.cs
----------------------------------------------------------------------
diff --git a/lang/cs/Org.Apache.REEF.Network/Group/Operators/IMpiOperator.cs b/lang/cs/Org.Apache.REEF.Network/Group/Operators/IMpiOperator.cs
new file mode 100644
index 0000000..661f348
--- /dev/null
+++ b/lang/cs/Org.Apache.REEF.Network/Group/Operators/IMpiOperator.cs
@@ -0,0 +1,49 @@
+/**
+ * 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.
+ */
+
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.Threading.Tasks;
+
+namespace Org.Apache.REEF.Network.Group.Operators
+{
+ /// <summary>
+ /// An Mpi Operator to be used in a Reef Task.
+ /// </summary>
+ /// <typeparam name="T">The message type</typeparam>
+ public interface IMpiOperator<T>
+ {
+ /// <summary>
+ /// The operator name.
+ /// </summary>
+ string OperatorName { get; }
+
+ /// <summary>
+ /// The name of the operator's CommunicationGroup.
+ /// </summary>
+ string GroupName { get; }
+
+ /// <summary>
+ /// The operator version number.
+ /// </summary>
+ int Version { get; }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/0292caf1/lang/cs/Org.Apache.REEF.Network/Group/Operators/IOperatorSpec.cs
----------------------------------------------------------------------
diff --git a/lang/cs/Org.Apache.REEF.Network/Group/Operators/IOperatorSpec.cs b/lang/cs/Org.Apache.REEF.Network/Group/Operators/IOperatorSpec.cs
new file mode 100644
index 0000000..5d18677
--- /dev/null
+++ b/lang/cs/Org.Apache.REEF.Network/Group/Operators/IOperatorSpec.cs
@@ -0,0 +1,36 @@
+/**
+ * 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.
+ */
+
+using System;
+using Org.Apache.REEF.Tang.Util;
+using Org.Apache.REEF.Wake.Remote;
+
+namespace Org.Apache.REEF.Network.Group.Operators
+{
+ /// <summary>
+ /// The specification used to define Broadcast Operators.
+ /// </summary>
+ public interface IOperatorSpec<T>
+ {
+ /// <summary>
+ /// Returns the codec used to serialize and deserialize messages.
+ /// </summary>
+ ICodec<T> Codec { get; }
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/0292caf1/lang/cs/Org.Apache.REEF.Network/Group/Operators/IReduceFunction.cs
----------------------------------------------------------------------
diff --git a/lang/cs/Org.Apache.REEF.Network/Group/Operators/IReduceFunction.cs b/lang/cs/Org.Apache.REEF.Network/Group/Operators/IReduceFunction.cs
new file mode 100644
index 0000000..020b09a
--- /dev/null
+++ b/lang/cs/Org.Apache.REEF.Network/Group/Operators/IReduceFunction.cs
@@ -0,0 +1,41 @@
+/**
+ * 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.
+ */
+
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.Threading.Tasks;
+
+namespace Org.Apache.REEF.Network.Group.Operators
+{
+ /// <summary>
+ /// The class used to aggregate messages sent by ReduceSenders.
+ /// </summary>
+ /// <typeparam name="T">The message type.</typeparam>
+ public interface IReduceFunction<T>
+ {
+ /// <summary>
+ /// Reduce the IEnumerable of messages into one message.
+ /// </summary>
+ /// <param name="elements">The messages to reduce</param>
+ /// <returns>The reduced message</returns>
+ T Reduce(IEnumerable<T> elements);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/0292caf1/lang/cs/Org.Apache.REEF.Network/Group/Operators/IReduceReceiver.cs
----------------------------------------------------------------------
diff --git a/lang/cs/Org.Apache.REEF.Network/Group/Operators/IReduceReceiver.cs b/lang/cs/Org.Apache.REEF.Network/Group/Operators/IReduceReceiver.cs
new file mode 100644
index 0000000..2305968
--- /dev/null
+++ b/lang/cs/Org.Apache.REEF.Network/Group/Operators/IReduceReceiver.cs
@@ -0,0 +1,46 @@
+/**
+ * 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.
+ */
+
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.Threading.Tasks;
+
+namespace Org.Apache.REEF.Network.Group.Operators
+{
+ /// <summary>
+ /// MPI operator used to receive and reduce messages.
+ /// </summary>
+ /// <typeparam name="T">The message type</typeparam>
+ public interface IReduceReceiver<T> : IMpiOperator<T>
+ {
+ /// <summary>
+ /// Returns the class used to reduce incoming messages sent by ReduceSenders.
+ /// </summary>
+ IReduceFunction<T> ReduceFunction { get; }
+
+ /// <summary>
+ /// Receives messages sent by all ReduceSenders and aggregates them
+ /// using the specified IReduceFunction.
+ /// </summary>
+ /// <returns>The single aggregated data</returns>
+ T Reduce();
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/0292caf1/lang/cs/Org.Apache.REEF.Network/Group/Operators/IReduceSender.cs
----------------------------------------------------------------------
diff --git a/lang/cs/Org.Apache.REEF.Network/Group/Operators/IReduceSender.cs b/lang/cs/Org.Apache.REEF.Network/Group/Operators/IReduceSender.cs
new file mode 100644
index 0000000..c15ded6
--- /dev/null
+++ b/lang/cs/Org.Apache.REEF.Network/Group/Operators/IReduceSender.cs
@@ -0,0 +1,40 @@
+/**
+ * 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.
+ */
+
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.Threading.Tasks;
+
+namespace Org.Apache.REEF.Network.Group.Operators
+{
+ /// <summary>
+ /// MPI Operator used to send messages to be reduced by the ReduceReceiver.
+ /// </summary>
+ /// <typeparam name="T">The message type</typeparam>
+ public interface IReduceSender<T> : IMpiOperator<T>
+ {
+ /// <summary>
+ /// Sends data to the operator's ReduceReceiver to be aggregated.
+ /// </summary>
+ /// <param name="data">The data to send</param>
+ void Send(T data);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/0292caf1/lang/cs/Org.Apache.REEF.Network/Group/Operators/IScatterReceiver.cs
----------------------------------------------------------------------
diff --git a/lang/cs/Org.Apache.REEF.Network/Group/Operators/IScatterReceiver.cs b/lang/cs/Org.Apache.REEF.Network/Group/Operators/IScatterReceiver.cs
new file mode 100644
index 0000000..7aa4e81
--- /dev/null
+++ b/lang/cs/Org.Apache.REEF.Network/Group/Operators/IScatterReceiver.cs
@@ -0,0 +1,41 @@
+/**
+ * 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.
+ */
+
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.Threading.Tasks;
+
+namespace Org.Apache.REEF.Network.Group.Operators
+{
+ /// <summary>
+ /// MPI operator used to receive a sublist of messages sent
+ /// from the IScatterSender.
+ /// </summary>
+ /// <typeparam name="T">The message type</typeparam>
+ public interface IScatterReceiver<T> : IMpiOperator<T>
+ {
+ /// <summary>
+ /// Receive a sublist of messages sent from the IScatterSender.
+ /// </summary>
+ /// <returns>The sublist of messages</returns>
+ List<T> Receive();
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/0292caf1/lang/cs/Org.Apache.REEF.Network/Group/Operators/IScatterSender.cs
----------------------------------------------------------------------
diff --git a/lang/cs/Org.Apache.REEF.Network/Group/Operators/IScatterSender.cs b/lang/cs/Org.Apache.REEF.Network/Group/Operators/IScatterSender.cs
new file mode 100644
index 0000000..c5a2c3d
--- /dev/null
+++ b/lang/cs/Org.Apache.REEF.Network/Group/Operators/IScatterSender.cs
@@ -0,0 +1,60 @@
+/**
+ * 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.
+ */
+
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.Threading.Tasks;
+
+namespace Org.Apache.REEF.Network.Group.Operators
+{
+ /// <summary>
+ /// MPI operator used to scatter a list of elements to all
+ /// of the IScatterReceivers.
+ /// </summary>
+ /// <typeparam name="T">The message type</typeparam>
+ public interface IScatterSender<T> : IMpiOperator<T>
+ {
+ /// <summary>
+ /// Split up the list of elements evenly and scatter each chunk
+ /// to the IScatterReceivers.
+ /// </summary>
+ /// <param name="elements">The list of elements to send.</param>
+ void Send(List<T> elements);
+
+ /// <summary>
+ /// Split up the list of elements and scatter each chunk
+ /// to the IScatterReceivers. Each receiver will receive
+ /// a sublist of the specified size.
+ /// </summary>
+ /// <param name="elements">The list of elements to send.</param>
+ /// <param name="count">The size of each sublist</param>
+ void Send(List<T> elements, int count);
+
+ /// <summary>
+ /// Split up the list of elements and scatter each chunk
+ /// to the IScatterReceivers in the specified task order.
+ /// </summary>
+ /// <param name="elements">The list of elements to send.</param>
+ /// <param name="order">The list of task identifiers representing
+ /// the order in which to scatter each sublist</param>
+ void Send(List<T> elements, List<string> order);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/0292caf1/lang/cs/Org.Apache.REEF.Network/Group/Operators/Impl/BroadcastOperatorSpec.cs
----------------------------------------------------------------------
diff --git a/lang/cs/Org.Apache.REEF.Network/Group/Operators/Impl/BroadcastOperatorSpec.cs b/lang/cs/Org.Apache.REEF.Network/Group/Operators/Impl/BroadcastOperatorSpec.cs
new file mode 100644
index 0000000..904e4ef
--- /dev/null
+++ b/lang/cs/Org.Apache.REEF.Network/Group/Operators/Impl/BroadcastOperatorSpec.cs
@@ -0,0 +1,50 @@
+/**
+ * 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.
+ */
+
+using Org.Apache.REEF.Wake.Remote;
+
+namespace Org.Apache.REEF.Network.Group.Operators.Impl
+{
+ /// <summary>
+ /// The specification used to define Broadcast Operators.
+ /// </summary>
+ public class BroadcastOperatorSpec<T> : IOperatorSpec<T>
+ {
+ /// <summary>
+ /// Create a new BroadcastOperatorSpec.
+ /// </summary>
+ /// <param name="senderId">The identifier of the root sending Task.</param>
+ /// <param name="codecType">The codec used to serialize messages.</param>
+ public BroadcastOperatorSpec(string senderId, ICodec<T> codecType)
+ {
+ SenderId = senderId;
+ Codec = codecType;
+ }
+
+ /// <summary>
+ /// Returns the identifier of the Task that will broadcast data to other Tasks.
+ /// </summary>
+ public string SenderId { get; private set; }
+
+ /// <summary>
+ /// Returns the ICodec used to serialize messages.
+ /// </summary>
+ public ICodec<T> Codec { get; private set; }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/0292caf1/lang/cs/Org.Apache.REEF.Network/Group/Operators/Impl/BroadcastReceiver.cs
----------------------------------------------------------------------
diff --git a/lang/cs/Org.Apache.REEF.Network/Group/Operators/Impl/BroadcastReceiver.cs b/lang/cs/Org.Apache.REEF.Network/Group/Operators/Impl/BroadcastReceiver.cs
new file mode 100644
index 0000000..4374ab5
--- /dev/null
+++ b/lang/cs/Org.Apache.REEF.Network/Group/Operators/Impl/BroadcastReceiver.cs
@@ -0,0 +1,92 @@
+/**
+ * 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.
+ */
+
+using Org.Apache.REEF.Network.Group.Config;
+using Org.Apache.REEF.Network.Group.Driver;
+using Org.Apache.REEF.Network.Group.Driver.Impl;
+using Org.Apache.REEF.Network.Group.Task;
+using Org.Apache.REEF.Network.Group.Task.Impl;
+using Org.Apache.REEF.Tang.Annotations;
+using System.Reactive;
+
+namespace Org.Apache.REEF.Network.Group.Operators.Impl
+{
+ /// <summary>
+ /// MPI Operator used to receive broadcast messages.
+ /// </summary>
+ /// <typeparam name="T">The type of message being sent.</typeparam>
+ public class BroadcastReceiver<T> : IBroadcastReceiver<T>
+ {
+ private const int DefaultVersion = 1;
+
+ private ICommunicationGroupNetworkObserver _networkHandler;
+ private OperatorTopology<T> _topology;
+
+ /// <summary>
+ /// Creates a new BroadcastReceiver.
+ /// </summary>
+ /// <param name="operatorName">The operator identifier</param>
+ /// <param name="groupName">The name of the CommunicationGroup that the
+ /// operator belongs to</param>
+ /// <param name="topology">The node's topology graph</param>
+ /// <param name="networkHandler">The incoming message handler</param>
+ [Inject]
+ public BroadcastReceiver(
+ [Parameter(typeof(MpiConfigurationOptions.OperatorName))] string operatorName,
+ [Parameter(typeof(MpiConfigurationOptions.CommunicationGroupName))] string groupName,
+ OperatorTopology<T> topology,
+ ICommunicationGroupNetworkObserver networkHandler)
+ {
+ OperatorName = operatorName;
+ GroupName = groupName;
+ Version = DefaultVersion;
+
+ _networkHandler = networkHandler;
+ _topology = topology;
+ _topology.Initialize();
+
+ var msgHandler = Observer.Create<GroupCommunicationMessage>(message => _topology.OnNext(message));
+ _networkHandler.Register(operatorName, msgHandler);
+ }
+
+ /// <summary>
+ /// Returns the operator identifier.
+ /// </summary>
+ public string OperatorName { get; private set; }
+
+ /// <summary>
+ /// Returns the name of the CommunicationGroup that the operator belongs to.
+ /// </summary>
+ public string GroupName { get; private set; }
+
+ /// <summary>
+ /// Returns the operator version.
+ /// </summary>
+ public int Version { get; private set; }
+
+ /// <summary>
+ /// Receive a message from parent BroadcastSender.
+ /// </summary>
+ /// <returns>The incoming message</returns>
+ public T Receive()
+ {
+ return _topology.ReceiveFromParent();
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/0292caf1/lang/cs/Org.Apache.REEF.Network/Group/Operators/Impl/BroadcastSender.cs
----------------------------------------------------------------------
diff --git a/lang/cs/Org.Apache.REEF.Network/Group/Operators/Impl/BroadcastSender.cs b/lang/cs/Org.Apache.REEF.Network/Group/Operators/Impl/BroadcastSender.cs
new file mode 100644
index 0000000..4e48428
--- /dev/null
+++ b/lang/cs/Org.Apache.REEF.Network/Group/Operators/Impl/BroadcastSender.cs
@@ -0,0 +1,98 @@
+/**
+ * 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.
+ */
+
+using Org.Apache.REEF.Network.Group.Config;
+using Org.Apache.REEF.Network.Group.Driver;
+using Org.Apache.REEF.Network.Group.Driver.Impl;
+using Org.Apache.REEF.Network.Group.Task;
+using Org.Apache.REEF.Network.Group.Task.Impl;
+using Org.Apache.REEF.Tang.Annotations;
+using System;
+using System.Reactive;
+
+namespace Org.Apache.REEF.Network.Group.Operators.Impl
+{
+ /// <summary>
+ /// MPI Operator used to send messages to child Tasks.
+ /// </summary>
+ /// <typeparam name="T">The message type</typeparam>
+ public class BroadcastSender<T> : IBroadcastSender<T>
+ {
+ private const int DefaultVersion = 1;
+
+ private ICommunicationGroupNetworkObserver _networkHandler;
+ private OperatorTopology<T> _topology;
+
+ /// <summary>
+ /// Creates a new BroadcastSender to send messages to other Tasks.
+ /// </summary>
+ /// <param name="operatorName">The identifier for the operator</param>
+ /// <param name="groupName">The name of the CommunicationGroup that the operator
+ /// belongs to</param>
+ /// <param name="topology">The node's topology graph</param>
+ /// <param name="networkHandler">The incoming message handler</param>
+ [Inject]
+ public BroadcastSender(
+ [Parameter(typeof(MpiConfigurationOptions.OperatorName))] string operatorName,
+ [Parameter(typeof(MpiConfigurationOptions.CommunicationGroupName))] string groupName,
+ OperatorTopology<T> topology,
+ ICommunicationGroupNetworkObserver networkHandler)
+ {
+ OperatorName = operatorName;
+ GroupName = groupName;
+ Version = DefaultVersion;
+
+ _networkHandler = networkHandler;
+ _topology = topology;
+ _topology.Initialize();
+
+ var msgHandler = Observer.Create<GroupCommunicationMessage>(message => _topology.OnNext(message));
+ _networkHandler.Register(operatorName, msgHandler);
+ }
+
+ /// <summary>
+ /// Returns the identifier for the MPI operator.
+ /// </summary>
+ public string OperatorName { get; private set; }
+
+ /// <summary>
+ /// Returns the name of the operator's CommunicationGroup.
+ /// </summary>
+ public string GroupName { get; private set; }
+
+ /// <summary>
+ /// Returns the operator version.
+ /// </summary>
+ public int Version { get; private set; }
+
+ /// <summary>
+ /// Send the data to all BroadcastReceivers.
+ /// </summary>
+ /// <param name="data">The data to send.</param>
+ public void Send(T data)
+ {
+ if (data == null)
+ {
+ throw new ArgumentNullException("data");
+ }
+
+ _topology.SendToChildren(data, MessageType.Data);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/0292caf1/lang/cs/Org.Apache.REEF.Network/Group/Operators/Impl/ReduceFunction.cs
----------------------------------------------------------------------
diff --git a/lang/cs/Org.Apache.REEF.Network/Group/Operators/Impl/ReduceFunction.cs b/lang/cs/Org.Apache.REEF.Network/Group/Operators/Impl/ReduceFunction.cs
new file mode 100644
index 0000000..bc60055
--- /dev/null
+++ b/lang/cs/Org.Apache.REEF.Network/Group/Operators/Impl/ReduceFunction.cs
@@ -0,0 +1,62 @@
+/**
+ * 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.
+ */
+
+using System;
+using System.Collections.Generic;
+using System.Linq;
+
+namespace Org.Apache.REEF.Network.Group.Operators.Impl
+{
+ public class ReduceFunction<T> : IReduceFunction<T>
+ {
+ private Func<T, T, T> _reduceFunction;
+ private T _initialValue;
+
+ private ReduceFunction(Func<T, T, T> reduceFunction)
+ {
+ _reduceFunction = reduceFunction;
+ }
+
+ private ReduceFunction(Func<T, T, T> reduceFunction, T initialValue)
+ {
+ _reduceFunction = reduceFunction;
+ _initialValue = initialValue;
+ }
+
+ public static IReduceFunction<T> Create(Func<T, T, T> reduceFunction)
+ {
+ return new ReduceFunction<T>(reduceFunction);
+ }
+
+ public static IReduceFunction<T> Create(Func<T, T, T> reduceFunction, T initialValue)
+ {
+ return new ReduceFunction<T>(reduceFunction, initialValue);
+ }
+
+ public T Reduce(IEnumerable<T> elements)
+ {
+ if (_initialValue == null)
+ {
+ return elements.Aggregate(_reduceFunction);
+ }
+
+ return elements.Aggregate(_initialValue, _reduceFunction);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/0292caf1/lang/cs/Org.Apache.REEF.Network/Group/Operators/Impl/ReduceOperatorSpec.cs
----------------------------------------------------------------------
diff --git a/lang/cs/Org.Apache.REEF.Network/Group/Operators/Impl/ReduceOperatorSpec.cs b/lang/cs/Org.Apache.REEF.Network/Group/Operators/Impl/ReduceOperatorSpec.cs
new file mode 100644
index 0000000..37b6ce7
--- /dev/null
+++ b/lang/cs/Org.Apache.REEF.Network/Group/Operators/Impl/ReduceOperatorSpec.cs
@@ -0,0 +1,62 @@
+/**
+ * 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.
+ */
+
+using Org.Apache.REEF.Wake.Remote;
+
+namespace Org.Apache.REEF.Network.Group.Operators.Impl
+{
+ /// <summary>
+ /// The specification used to define Reduce MPI Operators.
+ /// </summary>
+ public class ReduceOperatorSpec<T> : IOperatorSpec<T>
+ {
+ /// <summary>
+ /// Creates a new ReduceOperatorSpec.
+ /// </summary>
+ /// <param name="receiverId">The identifier of the task that
+ /// will receive and reduce incoming messages.</param>
+ /// <param name="codec">The codec used for serializing messages.</param>
+ /// <param name="reduceFunction">The class used to aggregate all messages.</param>
+ public ReduceOperatorSpec(
+ string receiverId,
+ ICodec<T> codec,
+ IReduceFunction<T> reduceFunction)
+ {
+ ReceiverId = receiverId;
+ Codec = codec;
+ ReduceFunction = reduceFunction;
+ }
+
+ /// <summary>
+ /// Returns the identifier for the task that receives and reduces
+ /// incoming messages.
+ /// </summary>
+ public string ReceiverId { get; private set; }
+
+ /// <summary>
+ /// The codec used to serialize and deserialize messages.
+ /// </summary>
+ public ICodec<T> Codec { get; private set; }
+
+ /// <summary>
+ /// The class used to aggregate incoming messages.
+ /// </summary>
+ public IReduceFunction<T> ReduceFunction { get; private set; }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/0292caf1/lang/cs/Org.Apache.REEF.Network/Group/Operators/Impl/ReduceReceiver.cs
----------------------------------------------------------------------
diff --git a/lang/cs/Org.Apache.REEF.Network/Group/Operators/Impl/ReduceReceiver.cs b/lang/cs/Org.Apache.REEF.Network/Group/Operators/Impl/ReduceReceiver.cs
new file mode 100644
index 0000000..3c722ef
--- /dev/null
+++ b/lang/cs/Org.Apache.REEF.Network/Group/Operators/Impl/ReduceReceiver.cs
@@ -0,0 +1,100 @@
+/**
+ * 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.
+ */
+
+using Org.Apache.REEF.Network.Group.Config;
+using Org.Apache.REEF.Network.Group.Driver;
+using Org.Apache.REEF.Network.Group.Driver.Impl;
+using Org.Apache.REEF.Network.Group.Task;
+using Org.Apache.REEF.Network.Group.Task.Impl;
+using Org.Apache.REEF.Tang.Annotations;
+using System.Reactive;
+
+namespace Org.Apache.REEF.Network.Group.Operators.Impl
+{
+ /// <summary>
+ /// MPI operator used to receive and reduce messages.
+ /// </summary>
+ /// <typeparam name="T">The message type</typeparam>
+ public class ReduceReceiver<T> : IReduceReceiver<T>
+ {
+ private const int DefaultVersion = 1;
+
+ private ICommunicationGroupNetworkObserver _networkHandler;
+ private OperatorTopology<T> _topology;
+
+ /// <summary>
+ /// Creates a new ReduceReceiver.
+ /// </summary>
+ /// <param name="operatorName">The name of the reduce operator</param>
+ /// <param name="groupName">The name of the operator's CommunicationGroup</param>
+ /// <param name="topology">The task's operator topology graph</param>
+ /// <param name="networkHandler">Handles incoming messages from other tasks</param>
+ /// <param name="reduceFunction">The class used to aggregate all incoming messages</param>
+ [Inject]
+ public ReduceReceiver(
+ [Parameter(typeof(MpiConfigurationOptions.OperatorName))] string operatorName,
+ [Parameter(typeof(MpiConfigurationOptions.CommunicationGroupName))] string groupName,
+ OperatorTopology<T> topology,
+ ICommunicationGroupNetworkObserver networkHandler,
+ IReduceFunction<T> reduceFunction)
+ {
+ OperatorName = operatorName;
+ GroupName = groupName;
+ Version = DefaultVersion;
+ ReduceFunction = reduceFunction;
+
+ _networkHandler = networkHandler;
+ _topology = topology;
+ _topology.Initialize();
+
+ var msgHandler = Observer.Create<GroupCommunicationMessage>(message => _topology.OnNext(message));
+ _networkHandler.Register(operatorName, msgHandler);
+ }
+
+ /// <summary>
+ /// Returns the name of the reduce operator
+ /// </summary>
+ public string OperatorName { get; private set; }
+
+ /// <summary>
+ /// Returns the name of the operator's CommunicationGroup.
+ /// </summary>
+ public string GroupName { get; private set; }
+
+ /// <summary>
+ /// Returns the operator version.
+ /// </summary>
+ public int Version { get; private set; }
+
+ /// <summary>
+ /// Returns the class used to reduce incoming messages sent by ReduceSenders.
+ /// </summary>
+ public IReduceFunction<T> ReduceFunction { get; private set; }
+
+ /// <summary>
+ /// Receives messages sent by all ReduceSenders and aggregates them
+ /// using the specified IReduceFunction.
+ /// </summary>
+ /// <returns>The single aggregated data</returns>
+ public T Reduce()
+ {
+ return _topology.ReceiveFromChildren(ReduceFunction);
+ }
+ }
+}
[4/4] incubator-reef git commit: [REEF-150] Adding group
communication to REEF .Net
Posted by tm...@apache.org.
[REEF-150] Adding group communication to REEF .Net
This is to port Group Communication to REEF .Net in Apache Git
Add source code, examples and tests
Updated namespace to follow conventions
IIRA: Reef-150. (https://issues.apache.org/jira/browse/REEF-150)
Author: Julia Wang Email: jwang98052@yahoo.com
Project: http://git-wip-us.apache.org/repos/asf/incubator-reef/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-reef/commit/0292caf1
Tree: http://git-wip-us.apache.org/repos/asf/incubator-reef/tree/0292caf1
Diff: http://git-wip-us.apache.org/repos/asf/incubator-reef/diff/0292caf1
Branch: refs/heads/master
Commit: 0292caf1437d61586d4e7ba1370710be833f5292
Parents: 7edb857
Author: Julia Wang <jw...@yahoo.com>
Authored: Tue Feb 10 19:16:14 2015 -0800
Committer: tmajest <tm...@microsoft.com>
Committed: Wed Feb 11 14:55:10 2015 -0800
----------------------------------------------------------------------
.../MachineLearning/KMeans/Centroids.cs | 44 ++
.../MachineLearning/KMeans/Constants.cs | 35 ++
.../KMeans/Contracts/CentroidsContract.cs | 49 ++
.../KMeans/Contracts/DataVectorContract.cs | 52 ++
.../KMeans/Contracts/PartialMeanContract.cs | 48 ++
.../Contracts/ProcessedResultsContract.cs | 57 +++
.../KMeans/DataPartitionCache.cs | 104 ++++
.../MachineLearning/KMeans/DataVector.cs | 260 ++++++++++
.../KMeans/KMeansConfiguratioinOptions.cs | 46 ++
.../KMeans/KMeansDriverHandlers.cs | 191 ++++++++
.../MachineLearning/KMeans/KMeansMasterTask.cs | 155 ++++++
.../MachineLearning/KMeans/KMeansSlaveTask.cs | 118 +++++
.../MachineLearning/KMeans/LegacyKMeansTask.cs | 113 +++++
.../MachineLearning/KMeans/PartialMean.cs | 124 +++++
.../MachineLearning/KMeans/ProcessedResults.cs | 54 +++
.../KMeans/codecs/CentroidsCodec.cs | 49 ++
.../KMeans/codecs/DataVectorCodec.cs | 46 ++
.../KMeans/codecs/ProcessedResultsCodec.cs | 57 +++
.../Org.Apache.REEF.Examples.csproj | 19 +
.../Group/Codec/GcmMessageProto.cs | 76 +++
.../Codec/GroupCommunicationMessageCodec.cs | 77 +++
.../Group/Config/MpiConfigurationOptions.cs | 71 +++
.../Group/Driver/ICommunicationGroupDriver.cs | 89 ++++
.../Group/Driver/IMpiDriver.cs | 93 ++++
.../Driver/Impl/CommunicationGroupDriver.cs | 260 ++++++++++
.../Driver/Impl/GroupCommunicationMessage.cs | 107 ++++
.../Group/Driver/Impl/MessageType.cs | 30 ++
.../Group/Driver/Impl/MpiDriver.cs | 239 +++++++++
.../Group/Driver/Impl/TaskStarter.cs | 135 ++++++
.../Group/Operators/IBroadcastReceiver.cs | 40 ++
.../Group/Operators/IBroadcastSender.cs | 40 ++
.../Group/Operators/IMpiOperator.cs | 49 ++
.../Group/Operators/IOperatorSpec.cs | 36 ++
.../Group/Operators/IReduceFunction.cs | 41 ++
.../Group/Operators/IReduceReceiver.cs | 46 ++
.../Group/Operators/IReduceSender.cs | 40 ++
.../Group/Operators/IScatterReceiver.cs | 41 ++
.../Group/Operators/IScatterSender.cs | 60 +++
.../Operators/Impl/BroadcastOperatorSpec.cs | 50 ++
.../Group/Operators/Impl/BroadcastReceiver.cs | 92 ++++
.../Group/Operators/Impl/BroadcastSender.cs | 98 ++++
.../Group/Operators/Impl/ReduceFunction.cs | 62 +++
.../Group/Operators/Impl/ReduceOperatorSpec.cs | 62 +++
.../Group/Operators/Impl/ReduceReceiver.cs | 100 ++++
.../Group/Operators/Impl/ReduceSender.cs | 97 ++++
.../Group/Operators/Impl/ScatterOperatorSpec.cs | 58 +++
.../Group/Operators/Impl/ScatterReceiver.cs | 101 ++++
.../Group/Operators/Impl/ScatterSender.cs | 112 +++++
.../Group/Operators/Impl/Sender.cs | 74 +++
.../Group/Task/ICommunicationGroupClient.cs | 90 ++++
.../Task/ICommunicationGroupNetworkObserver.cs | 49 ++
.../Group/Task/IMpiClient.cs | 44 ++
.../Group/Task/IMpiNetworkObserver.cs | 50 ++
.../Group/Task/Impl/CommunicationGroupClient.cs | 219 +++++++++
.../Impl/CommunicationGroupNetworkObserver.cs | 93 ++++
.../Group/Task/Impl/MpiClient.cs | 108 +++++
.../Group/Task/Impl/MpiNetworkObserver.cs | 109 +++++
.../Group/Task/Impl/NodeStruct.cs | 67 +++
.../Group/Task/Impl/OperatorTopology.cs | 484 +++++++++++++++++++
.../Group/Topology/FlatTopology.cs | 201 ++++++++
.../Group/Topology/ITopology.cs | 41 ++
.../Group/Topology/TaskNode.cs | 69 +++
.../Org.Apache.REEF.Network.csproj | 43 ++
.../Functional/ML/KMeans/TestKMeans.cs | 163 +++++++
.../BroadcastReduceDriver.cs | 187 +++++++
.../BroadcastReduceTest/BroadcastReduceTest.cs | 83 ++++
.../MPI/BroadcastReduceTest/MasterTask.cs | 89 ++++
.../MPI/BroadcastReduceTest/SlaveTask.cs | 80 +++
.../Functional/MPI/MpiTestConfig.cs | 36 ++
.../Functional/MPI/MpiTestConstants.cs | 33 ++
.../MPI/ScatterReduceTest/MasterTask.cs | 72 +++
.../ScatterReduceTest/ScatterReduceDriver.cs | 168 +++++++
.../MPI/ScatterReduceTest/ScatterReduceTest.cs | 80 +++
.../MPI/ScatterReduceTest/SlaveTask.cs | 67 +++
.../Org.Apache.REEF.Tests.csproj | 11 +
75 files changed, 6833 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/0292caf1/lang/cs/Org.Apache.REEF.Examples/MachineLearning/KMeans/Centroids.cs
----------------------------------------------------------------------
diff --git a/lang/cs/Org.Apache.REEF.Examples/MachineLearning/KMeans/Centroids.cs b/lang/cs/Org.Apache.REEF.Examples/MachineLearning/KMeans/Centroids.cs
new file mode 100644
index 0000000..0ceb4ed
--- /dev/null
+++ b/lang/cs/Org.Apache.REEF.Examples/MachineLearning/KMeans/Centroids.cs
@@ -0,0 +1,44 @@
+/**
+ * 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.
+ */
+
+using System.Collections.Generic;
+using Org.Apache.REEF.Examples.MachineLearning.KMeans.codecs;
+using Org.Apache.REEF.Utilities;
+
+namespace Org.Apache.REEF.Examples.MachineLearning.KMeans
+{
+ public class Centroids
+ {
+ public Centroids(List<DataVector> points)
+ {
+ Points = points;
+ }
+
+ public List<DataVector> Points { get; set; }
+
+ /// <summary>
+ /// helper function mostly used for logging
+ /// </summary>
+ /// <returns>the serialized string</returns>
+ public override string ToString()
+ {
+ return ByteUtilities.ByteArrarysToString(new CentroidsCodec().Encode(this));
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/0292caf1/lang/cs/Org.Apache.REEF.Examples/MachineLearning/KMeans/Constants.cs
----------------------------------------------------------------------
diff --git a/lang/cs/Org.Apache.REEF.Examples/MachineLearning/KMeans/Constants.cs b/lang/cs/Org.Apache.REEF.Examples/MachineLearning/KMeans/Constants.cs
new file mode 100644
index 0000000..526031e
--- /dev/null
+++ b/lang/cs/Org.Apache.REEF.Examples/MachineLearning/KMeans/Constants.cs
@@ -0,0 +1,35 @@
+/**
+ * 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 Org.Apache.REEF.Examples.MachineLearning.KMeans
+{
+ public class Constants
+ {
+ public const string KMeansExecutionBaseDirectory = @"KMeans";
+ public const string DataDirectory = "data";
+ public const string PartialMeanFilePrefix = "partialMeans_";
+ public const string CentroidsFile = "centroids";
+ public const string MasterTaskId = "KMeansMasterTaskId";
+ public const string SlaveTaskIdPrefix = "KMeansSlaveTask_";
+ public const string KMeansCommunicationGroupName = "KMeansBroadcastReduceGroup";
+ public const string CentroidsBroadcastOperatorName = "CentroidsBroadcast";
+ public const string ControlMessageBroadcastOperatorName = "ControlMessageBroadcast";
+ public const string MeansReduceOperatorName = "MeansReduce";
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/0292caf1/lang/cs/Org.Apache.REEF.Examples/MachineLearning/KMeans/Contracts/CentroidsContract.cs
----------------------------------------------------------------------
diff --git a/lang/cs/Org.Apache.REEF.Examples/MachineLearning/KMeans/Contracts/CentroidsContract.cs b/lang/cs/Org.Apache.REEF.Examples/MachineLearning/KMeans/Contracts/CentroidsContract.cs
new file mode 100644
index 0000000..8ea56c9
--- /dev/null
+++ b/lang/cs/Org.Apache.REEF.Examples/MachineLearning/KMeans/Contracts/CentroidsContract.cs
@@ -0,0 +1,49 @@
+/**
+ * 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.
+ */
+
+using System.Collections.Generic;
+using System.Linq;
+using System.Runtime.Serialization;
+
+namespace Org.Apache.REEF.Examples.MachineLearning.KMeans.Contracts
+{
+ [DataContract]
+ public class CentroidsContract
+ {
+ [DataMember]
+ public List<DataVectorContract> DataVectorContracts { get; set; }
+
+ public static CentroidsContract Create(Centroids centroids)
+ {
+ List<DataVectorContract> dataVectorContracts =
+ centroids.Points.Select(DataVectorContract.Create).ToList();
+
+ return new CentroidsContract { DataVectorContracts = dataVectorContracts };
+ }
+
+ public Centroids ToCentroids()
+ {
+ List<DataVector> dataVectors = DataVectorContracts
+ .Select(dv => dv.ToDataVector())
+ .ToList();
+
+ return new Centroids(dataVectors);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/0292caf1/lang/cs/Org.Apache.REEF.Examples/MachineLearning/KMeans/Contracts/DataVectorContract.cs
----------------------------------------------------------------------
diff --git a/lang/cs/Org.Apache.REEF.Examples/MachineLearning/KMeans/Contracts/DataVectorContract.cs b/lang/cs/Org.Apache.REEF.Examples/MachineLearning/KMeans/Contracts/DataVectorContract.cs
new file mode 100644
index 0000000..1e41944
--- /dev/null
+++ b/lang/cs/Org.Apache.REEF.Examples/MachineLearning/KMeans/Contracts/DataVectorContract.cs
@@ -0,0 +1,52 @@
+/**
+ * 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.
+ */
+
+using System.Collections.Generic;
+using System.Runtime.Serialization;
+
+namespace Org.Apache.REEF.Examples.MachineLearning.KMeans.Contracts
+{
+ [DataContract]
+ public class DataVectorContract
+ {
+ [DataMember]
+ public List<float> Data { get; private set; }
+
+ [DataMember]
+ public int Label { get; private set; }
+
+ [DataMember]
+ public int Dimension { get; private set; }
+
+ public static DataVectorContract Create(DataVector dataVector)
+ {
+ return new DataVectorContract()
+ {
+ Data = dataVector.Data,
+ Label = dataVector.Label,
+ Dimension = dataVector.Dimension
+ };
+ }
+
+ public DataVector ToDataVector()
+ {
+ return new DataVector(Data, Label);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/0292caf1/lang/cs/Org.Apache.REEF.Examples/MachineLearning/KMeans/Contracts/PartialMeanContract.cs
----------------------------------------------------------------------
diff --git a/lang/cs/Org.Apache.REEF.Examples/MachineLearning/KMeans/Contracts/PartialMeanContract.cs b/lang/cs/Org.Apache.REEF.Examples/MachineLearning/KMeans/Contracts/PartialMeanContract.cs
new file mode 100644
index 0000000..b62f8ff
--- /dev/null
+++ b/lang/cs/Org.Apache.REEF.Examples/MachineLearning/KMeans/Contracts/PartialMeanContract.cs
@@ -0,0 +1,48 @@
+/**
+ * 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.
+ */
+
+using System.Runtime.Serialization;
+
+namespace Org.Apache.REEF.Examples.MachineLearning.KMeans.Contracts
+{
+ [DataContract]
+ public class PartialMeanContract
+ {
+ [DataMember]
+ public DataVectorContract DataVectContract { get; set; }
+
+ [DataMember]
+ public int Size { get; set; }
+
+ public static PartialMeanContract Create(PartialMean partialMean)
+ {
+ return new PartialMeanContract
+ {
+ DataVectContract = DataVectorContract.Create(partialMean.Mean),
+ Size = partialMean.Size
+ };
+ }
+
+ public PartialMean ToPartialMean()
+ {
+ DataVector dataVector = DataVectContract.ToDataVector();
+ return new PartialMean(dataVector, Size);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/0292caf1/lang/cs/Org.Apache.REEF.Examples/MachineLearning/KMeans/Contracts/ProcessedResultsContract.cs
----------------------------------------------------------------------
diff --git a/lang/cs/Org.Apache.REEF.Examples/MachineLearning/KMeans/Contracts/ProcessedResultsContract.cs b/lang/cs/Org.Apache.REEF.Examples/MachineLearning/KMeans/Contracts/ProcessedResultsContract.cs
new file mode 100644
index 0000000..c78ee3c
--- /dev/null
+++ b/lang/cs/Org.Apache.REEF.Examples/MachineLearning/KMeans/Contracts/ProcessedResultsContract.cs
@@ -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.
+ */
+
+using System.Collections.Generic;
+using System.Linq;
+using System.Runtime.Serialization;
+
+namespace Org.Apache.REEF.Examples.MachineLearning.KMeans.Contracts
+{
+ [DataContract]
+ public class ProcessedResultsContract
+ {
+ [DataMember]
+ public List<PartialMeanContract> PartialMeanContracts { get; set; }
+
+ [DataMember]
+ public float Loss { get; set; }
+
+ public static ProcessedResultsContract Create(ProcessedResults obj)
+ {
+ List<PartialMeanContract> partialMeansContracts = obj.Means
+ .Select(PartialMeanContract.Create)
+ .ToList();
+
+ return new ProcessedResultsContract
+ {
+ PartialMeanContracts = partialMeansContracts,
+ Loss = obj.Loss
+ };
+ }
+
+ public ProcessedResults ToProcessedResults()
+ {
+ List<PartialMean> partialMeans = PartialMeanContracts
+ .Select(contract => contract.ToPartialMean())
+ .ToList();
+
+ return new ProcessedResults(partialMeans, Loss);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/0292caf1/lang/cs/Org.Apache.REEF.Examples/MachineLearning/KMeans/DataPartitionCache.cs
----------------------------------------------------------------------
diff --git a/lang/cs/Org.Apache.REEF.Examples/MachineLearning/KMeans/DataPartitionCache.cs b/lang/cs/Org.Apache.REEF.Examples/MachineLearning/KMeans/DataPartitionCache.cs
new file mode 100644
index 0000000..91c3ed0
--- /dev/null
+++ b/lang/cs/Org.Apache.REEF.Examples/MachineLearning/KMeans/DataPartitionCache.cs
@@ -0,0 +1,104 @@
+/**
+ * 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.
+ */
+
+using System.Collections.Generic;
+using System.Globalization;
+using System.IO;
+using Org.Apache.REEF.Common.Services;
+using Org.Apache.REEF.Tang.Annotations;
+using Org.Apache.REEF.Utilities.Logging;
+
+namespace Org.Apache.REEF.Examples.MachineLearning.KMeans
+{
+ // TODO: we should outsource some of the functionalites to a data loader implemenation
+ public class DataPartitionCache : IService
+ {
+ private static readonly Logger _Logger = Logger.GetLogger(typeof(DataPartitionCache));
+
+ [Inject]
+ public DataPartitionCache(
+ [Parameter(Value = typeof(PartitionIndex))] int partition,
+ [Parameter(Value = typeof(KMeansConfiguratioinOptions.ExecutionDirectory))] string executionDirectory)
+ {
+ Partition = partition;
+ if (Partition < 0)
+ {
+ _Logger.Log(Level.Info, "no data to load since partition = " + Partition);
+ }
+ else
+ {
+ _Logger.Log(Level.Info, "loading data for partition " + Partition);
+ DataVectors = loadData(partition, executionDirectory);
+ }
+ }
+
+ public List<DataVector> DataVectors { get; set; }
+
+ public int Partition { get; set; }
+
+ // read initial data from file and marked it as unlabeled (not associated with any centroid)
+ public static List<DataVector> ReadDataFile(string path, char seperator = ',')
+ {
+ List<DataVector> data = new List<DataVector>();
+ FileStream file = new FileStream(path, FileMode.Open, FileAccess.Read, FileShare.Read);
+ using (StreamReader reader = new StreamReader(file))
+ {
+ while (!reader.EndOfStream)
+ {
+ string line = reader.ReadLine();
+ if (!string.IsNullOrWhiteSpace(line))
+ {
+ data.Add(DataVector.FromString(line));
+ }
+ }
+ reader.Close();
+ }
+
+ return data;
+ }
+
+ public void LabelData(Centroids centroids)
+ {
+ foreach (DataVector vector in DataVectors)
+ {
+ float minimumDistance = float.MaxValue;
+ foreach (DataVector centroid in centroids.Points)
+ {
+ float d = vector.DistanceTo(centroid);
+ if (d < minimumDistance)
+ {
+ vector.Label = centroid.Label;
+ minimumDistance = d;
+ }
+ }
+ }
+ }
+
+ private List<DataVector> loadData(int partition, string executionDirectory)
+ {
+ string file = Path.Combine(executionDirectory, Constants.DataDirectory, partition.ToString(CultureInfo.InvariantCulture));
+ return ReadDataFile(file);
+ }
+
+ [NamedParameter("Data partition index", "partition", "")]
+ public class PartitionIndex : Name<int>
+ {
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/0292caf1/lang/cs/Org.Apache.REEF.Examples/MachineLearning/KMeans/DataVector.cs
----------------------------------------------------------------------
diff --git a/lang/cs/Org.Apache.REEF.Examples/MachineLearning/KMeans/DataVector.cs b/lang/cs/Org.Apache.REEF.Examples/MachineLearning/KMeans/DataVector.cs
new file mode 100644
index 0000000..fb6a16e
--- /dev/null
+++ b/lang/cs/Org.Apache.REEF.Examples/MachineLearning/KMeans/DataVector.cs
@@ -0,0 +1,260 @@
+/**
+ * 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.
+ */
+
+using System;
+using System.Collections.Generic;
+using System.Globalization;
+using System.IO;
+using System.Linq;
+
+namespace Org.Apache.REEF.Examples.MachineLearning.KMeans
+{
+ public class DataVector
+ {
+ public DataVector(int dimension, int label = -1)
+ {
+ Dimension = dimension;
+ Data = Enumerable.Repeat((float)0, Dimension).ToList();
+ Label = label;
+ }
+
+ // unlablered data
+ public DataVector(List<float> data) : this(data, -1)
+ {
+ }
+
+ public DataVector(List<float> data, int label)
+ {
+ if (data == null || data.Count == 0)
+ {
+ throw new ArgumentNullException("data");
+ }
+ Dimension = data.Count;
+ Data = data;
+ Label = label;
+ }
+
+ public List<float> Data { get; set; }
+
+ public int Label { get; set; }
+
+ public int Dimension { get; set; }
+
+ public static float TotalDistance(List<DataVector> list1, List<DataVector> list2)
+ {
+ if (list1 == null || list2 == null || list1.Count == 0 || list2.Count == 0)
+ {
+ throw new ArgumentException("one of the input list is null or empty");
+ }
+ if (list1.Count != list2.Count)
+ {
+ throw new ArgumentException("list 1's dimensionality does not mach list 2.");
+ }
+ float distance = 0;
+ for (int i = 0; i < list1.Count; i++)
+ {
+ distance += list1[i].DistanceTo(list2[i]);
+ }
+ return distance;
+ }
+
+ public static DataVector Mean(List<DataVector> vectors)
+ {
+ if (vectors == null || vectors.Count == 0)
+ {
+ throw new ArgumentNullException("vectors");
+ }
+ DataVector mean = new DataVector(vectors[0].Dimension);
+ for (int i = 0; i < vectors.Count; i++)
+ {
+ mean = mean.Add(vectors[i], ignoreLabel: true);
+ }
+ return mean.Normalize(vectors.Count);
+ }
+
+ // shuffle data and write them to different partions (different files on disk for now)
+ public static List<DataVector> ShuffleDataAndGetInitialCentriods(string originalDataFile, int partitionsNum, int clustersNum, string executionDirectory)
+ {
+ List<DataVector> data = DataPartitionCache.ReadDataFile(originalDataFile);
+ // shuffle, not truely random, but sufficient for our purpose
+ data = data.OrderBy(a => Guid.NewGuid()).ToList();
+ string dataDirectory = Path.Combine(executionDirectory, Constants.DataDirectory);
+ // clean things up first
+ if (Directory.Exists(dataDirectory))
+ {
+ Directory.Delete(dataDirectory, true);
+ }
+ Directory.CreateDirectory(dataDirectory);
+
+ int residualCount = data.Count;
+ int batchSize = data.Count / partitionsNum;
+ for (int i = 0; i < partitionsNum; i++)
+ {
+ int linesCount = residualCount > batchSize ? batchSize : residualCount;
+ using (StreamWriter writer = new StreamWriter(
+ File.OpenWrite(Path.Combine(executionDirectory, Constants.DataDirectory, i.ToString(CultureInfo.InvariantCulture)))))
+ {
+ for (int j = i * batchSize; j < (i * batchSize) + linesCount; j++)
+ {
+ writer.WriteLine(data[j].ToString());
+ }
+ writer.Close();
+ }
+ }
+ return InitializeCentroids(clustersNum, data, executionDirectory);
+ }
+
+ public static void WriteToCentroidFile(List<DataVector> centroids, string executionDirectory)
+ {
+ string centroidFile = Path.Combine(executionDirectory, Constants.CentroidsFile);
+ File.Delete(centroidFile);
+ using (StreamWriter writer = new StreamWriter(File.OpenWrite(centroidFile)))
+ {
+ foreach (DataVector dataVector in centroids)
+ {
+ writer.WriteLine(dataVector.ToString());
+ }
+ writer.Close();
+ }
+ }
+
+ // TODO: replace with proper deserialization
+ public static DataVector FromString(string str)
+ {
+ if (string.IsNullOrWhiteSpace(str))
+ {
+ throw new ArgumentException("str");
+ }
+ string[] dataAndLable = str.Split(';');
+ if (dataAndLable == null || dataAndLable.Length > 2)
+ {
+ throw new ArgumentException("Cannot deserialize DataVector from string " + str);
+ }
+ int label = -1;
+ if (dataAndLable.Length == 2)
+ {
+ label = int.Parse(dataAndLable[1], CultureInfo.InvariantCulture);
+ }
+ List<float> data = dataAndLable[0].Split(',').Select(float.Parse).ToList();
+ return new DataVector(data, label);
+ }
+
+ // by default use squared euclidean disatance
+ // a naive implemenation without considering things like data normalization or overflow
+ // and it is not particular about efficiency
+ public float DistanceTo(DataVector other)
+ {
+ VectorsArithmeticPrecondition(other);
+ float d = 0;
+ for (int i = 0; i < Data.Count; i++)
+ {
+ float diff = Data[i] - other.Data[i];
+ d += diff * diff;
+ }
+ return d;
+ }
+
+ public float DistanceTo(List<DataVector> list)
+ {
+ float distance = 0;
+ for (int i = 0; i < list.Count; i++)
+ {
+ distance += this.DistanceTo(list[i]);
+ }
+ return distance;
+ }
+
+ public DataVector Add(DataVector other, bool ignoreLabel = false)
+ {
+ VectorsArithmeticPrecondition(other);
+ if (!ignoreLabel)
+ {
+ if (Label != other.Label)
+ {
+ throw new InvalidOperationException("by default cannot apply addition operation on data of different labels.");
+ }
+ }
+ List<float> sumData = new List<float>(Data);
+ for (int i = 0; i < Data.Count; i++)
+ {
+ sumData[i] += other.Data[i];
+ }
+ return new DataVector(sumData, ignoreLabel ? -1 : Label);
+ }
+
+ public DataVector Normalize(float normalizationFactor)
+ {
+ if (normalizationFactor == 0)
+ {
+ throw new InvalidOperationException("normalizationFactor is zero");
+ }
+ DataVector result = new DataVector(Data, Label);
+ for (int i = 0; i < Data.Count; i++)
+ {
+ result.Data[i] /= normalizationFactor;
+ }
+ return result;
+ }
+
+ public DataVector MultiplyScalar(float scalar)
+ {
+ DataVector result = new DataVector(Data, Label);
+ for (int i = 0; i < Data.Count; i++)
+ {
+ result.Data[i] *= scalar;
+ }
+ return result;
+ }
+
+ // TODO: replace with proper serialization
+ public override string ToString()
+ {
+ return string.Join(",", Data.Select(i => i.ToString(CultureInfo.InvariantCulture)).ToArray()) + ";" + Label;
+ }
+
+ // normally centroids are picked as random points from the vector space
+ // here we just pick K random data samples
+ private static List<DataVector> InitializeCentroids(int clustersNum, List<DataVector> data, string executionDirectory)
+ {
+ // again we used the not-some-random guid trick,
+ // not truly random and not quite efficient, but easy to implement as v1
+ List<DataVector> centroids = data.OrderBy(a => Guid.NewGuid()).Take(clustersNum).ToList();
+
+ // add label to centroids
+ for (int i = 0; i < centroids.Count; i++)
+ {
+ centroids[i].Label = i;
+ }
+ WriteToCentroidFile(centroids, executionDirectory);
+ return centroids;
+ }
+
+ private void VectorsArithmeticPrecondition(DataVector other)
+ {
+ if (other == null || other.Data == null)
+ {
+ throw new ArgumentNullException("other");
+ }
+ if (Data.Count != other.Data.Count)
+ {
+ throw new InvalidOperationException("vector dimentionality mismatch");
+ }
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/0292caf1/lang/cs/Org.Apache.REEF.Examples/MachineLearning/KMeans/KMeansConfiguratioinOptions.cs
----------------------------------------------------------------------
diff --git a/lang/cs/Org.Apache.REEF.Examples/MachineLearning/KMeans/KMeansConfiguratioinOptions.cs b/lang/cs/Org.Apache.REEF.Examples/MachineLearning/KMeans/KMeansConfiguratioinOptions.cs
new file mode 100644
index 0000000..aeac77d
--- /dev/null
+++ b/lang/cs/Org.Apache.REEF.Examples/MachineLearning/KMeans/KMeansConfiguratioinOptions.cs
@@ -0,0 +1,46 @@
+/**
+ * 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.
+ */
+
+using Org.Apache.REEF.Tang.Annotations;
+
+namespace Org.Apache.REEF.Examples.MachineLearning.KMeans
+{
+ public class KMeansConfiguratioinOptions
+ {
+ [NamedParameter("Number of clusters", "K", "0")]
+ public class K : Name<int>
+ {
+ }
+
+ /// <summary>
+ /// This is for loading the initial data samples from file
+ /// currently it is assumed to load from local disk, we can easily extend this to
+ /// be an url that point to cloud storage, and have things downloaded from blob storage instead
+ /// </summary>
+ [NamedParameter("Directory for storing all execution data", "DD")]
+ public class ExecutionDirectory : Name<string>
+ {
+ }
+
+ [NamedParameter("Number of Evaluators")]
+ public class TotalNumEvaluators : Name<int>
+ {
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/0292caf1/lang/cs/Org.Apache.REEF.Examples/MachineLearning/KMeans/KMeansDriverHandlers.cs
----------------------------------------------------------------------
diff --git a/lang/cs/Org.Apache.REEF.Examples/MachineLearning/KMeans/KMeansDriverHandlers.cs b/lang/cs/Org.Apache.REEF.Examples/MachineLearning/KMeans/KMeansDriverHandlers.cs
new file mode 100644
index 0000000..235a268
--- /dev/null
+++ b/lang/cs/Org.Apache.REEF.Examples/MachineLearning/KMeans/KMeansDriverHandlers.cs
@@ -0,0 +1,191 @@
+/**
+ * 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.
+ */
+
+using System;
+using System.Collections.Generic;
+using System.Globalization;
+using System.IO;
+using System.Linq;
+using Org.Apache.REEF.Common.Io;
+using Org.Apache.REEF.Common.Services;
+using Org.Apache.REEF.Common.Tasks;
+using Org.Apache.REEF.Driver;
+using Org.Apache.REEF.Driver.Bridge;
+using Org.Apache.REEF.Driver.Context;
+using Org.Apache.REEF.Driver.Evaluator;
+using Org.Apache.REEF.Examples.MachineLearning.KMeans.codecs;
+using Org.Apache.REEF.Network.Group.Driver;
+using Org.Apache.REEF.Network.Group.Driver.Impl;
+using Org.Apache.REEF.Network.Group.Operators.Impl;
+using Org.Apache.REEF.Network.NetworkService;
+using Org.Apache.REEF.Network.NetworkService.Codec;
+using Org.Apache.REEF.Tang.Annotations;
+using Org.Apache.REEF.Tang.Formats;
+using Org.Apache.REEF.Tang.Implementations.Configuration;
+using Org.Apache.REEF.Tang.Implementations.Tang;
+using Org.Apache.REEF.Tang.Interface;
+using Org.Apache.REEF.Tang.Util;
+using Org.Apache.REEF.Utilities.Logging;
+
+namespace Org.Apache.REEF.Examples.MachineLearning.KMeans
+{
+ public class KMeansDriverHandlers :
+ IStartHandler,
+ IObserver<IEvaluatorRequestor>,
+ IObserver<IAllocatedEvaluator>,
+ IObserver<IActiveContext>
+ {
+ private static readonly Logger _Logger = Logger.GetLogger(typeof(KMeansDriverHandlers));
+ private readonly object _lockObj = new object();
+ private string _executionDirectory;
+
+ // TODO: we may want to make this injectable
+ private int _partitionsNumber = 2;
+ private int _clustersNumber = 3;
+ private int _totalEvaluators;
+ private int _partitionInex = 0;
+ private IMpiDriver _mpiDriver;
+ private ICommunicationGroupDriver _commGroup;
+ private TaskStarter _mpiTaskStarter;
+
+ [Inject]
+ public KMeansDriverHandlers()
+ {
+ Identifier = "KMeansDriverId";
+ _executionDirectory = Path.Combine(Directory.GetCurrentDirectory(), Constants.KMeansExecutionBaseDirectory, Guid.NewGuid().ToString("N").Substring(0, 4));
+ ISet<string> arguments = ClrHandlerHelper.GetCommandLineArguments();
+ string dataFile = arguments.Single(a => a.StartsWith("DataFile", StringComparison.Ordinal)).Split(':')[1];
+ DataVector.ShuffleDataAndGetInitialCentriods(
+ Path.Combine(Directory.GetCurrentDirectory(), "reef", "global", dataFile),
+ _partitionsNumber,
+ _clustersNumber,
+ _executionDirectory);
+
+ _totalEvaluators = _partitionsNumber + 1;
+ _mpiDriver = new MpiDriver(Identifier, Constants.MasterTaskId, new AvroConfigurationSerializer());
+
+ _commGroup = _mpiDriver.NewCommunicationGroup(
+ Constants.KMeansCommunicationGroupName,
+ _totalEvaluators)
+ .AddBroadcast(Constants.CentroidsBroadcastOperatorName, new BroadcastOperatorSpec<Centroids>(Constants.MasterTaskId, new CentroidsCodec()))
+ .AddBroadcast(Constants.ControlMessageBroadcastOperatorName, new BroadcastOperatorSpec<ControlMessage>(Constants.MasterTaskId, new ControlMessageCodec()))
+ .AddReduce(Constants.MeansReduceOperatorName, new ReduceOperatorSpec<ProcessedResults>(Constants.MasterTaskId, new ProcessedResultsCodec(), new KMeansMasterTask.AggregateMeans()))
+ .Build();
+ _mpiTaskStarter = new TaskStarter(_mpiDriver, _totalEvaluators);
+
+ CreateClassHierarchy();
+ }
+
+ public string Identifier { get; set; }
+
+ public void OnNext(IEvaluatorRequestor evalutorRequestor)
+ {
+ int evaluatorsNumber = _totalEvaluators;
+ int memory = 2048;
+ int core = 1;
+ EvaluatorRequest request = new EvaluatorRequest(evaluatorsNumber, memory, core);
+
+ evalutorRequestor.Submit(request);
+ }
+
+ public void OnNext(IAllocatedEvaluator allocatedEvaluator)
+ {
+ IConfiguration contextConfiguration = _mpiDriver.GetContextConfiguration();
+
+ int partitionNum;
+ if (_mpiDriver.IsMasterContextConfiguration(contextConfiguration))
+ {
+ partitionNum = -1;
+ }
+ else
+ {
+ lock (_lockObj)
+ {
+ partitionNum = _partitionInex;
+ _partitionInex++;
+ }
+ }
+
+ IConfiguration gcServiceConfiguration = _mpiDriver.GetServiceConfiguration();
+
+ IConfiguration commonServiceConfiguration = TangFactory.GetTang().NewConfigurationBuilder(gcServiceConfiguration)
+ .BindNamedParameter<DataPartitionCache.PartitionIndex, int>(GenericType<DataPartitionCache.PartitionIndex>.Class, partitionNum.ToString(CultureInfo.InvariantCulture))
+ .BindNamedParameter<KMeansConfiguratioinOptions.ExecutionDirectory, string>(GenericType<KMeansConfiguratioinOptions.ExecutionDirectory>.Class, _executionDirectory)
+ .BindNamedParameter<KMeansConfiguratioinOptions.TotalNumEvaluators, int>(GenericType<KMeansConfiguratioinOptions.TotalNumEvaluators>.Class, _totalEvaluators.ToString(CultureInfo.InvariantCulture))
+ .BindNamedParameter<KMeansConfiguratioinOptions.K, int>(GenericType<KMeansConfiguratioinOptions.K>.Class, _clustersNumber.ToString(CultureInfo.InvariantCulture))
+ .Build();
+
+ IConfiguration dataCacheServiceConfiguration = ServiceConfiguration.ConfigurationModule
+ .Set(ServiceConfiguration.Services, GenericType<DataPartitionCache>.Class)
+ .Build();
+
+ allocatedEvaluator.SubmitContextAndService(contextConfiguration, Configurations.Merge(commonServiceConfiguration, dataCacheServiceConfiguration));
+ }
+
+ public void OnNext(IActiveContext activeContext)
+ {
+ IConfiguration taskConfiguration;
+
+ if (_mpiDriver.IsMasterTaskContext(activeContext))
+ {
+ // Configure Master Task
+ taskConfiguration = TaskConfiguration.ConfigurationModule
+ .Set(TaskConfiguration.Identifier, Constants.MasterTaskId)
+ .Set(TaskConfiguration.Task, GenericType<KMeansMasterTask>.Class)
+ .Build();
+
+ _commGroup.AddTask(Constants.MasterTaskId);
+ }
+ else
+ {
+ string slaveTaskId = Constants.SlaveTaskIdPrefix + activeContext.Id;
+ // Configure Slave Task
+ taskConfiguration = TaskConfiguration.ConfigurationModule
+ .Set(TaskConfiguration.Identifier, Constants.SlaveTaskIdPrefix + activeContext.Id)
+ .Set(TaskConfiguration.Task, GenericType<KMeansSlaveTask>.Class)
+ .Build();
+
+ _commGroup.AddTask(slaveTaskId);
+ }
+ _mpiTaskStarter.QueueTask(taskConfiguration, activeContext);
+ }
+
+ public void OnError(Exception error)
+ {
+ throw new NotImplementedException();
+ }
+
+ public void OnCompleted()
+ {
+ throw new NotImplementedException();
+ }
+
+ private void CreateClassHierarchy()
+ {
+ HashSet<string> clrDlls = new HashSet<string>();
+ clrDlls.Add(typeof(IDriver).Assembly.GetName().Name);
+ clrDlls.Add(typeof(ITask).Assembly.GetName().Name);
+ clrDlls.Add(typeof(LegacyKMeansTask).Assembly.GetName().Name);
+ clrDlls.Add(typeof(INameClient).Assembly.GetName().Name);
+ clrDlls.Add(typeof(INetworkService<>).Assembly.GetName().Name);
+
+ ClrHandlerHelper.GenerateClassHierarchy(clrDlls);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/0292caf1/lang/cs/Org.Apache.REEF.Examples/MachineLearning/KMeans/KMeansMasterTask.cs
----------------------------------------------------------------------
diff --git a/lang/cs/Org.Apache.REEF.Examples/MachineLearning/KMeans/KMeansMasterTask.cs b/lang/cs/Org.Apache.REEF.Examples/MachineLearning/KMeans/KMeansMasterTask.cs
new file mode 100644
index 0000000..3dd7adb
--- /dev/null
+++ b/lang/cs/Org.Apache.REEF.Examples/MachineLearning/KMeans/KMeansMasterTask.cs
@@ -0,0 +1,155 @@
+/**
+ * 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.
+ */
+
+using System;
+using System.Collections.Generic;
+using System.Globalization;
+using System.IO;
+using System.Linq;
+using Org.Apache.REEF.Common.Tasks;
+using Org.Apache.REEF.Network.Group.Operators;
+using Org.Apache.REEF.Network.Group.Task;
+using Org.Apache.REEF.Network.NetworkService;
+using Org.Apache.REEF.Tang.Annotations;
+using Org.Apache.REEF.Utilities.Logging;
+
+namespace Org.Apache.REEF.Examples.MachineLearning.KMeans
+{
+ public class KMeansMasterTask : ITask
+ {
+ private static Logger _logger = Logger.GetLogger(typeof(KMeansMasterTask));
+
+ private int _iteration = 0;
+
+ private ICommunicationGroupClient _commGroup;
+ private IBroadcastSender<Centroids> _dataBroadcastSender;
+ private IBroadcastSender<ControlMessage> _controlBroadcastSender;
+ private IReduceReceiver<ProcessedResults> _meansReducerReceiver;
+ private string _kMeansExecutionDirectory;
+ private Centroids _centroids;
+ private bool _isInitialIteration;
+
+ [Inject]
+ public KMeansMasterTask(
+ [Parameter(typeof(KMeansConfiguratioinOptions.TotalNumEvaluators))] int totalNumEvaluators,
+ [Parameter(Value = typeof(KMeansConfiguratioinOptions.ExecutionDirectory))] string executionDirectory,
+ IMpiClient mpiClient)
+ {
+ using (_logger.LogFunction("KMeansMasterTask"))
+ {
+ if (totalNumEvaluators <= 1)
+ {
+ throw new ArgumentException("There must be more than 1 Evaluators in total, but the total evaluators number provided is " + totalNumEvaluators);
+ }
+ _commGroup = mpiClient.GetCommunicationGroup(Constants.KMeansCommunicationGroupName);
+ _dataBroadcastSender = _commGroup.GetBroadcastSender<Centroids>(Constants.CentroidsBroadcastOperatorName);
+ _meansReducerReceiver = _commGroup.GetReduceReceiver<ProcessedResults>(Constants.MeansReduceOperatorName);
+ _controlBroadcastSender = _commGroup.GetBroadcastSender<ControlMessage>(Constants.ControlMessageBroadcastOperatorName);
+ _kMeansExecutionDirectory = executionDirectory;
+ _isInitialIteration = true;
+ }
+ }
+
+ public byte[] Call(byte[] memento)
+ {
+ // TODO: this belongs to dedicated dataloader layer, will refactor once we have that
+ string centroidFile = Path.Combine(_kMeansExecutionDirectory, Constants.CentroidsFile);
+ _centroids = new Centroids(DataPartitionCache.ReadDataFile(centroidFile));
+
+ float loss = float.MaxValue;
+ float newLoss;
+
+ while (true)
+ {
+ if (_isInitialIteration)
+ {
+ // broadcast initial centroids to all slave nodes
+ _logger.Log(Level.Info, "Broadcasting INITIAL centroids to all slave nodes: " + _centroids);
+ _isInitialIteration = false;
+ }
+ else
+ {
+ ProcessedResults results = _meansReducerReceiver.Reduce();
+ _centroids = new Centroids(results.Means.Select(m => m.Mean).ToList());
+ _logger.Log(Level.Info, "Broadcasting new centroids to all slave nodes: " + _centroids);
+ newLoss = results.Loss;
+ _logger.Log(Level.Info, string.Format(CultureInfo.InvariantCulture, "The new loss value {0} at iteration {1} ", newLoss, _iteration));
+ if (newLoss > loss)
+ {
+ _controlBroadcastSender.Send(ControlMessage.STOP);
+ throw new InvalidOperationException(
+ string.Format(CultureInfo.InvariantCulture, "The new loss {0} is larger than previous loss {1}, while loss function must be monotonically decreasing across iterations", newLoss, loss));
+ }
+ else if (newLoss.Equals(loss))
+ {
+ _logger.Log(Level.Info, string.Format(CultureInfo.InvariantCulture, "KMeans clustering has converged with a loss value of {0} at iteration {1} ", newLoss, _iteration));
+ break;
+ }
+ else
+ {
+ loss = newLoss;
+ }
+ }
+ _controlBroadcastSender.Send(ControlMessage.RECEIVE);
+ _dataBroadcastSender.Send(_centroids);
+ _iteration++;
+ }
+ _controlBroadcastSender.Send(ControlMessage.STOP);
+ return null;
+ }
+
+ public void Dispose()
+ {
+ }
+
+ public class AggregateMeans : IReduceFunction<ProcessedResults>
+ {
+ [Inject]
+ public AggregateMeans()
+ {
+ }
+
+ public ProcessedResults Reduce(IEnumerable<ProcessedResults> elements)
+ {
+ List<PartialMean> aggregatedMeans = new List<PartialMean>();
+ List<PartialMean> totalList = new List<PartialMean>();
+ float aggregatedLoss = 0;
+
+ foreach (var element in elements)
+ {
+ totalList.AddRange(element.Means);
+ aggregatedLoss += element.Loss;
+ }
+
+ // we infer the value of K from the labeled data
+ int clustersNum = totalList.Max(p => p.Mean.Label) + 1;
+ for (int i = 0; i < clustersNum; i++)
+ {
+ List<PartialMean> means = totalList.Where(m => m.Mean.Label == i).ToList();
+ aggregatedMeans.Add(new PartialMean(PartialMean.AggreatedMean(means), means.Count));
+ }
+
+ ProcessedResults returnMeans = new ProcessedResults(aggregatedMeans, aggregatedLoss);
+
+ _logger.Log(Level.Info, "The true means aggregated by the reduce function: " + returnMeans);
+ return returnMeans;
+ }
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/0292caf1/lang/cs/Org.Apache.REEF.Examples/MachineLearning/KMeans/KMeansSlaveTask.cs
----------------------------------------------------------------------
diff --git a/lang/cs/Org.Apache.REEF.Examples/MachineLearning/KMeans/KMeansSlaveTask.cs b/lang/cs/Org.Apache.REEF.Examples/MachineLearning/KMeans/KMeansSlaveTask.cs
new file mode 100644
index 0000000..a36fbcb
--- /dev/null
+++ b/lang/cs/Org.Apache.REEF.Examples/MachineLearning/KMeans/KMeansSlaveTask.cs
@@ -0,0 +1,118 @@
+/**
+ * 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.
+ */
+
+using System.Collections.Generic;
+using System.Linq;
+using Org.Apache.REEF.Common.Tasks;
+using Org.Apache.REEF.Network.Group.Operators;
+using Org.Apache.REEF.Network.Group.Task;
+using Org.Apache.REEF.Network.NetworkService;
+using Org.Apache.REEF.Tang.Annotations;
+using Org.Apache.REEF.Utilities.Logging;
+
+namespace Org.Apache.REEF.Examples.MachineLearning.KMeans
+{
+ public class KMeansSlaveTask : ITask
+ {
+ private static Logger _logger = Logger.GetLogger(typeof(KMeansSlaveTask));
+ private int _clustersNum;
+ private IMpiClient _mpiClient;
+ private ICommunicationGroupClient _commGroup;
+ private IBroadcastReceiver<Centroids> _dataBroadcastReceiver;
+ private IBroadcastReceiver<ControlMessage> _controlBroadcastReceiver;
+ private IReduceSender<ProcessedResults> _partialMeansSender;
+ private DataPartitionCache _dataPartition;
+
+ [Inject]
+ public KMeansSlaveTask(
+ DataPartitionCache dataPartition,
+ [Parameter(typeof(KMeansConfiguratioinOptions.TotalNumEvaluators))] int clustersNumber,
+ IMpiClient mpiClient)
+ {
+ using (_logger.LogFunction("KMeansSlaveTask::KMeansSlaveTask"))
+ {
+ _dataPartition = dataPartition;
+ _mpiClient = mpiClient;
+ _clustersNum = clustersNumber;
+ _commGroup = _mpiClient.GetCommunicationGroup(Constants.KMeansCommunicationGroupName);
+ _dataBroadcastReceiver = _commGroup.GetBroadcastReceiver<Centroids>(Constants.CentroidsBroadcastOperatorName);
+ _partialMeansSender = _commGroup.GetReduceSender<ProcessedResults>(Constants.MeansReduceOperatorName);
+ _controlBroadcastReceiver = _commGroup.GetBroadcastReceiver<ControlMessage>(Constants.ControlMessageBroadcastOperatorName);
+ }
+ }
+
+ public byte[] Call(byte[] memento)
+ {
+ while (true)
+ {
+ if (_controlBroadcastReceiver.Receive() == ControlMessage.STOP)
+ {
+ break;
+ }
+ Centroids centroids = _dataBroadcastReceiver.Receive();
+ // we compute the loss here before data is relabled, this does not reflect the latest clustering result at the end of current iteration,
+ // but it will save another round of group communications in each iteration
+ _logger.Log(Level.Info, "Received centroids from master: " + centroids);
+ _dataPartition.LabelData(centroids);
+ ProcessedResults partialMeans = new ProcessedResults(ComputePartialMeans(), ComputeLossFunction(centroids, _dataPartition.DataVectors));
+ _logger.Log(Level.Info, "Sending partial means: " + partialMeans);
+ _partialMeansSender.Send(partialMeans);
+ }
+
+ return null;
+ }
+
+ public void Dispose()
+ {
+ _mpiClient.Dispose();
+ }
+
+ private List<PartialMean> ComputePartialMeans()
+ {
+ _logger.Log(Level.Verbose, "cluster number " + _clustersNum);
+ List<PartialMean> partialMeans = new PartialMean[_clustersNum].ToList();
+ for (int i = 0; i < _clustersNum; i++)
+ {
+ List<DataVector> slices = _dataPartition.DataVectors.Where(d => d.Label == i).ToList();
+ DataVector average = new DataVector(_dataPartition.DataVectors[0].Dimension);
+
+ if (slices.Count > 1)
+ {
+ average = DataVector.Mean(slices);
+ }
+ average.Label = i;
+ partialMeans[i] = new PartialMean(average, slices.Count);
+ _logger.Log(Level.Info, "Adding to partial means list: " + partialMeans[i]);
+ }
+ return partialMeans;
+ }
+
+ private float ComputeLossFunction(Centroids centroids, List<DataVector> labeledData)
+ {
+ float d = 0;
+ for (int i = 0; i < centroids.Points.Count; i++)
+ {
+ DataVector centroid = centroids.Points[i];
+ List<DataVector> slice = labeledData.Where(v => v.Label == centroid.Label).ToList();
+ d += centroid.DistanceTo(slice);
+ }
+ return d;
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/0292caf1/lang/cs/Org.Apache.REEF.Examples/MachineLearning/KMeans/LegacyKMeansTask.cs
----------------------------------------------------------------------
diff --git a/lang/cs/Org.Apache.REEF.Examples/MachineLearning/KMeans/LegacyKMeansTask.cs b/lang/cs/Org.Apache.REEF.Examples/MachineLearning/KMeans/LegacyKMeansTask.cs
new file mode 100644
index 0000000..b674d84
--- /dev/null
+++ b/lang/cs/Org.Apache.REEF.Examples/MachineLearning/KMeans/LegacyKMeansTask.cs
@@ -0,0 +1,113 @@
+/**
+ * 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.
+ */
+
+using System.Collections.Generic;
+using System.IO;
+using System.Linq;
+using Org.Apache.REEF.Tang.Annotations;
+
+namespace Org.Apache.REEF.Examples.MachineLearning.KMeans
+{
+ /// <summary>
+ /// This is the legacy KmeansTask implmented when group communications are not available
+ /// It is still being used for plain KMeans without REEF, we probably want to refactor it later
+ /// to reflect that
+ /// </summary>
+ public class LegacyKMeansTask
+ {
+ private int _clustersNum;
+ private DataPartitionCache _dataPartition;
+ private string _kMeansExecutionDirectory;
+
+ private Centroids _centroids;
+ private List<PartialMean> _partialMeans;
+
+ [Inject]
+ public LegacyKMeansTask(
+ DataPartitionCache dataPartition,
+ [Parameter(Value = typeof(KMeansConfiguratioinOptions.K))] int clustersNumber,
+ [Parameter(Value = typeof(KMeansConfiguratioinOptions.ExecutionDirectory))] string executionDirectory)
+ {
+ _dataPartition = dataPartition;
+ _clustersNum = clustersNumber;
+ _kMeansExecutionDirectory = executionDirectory;
+ if (_centroids == null)
+ {
+ string centroidFile = Path.Combine(_kMeansExecutionDirectory, Constants.CentroidsFile);
+ _centroids = new Centroids(DataPartitionCache.ReadDataFile(centroidFile));
+ }
+ }
+
+ public static float ComputeLossFunction(List<DataVector> centroids, List<DataVector> labeledData)
+ {
+ float d = 0;
+ for (int i = 0; i < centroids.Count; i++)
+ {
+ DataVector centroid = centroids[i];
+ List<DataVector> slice = labeledData.Where(v => v.Label == centroid.Label).ToList();
+ d += centroid.DistanceTo(slice);
+ }
+ return d;
+ }
+
+ public byte[] CallWithWritingToFileSystem(byte[] memento)
+ {
+ string centroidFile = Path.Combine(_kMeansExecutionDirectory, Constants.CentroidsFile);
+ _centroids = new Centroids(DataPartitionCache.ReadDataFile(centroidFile));
+
+ _dataPartition.LabelData(_centroids);
+ _partialMeans = ComputePartialMeans();
+
+ // should be replaced with MPI
+ using (StreamWriter writer = new StreamWriter(
+ File.OpenWrite(Path.Combine(_kMeansExecutionDirectory, Constants.DataDirectory, Constants.PartialMeanFilePrefix + _dataPartition.Partition))))
+ {
+ for (int i = 0; i < _partialMeans.Count; i++)
+ {
+ writer.WriteLine(_partialMeans[i].ToString());
+ }
+ writer.Close();
+ }
+
+ return null;
+ }
+
+ public List<PartialMean> ComputePartialMeans()
+ {
+ List<PartialMean> partialMeans = new PartialMean[_clustersNum].ToList();
+ for (int i = 0; i < _clustersNum; i++)
+ {
+ List<DataVector> slices = _dataPartition.DataVectors.Where(d => d.Label == i).ToList();
+ DataVector average = new DataVector(_dataPartition.DataVectors[0].Dimension);
+
+ if (slices.Count > 1)
+ {
+ average = DataVector.Mean(slices);
+ }
+ average.Label = i;
+ partialMeans[i] = new PartialMean(average, slices.Count);
+ }
+ return partialMeans;
+ }
+
+ public void Dispose()
+ {
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/0292caf1/lang/cs/Org.Apache.REEF.Examples/MachineLearning/KMeans/PartialMean.cs
----------------------------------------------------------------------
diff --git a/lang/cs/Org.Apache.REEF.Examples/MachineLearning/KMeans/PartialMean.cs b/lang/cs/Org.Apache.REEF.Examples/MachineLearning/KMeans/PartialMean.cs
new file mode 100644
index 0000000..6f44167
--- /dev/null
+++ b/lang/cs/Org.Apache.REEF.Examples/MachineLearning/KMeans/PartialMean.cs
@@ -0,0 +1,124 @@
+/**
+ * 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.
+ */
+
+using System;
+using System.Collections.Generic;
+using System.Globalization;
+using System.IO;
+using System.Linq;
+
+namespace Org.Apache.REEF.Examples.MachineLearning.KMeans
+{
+ public class PartialMean
+ {
+ public PartialMean(DataVector vector, int size)
+ {
+ Mean = vector;
+ Size = size;
+ }
+
+ public PartialMean()
+ {
+ }
+
+ public DataVector Mean { get; set; }
+
+ public int Size { get; set; }
+
+ public static PartialMean FromString(string str)
+ {
+ if (string.IsNullOrWhiteSpace(str))
+ {
+ throw new ArgumentException("str");
+ }
+ string[] parts = str.Split('#');
+ if (parts == null || parts.Length != 2)
+ {
+ throw new ArgumentException("Cannot deserialize PartialMean from string " + str);
+ }
+ return new PartialMean(DataVector.FromString(parts[0]), int.Parse(parts[1], CultureInfo.InvariantCulture));
+ }
+
+ public static DataVector AggreatedMean(List<PartialMean> means)
+ {
+ if (means == null || means.Count == 0)
+ {
+ throw new ArgumentException("means");
+ }
+ PartialMean mean = means[0];
+ for (int i = 1; i < means.Count; i++)
+ {
+ mean = mean.CombinePartialMean(means[i]);
+ }
+ return mean.Mean;
+ }
+
+ public static List<DataVector> AggregateTrueMeansToFileSystem(int partitionsNum, int clustersNum, string executionDirectory)
+ {
+ List<PartialMean> partialMeans = new List<PartialMean>();
+ for (int i = 0; i < partitionsNum; i++)
+ {
+ // should be replaced with MPI
+ string path = Path.Combine(executionDirectory, Constants.DataDirectory, Constants.PartialMeanFilePrefix + i.ToString(CultureInfo.InvariantCulture));
+ FileStream file = new FileStream(path, FileMode.Open, FileAccess.Read, FileShare.Read);
+ using (StreamReader reader = new StreamReader(file))
+ {
+ int index = 0;
+ while (!reader.EndOfStream)
+ {
+ string line = reader.ReadLine();
+ if (index++ < clustersNum)
+ {
+ partialMeans.Add(PartialMean.FromString(line));
+ }
+ }
+ reader.Close();
+ }
+ }
+ List<DataVector> newCentroids = new List<DataVector>();
+ for (int i = 0; i < clustersNum; i++)
+ {
+ List<PartialMean> means = partialMeans.Where(m => m.Mean.Label == i).ToList();
+ newCentroids.Add(PartialMean.AggreatedMean(means));
+ }
+ return newCentroids;
+ }
+
+ public override string ToString()
+ {
+ return Mean.ToString() + "#" + Size;
+ }
+
+ private PartialMean CombinePartialMean(PartialMean other)
+ {
+ PartialMean aggreatedMean = new PartialMean();
+ if (other == null)
+ {
+ throw new ArgumentNullException("other");
+ }
+ if (Mean.Label != other.Mean.Label)
+ {
+ throw new ArgumentException("cannot combine partial means with different labels");
+ }
+ aggreatedMean.Size = Size + other.Size;
+ aggreatedMean.Mean = (Mean.MultiplyScalar(Size).Add(other.Mean.MultiplyScalar(other.Size))).Normalize(aggreatedMean.Size);
+ return aggreatedMean;
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/0292caf1/lang/cs/Org.Apache.REEF.Examples/MachineLearning/KMeans/ProcessedResults.cs
----------------------------------------------------------------------
diff --git a/lang/cs/Org.Apache.REEF.Examples/MachineLearning/KMeans/ProcessedResults.cs b/lang/cs/Org.Apache.REEF.Examples/MachineLearning/KMeans/ProcessedResults.cs
new file mode 100644
index 0000000..3e3394f
--- /dev/null
+++ b/lang/cs/Org.Apache.REEF.Examples/MachineLearning/KMeans/ProcessedResults.cs
@@ -0,0 +1,54 @@
+/**
+ * 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.
+ */
+
+using System.Collections.Generic;
+using Org.Apache.REEF.Examples.MachineLearning.KMeans.codecs;
+using Org.Apache.REEF.Utilities;
+
+namespace Org.Apache.REEF.Examples.MachineLearning.KMeans
+{
+ /// <summary>
+ /// ProcessedResults includes a list of "PartialMeans" and "Partial loss", but it can be used to denote
+ /// the "whole" means as well, aggreated from all PartialMeans
+ /// </summary>
+ public class ProcessedResults
+ {
+ public ProcessedResults(List<PartialMean> means, float loss)
+ {
+ Means = means;
+ Loss = loss;
+ }
+
+ public List<PartialMean> Means { get; set; }
+
+ /// <summary>
+ /// the loss for current slice computed from
+ /// </summary>
+ public float Loss { get; set; }
+
+ /// <summary>
+ /// helper function mostly used for logging
+ /// </summary>
+ /// <returns>seralized string</returns>
+ public override string ToString()
+ {
+ return ByteUtilities.ByteArrarysToString(new ProcessedResultsCodec().Encode(this));
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/0292caf1/lang/cs/Org.Apache.REEF.Examples/MachineLearning/KMeans/codecs/CentroidsCodec.cs
----------------------------------------------------------------------
diff --git a/lang/cs/Org.Apache.REEF.Examples/MachineLearning/KMeans/codecs/CentroidsCodec.cs b/lang/cs/Org.Apache.REEF.Examples/MachineLearning/KMeans/codecs/CentroidsCodec.cs
new file mode 100644
index 0000000..19480b6
--- /dev/null
+++ b/lang/cs/Org.Apache.REEF.Examples/MachineLearning/KMeans/codecs/CentroidsCodec.cs
@@ -0,0 +1,49 @@
+/**
+ * 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.
+ */
+
+using Org.Apache.REEF.Examples.MachineLearning.KMeans.Contracts;
+using Org.Apache.REEF.Tang.Annotations;
+using Org.Apache.REEF.Utilities;
+using Org.Apache.REEF.Wake.Remote;
+
+namespace Org.Apache.REEF.Examples.MachineLearning.KMeans.codecs
+{
+ /// <summary>
+ /// Used to serialize and deserialize Centroids objects.
+ /// </summary>
+ public class CentroidsCodec : ICodec<Centroids>
+ {
+ [Inject]
+ public CentroidsCodec()
+ {
+ }
+
+ public byte[] Encode(Centroids centroids)
+ {
+ CentroidsContract contract = CentroidsContract.Create(centroids);
+ return AvroUtils.AvroSerialize(contract);
+ }
+
+ public Centroids Decode(byte[] data)
+ {
+ CentroidsContract contract = AvroUtils.AvroDeserialize<CentroidsContract>(data);
+ return contract.ToCentroids();
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/0292caf1/lang/cs/Org.Apache.REEF.Examples/MachineLearning/KMeans/codecs/DataVectorCodec.cs
----------------------------------------------------------------------
diff --git a/lang/cs/Org.Apache.REEF.Examples/MachineLearning/KMeans/codecs/DataVectorCodec.cs b/lang/cs/Org.Apache.REEF.Examples/MachineLearning/KMeans/codecs/DataVectorCodec.cs
new file mode 100644
index 0000000..820162e
--- /dev/null
+++ b/lang/cs/Org.Apache.REEF.Examples/MachineLearning/KMeans/codecs/DataVectorCodec.cs
@@ -0,0 +1,46 @@
+/**
+ * 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.
+ */
+
+using Org.Apache.REEF.Examples.MachineLearning.KMeans.Contracts;
+using Org.Apache.REEF.Tang.Annotations;
+using Org.Apache.REEF.Utilities;
+using Org.Apache.REEF.Wake.Remote;
+
+namespace Org.Apache.REEF.Examples.MachineLearning.KMeans.codecs
+{
+ public class DataVectorCodec : ICodec<DataVector>
+ {
+ [Inject]
+ public DataVectorCodec()
+ {
+ }
+
+ public byte[] Encode(DataVector obj)
+ {
+ DataVectorContract contract = DataVectorContract.Create(obj);
+ return AvroUtils.AvroSerialize(contract);
+ }
+
+ public DataVector Decode(byte[] data)
+ {
+ DataVectorContract contract = AvroUtils.AvroDeserialize<DataVectorContract>(data);
+ return contract.ToDataVector();
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/0292caf1/lang/cs/Org.Apache.REEF.Examples/MachineLearning/KMeans/codecs/ProcessedResultsCodec.cs
----------------------------------------------------------------------
diff --git a/lang/cs/Org.Apache.REEF.Examples/MachineLearning/KMeans/codecs/ProcessedResultsCodec.cs b/lang/cs/Org.Apache.REEF.Examples/MachineLearning/KMeans/codecs/ProcessedResultsCodec.cs
new file mode 100644
index 0000000..8166030
--- /dev/null
+++ b/lang/cs/Org.Apache.REEF.Examples/MachineLearning/KMeans/codecs/ProcessedResultsCodec.cs
@@ -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.
+ */
+
+using System;
+using System.Collections.Generic;
+using System.Globalization;
+using System.Linq;
+using Org.Apache.REEF.Tang.Annotations;
+using Org.Apache.REEF.Utilities;
+using Org.Apache.REEF.Wake.Remote;
+
+namespace Org.Apache.REEF.Examples.MachineLearning.KMeans.codecs
+{
+ /// <summary>
+ /// TODO: use proper avro scheme to do encode/decode
+ /// </summary>
+ public class ProcessedResultsCodec : ICodec<ProcessedResults>
+ {
+ [Inject]
+ public ProcessedResultsCodec()
+ {
+ }
+
+ public byte[] Encode(ProcessedResults results)
+ {
+ return ByteUtilities.StringToByteArrays(results.Loss + "+" + string.Join("@", results.Means.Select(m => m.ToString())));
+ }
+
+ public ProcessedResults Decode(byte[] data)
+ {
+ string[] parts = ByteUtilities.ByteArrarysToString(data).Split('+');
+ if (parts.Count() != 2)
+ {
+ throw new ArgumentException("cannot deserialize from" + ByteUtilities.ByteArrarysToString(data));
+ }
+ float loss = float.Parse(parts[0], CultureInfo.InvariantCulture);
+ List<PartialMean> means = parts[1].Split('@').Select(PartialMean.FromString).ToList();
+ return new ProcessedResults(means, loss);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/0292caf1/lang/cs/Org.Apache.REEF.Examples/Org.Apache.REEF.Examples.csproj
----------------------------------------------------------------------
diff --git a/lang/cs/Org.Apache.REEF.Examples/Org.Apache.REEF.Examples.csproj b/lang/cs/Org.Apache.REEF.Examples/Org.Apache.REEF.Examples.csproj
index e2691c5..e3ca589 100644
--- a/lang/cs/Org.Apache.REEF.Examples/Org.Apache.REEF.Examples.csproj
+++ b/lang/cs/Org.Apache.REEF.Examples/Org.Apache.REEF.Examples.csproj
@@ -72,6 +72,7 @@ under the License.
<ItemGroup>
<Reference Include="System" />
<Reference Include="System.Core" />
+ <Reference Include="System.Runtime.Serialization" />
<Reference Include="System.Xml.Linq" />
<Reference Include="System.Data.DataSetExtensions" />
<Reference Include="Microsoft.CSharp" />
@@ -95,6 +96,24 @@ under the License.
<Compile Include="HelloCLRBridge\Handlers\HelloStartHandler.cs" />
<Compile Include="HelloCLRBridge\Handlers\HelloTaskMessageHandler.cs" />
<Compile Include="HelloCLRBridge\HelloTraceListener.cs" />
+ <Compile Include="MachineLearning\KMeans\Centroids.cs" />
+ <Compile Include="MachineLearning\KMeans\codecs\CentroidsCodec.cs" />
+ <Compile Include="MachineLearning\KMeans\codecs\DataVectorCodec.cs" />
+ <Compile Include="MachineLearning\KMeans\codecs\ProcessedResultsCodec.cs" />
+ <Compile Include="MachineLearning\KMeans\Constants.cs" />
+ <Compile Include="MachineLearning\KMeans\Contracts\CentroidsContract.cs" />
+ <Compile Include="MachineLearning\KMeans\Contracts\DataVectorContract.cs" />
+ <Compile Include="MachineLearning\KMeans\Contracts\PartialMeanContract.cs" />
+ <Compile Include="MachineLearning\KMeans\Contracts\ProcessedResultsContract.cs" />
+ <Compile Include="MachineLearning\KMeans\DataPartitionCache.cs" />
+ <Compile Include="MachineLearning\KMeans\DataVector.cs" />
+ <Compile Include="MachineLearning\KMeans\KMeansConfiguratioinOptions.cs" />
+ <Compile Include="MachineLearning\KMeans\KMeansDriverHandlers.cs" />
+ <Compile Include="MachineLearning\KMeans\KMeansMasterTask.cs" />
+ <Compile Include="MachineLearning\KMeans\KMeansSlaveTask.cs" />
+ <Compile Include="MachineLearning\KMeans\LegacyKMeansTask.cs" />
+ <Compile Include="MachineLearning\KMeans\PartialMean.cs" />
+ <Compile Include="MachineLearning\KMeans\ProcessedResults.cs" />
<Compile Include="RetainedEvalCLRBridge\Handlers\RetainedEvalActiveContextHandler.cs" />
<Compile Include="RetainedEvalCLRBridge\Handlers\RetainedEvalAllocatedEvaluatorHandler.cs" />
<Compile Include="RetainedEvalCLRBridge\Handlers\RetainedEvalEvaluatorRequestorHandler.cs" />
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/0292caf1/lang/cs/Org.Apache.REEF.Network/Group/Codec/GcmMessageProto.cs
----------------------------------------------------------------------
diff --git a/lang/cs/Org.Apache.REEF.Network/Group/Codec/GcmMessageProto.cs b/lang/cs/Org.Apache.REEF.Network/Group/Codec/GcmMessageProto.cs
new file mode 100644
index 0000000..8a5a726
--- /dev/null
+++ b/lang/cs/Org.Apache.REEF.Network/Group/Codec/GcmMessageProto.cs
@@ -0,0 +1,76 @@
+/**
+ * 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.
+ */
+
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.Threading.Tasks;
+using Org.Apache.REEF.Network.Group.Driver;
+using Org.Apache.REEF.Network.Group.Driver.Impl;
+using ProtoBuf;
+
+namespace Org.Apache.REEF.Network.Group.Codec
+{
+ [ProtoContract]
+ public class GcmMessageProto
+ {
+ [ProtoMember(1)]
+ public byte[][] Data { get; set; }
+
+ [ProtoMember(2)]
+ public string OperatorName { get; set; }
+
+ [ProtoMember(3)]
+ public string GroupName { get; set; }
+
+ [ProtoMember(4)]
+ public string Source { get; set; }
+
+ [ProtoMember(5)]
+ public string Destination { get; set; }
+
+ [ProtoMember(6)]
+ public MessageType MsgType { get; set; }
+
+ public static GcmMessageProto Create(GroupCommunicationMessage gcm)
+ {
+ return new GcmMessageProto()
+ {
+ Data = gcm.Data,
+ OperatorName = gcm.OperatorName,
+ GroupName = gcm.GroupName,
+ Source = gcm.Source,
+ Destination = gcm.Destination,
+ MsgType = gcm.MsgType,
+ };
+ }
+
+ public GroupCommunicationMessage ToGcm()
+ {
+ return new GroupCommunicationMessage(
+ GroupName,
+ OperatorName,
+ Source,
+ Destination,
+ Data,
+ MsgType);
+ }
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/0292caf1/lang/cs/Org.Apache.REEF.Network/Group/Codec/GroupCommunicationMessageCodec.cs
----------------------------------------------------------------------
diff --git a/lang/cs/Org.Apache.REEF.Network/Group/Codec/GroupCommunicationMessageCodec.cs b/lang/cs/Org.Apache.REEF.Network/Group/Codec/GroupCommunicationMessageCodec.cs
new file mode 100644
index 0000000..a8f884f
--- /dev/null
+++ b/lang/cs/Org.Apache.REEF.Network/Group/Codec/GroupCommunicationMessageCodec.cs
@@ -0,0 +1,77 @@
+/**
+ * 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.
+ */
+
+using System;
+using System.Collections.Generic;
+using System.IO;
+using System.Linq;
+using System.Text;
+using System.Threading.Tasks;
+using Org.Apache.REEF.Network.Group.Codec;
+using Org.Apache.REEF.Network.Group.Driver;
+using Org.Apache.REEF.Network.Group.Driver.Impl;
+using Org.Apache.REEF.Tang.Annotations;
+using Org.Apache.REEF.Wake.Remote;
+using ProtoBuf;
+
+namespace Org.Apache.REEF.Network.Group.Codec
+{
+ /// <summary>
+ /// Used to serialize GroupCommunicationMessages.
+ /// </summary>
+ public class GroupCommunicationMessageCodec : ICodec<GroupCommunicationMessage>
+ {
+ /// <summary>
+ /// Create a new GroupCommunicationMessageCodec.
+ /// </summary>
+ [Inject]
+ public GroupCommunicationMessageCodec()
+ {
+ }
+
+ /// <summary>
+ /// Serialize the GroupCommunicationObject into a byte array using Protobuf.
+ /// </summary>
+ /// <param name="obj">The object to serialize.</param>
+ /// <returns>The serialized GroupCommunicationMessage in byte array form</returns>
+ public byte[] Encode(GroupCommunicationMessage obj)
+ {
+ GcmMessageProto proto = GcmMessageProto.Create(obj);
+ using (var stream = new MemoryStream())
+ {
+ Serializer.Serialize(stream, proto);
+ return stream.ToArray();
+ }
+ }
+
+ /// <summary>
+ /// Deserialize the byte array into a GroupCommunicationMessage using Protobuf.
+ /// </summary>
+ /// <param name="data">The byte array to deserialize</param>
+ /// <returns>The deserialized GroupCommunicationMessage object.</returns>
+ public GroupCommunicationMessage Decode(byte[] data)
+ {
+ using (var stream = new MemoryStream(data))
+ {
+ GcmMessageProto proto = Serializer.Deserialize<GcmMessageProto>(stream);
+ return proto.ToGcm();
+ }
+ }
+ }
+}