You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@reef.apache.org by we...@apache.org on 2015/04/16 23:38:30 UTC
[2/3] incubator-reef git commit: [REEF-256] Rename MPI to Group
Communication
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/c330dcff/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
index 2e11441..b47e076 100644
--- a/lang/cs/Org.Apache.REEF.Network/Group/Driver/ICommunicationGroupDriver.cs
+++ b/lang/cs/Org.Apache.REEF.Network/Group/Driver/ICommunicationGroupDriver.cs
@@ -29,7 +29,7 @@ using Org.Apache.REEF.Wake.Remote;
namespace Org.Apache.REEF.Network.Group.Driver
{
/// <summary>
- /// Used to configure MPI operators in Reef driver.
+ /// Used to configure Group Communication operators in Reef driver.
/// All operators in the same Communication Group run on the the
/// same set of tasks.
/// </summary>
@@ -41,7 +41,7 @@ namespace Org.Apache.REEF.Network.Group.Driver
List<string> TaskIds { get; }
/// <summary>
- /// Adds the Broadcast MPI operator to the communication group.
+ /// Adds the Broadcast Group Communication operator to the communication group.
/// </summary>
/// <typeparam name="TMessage">The type of messages that operators will send</typeparam>
/// <typeparam name="TMessageCodec">The codec used for serializing messages</typeparam>
@@ -53,7 +53,7 @@ namespace Org.Apache.REEF.Network.Group.Driver
ICommunicationGroupDriver AddBroadcast<TMessage, TMessageCodec>(string operatorName, string masterTaskId, TopologyTypes topologyType, IPipelineDataConverter<TMessage> pipelineDataConverter) where TMessageCodec : ICodec<TMessage>;
/// <summary>
- /// Adds the Broadcast MPI operator to the communication group.
+ /// Adds the Broadcast Group Communication operator to the communication group.
/// </summary>
/// <typeparam name="TMessage">The type of messages that operators will send</typeparam>
/// <typeparam name="TMessageCodec">The codec used for serializing messages</typeparam>
@@ -64,7 +64,7 @@ namespace Org.Apache.REEF.Network.Group.Driver
ICommunicationGroupDriver AddBroadcast<TMessage, TMessageCodec>(string operatorName, string masterTaskId, TopologyTypes topologyType = TopologyTypes.Flat) where TMessageCodec : ICodec<TMessage>;
/// <summary>
- /// Adds the Broadcast MPI operator to the communication group. Default to IntCodec
+ /// Adds the Broadcast Group Communication operator to the communication group. Default to IntCodec
/// </summary>
/// <param name="operatorName">The name of the broadcast operator</param>
/// <param name="masterTaskId">The master task id in broadcast operator</param>
@@ -73,7 +73,7 @@ namespace Org.Apache.REEF.Network.Group.Driver
ICommunicationGroupDriver AddBroadcast(string operatorName, string masterTaskId, TopologyTypes topologyType = TopologyTypes.Flat);
/// <summary>
- /// Adds the Reduce MPI operator to the communication group.
+ /// Adds the Reduce Group Communication operator to the communication group.
/// </summary>
/// <typeparam name="TMessage">The type of messages that operators will send</typeparam>
/// <typeparam name="TMessageCodec">The codec used for serializing messages</typeparam>
@@ -85,7 +85,7 @@ namespace Org.Apache.REEF.Network.Group.Driver
ICommunicationGroupDriver AddReduce<TMessage, TMessageCodec>(string operatorName, string masterTaskId, IReduceFunction<TMessage> reduceFunction, TopologyTypes topologyType, IPipelineDataConverter<TMessage> pipelineDataConverter) where TMessageCodec : ICodec<TMessage>;
/// <summary>
- /// Adds the Reduce MPI operator to the communication group.
+ /// Adds the Reduce Group Communication operator to the communication group.
/// </summary>
/// <typeparam name="TMessage">The type of messages that operators will send</typeparam>
/// <typeparam name="TMessageCodec">The codec used for serializing messages</typeparam>
@@ -98,7 +98,7 @@ namespace Org.Apache.REEF.Network.Group.Driver
/// <summary>
- /// Adds the Reduce MPI operator to the communication group with default IntCodec
+ /// Adds the Reduce Group Communication operator to the communication group with default IntCodec
/// </summary>
/// <param name="operatorName">The name of the reduce operator</param>
/// <param name="masterTaskId">The master task id for the typology</param>
@@ -108,7 +108,7 @@ namespace Org.Apache.REEF.Network.Group.Driver
ICommunicationGroupDriver AddReduce(string operatorName, string masterTaskId, IReduceFunction<int> reduceFunction, TopologyTypes topologyType = TopologyTypes.Flat);
/// <summary>
- /// Adds the Scatter MPI operator to the communication group.
+ /// Adds the Scatter Group Communication operator to the communication group.
/// </summary>
/// <typeparam name="TMessage">The type of messages that operators will send</typeparam>
/// <typeparam name="TMessageCodec">The codec used for serializing messages</typeparam>
@@ -119,7 +119,7 @@ namespace Org.Apache.REEF.Network.Group.Driver
ICommunicationGroupDriver AddScatter<TMessage, TMessageCodec>(string operatorName, string senderId, TopologyTypes topologyType = TopologyTypes.Flat) where TMessageCodec : ICodec<TMessage>;
/// <summary>
- /// Adds the Scatter MPI operator to the communication group with default Codec
+ /// Adds the Scatter Group Communication operator to the communication group with default Codec
/// </summary>
/// <param name="operatorName">The name of the scatter operator</param>
/// <param name="senderId">The sender id</param>
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/c330dcff/lang/cs/Org.Apache.REEF.Network/Group/Driver/IGroupCommDriver.cs
----------------------------------------------------------------------
diff --git a/lang/cs/Org.Apache.REEF.Network/Group/Driver/IGroupCommDriver.cs b/lang/cs/Org.Apache.REEF.Network/Group/Driver/IGroupCommDriver.cs
new file mode 100644
index 0000000..71b47b6
--- /dev/null
+++ b/lang/cs/Org.Apache.REEF.Network/Group/Driver/IGroupCommDriver.cs
@@ -0,0 +1,96 @@
+/**
+ * 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 Group Communication Operators.
+ /// Also manages configuration for Group Communication tasks/services.
+ /// </summary>
+ public interface IGroupCommDriver
+ {
+ /// <summary>
+ /// Returns the identifier for the master task
+ /// </summary>
+ string MasterTaskId { get; }
+
+ ICommunicationGroupDriver DefaultGroup { 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 Group Communication 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 Group Communication task configuration.
+ /// </summary>
+ /// <param name="taskId">The id of the task Configuration to generate</param>
+ /// <returns>The Group Communication task configuration with communication group and
+ /// operator configuration set.</returns>
+ IConfiguration GetGroupCommTaskConfiguration(string taskId);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/c330dcff/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
deleted file mode 100644
index 9c6eef2..0000000
--- a/lang/cs/Org.Apache.REEF.Network/Group/Driver/IMpiDriver.cs
+++ /dev/null
@@ -1,96 +0,0 @@
-/**
- * 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; }
-
- ICommunicationGroupDriver DefaultGroup { 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/c330dcff/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
index 065c158..c3923ae 100644
--- a/lang/cs/Org.Apache.REEF.Network/Group/Driver/Impl/CommunicationGroupDriver.cs
+++ b/lang/cs/Org.Apache.REEF.Network/Group/Driver/Impl/CommunicationGroupDriver.cs
@@ -37,7 +37,7 @@ using Org.Apache.REEF.Wake.Remote.Impl;
namespace Org.Apache.REEF.Network.Group.Driver.Impl
{
/// <summary>
- /// Used to configure MPI operators in Reef driver.
+ /// Used to configure Group Communication operators in Reef driver.
/// All operators in the same Communication Group run on the the
/// same set of tasks.
/// </summary>
@@ -168,7 +168,7 @@ namespace Org.Apache.REEF.Network.Group.Driver.Impl
}
/// <summary>
- /// Adds the Broadcast MPI operator to the communication group. Default to IntCodec
+ /// Adds the Broadcast Group Communication operator to the communication group. Default to IntCodec
/// </summary>
/// <param name="operatorName">The name of the broadcast operator</param>
/// <param name="masterTaskId">The master task id in broadcast operator</param>
@@ -181,7 +181,7 @@ namespace Org.Apache.REEF.Network.Group.Driver.Impl
}
/// <summary>
- /// Adds the Reduce MPI operator to the communication group.
+ /// Adds the Reduce Group Communication operator to the communication group.
/// </summary>
/// <typeparam name="TMessage">The type of messages that operators will send</typeparam>
/// <typeparam name="TMessageCodec">The codec used for serializing messages</typeparam>
@@ -226,7 +226,7 @@ namespace Org.Apache.REEF.Network.Group.Driver.Impl
}
/// <summary>
- /// Adds the Reduce MPI operator to the communication group.
+ /// Adds the Reduce Group Communication operator to the communication group.
/// </summary>
/// <typeparam name="TMessage">The type of messages that operators will send</typeparam>
/// <typeparam name="TMessageCodec">The codec used for serializing messages</typeparam>
@@ -270,7 +270,7 @@ namespace Org.Apache.REEF.Network.Group.Driver.Impl
}
/// <summary>
- /// Adds the Reduce MPI operator to the communication group with default IntCodec
+ /// Adds the Reduce Group Communication operator to the communication group with default IntCodec
/// </summary>
/// <param name="operatorName">The name of the reduce operator</param>
/// <param name="masterTaskId">The master task id for the typology</param>
@@ -287,7 +287,7 @@ namespace Org.Apache.REEF.Network.Group.Driver.Impl
}
/// <summary>
- /// Adds the Scatter MPI operator to the communication group.
+ /// Adds the Scatter Group Communication operator to the communication group.
/// </summary>
/// <typeparam name="TMessage">The type of messages that operators will send</typeparam>
/// <typeparam name="TMessageCodec">The codec used for serializing messages</typeparam>
@@ -322,7 +322,7 @@ namespace Org.Apache.REEF.Network.Group.Driver.Impl
}
/// <summary>
- /// Adds the Scatter MPI operator to the communication group with default IntCodec
+ /// Adds the Scatter Group Communication operator to the communication group with default IntCodec
/// </summary>
/// <param name="operatorName">The name of the scatter operator</param>
/// <param name="senderId">The sender id</param>
@@ -397,26 +397,26 @@ namespace Org.Apache.REEF.Network.Group.Driver.Impl
}
var confBuilder = TangFactory.GetTang().NewConfigurationBuilder()
- .BindNamedParameter<MpiConfigurationOptions.DriverId, string>(
- GenericType<MpiConfigurationOptions.DriverId>.Class,
+ .BindNamedParameter<GroupCommConfigurationOptions.DriverId, string>(
+ GenericType<GroupCommConfigurationOptions.DriverId>.Class,
_driverId)
- .BindNamedParameter<MpiConfigurationOptions.CommunicationGroupName, string>(
- GenericType<MpiConfigurationOptions.CommunicationGroupName>.Class,
+ .BindNamedParameter<GroupCommConfigurationOptions.CommunicationGroupName, string>(
+ GenericType<GroupCommConfigurationOptions.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,
+ .BindNamedParameter<GroupCommConfigurationOptions.DriverId, string>(
+ GenericType<GroupCommConfigurationOptions.DriverId>.Class,
_driverId)
- .BindNamedParameter<MpiConfigurationOptions.OperatorName, string>(
- GenericType<MpiConfigurationOptions.OperatorName>.Class,
+ .BindNamedParameter<GroupCommConfigurationOptions.OperatorName, string>(
+ GenericType<GroupCommConfigurationOptions.OperatorName>.Class,
operatorName)
.Build();
- confBuilder.BindSetEntry<MpiConfigurationOptions.SerializedOperatorConfigs, string>(
- GenericType<MpiConfigurationOptions.SerializedOperatorConfigs>.Class,
+ confBuilder.BindSetEntry<GroupCommConfigurationOptions.SerializedOperatorConfigs, string>(
+ GenericType<GroupCommConfigurationOptions.SerializedOperatorConfigs>.Class,
_confSerializer.ToString(innerConf));
}
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/c330dcff/lang/cs/Org.Apache.REEF.Network/Group/Driver/Impl/GroupCommDriver.cs
----------------------------------------------------------------------
diff --git a/lang/cs/Org.Apache.REEF.Network/Group/Driver/Impl/GroupCommDriver.cs b/lang/cs/Org.Apache.REEF.Network/Group/Driver/Impl/GroupCommDriver.cs
new file mode 100644
index 0000000..42394ab
--- /dev/null
+++ b/lang/cs/Org.Apache.REEF.Network/Group/Driver/Impl/GroupCommDriver.cs
@@ -0,0 +1,285 @@
+/**
+ * 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.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;
+
+namespace Org.Apache.REEF.Network.Group.Driver.Impl
+{
+ /// <summary>
+ /// Used to create Communication Groups for Group Communication Operators on the Reef driver.
+ /// Also manages configuration for Group Communication tasks/services.
+ /// </summary>
+ public class GroupCommDriver : IGroupCommDriver
+ {
+ private const string MasterTaskContextName = "MasterTaskContext";
+ private const string SlaveTaskContextName = "SlaveTaskContext";
+
+ private static Logger LOGGER = Logger.GetLogger(typeof(GroupCommDriver));
+
+ private readonly string _driverId;
+ private readonly string _nameServerAddr;
+ private readonly int _nameServerPort;
+ private int _contextIds;
+ private int _fanOut;
+ private string _groupName;
+
+ private readonly Dictionary<string, ICommunicationGroupDriver> _commGroups;
+ private readonly AvroConfigurationSerializer _configSerializer;
+ private readonly NameServer _nameServer;
+
+ /// <summary>
+ /// Create a new GroupCommunicationDriver object.
+ /// </summary>
+ /// <param name="driverId">Identifer for the REEF driver</param>
+ /// <param name="masterTaskId">Identifer for Group Communication master task</param>
+ /// <param name="fanOut">fanOut for tree topology</param>
+ /// <param name="configSerializer">Used to serialize task configuration</param>
+ [System.Obsolete("user the other constructor")]
+ [Inject]
+ public GroupCommDriver(
+ [Parameter(typeof(GroupCommConfigurationOptions.DriverId))] string driverId,
+ [Parameter(typeof(GroupCommConfigurationOptions.MasterTaskId))] string masterTaskId,
+ [Parameter(typeof(GroupCommConfigurationOptions.FanOut))] int fanOut,
+ AvroConfigurationSerializer configSerializer)
+ {
+ _driverId = driverId;
+ _contextIds = -1;
+ _fanOut = fanOut;
+ 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>
+ /// Create a new GroupCommunicationDriver object.
+ /// </summary>
+ /// <param name="driverId">Identifer for the REEF driver</param>
+ /// <param name="masterTaskId">Identifer for Group Communication master task</param>
+ /// <param name="fanOut">fanOut for tree topology</param>
+ /// <param name="groupName">default communication group name</param>
+ /// <param name="numberOfTasks">Number of tasks in the default group</param>
+ /// <param name="configSerializer">Used to serialize task configuration</param>
+ [Inject]
+ public GroupCommDriver(
+ [Parameter(typeof(GroupCommConfigurationOptions.DriverId))] string driverId,
+ [Parameter(typeof(GroupCommConfigurationOptions.MasterTaskId))] string masterTaskId,
+ [Parameter(typeof(GroupCommConfigurationOptions.FanOut))] int fanOut,
+ [Parameter(typeof(GroupCommConfigurationOptions.GroupName))] string groupName,
+ [Parameter(typeof(GroupCommConfigurationOptions.NumberOfTasks))] int numberOfTasks,
+ AvroConfigurationSerializer configSerializer)
+ {
+ _driverId = driverId;
+ _contextIds = -1;
+ _fanOut = fanOut;
+ MasterTaskId = masterTaskId;
+ _groupName = groupName;
+
+ _configSerializer = configSerializer;
+ _commGroups = new Dictionary<string, ICommunicationGroupDriver>();
+ _nameServer = new NameServer(0);
+
+ IPEndPoint localEndpoint = _nameServer.LocalEndpoint;
+ _nameServerAddr = localEndpoint.Address.ToString();
+ _nameServerPort = localEndpoint.Port;
+
+ NewCommunicationGroup(groupName, numberOfTasks);
+ }
+
+ /// <summary>
+ /// Returns the identifier for the master task
+ /// </summary>
+ public string MasterTaskId { get; private set; }
+
+ public ICommunicationGroupDriver DefaultGroup
+ {
+ get { return _commGroups[_groupName]; }
+ }
+
+ /// <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 GroupCommunicationDriver");
+ }
+
+ var commGroup = new CommunicationGroupDriver(groupName, _driverId, numTasks, _fanOut, _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 Group Communication 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<GroupCommNetworkObserver>.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))
+ .BindImplementation(GenericType<INameClient>.Class,
+ GenericType<NameClient>.Class)
+ .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 Group Communication task configuration.
+ /// </summary>
+ /// <param name="taskId">The id of the task Configuration to generate</param>
+ /// <returns>The Group Communication task configuration with communication group and
+ /// operator configuration set.</returns>
+ public IConfiguration GetGroupCommTaskConfiguration(string taskId)
+ {
+ var confBuilder = TangFactory.GetTang().NewConfigurationBuilder();
+
+ foreach (ICommunicationGroupDriver commGroup in _commGroups.Values)
+ {
+ var taskConf = commGroup.GetGroupTaskConfiguration(taskId);
+ if (taskConf != null)
+ {
+ confBuilder.BindSetEntry<GroupCommConfigurationOptions.SerializedGroupConfigs, string>(
+ GenericType<GroupCommConfigurationOptions.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/c330dcff/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
index 1439a36..dd67f8c 100644
--- a/lang/cs/Org.Apache.REEF.Network/Group/Driver/Impl/GroupCommunicationMessage.cs
+++ b/lang/cs/Org.Apache.REEF.Network/Group/Driver/Impl/GroupCommunicationMessage.cs
@@ -20,7 +20,7 @@
namespace Org.Apache.REEF.Network.Group.Driver.Impl
{
/// <summary>
- /// Messages sent by MPI Operators
+ /// Messages sent by Group Communication Operators
/// </summary>
public class GroupCommunicationMessage
{
@@ -28,7 +28,7 @@ namespace Org.Apache.REEF.Network.Group.Driver.Impl
/// 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="operatorName">The name of the Group Communication 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>
@@ -53,7 +53,7 @@ namespace Org.Apache.REEF.Network.Group.Driver.Impl
/// 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="operatorName">The name of the Group Communication 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>
@@ -80,7 +80,7 @@ namespace Org.Apache.REEF.Network.Group.Driver.Impl
public string GroupName { get; private set; }
/// <summary>
- /// Returns the MPI Operator name.
+ /// Returns the Group Communication Operator name.
/// </summary>
public string OperatorName { get; private set; }
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/c330dcff/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
index cd8ace2..26fabc0 100644
--- a/lang/cs/Org.Apache.REEF.Network/Group/Driver/Impl/MessageType.cs
+++ b/lang/cs/Org.Apache.REEF.Network/Group/Driver/Impl/MessageType.cs
@@ -20,7 +20,7 @@
namespace Org.Apache.REEF.Network.Group.Driver.Impl
{
/// <summary>
- /// Represents the different types of messages that Mpi Tasks can
+ /// Represents the different types of messages that Group Comm Tasks can
/// send to each other.
/// </summary>
public enum MessageType
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/c330dcff/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
deleted file mode 100644
index 77944a7..0000000
--- a/lang/cs/Org.Apache.REEF.Network/Group/Driver/Impl/MpiDriver.cs
+++ /dev/null
@@ -1,285 +0,0 @@
-/**
- * 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.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;
-
-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 readonly string _driverId;
- private readonly string _nameServerAddr;
- private readonly int _nameServerPort;
- private int _contextIds;
- private int _fanOut;
- private string _groupName;
-
- private readonly Dictionary<string, ICommunicationGroupDriver> _commGroups;
- private readonly AvroConfigurationSerializer _configSerializer;
- private readonly 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="fanOut">fanOut for tree topology</param>
- /// <param name="configSerializer">Used to serialize task configuration</param>
- [System.Obsolete("user the other constructor")]
- [Inject]
- public MpiDriver(
- [Parameter(typeof(MpiConfigurationOptions.DriverId))] string driverId,
- [Parameter(typeof(MpiConfigurationOptions.MasterTaskId))] string masterTaskId,
- [Parameter(typeof(MpiConfigurationOptions.FanOut))] int fanOut,
- AvroConfigurationSerializer configSerializer)
- {
- _driverId = driverId;
- _contextIds = -1;
- _fanOut = fanOut;
- 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>
- /// 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="fanOut">fanOut for tree topology</param>
- /// <param name="groupName">default communication group name</param>
- /// <param name="numberOfTasks">Number of tasks in the default group</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,
- [Parameter(typeof(MpiConfigurationOptions.FanOut))] int fanOut,
- [Parameter(typeof(MpiConfigurationOptions.GroupName))] string groupName,
- [Parameter(typeof(MpiConfigurationOptions.NumberOfTasks))] int numberOfTasks,
- AvroConfigurationSerializer configSerializer)
- {
- _driverId = driverId;
- _contextIds = -1;
- _fanOut = fanOut;
- MasterTaskId = masterTaskId;
- _groupName = groupName;
-
- _configSerializer = configSerializer;
- _commGroups = new Dictionary<string, ICommunicationGroupDriver>();
- _nameServer = new NameServer(0);
-
- IPEndPoint localEndpoint = _nameServer.LocalEndpoint;
- _nameServerAddr = localEndpoint.Address.ToString();
- _nameServerPort = localEndpoint.Port;
-
- NewCommunicationGroup(groupName, numberOfTasks);
- }
-
- /// <summary>
- /// Returns the identifier for the master task
- /// </summary>
- public string MasterTaskId { get; private set; }
-
- public ICommunicationGroupDriver DefaultGroup
- {
- get { return _commGroups[_groupName]; }
- }
-
- /// <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, _fanOut, _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))
- .BindImplementation(GenericType<INameClient>.Class,
- GenericType<NameClient>.Class)
- .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/c330dcff/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
index ab947f0..bad2bb6 100644
--- a/lang/cs/Org.Apache.REEF.Network/Group/Driver/Impl/TaskStarter.cs
+++ b/lang/cs/Org.Apache.REEF.Network/Group/Driver/Impl/TaskStarter.cs
@@ -30,7 +30,7 @@ using Org.Apache.REEF.Utilities.Logging;
namespace Org.Apache.REEF.Network.Group.Driver.Impl
{
/// <summary>
- /// Helper class to start MPI tasks.
+ /// Helper class to start Group Communication tasks.
/// </summary>
public class TaskStarter
{
@@ -41,7 +41,7 @@ namespace Org.Apache.REEF.Network.Group.Driver.Impl
private int _tasksAdded;
private readonly string _masterTaskId;
- private readonly IMpiDriver _mpiDriver;
+ private readonly IGroupCommDriver _groupCommDriver;
private readonly List<Tuple<string, IConfiguration, IActiveContext>> _taskTuples;
/// <summary>
@@ -49,18 +49,18 @@ namespace Org.Apache.REEF.Network.Group.Driver.Impl
/// 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="groupCommDriver">The IGroupCommuDriver for the Group Communication 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)
+ public TaskStarter(IGroupCommDriver groupCommDriver, int numTasks)
{
LOGGER.Log(Level.Verbose, "Creating TaskStarter");
- _masterTaskId = mpiDriver.MasterTaskId;
+ _masterTaskId = groupCommDriver.MasterTaskId;
_numTasks = numTasks;
_tasksAdded = 0;
_lock = new object();
- _mpiDriver = mpiDriver;
+ _groupCommDriver = groupCommDriver;
_taskTuples = new List<Tuple<string, IConfiguration, IActiveContext>>();
}
@@ -127,8 +127,8 @@ namespace Org.Apache.REEF.Network.Group.Driver.Impl
IConfiguration userPartialTaskConf,
IActiveContext activeContext)
{
- IConfiguration mpiTaskConfiguration = _mpiDriver.GetMpiTaskConfiguration(taskId);
- IConfiguration mergedTaskConf = Configurations.Merge(userPartialTaskConf, mpiTaskConfiguration);
+ IConfiguration groupCommTaskConfiguration = _groupCommDriver.GetGroupCommTaskConfiguration(taskId);
+ IConfiguration mergedTaskConf = Configurations.Merge(userPartialTaskConf, groupCommTaskConfiguration);
activeContext.SubmitTask(mergedTaskConf);
}
}
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/c330dcff/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
index 6e2fc68..4ea9579 100644
--- a/lang/cs/Org.Apache.REEF.Network/Group/Operators/IBroadcastReceiver.cs
+++ b/lang/cs/Org.Apache.REEF.Network/Group/Operators/IBroadcastReceiver.cs
@@ -20,10 +20,10 @@
namespace Org.Apache.REEF.Network.Group.Operators
{
/// <summary>
- /// MPI Operator used to receive broadcast messages.
+ /// Group Communication Operator used to receive broadcast messages.
/// </summary>
/// <typeparam name="T">The type of message being sent.</typeparam>
- public interface IBroadcastReceiver<T> : IMpiOperator<T>
+ public interface IBroadcastReceiver<T> : IGroupCommOperator<T>
{
/// <summary>
/// Receive a message from parent BroadcastSender.
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/c330dcff/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
index f62702d..826897e 100644
--- a/lang/cs/Org.Apache.REEF.Network/Group/Operators/IBroadcastSender.cs
+++ b/lang/cs/Org.Apache.REEF.Network/Group/Operators/IBroadcastSender.cs
@@ -20,10 +20,10 @@
namespace Org.Apache.REEF.Network.Group.Operators
{
/// <summary>
- /// MPI Operator used to send messages to child Tasks.
+ /// Group Communication Operator used to send messages to child Tasks.
/// </summary>
/// <typeparam name="T">The message type</typeparam>
- public interface IBroadcastSender<T> : IMpiOperator<T>
+ public interface IBroadcastSender<T> : IGroupCommOperator<T>
{
/// <summary>
/// Send the data to all BroadcastReceivers.
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/c330dcff/lang/cs/Org.Apache.REEF.Network/Group/Operators/IGroupCommOperator.cs
----------------------------------------------------------------------
diff --git a/lang/cs/Org.Apache.REEF.Network/Group/Operators/IGroupCommOperator.cs b/lang/cs/Org.Apache.REEF.Network/Group/Operators/IGroupCommOperator.cs
new file mode 100644
index 0000000..c779333
--- /dev/null
+++ b/lang/cs/Org.Apache.REEF.Network/Group/Operators/IGroupCommOperator.cs
@@ -0,0 +1,43 @@
+/**
+ * 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.Operators
+{
+ /// <summary>
+ /// An Group Communication Operator to be used in a Reef Task.
+ /// </summary>
+ /// <typeparam name="T">The message type</typeparam>
+ public interface IGroupCommOperator<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/c330dcff/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
deleted file mode 100644
index 58ad1b2..0000000
--- a/lang/cs/Org.Apache.REEF.Network/Group/Operators/IMpiOperator.cs
+++ /dev/null
@@ -1,43 +0,0 @@
-/**
- * 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.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/c330dcff/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
index 64f366a..27a164d 100644
--- a/lang/cs/Org.Apache.REEF.Network/Group/Operators/IReduceReceiver.cs
+++ b/lang/cs/Org.Apache.REEF.Network/Group/Operators/IReduceReceiver.cs
@@ -20,10 +20,10 @@
namespace Org.Apache.REEF.Network.Group.Operators
{
/// <summary>
- /// MPI operator used to receive and reduce messages.
+ /// Group Communication operator used to receive and reduce messages.
/// </summary>
/// <typeparam name="T">The message type</typeparam>
- public interface IReduceReceiver<T> : IMpiOperator<T>
+ public interface IReduceReceiver<T> : IGroupCommOperator<T>
{
/// <summary>
/// Returns the class used to reduce incoming messages sent by ReduceSenders.
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/c330dcff/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
index 97990bd..9d0106b 100644
--- a/lang/cs/Org.Apache.REEF.Network/Group/Operators/IReduceSender.cs
+++ b/lang/cs/Org.Apache.REEF.Network/Group/Operators/IReduceSender.cs
@@ -20,10 +20,10 @@
namespace Org.Apache.REEF.Network.Group.Operators
{
/// <summary>
- /// MPI Operator used to send messages to be reduced by the ReduceReceiver.
+ /// Group Communication 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>
+ public interface IReduceSender<T> : IGroupCommOperator<T>
{
/// <summary>
/// Get reduced data from children, reduce with the data given, then sends reduced data to parent
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/c330dcff/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
index 204ea88..f62ed3b 100644
--- a/lang/cs/Org.Apache.REEF.Network/Group/Operators/IScatterReceiver.cs
+++ b/lang/cs/Org.Apache.REEF.Network/Group/Operators/IScatterReceiver.cs
@@ -22,11 +22,11 @@ using System.Collections.Generic;
namespace Org.Apache.REEF.Network.Group.Operators
{
/// <summary>
- /// MPI operator used to receive a sublist of messages sent
+ /// Group Communication 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>
+ public interface IScatterReceiver<T> : IGroupCommOperator<T>
{
/// <summary>
/// Receive a sublist of messages sent from the IScatterSender.
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/c330dcff/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
index 2b020a5..ce78a6b 100644
--- a/lang/cs/Org.Apache.REEF.Network/Group/Operators/IScatterSender.cs
+++ b/lang/cs/Org.Apache.REEF.Network/Group/Operators/IScatterSender.cs
@@ -22,11 +22,11 @@ using System.Collections.Generic;
namespace Org.Apache.REEF.Network.Group.Operators
{
/// <summary>
- /// MPI operator used to scatter a list of elements to all
+ /// Group Communication 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>
+ public interface IScatterSender<T> : IGroupCommOperator<T>
{
/// <summary>
/// Split up the list of elements evenly and scatter each chunk
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/c330dcff/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
index b8b2a5a..75ab88e 100644
--- a/lang/cs/Org.Apache.REEF.Network/Group/Operators/Impl/BroadcastReceiver.cs
+++ b/lang/cs/Org.Apache.REEF.Network/Group/Operators/Impl/BroadcastReceiver.cs
@@ -30,7 +30,7 @@ using Org.Apache.REEF.Utilities.Logging;
namespace Org.Apache.REEF.Network.Group.Operators.Impl
{
/// <summary>
- /// MPI Operator used to receive broadcast messages in pipelined fashion.
+ /// Group Communication Operator used to receive broadcast messages in pipelined fashion.
/// </summary>
/// <typeparam name="T">The type of message being sent.</typeparam>
public class BroadcastReceiver<T> : IBroadcastReceiver<T>
@@ -51,8 +51,8 @@ namespace Org.Apache.REEF.Network.Group.Operators.Impl
/// message to pipelined ones and vice versa.</param>
[Inject]
public BroadcastReceiver(
- [Parameter(typeof(MpiConfigurationOptions.OperatorName))] string operatorName,
- [Parameter(typeof(MpiConfigurationOptions.CommunicationGroupName))] string groupName,
+ [Parameter(typeof(GroupCommConfigurationOptions.OperatorName))] string operatorName,
+ [Parameter(typeof(GroupCommConfigurationOptions.CommunicationGroupName))] string groupName,
OperatorTopology<PipelineMessage<T>> topology,
ICommunicationGroupNetworkObserver networkHandler,
IPipelineDataConverter<T> dataConverter)
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/c330dcff/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
index dc0142b..21701ea 100644
--- a/lang/cs/Org.Apache.REEF.Network/Group/Operators/Impl/BroadcastSender.cs
+++ b/lang/cs/Org.Apache.REEF.Network/Group/Operators/Impl/BroadcastSender.cs
@@ -30,7 +30,7 @@ using Org.Apache.REEF.Utilities.Logging;
namespace Org.Apache.REEF.Network.Group.Operators.Impl
{
/// <summary>
- /// MPI Operator used to send messages to child Tasks in pipelined fashion.
+ /// Group Communication Operator used to send messages to child Tasks in pipelined fashion.
/// </summary>
/// <typeparam name="T">The message type</typeparam>
public class BroadcastSender<T> : IBroadcastSender<T>
@@ -51,8 +51,8 @@ namespace Org.Apache.REEF.Network.Group.Operators.Impl
/// message to pipelined ones and vice versa.</param>
[Inject]
public BroadcastSender(
- [Parameter(typeof(MpiConfigurationOptions.OperatorName))] string operatorName,
- [Parameter(typeof(MpiConfigurationOptions.CommunicationGroupName))] string groupName,
+ [Parameter(typeof(GroupCommConfigurationOptions.OperatorName))] string operatorName,
+ [Parameter(typeof(GroupCommConfigurationOptions.CommunicationGroupName))] string groupName,
OperatorTopology<PipelineMessage<T>> topology,
ICommunicationGroupNetworkObserver networkHandler,
IPipelineDataConverter<T> dataConverter)
@@ -71,7 +71,7 @@ namespace Org.Apache.REEF.Network.Group.Operators.Impl
}
/// <summary>
- /// Returns the identifier for the MPI operator.
+ /// Returns the identifier for the Group Communication operator.
/// </summary>
public string OperatorName { get; private set; }
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/c330dcff/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
index bf60841..dbfd1c7 100644
--- a/lang/cs/Org.Apache.REEF.Network/Group/Operators/Impl/ReduceOperatorSpec.cs
+++ b/lang/cs/Org.Apache.REEF.Network/Group/Operators/Impl/ReduceOperatorSpec.cs
@@ -25,7 +25,7 @@ using Org.Apache.REEF.Network.Group.Pipelining;
namespace Org.Apache.REEF.Network.Group.Operators.Impl
{
/// <summary>
- /// The specification used to define Reduce MPI Operators.
+ /// The specification used to define Reduce Group Communication Operators.
/// </summary>
public class ReduceOperatorSpec<T1, T2> : IOperatorSpec<T1, T2> where T2 : ICodec<T1>
{
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/c330dcff/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
index 70ed1ae..2242368 100644
--- a/lang/cs/Org.Apache.REEF.Network/Group/Operators/Impl/ReduceReceiver.cs
+++ b/lang/cs/Org.Apache.REEF.Network/Group/Operators/Impl/ReduceReceiver.cs
@@ -30,7 +30,7 @@ using Org.Apache.REEF.Utilities.Logging;
namespace Org.Apache.REEF.Network.Group.Operators.Impl
{
/// <summary>
- /// MPI operator used to receive and reduce messages in pipelined fashion.
+ /// Group Communication operator used to receive and reduce messages in pipelined fashion.
/// </summary>
/// <typeparam name="T">The message type</typeparam>
public class ReduceReceiver<T> : IReduceReceiver<T>
@@ -52,8 +52,8 @@ namespace Org.Apache.REEF.Network.Group.Operators.Impl
/// message to pipelined ones and vice versa.</param>
[Inject]
public ReduceReceiver(
- [Parameter(typeof (MpiConfigurationOptions.OperatorName))] string operatorName,
- [Parameter(typeof (MpiConfigurationOptions.CommunicationGroupName))] string groupName,
+ [Parameter(typeof (GroupCommConfigurationOptions.OperatorName))] string operatorName,
+ [Parameter(typeof (GroupCommConfigurationOptions.CommunicationGroupName))] string groupName,
OperatorTopology<PipelineMessage<T>> topology,
ICommunicationGroupNetworkObserver networkHandler,
IReduceFunction<T> reduceFunction,
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/c330dcff/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
index 4d73e04..d61657f 100644
--- a/lang/cs/Org.Apache.REEF.Network/Group/Operators/Impl/ReduceSender.cs
+++ b/lang/cs/Org.Apache.REEF.Network/Group/Operators/Impl/ReduceSender.cs
@@ -31,7 +31,7 @@ using Org.Apache.REEF.Utilities.Logging;
namespace Org.Apache.REEF.Network.Group.Operators.Impl
{
/// <summary>
- /// MPI Operator used to send messages to be reduced by the ReduceReceiver in pipelined fashion.
+ /// Group Communication Operator used to send messages to be reduced by the ReduceReceiver in pipelined fashion.
/// </summary>
/// <typeparam name="T">The message type</typeparam>
public class ReduceSender<T> : IReduceSender<T>
@@ -53,8 +53,8 @@ namespace Org.Apache.REEF.Network.Group.Operators.Impl
/// message to pipelined ones and vice versa.</param>
[Inject]
public ReduceSender(
- [Parameter(typeof(MpiConfigurationOptions.OperatorName))] string operatorName,
- [Parameter(typeof(MpiConfigurationOptions.CommunicationGroupName))] string groupName,
+ [Parameter(typeof(GroupCommConfigurationOptions.OperatorName))] string operatorName,
+ [Parameter(typeof(GroupCommConfigurationOptions.CommunicationGroupName))] string groupName,
OperatorTopology<PipelineMessage<T>> topology,
ICommunicationGroupNetworkObserver networkHandler,
IReduceFunction<T> reduceFunction,
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/c330dcff/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
index 5961615..7807f3b 100644
--- a/lang/cs/Org.Apache.REEF.Network/Group/Operators/Impl/ScatterOperatorSpec.cs
+++ b/lang/cs/Org.Apache.REEF.Network/Group/Operators/Impl/ScatterOperatorSpec.cs
@@ -24,7 +24,7 @@ using Org.Apache.REEF.Network.Group.Pipelining;
namespace Org.Apache.REEF.Network.Group.Operators.Impl
{
/// <summary>
- /// The specification used to define Scatter MPI Operators.
+ /// The specification used to define Scatter Group Communication Operators.
/// </summary>
public class ScatterOperatorSpec<T1, T2> : IOperatorSpec<T1, T2> where T2 : ICodec<T1>
{
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/c330dcff/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
index 6a9207d..b40ff68 100644
--- a/lang/cs/Org.Apache.REEF.Network/Group/Operators/Impl/ScatterReceiver.cs
+++ b/lang/cs/Org.Apache.REEF.Network/Group/Operators/Impl/ScatterReceiver.cs
@@ -28,7 +28,7 @@ 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
+ /// Group Communication operator used to receive a sublist of messages sent
/// from the IScatterSender.
/// </summary>
/// <typeparam name="T">The message type</typeparam>
@@ -48,8 +48,8 @@ namespace Org.Apache.REEF.Network.Group.Operators.Impl
/// <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,
+ [Parameter(typeof(GroupCommConfigurationOptions.OperatorName))] string operatorName,
+ [Parameter(typeof(GroupCommConfigurationOptions.CommunicationGroupName))] string groupName,
OperatorTopology<T> topology,
ICommunicationGroupNetworkObserver networkHandler)
{
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/c330dcff/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
index 6b4852a..2c664b8 100644
--- a/lang/cs/Org.Apache.REEF.Network/Group/Operators/Impl/ScatterSender.cs
+++ b/lang/cs/Org.Apache.REEF.Network/Group/Operators/Impl/ScatterSender.cs
@@ -28,7 +28,7 @@ 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
+ /// Group Communication operator used to scatter a list of elements to all
/// of the IScatterReceivers.
/// </summary>
/// <typeparam name="T">The message type</typeparam>
@@ -48,8 +48,8 @@ namespace Org.Apache.REEF.Network.Group.Operators.Impl
/// <param name="networkHandler">The network handler</param>
[Inject]
public ScatterSender(
- [Parameter(typeof(MpiConfigurationOptions.OperatorName))] string operatorName,
- [Parameter(typeof(MpiConfigurationOptions.CommunicationGroupName))] string groupName,
+ [Parameter(typeof(GroupCommConfigurationOptions.OperatorName))] string operatorName,
+ [Parameter(typeof(GroupCommConfigurationOptions.CommunicationGroupName))] string groupName,
OperatorTopology<T> topology,
ICommunicationGroupNetworkObserver networkHandler)
{
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/c330dcff/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
index 2f03a7a..7b37c07 100644
--- a/lang/cs/Org.Apache.REEF.Network/Group/Operators/Impl/Sender.cs
+++ b/lang/cs/Org.Apache.REEF.Network/Group/Operators/Impl/Sender.cs
@@ -26,7 +26,7 @@ using Org.Apache.REEF.Wake;
namespace Org.Apache.REEF.Network.Group.Operators.Impl
{
/// <summary>
- /// MPI operator used to do point-to-point communication between named Tasks.
+ /// Group Communication operator used to do point-to-point communication between named Tasks.
/// </summary>
public class Sender
{
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/c330dcff/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
index 3630070..4f83e2f 100644
--- a/lang/cs/Org.Apache.REEF.Network/Group/Task/ICommunicationGroupClient.cs
+++ b/lang/cs/Org.Apache.REEF.Network/Group/Task/ICommunicationGroupClient.cs
@@ -24,7 +24,7 @@ using Org.Apache.REEF.Tang.Annotations;
namespace Org.Apache.REEF.Network.Group.Task
{
/// <summary>
- /// Used by Tasks to fetch MPI Operators in the group configured by the driver.
+ /// Used by Tasks to fetch Group Communication Operators in the group configured by the driver.
/// </summary>
[DefaultImplementation(typeof(CommunicationGroupClient))]
public interface ICommunicationGroupClient
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/c330dcff/lang/cs/Org.Apache.REEF.Network/Group/Task/IGroupCommClient.cs
----------------------------------------------------------------------
diff --git a/lang/cs/Org.Apache.REEF.Network/Group/Task/IGroupCommClient.cs b/lang/cs/Org.Apache.REEF.Network/Group/Task/IGroupCommClient.cs
new file mode 100644
index 0000000..82264bd
--- /dev/null
+++ b/lang/cs/Org.Apache.REEF.Network/Group/Task/IGroupCommClient.cs
@@ -0,0 +1,39 @@
+/**
+ * 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.Task.Impl;
+using Org.Apache.REEF.Tang.Annotations;
+
+namespace Org.Apache.REEF.Network.Group.Task
+{
+ /// <summary>
+ /// Used by Tasks to fetch CommunicationGroupClients.
+ /// </summary>
+ [DefaultImplementation(typeof(GroupCommClient))]
+ public interface IGroupCommClient : 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/c330dcff/lang/cs/Org.Apache.REEF.Network/Group/Task/IGroupCommNetworkObserver.cs
----------------------------------------------------------------------
diff --git a/lang/cs/Org.Apache.REEF.Network/Group/Task/IGroupCommNetworkObserver.cs b/lang/cs/Org.Apache.REEF.Network/Group/Task/IGroupCommNetworkObserver.cs
new file mode 100644
index 0000000..8416cca
--- /dev/null
+++ b/lang/cs/Org.Apache.REEF.Network/Group/Task/IGroupCommNetworkObserver.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;
+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;
+
+namespace Org.Apache.REEF.Network.Group.Task
+{
+ /// <summary>
+ /// Handles all incoming messages for this Task.
+ /// </summary>
+ [DefaultImplementation(typeof(GroupCommNetworkObserver))]
+ public interface IGroupCommNetworkObserver : 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/c330dcff/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
deleted file mode 100644
index 159605d..0000000
--- a/lang/cs/Org.Apache.REEF.Network/Group/Task/IMpiClient.cs
+++ /dev/null
@@ -1,39 +0,0 @@
-/**
- * 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.Task.Impl;
-using Org.Apache.REEF.Tang.Annotations;
-
-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/c330dcff/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
deleted file mode 100644
index f0c1b3a..0000000
--- a/lang/cs/Org.Apache.REEF.Network/Group/Task/IMpiNetworkObserver.cs
+++ /dev/null
@@ -1,44 +0,0 @@
-/**
- * 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.Impl;
-using Org.Apache.REEF.Network.Group.Task.Impl;
-using Org.Apache.REEF.Network.NetworkService;
-using Org.Apache.REEF.Tang.Annotations;
-
-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);
- }
-}