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:29 UTC
[1/3] incubator-reef git commit: [REEF-256] Rename MPI to Group
Communication
Repository: incubator-reef
Updated Branches:
refs/heads/master e1be758b5 -> c330dcff5
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/c330dcff/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
index a4d9e70..3fcdc2f 100644
--- a/lang/cs/Org.Apache.REEF.Network/Group/Task/Impl/CommunicationGroupClient.cs
+++ b/lang/cs/Org.Apache.REEF.Network/Group/Task/Impl/CommunicationGroupClient.cs
@@ -36,7 +36,7 @@ using Org.Apache.REEF.Utilities.Logging;
namespace Org.Apache.REEF.Network.Group.Task.Impl
{
/// <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>
public class CommunicationGroupClient : ICommunicationGroupClient
{
@@ -48,7 +48,7 @@ namespace Org.Apache.REEF.Network.Group.Task.Impl
private readonly Dictionary<string, IInjector> _operatorInjectors;
private readonly Dictionary<string, object> _operators;
private readonly NetworkService<GroupCommunicationMessage> _networkService;
- private readonly IMpiNetworkObserver _mpiNetworkHandler;
+ private readonly IGroupCommNetworkObserver _groupCommNetworkHandler;
private readonly ICommunicationGroupNetworkObserver _commGroupNetworkHandler;
/// <summary>
@@ -58,17 +58,17 @@ namespace Org.Apache.REEF.Network.Group.Task.Impl
/// <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
+ /// <param name="groupCommNetworkObserver">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,
+ [Parameter(typeof(GroupCommConfigurationOptions.CommunicationGroupName))] string groupName,
+ [Parameter(typeof(GroupCommConfigurationOptions.DriverId))] string driverId,
+ [Parameter(typeof(GroupCommConfigurationOptions.SerializedOperatorConfigs))] ISet<string> operatorConfigs,
+ IGroupCommNetworkObserver groupCommNetworkObserver,
NetworkService<GroupCommunicationMessage> networkService,
AvroConfigurationSerializer configSerializer,
CommunicationGroupNetworkObserver commGroupNetworkHandler)
@@ -81,20 +81,20 @@ namespace Org.Apache.REEF.Network.Group.Task.Impl
_operatorInjectors = new Dictionary<string, IInjector>();
_networkService = networkService;
- _mpiNetworkHandler = mpiNetworkObserver;
+ _groupCommNetworkHandler = groupCommNetworkObserver;
_commGroupNetworkHandler = commGroupNetworkHandler;
- _mpiNetworkHandler.Register(groupName, _commGroupNetworkHandler);
+ _groupCommNetworkHandler.Register(groupName, _commGroupNetworkHandler);
// Deserialize operator configuration and store each injector.
- // When user requests the MPI Operator, use type information to
+ // When user requests the Group Communication 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);
+ string operatorName = injector.GetNamedInstance<GroupCommConfigurationOptions.OperatorName, string>(
+ GenericType<GroupCommConfigurationOptions.OperatorName>.Class);
_operatorInjectors[operatorName] = injector;
}
}
@@ -171,14 +171,14 @@ namespace Org.Apache.REEF.Network.Group.Task.Impl
}
/// <summary>
- /// Gets the MPI operator with the specified name and type.
+ /// Gets the Group Communication 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>
+ /// <returns>The newly created Group Communication Operator</returns>
private T GetOperatorInstance<T>(string operatorName) where T : class
{
if (string.IsNullOrEmpty(operatorName))
@@ -196,7 +196,7 @@ namespace Org.Apache.REEF.Network.Group.Task.Impl
IInjector injector = _operatorInjectors[operatorName];
injector.BindVolatileParameter(GenericType<TaskConfigurationOptions.Identifier>.Class, _taskId);
- injector.BindVolatileParameter(GenericType<MpiConfigurationOptions.CommunicationGroupName>.Class, GroupName);
+ injector.BindVolatileParameter(GenericType<GroupCommConfigurationOptions.CommunicationGroupName>.Class, GroupName);
injector.BindVolatileInstance(GenericType<ICommunicationGroupNetworkObserver>.Class, _commGroupNetworkHandler);
injector.BindVolatileInstance(GenericType<NetworkService<GroupCommunicationMessage>>.Class, _networkService);
injector.BindVolatileInstance(GenericType<ICommunicationGroupClient>.Class, this);
@@ -208,7 +208,7 @@ namespace Org.Apache.REEF.Network.Group.Task.Impl
}
catch (InjectionException)
{
- LOGGER.Log(Level.Error, "Cannot inject MPI operator: No known operator of type: {0}", typeof(T));
+ LOGGER.Log(Level.Error, "Cannot inject Group Communication operator: No known operator of type: {0}", typeof(T));
throw;
}
}
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/c330dcff/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
index 20f17ec..444c4a1 100644
--- a/lang/cs/Org.Apache.REEF.Network/Group/Task/Impl/CommunicationGroupNetworkObserver.cs
+++ b/lang/cs/Org.Apache.REEF.Network/Group/Task/Impl/CommunicationGroupNetworkObserver.cs
@@ -43,8 +43,8 @@ namespace Org.Apache.REEF.Network.Group.Task.Impl
/// </summary>
[Inject]
public CommunicationGroupNetworkObserver(
- [Parameter(typeof(MpiConfigurationOptions.RetryCountWaitingForHanler))] int retryCount,
- [Parameter(typeof(MpiConfigurationOptions.SleepTimeWaitingForHandler))] int sleepTime)
+ [Parameter(typeof(GroupCommConfigurationOptions.RetryCountWaitingForHanler))] int retryCount,
+ [Parameter(typeof(GroupCommConfigurationOptions.SleepTimeWaitingForHandler))] int sleepTime)
{
_handlers = new Dictionary<string, IObserver<GroupCommunicationMessage>>();
_retryCount = retryCount;
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/c330dcff/lang/cs/Org.Apache.REEF.Network/Group/Task/Impl/GroupCommClient.cs
----------------------------------------------------------------------
diff --git a/lang/cs/Org.Apache.REEF.Network/Group/Task/Impl/GroupCommClient.cs b/lang/cs/Org.Apache.REEF.Network/Group/Task/Impl/GroupCommClient.cs
new file mode 100644
index 0000000..4f0f283
--- /dev/null
+++ b/lang/cs/Org.Apache.REEF.Network/Group/Task/Impl/GroupCommClient.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.
+ */
+
+using System;
+using System.Collections.Generic;
+using Org.Apache.REEF.Common.Tasks;
+using Org.Apache.REEF.Network.Group.Config;
+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.Implementations.Tang;
+using Org.Apache.REEF.Tang.Interface;
+using Org.Apache.REEF.Tang.Util;
+using Org.Apache.REEF.Wake.Remote.Impl;
+
+namespace Org.Apache.REEF.Network.Group.Task.Impl
+{
+ /// <summary>
+ /// Used by Tasks to fetch CommunicationGroupClients.
+ /// </summary>
+ public class GroupCommClient : IGroupCommClient
+ {
+ private readonly Dictionary<string, ICommunicationGroupClient> _commGroups;
+
+ private readonly INetworkService<GroupCommunicationMessage> _networkService;
+
+ /// <summary>
+ /// Creates a new GroupCommClient 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="groupCommNetworkObserver">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 GroupCommClient(
+ [Parameter(typeof(GroupCommConfigurationOptions.SerializedGroupConfigs))] ISet<string> groupConfigs,
+ [Parameter(typeof(TaskConfigurationOptions.Identifier))] string taskId,
+ IGroupCommNetworkObserver groupCommNetworkObserver,
+ 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<IGroupCommNetworkObserver>.Class, groupCommNetworkObserver);
+ 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 GroupCommClient's services.
+ /// </summary>
+ public void Dispose()
+ {
+ _networkService.Unregister();
+ _networkService.Dispose();
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/c330dcff/lang/cs/Org.Apache.REEF.Network/Group/Task/Impl/GroupCommNetworkObserver.cs
----------------------------------------------------------------------
diff --git a/lang/cs/Org.Apache.REEF.Network/Group/Task/Impl/GroupCommNetworkObserver.cs b/lang/cs/Org.Apache.REEF.Network/Group/Task/Impl/GroupCommNetworkObserver.cs
new file mode 100644
index 0000000..5604885
--- /dev/null
+++ b/lang/cs/Org.Apache.REEF.Network/Group/Task/Impl/GroupCommNetworkObserver.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 System;
+using System.Collections.Generic;
+using System.Linq;
+using Org.Apache.REEF.Network.Group.Driver.Impl;
+using Org.Apache.REEF.Network.NetworkService;
+using Org.Apache.REEF.Tang.Annotations;
+using Org.Apache.REEF.Utilities.Logging;
+
+namespace Org.Apache.REEF.Network.Group.Task.Impl
+{
+ /// <summary>
+ /// Handles all incoming messages for this Task.
+ /// </summary>
+ public class GroupCommNetworkObserver : IGroupCommNetworkObserver
+ {
+ private static readonly Logger LOGGER = Logger.GetLogger(typeof(GroupCommNetworkObserver));
+
+ private readonly Dictionary<string, IObserver<GroupCommunicationMessage>> _commGroupHandlers;
+
+ /// <summary>
+ /// Creates a new GroupCommNetworkObserver.
+ /// </summary>
+ [Inject]
+ public GroupCommNetworkObserver()
+ {
+ _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, "Group Communication Network Handler received message with no data");
+ throw;
+ }
+ catch (KeyNotFoundException)
+ {
+ LOGGER.Log(Level.Error, "Group Communication 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/c330dcff/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
deleted file mode 100644
index 97a43ea..0000000
--- a/lang/cs/Org.Apache.REEF.Network/Group/Task/Impl/MpiClient.cs
+++ /dev/null
@@ -1,107 +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 Org.Apache.REEF.Common.Tasks;
-using Org.Apache.REEF.Network.Group.Config;
-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.Implementations.Tang;
-using Org.Apache.REEF.Tang.Interface;
-using Org.Apache.REEF.Tang.Util;
-using Org.Apache.REEF.Wake.Remote.Impl;
-
-namespace Org.Apache.REEF.Network.Group.Task.Impl
-{
- /// <summary>
- /// Used by Tasks to fetch CommunicationGroupClients.
- /// </summary>
- public class MpiClient : IMpiClient
- {
- private readonly Dictionary<string, ICommunicationGroupClient> _commGroups;
-
- private readonly 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/c330dcff/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
deleted file mode 100644
index 10a2ba5..0000000
--- a/lang/cs/Org.Apache.REEF.Network/Group/Task/Impl/MpiNetworkObserver.cs
+++ /dev/null
@@ -1,108 +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.Linq;
-using Org.Apache.REEF.Network.Group.Driver.Impl;
-using Org.Apache.REEF.Network.NetworkService;
-using Org.Apache.REEF.Tang.Annotations;
-using Org.Apache.REEF.Utilities.Logging;
-
-namespace Org.Apache.REEF.Network.Group.Task.Impl
-{
- /// <summary>
- /// Handles all incoming messages for this Task.
- /// </summary>
- public class MpiNetworkObserver : IMpiNetworkObserver
- {
- private static readonly Logger LOGGER = Logger.GetLogger(typeof(MpiNetworkObserver));
-
- private readonly 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/c330dcff/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
index 2e98d3d..315cf64 100644
--- a/lang/cs/Org.Apache.REEF.Network/Group/Task/Impl/OperatorTopology.cs
+++ b/lang/cs/Org.Apache.REEF.Network/Group/Task/Impl/OperatorTopology.cs
@@ -68,7 +68,7 @@ namespace Org.Apache.REEF.Network.Group.Task.Impl
/// <summary>
/// Creates a new OperatorTopology object.
/// </summary>
- /// <param name="operatorName">The name of the MPI Operator</param>
+ /// <param name="operatorName">The name of the Group Communication 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>
@@ -79,14 +79,14 @@ namespace Org.Apache.REEF.Network.Group.Task.Impl
/// <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(GroupCommConfigurationOptions.OperatorName))] string operatorName,
+ [Parameter(typeof(GroupCommConfigurationOptions.CommunicationGroupName))] string groupName,
[Parameter(typeof(TaskConfigurationOptions.Identifier))] string taskId,
- [Parameter(typeof(MpiConfigurationOptions.DriverId))] string driverId,
- [Parameter(typeof(MpiConfigurationOptions.Timeout))] int timrout,
- [Parameter(typeof(MpiConfigurationOptions.RetryCount))] int retryCount,
- [Parameter(typeof(MpiConfigurationOptions.TopologyRootTaskId))] string rootId,
- [Parameter(typeof(MpiConfigurationOptions.TopologyChildTaskIds))] ISet<string> childIds,
+ [Parameter(typeof(GroupCommConfigurationOptions.DriverId))] string driverId,
+ [Parameter(typeof(GroupCommConfigurationOptions.Timeout))] int timrout,
+ [Parameter(typeof(GroupCommConfigurationOptions.RetryCount))] int retryCount,
+ [Parameter(typeof(GroupCommConfigurationOptions.TopologyRootTaskId))] string rootId,
+ [Parameter(typeof(GroupCommConfigurationOptions.TopologyChildTaskIds))] ISet<string> childIds,
NetworkService<GroupCommunicationMessage> networkService,
ICodec<T> codec,
Sender sender)
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/c330dcff/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
index 909dc4c..d5c84c8 100644
--- a/lang/cs/Org.Apache.REEF.Network/Group/Topology/FlatTopology.cs
+++ b/lang/cs/Org.Apache.REEF.Network/Group/Topology/FlatTopology.cs
@@ -31,7 +31,7 @@ using Org.Apache.REEF.Network.Group.Pipelining;
namespace Org.Apache.REEF.Network.Group.Topology
{
/// <summary>
- /// Represents a graph of MPI Operators where there are only two levels of
+ /// Represents a graph of Group Communication 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>
@@ -85,8 +85,8 @@ namespace Org.Apache.REEF.Network.Group.Topology
{
var confBuilder = TangFactory.GetTang().NewConfigurationBuilder()
.BindImplementation(typeof(ICodec<T1>), OperatorSpec.Codec)
- .BindNamedParameter<MpiConfigurationOptions.TopologyRootTaskId, string>(
- GenericType<MpiConfigurationOptions.TopologyRootTaskId>.Class,
+ .BindNamedParameter<GroupCommConfigurationOptions.TopologyRootTaskId, string>(
+ GenericType<GroupCommConfigurationOptions.TopologyRootTaskId>.Class,
_rootId);
if (taskId.Equals(_rootId))
@@ -95,8 +95,8 @@ namespace Org.Apache.REEF.Network.Group.Topology
{
if (!tId.Equals(_rootId))
{
- confBuilder.BindSetEntry<MpiConfigurationOptions.TopologyChildTaskIds, string>(
- GenericType<MpiConfigurationOptions.TopologyChildTaskIds>.Class,
+ confBuilder.BindSetEntry<GroupCommConfigurationOptions.TopologyChildTaskIds, string>(
+ GenericType<GroupCommConfigurationOptions.TopologyChildTaskIds>.Class,
tId);
}
}
@@ -112,11 +112,11 @@ namespace Org.Apache.REEF.Network.Group.Topology
if (taskId.Equals(broadcastSpec.SenderId))
{
- confBuilder.BindImplementation(GenericType<IMpiOperator<T1>>.Class, GenericType<BroadcastSender<T1>>.Class);
+ confBuilder.BindImplementation(GenericType<IGroupCommOperator<T1>>.Class, GenericType<BroadcastSender<T1>>.Class);
}
else
{
- confBuilder.BindImplementation(GenericType<IMpiOperator<T1>>.Class, GenericType<BroadcastReceiver<T1>>.Class);
+ confBuilder.BindImplementation(GenericType<IGroupCommOperator<T1>>.Class, GenericType<BroadcastReceiver<T1>>.Class);
}
}
else if (OperatorSpec is ReduceOperatorSpec<T1, T2>)
@@ -129,11 +129,11 @@ namespace Org.Apache.REEF.Network.Group.Topology
if (taskId.Equals(reduceSpec.ReceiverId))
{
- confBuilder.BindImplementation(GenericType<IMpiOperator<T1>>.Class, GenericType<ReduceReceiver<T1>>.Class);
+ confBuilder.BindImplementation(GenericType<IGroupCommOperator<T1>>.Class, GenericType<ReduceReceiver<T1>>.Class);
}
else
{
- confBuilder.BindImplementation(GenericType<IMpiOperator<T1>>.Class, GenericType<ReduceSender<T1>>.Class);
+ confBuilder.BindImplementation(GenericType<IGroupCommOperator<T1>>.Class, GenericType<ReduceSender<T1>>.Class);
}
}
else if (OperatorSpec is ScatterOperatorSpec<T1, T2>)
@@ -141,11 +141,11 @@ namespace Org.Apache.REEF.Network.Group.Topology
ScatterOperatorSpec<T1, T2> scatterSpec = OperatorSpec as ScatterOperatorSpec<T1, T2>;
if (taskId.Equals(scatterSpec.SenderId))
{
- confBuilder.BindImplementation(GenericType<IMpiOperator<T1>>.Class, GenericType<ScatterSender<T1>>.Class);
+ confBuilder.BindImplementation(GenericType<IGroupCommOperator<T1>>.Class, GenericType<ScatterSender<T1>>.Class);
}
else
{
- confBuilder.BindImplementation(GenericType<IMpiOperator<T1>>.Class, GenericType<ScatterReceiver<T1>>.Class);
+ confBuilder.BindImplementation(GenericType<IGroupCommOperator<T1>>.Class, GenericType<ScatterReceiver<T1>>.Class);
}
}
else
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/c330dcff/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
index 3f15318..083330d 100644
--- a/lang/cs/Org.Apache.REEF.Network/Group/Topology/ITopology.cs
+++ b/lang/cs/Org.Apache.REEF.Network/Group/Topology/ITopology.cs
@@ -24,7 +24,7 @@ using Org.Apache.REEF.Wake.Remote;
namespace Org.Apache.REEF.Network.Group.Topology
{
/// <summary>
- /// Represents a topology graph for IMpiOperators.
+ /// Represents a topology graph for IGroupCommOperators.
/// </summary>
public interface ITopology<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/Topology/TreeTopology.cs
----------------------------------------------------------------------
diff --git a/lang/cs/Org.Apache.REEF.Network/Group/Topology/TreeTopology.cs b/lang/cs/Org.Apache.REEF.Network/Group/Topology/TreeTopology.cs
index 1ee459e..b129f8a 100644
--- a/lang/cs/Org.Apache.REEF.Network/Group/Topology/TreeTopology.cs
+++ b/lang/cs/Org.Apache.REEF.Network/Group/Topology/TreeTopology.cs
@@ -107,15 +107,15 @@ namespace Org.Apache.REEF.Network.Group.Topology
//add parentid, if no parent, add itself
var confBuilder = TangFactory.GetTang().NewConfigurationBuilder()
.BindImplementation(typeof(ICodec<T1>), OperatorSpec.Codec)
- .BindNamedParameter<MpiConfigurationOptions.TopologyRootTaskId, string>(
- GenericType<MpiConfigurationOptions.TopologyRootTaskId>.Class,
+ .BindNamedParameter<GroupCommConfigurationOptions.TopologyRootTaskId, string>(
+ GenericType<GroupCommConfigurationOptions.TopologyRootTaskId>.Class,
parentId);
//add all its children
foreach (TaskNode childNode in selfTaskNode.GetChildren())
{
- confBuilder.BindSetEntry<MpiConfigurationOptions.TopologyChildTaskIds, string>(
- GenericType<MpiConfigurationOptions.TopologyChildTaskIds>.Class,
+ confBuilder.BindSetEntry<GroupCommConfigurationOptions.TopologyChildTaskIds, string>(
+ GenericType<GroupCommConfigurationOptions.TopologyChildTaskIds>.Class,
childNode.TaskId);
}
@@ -127,11 +127,11 @@ namespace Org.Apache.REEF.Network.Group.Topology
.BindImplementation(GenericType<ICodec<PipelineMessage<T1>>>.Class, GenericType<PipelineMessageCodec<T1>>.Class);
if (taskId.Equals(broadcastSpec.SenderId))
{
- confBuilder.BindImplementation(GenericType<IMpiOperator<T1>>.Class, GenericType<BroadcastSender<T1>>.Class);
+ confBuilder.BindImplementation(GenericType<IGroupCommOperator<T1>>.Class, GenericType<BroadcastSender<T1>>.Class);
}
else
{
- confBuilder.BindImplementation(GenericType<IMpiOperator<T1>>.Class, GenericType<BroadcastReceiver<T1>>.Class);
+ confBuilder.BindImplementation(GenericType<IGroupCommOperator<T1>>.Class, GenericType<BroadcastReceiver<T1>>.Class);
}
}
else if (OperatorSpec is ReduceOperatorSpec<T1, T2>)
@@ -144,11 +144,11 @@ namespace Org.Apache.REEF.Network.Group.Topology
if (taskId.Equals(reduceSpec.ReceiverId))
{
- confBuilder.BindImplementation(GenericType<IMpiOperator<T1>>.Class, GenericType<ReduceReceiver<T1>>.Class);
+ confBuilder.BindImplementation(GenericType<IGroupCommOperator<T1>>.Class, GenericType<ReduceReceiver<T1>>.Class);
}
else
{
- confBuilder.BindImplementation(GenericType<IMpiOperator<T1>>.Class, GenericType<ReduceSender<T1>>.Class);
+ confBuilder.BindImplementation(GenericType<IGroupCommOperator<T1>>.Class, GenericType<ReduceSender<T1>>.Class);
}
}
else if (OperatorSpec is ScatterOperatorSpec<T1, T2>)
@@ -156,11 +156,11 @@ namespace Org.Apache.REEF.Network.Group.Topology
ScatterOperatorSpec<T1, T2> scatterSpec = OperatorSpec as ScatterOperatorSpec<T1, T2>;
if (taskId.Equals(scatterSpec.SenderId))
{
- confBuilder.BindImplementation(GenericType<IMpiOperator<T1>>.Class, GenericType<ScatterSender<T1>>.Class);
+ confBuilder.BindImplementation(GenericType<IGroupCommOperator<T1>>.Class, GenericType<ScatterSender<T1>>.Class);
}
else
{
- confBuilder.BindImplementation(GenericType<IMpiOperator<T1>>.Class, GenericType<ScatterReceiver<T1>>.Class);
+ confBuilder.BindImplementation(GenericType<IGroupCommOperator<T1>>.Class, GenericType<ScatterReceiver<T1>>.Class);
}
}
else
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/c330dcff/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 58e41a8..9834269 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
@@ -52,17 +52,17 @@ under the License.
<ItemGroup>
<Compile Include="Group\Codec\GcmMessageProto.cs" />
<Compile Include="Group\Codec\GroupCommunicationMessageCodec.cs" />
- <Compile Include="Group\Config\MpiConfigurationOptions.cs" />
+ <Compile Include="Group\Config\GroupCommConfigurationOptions.cs" />
<Compile Include="Group\Driver\ICommunicationGroupDriver.cs" />
- <Compile Include="Group\Driver\IMpiDriver.cs" />
+ <Compile Include="Group\Driver\IGroupCommDriver.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\GroupCommDriver.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\IGroupCommOperator.cs" />
<Compile Include="Group\Operators\Impl\BroadcastOperatorSpec.cs" />
<Compile Include="Group\Operators\Impl\BroadcastReceiver.cs" />
<Compile Include="Group\Operators\Impl\BroadcastSender.cs" />
@@ -87,12 +87,12 @@ under the License.
<Compile Include="Group\Pipelining\PipelineMessageCodec.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\IGroupCommClient.cs" />
+ <Compile Include="Group\Task\IGroupCommNetworkObserver.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\GroupCommClient.cs" />
+ <Compile Include="Group\Task\Impl\GroupCommNetworkObserver.cs" />
<Compile Include="Group\Task\Impl\NodeStruct.cs" />
<Compile Include="Group\Task\Impl\OperatorTopology.cs" />
<Compile Include="Group\Topology\FlatTopology.cs" />
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/c330dcff/lang/cs/Org.Apache.REEF.Tests/Functional/Group/BroadcastReduceTest.cs
----------------------------------------------------------------------
diff --git a/lang/cs/Org.Apache.REEF.Tests/Functional/Group/BroadcastReduceTest.cs b/lang/cs/Org.Apache.REEF.Tests/Functional/Group/BroadcastReduceTest.cs
index 94529c1..7dc3423 100644
--- a/lang/cs/Org.Apache.REEF.Tests/Functional/Group/BroadcastReduceTest.cs
+++ b/lang/cs/Org.Apache.REEF.Tests/Functional/Group/BroadcastReduceTest.cs
@@ -87,15 +87,15 @@ namespace Org.Apache.REEF.Tests.Functional.Group
numTasks.ToString(CultureInfo.InvariantCulture))
.Build();
- IConfiguration mpiDriverConfig = TangFactory.GetTang().NewConfigurationBuilder()
- .BindStringNamedParam<MpiConfigurationOptions.DriverId>(GroupTestConstants.DriverId)
- .BindStringNamedParam<MpiConfigurationOptions.MasterTaskId>(GroupTestConstants.MasterTaskId)
- .BindStringNamedParam<MpiConfigurationOptions.GroupName>(GroupTestConstants.GroupName)
- .BindIntNamedParam<MpiConfigurationOptions.FanOut>(GroupTestConstants.FanOut.ToString(CultureInfo.InvariantCulture).ToString(CultureInfo.InvariantCulture))
- .BindIntNamedParam<MpiConfigurationOptions.NumberOfTasks>(numTasks.ToString(CultureInfo.InvariantCulture))
+ IConfiguration groupCommDriverConfig = TangFactory.GetTang().NewConfigurationBuilder()
+ .BindStringNamedParam<GroupCommConfigurationOptions.DriverId>(GroupTestConstants.DriverId)
+ .BindStringNamedParam<GroupCommConfigurationOptions.MasterTaskId>(GroupTestConstants.MasterTaskId)
+ .BindStringNamedParam<GroupCommConfigurationOptions.GroupName>(GroupTestConstants.GroupName)
+ .BindIntNamedParam<GroupCommConfigurationOptions.FanOut>(GroupTestConstants.FanOut.ToString(CultureInfo.InvariantCulture).ToString(CultureInfo.InvariantCulture))
+ .BindIntNamedParam<GroupCommConfigurationOptions.NumberOfTasks>(numTasks.ToString(CultureInfo.InvariantCulture))
.Build();
- IConfiguration merged = Configurations.Merge(driverConfig, mpiDriverConfig);
+ IConfiguration merged = Configurations.Merge(driverConfig, groupCommDriverConfig);
HashSet<string> appDlls = new HashSet<string>();
appDlls.Add(typeof(IDriver).Assembly.GetName().Name);
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/c330dcff/lang/cs/Org.Apache.REEF.Tests/Functional/Group/PipelinedBroadcastReduceTest.cs
----------------------------------------------------------------------
diff --git a/lang/cs/Org.Apache.REEF.Tests/Functional/Group/PipelinedBroadcastReduceTest.cs b/lang/cs/Org.Apache.REEF.Tests/Functional/Group/PipelinedBroadcastReduceTest.cs
index 0c918b5..fe098a8 100644
--- a/lang/cs/Org.Apache.REEF.Tests/Functional/Group/PipelinedBroadcastReduceTest.cs
+++ b/lang/cs/Org.Apache.REEF.Tests/Functional/Group/PipelinedBroadcastReduceTest.cs
@@ -89,15 +89,15 @@ namespace Org.Apache.REEF.Tests.Functional.Group
GroupTestConstants.ChunkSize.ToString(CultureInfo.InvariantCulture))
.Build();
- IConfiguration mpiDriverConfig = TangFactory.GetTang().NewConfigurationBuilder()
- .BindStringNamedParam<MpiConfigurationOptions.DriverId>(GroupTestConstants.DriverId)
- .BindStringNamedParam<MpiConfigurationOptions.MasterTaskId>(GroupTestConstants.MasterTaskId)
- .BindStringNamedParam<MpiConfigurationOptions.GroupName>(GroupTestConstants.GroupName)
- .BindIntNamedParam<MpiConfigurationOptions.FanOut>(GroupTestConstants.FanOut.ToString(CultureInfo.InvariantCulture).ToString(CultureInfo.InvariantCulture))
- .BindIntNamedParam<MpiConfigurationOptions.NumberOfTasks>(numTasks.ToString(CultureInfo.InvariantCulture))
+ IConfiguration groupCommDriverConfig = TangFactory.GetTang().NewConfigurationBuilder()
+ .BindStringNamedParam<GroupCommConfigurationOptions.DriverId>(GroupTestConstants.DriverId)
+ .BindStringNamedParam<GroupCommConfigurationOptions.MasterTaskId>(GroupTestConstants.MasterTaskId)
+ .BindStringNamedParam<GroupCommConfigurationOptions.GroupName>(GroupTestConstants.GroupName)
+ .BindIntNamedParam<GroupCommConfigurationOptions.FanOut>(GroupTestConstants.FanOut.ToString(CultureInfo.InvariantCulture).ToString(CultureInfo.InvariantCulture))
+ .BindIntNamedParam<GroupCommConfigurationOptions.NumberOfTasks>(numTasks.ToString(CultureInfo.InvariantCulture))
.Build();
- IConfiguration merged = Configurations.Merge(driverConfig, mpiDriverConfig);
+ IConfiguration merged = Configurations.Merge(driverConfig, groupCommDriverConfig);
HashSet<string> appDlls = new HashSet<string>();
appDlls.Add(typeof(IDriver).Assembly.GetName().Name);
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/c330dcff/lang/cs/Org.Apache.REEF.Tests/Functional/Group/ScatterReduceTest.cs
----------------------------------------------------------------------
diff --git a/lang/cs/Org.Apache.REEF.Tests/Functional/Group/ScatterReduceTest.cs b/lang/cs/Org.Apache.REEF.Tests/Functional/Group/ScatterReduceTest.cs
index 083261a..868b3a1 100644
--- a/lang/cs/Org.Apache.REEF.Tests/Functional/Group/ScatterReduceTest.cs
+++ b/lang/cs/Org.Apache.REEF.Tests/Functional/Group/ScatterReduceTest.cs
@@ -84,15 +84,15 @@ namespace Org.Apache.REEF.Tests.Functional.Group
numTasks.ToString(CultureInfo.InvariantCulture))
.Build();
- IConfiguration mpiDriverConfig = TangFactory.GetTang().NewConfigurationBuilder()
- .BindStringNamedParam<MpiConfigurationOptions.DriverId>(GroupTestConstants.DriverId)
- .BindStringNamedParam<MpiConfigurationOptions.MasterTaskId>(GroupTestConstants.MasterTaskId)
- .BindStringNamedParam<MpiConfigurationOptions.GroupName>(GroupTestConstants.GroupName)
- .BindIntNamedParam<MpiConfigurationOptions.FanOut>(GroupTestConstants.FanOut.ToString(CultureInfo.InvariantCulture).ToString(CultureInfo.InvariantCulture))
- .BindIntNamedParam<MpiConfigurationOptions.NumberOfTasks>(numTasks.ToString())
+ IConfiguration groupCommDriverConfig = TangFactory.GetTang().NewConfigurationBuilder()
+ .BindStringNamedParam<GroupCommConfigurationOptions.DriverId>(GroupTestConstants.DriverId)
+ .BindStringNamedParam<GroupCommConfigurationOptions.MasterTaskId>(GroupTestConstants.MasterTaskId)
+ .BindStringNamedParam<GroupCommConfigurationOptions.GroupName>(GroupTestConstants.GroupName)
+ .BindIntNamedParam<GroupCommConfigurationOptions.FanOut>(GroupTestConstants.FanOut.ToString(CultureInfo.InvariantCulture).ToString(CultureInfo.InvariantCulture))
+ .BindIntNamedParam<GroupCommConfigurationOptions.NumberOfTasks>(numTasks.ToString())
.Build();
- IConfiguration merged = Configurations.Merge(driverConfig, mpiDriverConfig);
+ IConfiguration merged = Configurations.Merge(driverConfig, groupCommDriverConfig);
HashSet<string> appDlls = new HashSet<string>();
appDlls.Add(typeof(IDriver).Assembly.GetName().Name);
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/c330dcff/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
index 6b3b26c..8044599 100644
--- a/lang/cs/Org.Apache.REEF.Tests/Functional/ML/KMeans/TestKMeans.cs
+++ b/lang/cs/Org.Apache.REEF.Tests/Functional/ML/KMeans/TestKMeans.cs
@@ -160,15 +160,15 @@ namespace Org.Apache.REEF.Tests.Functional.ML.KMeans
.BindIntNamedParam<NumPartitions>(Partitions.ToString())
.Build();
- IConfiguration mpiDriverConfig = TangFactory.GetTang().NewConfigurationBuilder()
- .BindStringNamedParam<MpiConfigurationOptions.DriverId>(Identifier)
- .BindStringNamedParam<MpiConfigurationOptions.MasterTaskId>(Constants.MasterTaskId)
- .BindStringNamedParam<MpiConfigurationOptions.GroupName>(Constants.KMeansCommunicationGroupName)
- .BindIntNamedParam<MpiConfigurationOptions.FanOut>(fanOut.ToString(CultureInfo.InvariantCulture).ToString(CultureInfo.InvariantCulture))
- .BindIntNamedParam<MpiConfigurationOptions.NumberOfTasks>(totalEvaluators.ToString())
+ IConfiguration groupCommunicationDriverConfig = TangFactory.GetTang().NewConfigurationBuilder()
+ .BindStringNamedParam<GroupCommConfigurationOptions.DriverId>(Identifier)
+ .BindStringNamedParam<GroupCommConfigurationOptions.MasterTaskId>(Constants.MasterTaskId)
+ .BindStringNamedParam<GroupCommConfigurationOptions.GroupName>(Constants.KMeansCommunicationGroupName)
+ .BindIntNamedParam<GroupCommConfigurationOptions.FanOut>(fanOut.ToString(CultureInfo.InvariantCulture).ToString(CultureInfo.InvariantCulture))
+ .BindIntNamedParam<GroupCommConfigurationOptions.NumberOfTasks>(totalEvaluators.ToString())
.Build();
- return Configurations.Merge(driverConfig, mpiDriverConfig);
+ return Configurations.Merge(driverConfig, groupCommunicationDriverConfig);
}
private HashSet<string> AssembliesToCopy()
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/c330dcff/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 3827941..44fdfe3 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
@@ -119,10 +119,6 @@ under the License.
</ProjectReference>
</ItemGroup>
<ItemGroup>
- <Folder Include="bin\Debug\" />
- <Folder Include="bin\Release\" />
- </ItemGroup>
- <ItemGroup>
<Service Include="{82A7F48D-3B50-4B1E-B82E-3ADA8210C358}" />
</ItemGroup>
<Import Project="$(MSBuildToolsPath)\Microsoft.CSharp.targets" />
[2/3] incubator-reef git commit: [REEF-256] Rename MPI to Group
Communication
Posted by we...@apache.org.
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);
- }
-}
[3/3] incubator-reef git commit: [REEF-256] Rename MPI to Group
Communication
Posted by we...@apache.org.
[REEF-256] Rename MPI to Group Communication
Some of the classes in Group Communication are names as Mpi when it was
created initially. We need to rename them to group communication. This
PR renamed classes, variables and tests to use GroupComm or
GroupCommunication
JIRA:
[REEF-256](https://issues.apache.org/jira/browse/REEF-256)
Pull Request:
This closes #148
Project: http://git-wip-us.apache.org/repos/asf/incubator-reef/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-reef/commit/c330dcff
Tree: http://git-wip-us.apache.org/repos/asf/incubator-reef/tree/c330dcff
Diff: http://git-wip-us.apache.org/repos/asf/incubator-reef/diff/c330dcff
Branch: refs/heads/master
Commit: c330dcff5157f53cedaf8db4c5cbdda7570a0c48
Parents: e1be758
Author: Julia Wang <jw...@yahoo.com>
Authored: Thu Apr 16 13:12:38 2015 -0700
Committer: Markus Weimer <we...@apache.org>
Committed: Thu Apr 16 14:33:27 2015 -0700
----------------------------------------------------------------------
.../KMeans/KMeansDriverHandlers.cs | 22 +-
.../MachineLearning/KMeans/KMeansMasterTask.cs | 4 +-
.../MachineLearning/KMeans/KMeansSlaveTask.cs | 10 +-
.../MachineLearning/KMeans/LegacyKMeansTask.cs | 2 +-
.../MachineLearning/KMeans/PartialMean.cs | 2 +-
.../BroadcastAndReduceClient.cs | 14 +-
.../PipelineBroadcastAndReduceClient.cs | 14 +-
.../BroadcastReduceDriver.cs | 22 +-
.../BroadcastReduceDriverAndTasks/MasterTask.cs | 10 +-
.../BroadcastReduceDriverAndTasks/SlaveTask.cs | 10 +-
.../PipelinedBroadcastReduceDriver.cs | 22 +-
.../PipelinedMasterTask.cs | 10 +-
.../PipelinedSlaveTask.cs | 10 +-
.../ScatterReduceDriverAndTasks/MasterTask.cs | 10 +-
.../ScatterReduceDriver.cs | 24 +-
.../ScatterReduceDriverAndTasks/SlaveTask.cs | 10 +-
.../GroupCommunicationTests.cs | 100 +++----
.../GroupCommunicationTreeTopologyTests.cs | 54 ++--
.../Config/GroupCommConfigurationOptions.cs | 102 +++++++
.../Group/Config/MpiConfigurationOptions.cs | 102 -------
.../Group/Driver/ICommunicationGroupDriver.cs | 18 +-
.../Group/Driver/IGroupCommDriver.cs | 96 +++++++
.../Group/Driver/IMpiDriver.cs | 96 -------
.../Driver/Impl/CommunicationGroupDriver.cs | 34 +--
.../Group/Driver/Impl/GroupCommDriver.cs | 285 +++++++++++++++++++
.../Driver/Impl/GroupCommunicationMessage.cs | 8 +-
.../Group/Driver/Impl/MessageType.cs | 2 +-
.../Group/Driver/Impl/MpiDriver.cs | 285 -------------------
.../Group/Driver/Impl/TaskStarter.cs | 16 +-
.../Group/Operators/IBroadcastReceiver.cs | 4 +-
.../Group/Operators/IBroadcastSender.cs | 4 +-
.../Group/Operators/IGroupCommOperator.cs | 43 +++
.../Group/Operators/IMpiOperator.cs | 43 ---
.../Group/Operators/IReduceReceiver.cs | 4 +-
.../Group/Operators/IReduceSender.cs | 4 +-
.../Group/Operators/IScatterReceiver.cs | 4 +-
.../Group/Operators/IScatterSender.cs | 4 +-
.../Group/Operators/Impl/BroadcastReceiver.cs | 6 +-
.../Group/Operators/Impl/BroadcastSender.cs | 8 +-
.../Group/Operators/Impl/ReduceOperatorSpec.cs | 2 +-
.../Group/Operators/Impl/ReduceReceiver.cs | 6 +-
.../Group/Operators/Impl/ReduceSender.cs | 6 +-
.../Group/Operators/Impl/ScatterOperatorSpec.cs | 2 +-
.../Group/Operators/Impl/ScatterReceiver.cs | 6 +-
.../Group/Operators/Impl/ScatterSender.cs | 6 +-
.../Group/Operators/Impl/Sender.cs | 2 +-
.../Group/Task/ICommunicationGroupClient.cs | 2 +-
.../Group/Task/IGroupCommClient.cs | 39 +++
.../Group/Task/IGroupCommNetworkObserver.cs | 44 +++
.../Group/Task/IMpiClient.cs | 39 ---
.../Group/Task/IMpiNetworkObserver.cs | 44 ---
.../Group/Task/Impl/CommunicationGroupClient.cs | 32 +--
.../Impl/CommunicationGroupNetworkObserver.cs | 4 +-
.../Group/Task/Impl/GroupCommClient.cs | 107 +++++++
.../Group/Task/Impl/GroupCommNetworkObserver.cs | 108 +++++++
.../Group/Task/Impl/MpiClient.cs | 107 -------
.../Group/Task/Impl/MpiNetworkObserver.cs | 108 -------
.../Group/Task/Impl/OperatorTopology.cs | 16 +-
.../Group/Topology/FlatTopology.cs | 22 +-
.../Group/Topology/ITopology.cs | 2 +-
.../Group/Topology/TreeTopology.cs | 20 +-
.../Org.Apache.REEF.Network.csproj | 16 +-
.../Functional/Group/BroadcastReduceTest.cs | 14 +-
.../Group/PipelinedBroadcastReduceTest.cs | 14 +-
.../Functional/Group/ScatterReduceTest.cs | 14 +-
.../Functional/ML/KMeans/TestKMeans.cs | 14 +-
.../Org.Apache.REEF.Tests.csproj | 4 -
67 files changed, 1157 insertions(+), 1161 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/c330dcff/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
index 358e3e1..08173cc 100644
--- a/lang/cs/Org.Apache.REEF.Examples/MachineLearning/KMeans/KMeansDriverHandlers.cs
+++ b/lang/cs/Org.Apache.REEF.Examples/MachineLearning/KMeans/KMeansDriverHandlers.cs
@@ -60,12 +60,12 @@ namespace Org.Apache.REEF.Examples.MachineLearning.KMeans
private readonly int _clustersNumber = 3;
private readonly int _totalEvaluators;
private int _partitionInex = 0;
- private readonly IMpiDriver _mpiDriver;
+ private readonly IGroupCommDriver _groupCommDriver;
private readonly ICommunicationGroupDriver _commGroup;
- private readonly TaskStarter _mpiTaskStarter;
+ private readonly TaskStarter _groupCommTaskStarter;
[Inject]
- public KMeansDriverHandlers([Parameter(typeof(NumPartitions))] int numPartitions, MpiDriver mpiDriver)
+ public KMeansDriverHandlers([Parameter(typeof(NumPartitions))] int numPartitions, GroupCommDriver groupCommDriver)
{
Identifier = "KMeansDriverId";
_executionDirectory = Path.Combine(Directory.GetCurrentDirectory(), Constants.KMeansExecutionBaseDirectory, Guid.NewGuid().ToString("N").Substring(0, 4));
@@ -79,15 +79,15 @@ namespace Org.Apache.REEF.Examples.MachineLearning.KMeans
_totalEvaluators = numPartitions + 1;
- _mpiDriver = mpiDriver;
+ _groupCommDriver = groupCommDriver;
- _commGroup = _mpiDriver.DefaultGroup
+ _commGroup = _groupCommDriver.DefaultGroup
.AddBroadcast<Centroids, CentroidsCodec>(Constants.CentroidsBroadcastOperatorName, Constants.MasterTaskId)
.AddBroadcast<ControlMessage, ControlMessageCodec>(Constants.ControlMessageBroadcastOperatorName, Constants.MasterTaskId)
.AddReduce<ProcessedResults, ProcessedResultsCodec>(Constants.MeansReduceOperatorName, Constants.MasterTaskId, new KMeansMasterTask.AggregateMeans())
.Build();
- _mpiTaskStarter = new TaskStarter(_mpiDriver, _totalEvaluators);
+ _groupCommTaskStarter = new TaskStarter(_groupCommDriver, _totalEvaluators);
CreateClassHierarchy();
}
@@ -105,10 +105,10 @@ namespace Org.Apache.REEF.Examples.MachineLearning.KMeans
public void OnNext(IAllocatedEvaluator allocatedEvaluator)
{
- IConfiguration contextConfiguration = _mpiDriver.GetContextConfiguration();
+ IConfiguration contextConfiguration = _groupCommDriver.GetContextConfiguration();
int partitionNum;
- if (_mpiDriver.IsMasterContextConfiguration(contextConfiguration))
+ if (_groupCommDriver.IsMasterContextConfiguration(contextConfiguration))
{
partitionNum = -1;
}
@@ -121,7 +121,7 @@ namespace Org.Apache.REEF.Examples.MachineLearning.KMeans
}
}
- IConfiguration gcServiceConfiguration = _mpiDriver.GetServiceConfiguration();
+ IConfiguration gcServiceConfiguration = _groupCommDriver.GetServiceConfiguration();
IConfiguration commonServiceConfiguration = TangFactory.GetTang().NewConfigurationBuilder(gcServiceConfiguration)
.BindNamedParameter<DataPartitionCache.PartitionIndex, int>(GenericType<DataPartitionCache.PartitionIndex>.Class, partitionNum.ToString(CultureInfo.InvariantCulture))
@@ -141,7 +141,7 @@ namespace Org.Apache.REEF.Examples.MachineLearning.KMeans
{
IConfiguration taskConfiguration;
- if (_mpiDriver.IsMasterTaskContext(activeContext))
+ if (_groupCommDriver.IsMasterTaskContext(activeContext))
{
// Configure Master Task
taskConfiguration = TaskConfiguration.ConfigurationModule
@@ -162,7 +162,7 @@ namespace Org.Apache.REEF.Examples.MachineLearning.KMeans
_commGroup.AddTask(slaveTaskId);
}
- _mpiTaskStarter.QueueTask(taskConfiguration, activeContext);
+ _groupCommTaskStarter.QueueTask(taskConfiguration, activeContext);
}
public void OnError(Exception error)
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/c330dcff/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
index fd096ea..375dbc6 100644
--- a/lang/cs/Org.Apache.REEF.Examples/MachineLearning/KMeans/KMeansMasterTask.cs
+++ b/lang/cs/Org.Apache.REEF.Examples/MachineLearning/KMeans/KMeansMasterTask.cs
@@ -49,7 +49,7 @@ namespace Org.Apache.REEF.Examples.MachineLearning.KMeans
public KMeansMasterTask(
[Parameter(typeof(KMeansConfiguratioinOptions.TotalNumEvaluators))] int totalNumEvaluators,
[Parameter(Value = typeof(KMeansConfiguratioinOptions.ExecutionDirectory))] string executionDirectory,
- IMpiClient mpiClient)
+ IGroupCommClient groupCommClient)
{
using (_logger.LogFunction("KMeansMasterTask"))
{
@@ -57,7 +57,7 @@ namespace Org.Apache.REEF.Examples.MachineLearning.KMeans
{
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);
+ _commGroup = groupCommClient.GetCommunicationGroup(Constants.KMeansCommunicationGroupName);
_dataBroadcastSender = _commGroup.GetBroadcastSender<Centroids>(Constants.CentroidsBroadcastOperatorName);
_meansReducerReceiver = _commGroup.GetReduceReceiver<ProcessedResults>(Constants.MeansReduceOperatorName);
_controlBroadcastSender = _commGroup.GetBroadcastSender<ControlMessage>(Constants.ControlMessageBroadcastOperatorName);
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/c330dcff/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
index 8d45245..6e037fe 100644
--- a/lang/cs/Org.Apache.REEF.Examples/MachineLearning/KMeans/KMeansSlaveTask.cs
+++ b/lang/cs/Org.Apache.REEF.Examples/MachineLearning/KMeans/KMeansSlaveTask.cs
@@ -32,7 +32,7 @@ namespace Org.Apache.REEF.Examples.MachineLearning.KMeans
{
private static readonly Logger _logger = Logger.GetLogger(typeof(KMeansSlaveTask));
private readonly int _clustersNum;
- private readonly IMpiClient _mpiClient;
+ private readonly IGroupCommClient _groupCommClient;
private readonly ICommunicationGroupClient _commGroup;
private readonly IBroadcastReceiver<Centroids> _dataBroadcastReceiver;
private readonly IBroadcastReceiver<ControlMessage> _controlBroadcastReceiver;
@@ -43,14 +43,14 @@ namespace Org.Apache.REEF.Examples.MachineLearning.KMeans
public KMeansSlaveTask(
DataPartitionCache dataPartition,
[Parameter(typeof(KMeansConfiguratioinOptions.TotalNumEvaluators))] int clustersNumber,
- IMpiClient mpiClient)
+ IGroupCommClient groupCommClient)
{
using (_logger.LogFunction("KMeansSlaveTask::KMeansSlaveTask"))
{
_dataPartition = dataPartition;
- _mpiClient = mpiClient;
+ _groupCommClient = groupCommClient;
_clustersNum = clustersNumber;
- _commGroup = _mpiClient.GetCommunicationGroup(Constants.KMeansCommunicationGroupName);
+ _commGroup = _groupCommClient.GetCommunicationGroup(Constants.KMeansCommunicationGroupName);
_dataBroadcastReceiver = _commGroup.GetBroadcastReceiver<Centroids>(Constants.CentroidsBroadcastOperatorName);
_partialMeansSender = _commGroup.GetReduceSender<ProcessedResults>(Constants.MeansReduceOperatorName);
_controlBroadcastReceiver = _commGroup.GetBroadcastReceiver<ControlMessage>(Constants.ControlMessageBroadcastOperatorName);
@@ -80,7 +80,7 @@ namespace Org.Apache.REEF.Examples.MachineLearning.KMeans
public void Dispose()
{
- _mpiClient.Dispose();
+ _groupCommClient.Dispose();
}
private List<PartialMean> ComputePartialMeans()
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/c330dcff/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
index c6c3ffa..6a6634a 100644
--- a/lang/cs/Org.Apache.REEF.Examples/MachineLearning/KMeans/LegacyKMeansTask.cs
+++ b/lang/cs/Org.Apache.REEF.Examples/MachineLearning/KMeans/LegacyKMeansTask.cs
@@ -74,7 +74,7 @@ namespace Org.Apache.REEF.Examples.MachineLearning.KMeans
_dataPartition.LabelData(_centroids);
_partialMeans = ComputePartialMeans();
- // should be replaced with MPI
+ // should be replaced with Group Communication
using (StreamWriter writer = new StreamWriter(
File.OpenWrite(Path.Combine(_kMeansExecutionDirectory, Constants.DataDirectory, Constants.PartialMeanFilePrefix + _dataPartition.Partition))))
{
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/c330dcff/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
index 6f44167..3a6ece0 100644
--- a/lang/cs/Org.Apache.REEF.Examples/MachineLearning/KMeans/PartialMean.cs
+++ b/lang/cs/Org.Apache.REEF.Examples/MachineLearning/KMeans/PartialMean.cs
@@ -74,7 +74,7 @@ namespace Org.Apache.REEF.Examples.MachineLearning.KMeans
List<PartialMean> partialMeans = new List<PartialMean>();
for (int i = 0; i < partitionsNum; i++)
{
- // should be replaced with MPI
+ // should be replaced with Group Communication
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))
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/c330dcff/lang/cs/Org.Apache.REEF.Network.Examples.Client/BroadcastAndReduceClient.cs
----------------------------------------------------------------------
diff --git a/lang/cs/Org.Apache.REEF.Network.Examples.Client/BroadcastAndReduceClient.cs b/lang/cs/Org.Apache.REEF.Network.Examples.Client/BroadcastAndReduceClient.cs
index 0db6ce0..559d936 100644
--- a/lang/cs/Org.Apache.REEF.Network.Examples.Client/BroadcastAndReduceClient.cs
+++ b/lang/cs/Org.Apache.REEF.Network.Examples.Client/BroadcastAndReduceClient.cs
@@ -62,15 +62,15 @@ namespace Org.Apache.REEF.Network.Examples.Client
numTasks.ToString(CultureInfo.InvariantCulture))
.Build();
- IConfiguration mpiDriverConfig = TangFactory.GetTang().NewConfigurationBuilder()
- .BindStringNamedParam<MpiConfigurationOptions.DriverId>(driverId)
- .BindStringNamedParam<MpiConfigurationOptions.MasterTaskId>(masterTaskId)
- .BindStringNamedParam<MpiConfigurationOptions.GroupName>(groupName)
- .BindIntNamedParam<MpiConfigurationOptions.FanOut>(fanOut.ToString(CultureInfo.InvariantCulture).ToString(CultureInfo.InvariantCulture))
- .BindIntNamedParam<MpiConfigurationOptions.NumberOfTasks>(numTasks.ToString(CultureInfo.InvariantCulture))
+ IConfiguration groupCommDriverConfig = TangFactory.GetTang().NewConfigurationBuilder()
+ .BindStringNamedParam<GroupCommConfigurationOptions.DriverId>(driverId)
+ .BindStringNamedParam<GroupCommConfigurationOptions.MasterTaskId>(masterTaskId)
+ .BindStringNamedParam<GroupCommConfigurationOptions.GroupName>(groupName)
+ .BindIntNamedParam<GroupCommConfigurationOptions.FanOut>(fanOut.ToString(CultureInfo.InvariantCulture).ToString(CultureInfo.InvariantCulture))
+ .BindIntNamedParam<GroupCommConfigurationOptions.NumberOfTasks>(numTasks.ToString(CultureInfo.InvariantCulture))
.Build();
- IConfiguration merged = Configurations.Merge(driverConfig, mpiDriverConfig);
+ IConfiguration merged = Configurations.Merge(driverConfig, groupCommDriverConfig);
HashSet<string> appDlls = new HashSet<string>();
appDlls.Add(typeof(IDriver).Assembly.GetName().Name);
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/c330dcff/lang/cs/Org.Apache.REEF.Network.Examples.Client/PipelineBroadcastAndReduceClient.cs
----------------------------------------------------------------------
diff --git a/lang/cs/Org.Apache.REEF.Network.Examples.Client/PipelineBroadcastAndReduceClient.cs b/lang/cs/Org.Apache.REEF.Network.Examples.Client/PipelineBroadcastAndReduceClient.cs
index 570b227..c86697b 100644
--- a/lang/cs/Org.Apache.REEF.Network.Examples.Client/PipelineBroadcastAndReduceClient.cs
+++ b/lang/cs/Org.Apache.REEF.Network.Examples.Client/PipelineBroadcastAndReduceClient.cs
@@ -63,15 +63,15 @@ namespace Org.Apache.REEF.Network.Examples.Client
GroupTestConstants.ChunkSize.ToString(CultureInfo.InvariantCulture))
.Build();
- IConfiguration mpiDriverConfig = TangFactory.GetTang().NewConfigurationBuilder()
- .BindStringNamedParam<MpiConfigurationOptions.DriverId>(GroupTestConstants.DriverId)
- .BindStringNamedParam<MpiConfigurationOptions.MasterTaskId>(GroupTestConstants.MasterTaskId)
- .BindStringNamedParam<MpiConfigurationOptions.GroupName>(GroupTestConstants.GroupName)
- .BindIntNamedParam<MpiConfigurationOptions.FanOut>(GroupTestConstants.FanOut.ToString(CultureInfo.InvariantCulture).ToString(CultureInfo.InvariantCulture))
- .BindIntNamedParam<MpiConfigurationOptions.NumberOfTasks>(numTasks.ToString(CultureInfo.InvariantCulture))
+ IConfiguration groupCommDriverConfig = TangFactory.GetTang().NewConfigurationBuilder()
+ .BindStringNamedParam<GroupCommConfigurationOptions.DriverId>(GroupTestConstants.DriverId)
+ .BindStringNamedParam<GroupCommConfigurationOptions.MasterTaskId>(GroupTestConstants.MasterTaskId)
+ .BindStringNamedParam<GroupCommConfigurationOptions.GroupName>(GroupTestConstants.GroupName)
+ .BindIntNamedParam<GroupCommConfigurationOptions.FanOut>(GroupTestConstants.FanOut.ToString(CultureInfo.InvariantCulture).ToString(CultureInfo.InvariantCulture))
+ .BindIntNamedParam<GroupCommConfigurationOptions.NumberOfTasks>(numTasks.ToString(CultureInfo.InvariantCulture))
.Build();
- IConfiguration merged = Configurations.Merge(driverConfig, mpiDriverConfig);
+ IConfiguration merged = Configurations.Merge(driverConfig, groupCommDriverConfig);
HashSet<string> appDlls = new HashSet<string>();
appDlls.Add(typeof(IDriver).Assembly.GetName().Name);
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/c330dcff/lang/cs/Org.Apache.REEF.Network.Examples/GroupCommunication/BroadcastReduceDriverAndTasks/BroadcastReduceDriver.cs
----------------------------------------------------------------------
diff --git a/lang/cs/Org.Apache.REEF.Network.Examples/GroupCommunication/BroadcastReduceDriverAndTasks/BroadcastReduceDriver.cs b/lang/cs/Org.Apache.REEF.Network.Examples/GroupCommunication/BroadcastReduceDriverAndTasks/BroadcastReduceDriver.cs
index c6b8578..760d3b8 100644
--- a/lang/cs/Org.Apache.REEF.Network.Examples/GroupCommunication/BroadcastReduceDriverAndTasks/BroadcastReduceDriver.cs
+++ b/lang/cs/Org.Apache.REEF.Network.Examples/GroupCommunication/BroadcastReduceDriverAndTasks/BroadcastReduceDriver.cs
@@ -47,21 +47,21 @@ namespace Org.Apache.REEF.Network.Examples.GroupCommunication.BroadcastReduceDri
private readonly int _numEvaluators;
private readonly int _numIterations;
- private readonly IMpiDriver _mpiDriver;
+ private readonly IGroupCommDriver _groupCommDriver;
private readonly ICommunicationGroupDriver _commGroup;
- private readonly TaskStarter _mpiTaskStarter;
+ private readonly TaskStarter _groupCommTaskStarter;
[Inject]
public BroadcastReduceDriver(
[Parameter(typeof(GroupTestConfig.NumEvaluators))] int numEvaluators,
[Parameter(typeof(GroupTestConfig.NumIterations))] int numIterations,
- MpiDriver mpiDriver)
+ GroupCommDriver groupCommDriver)
{
Identifier = "BroadcastStartHandler";
_numEvaluators = numEvaluators;
_numIterations = numIterations;
- _mpiDriver = mpiDriver;
- _commGroup = _mpiDriver.DefaultGroup
+ _groupCommDriver = groupCommDriver;
+ _commGroup = _groupCommDriver.DefaultGroup
.AddBroadcast<int, IntCodec>(
GroupTestConstants.BroadcastOperatorName,
GroupTestConstants.MasterTaskId)
@@ -71,7 +71,7 @@ namespace Org.Apache.REEF.Network.Examples.GroupCommunication.BroadcastReduceDri
new SumFunction())
.Build();
- _mpiTaskStarter = new TaskStarter(_mpiDriver, numEvaluators);
+ _groupCommTaskStarter = new TaskStarter(_groupCommDriver, numEvaluators);
CreateClassHierarchy();
}
@@ -86,14 +86,14 @@ namespace Org.Apache.REEF.Network.Examples.GroupCommunication.BroadcastReduceDri
public void OnNext(IAllocatedEvaluator allocatedEvaluator)
{
- IConfiguration contextConf = _mpiDriver.GetContextConfiguration();
- IConfiguration serviceConf = _mpiDriver.GetServiceConfiguration();
+ IConfiguration contextConf = _groupCommDriver.GetContextConfiguration();
+ IConfiguration serviceConf = _groupCommDriver.GetServiceConfiguration();
allocatedEvaluator.SubmitContextAndService(contextConf, serviceConf);
}
public void OnNext(IActiveContext activeContext)
{
- if (_mpiDriver.IsMasterTaskContext(activeContext))
+ if (_groupCommDriver.IsMasterTaskContext(activeContext))
{
// Configure Master Task
IConfiguration partialTaskConf = TangFactory.GetTang().NewConfigurationBuilder(
@@ -110,7 +110,7 @@ namespace Org.Apache.REEF.Network.Examples.GroupCommunication.BroadcastReduceDri
.Build();
_commGroup.AddTask(GroupTestConstants.MasterTaskId);
- _mpiTaskStarter.QueueTask(partialTaskConf, activeContext);
+ _groupCommTaskStarter.QueueTask(partialTaskConf, activeContext);
}
else
{
@@ -130,7 +130,7 @@ namespace Org.Apache.REEF.Network.Examples.GroupCommunication.BroadcastReduceDri
.Build();
_commGroup.AddTask(slaveTaskId);
- _mpiTaskStarter.QueueTask(partialTaskConf, activeContext);
+ _groupCommTaskStarter.QueueTask(partialTaskConf, activeContext);
}
}
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/c330dcff/lang/cs/Org.Apache.REEF.Network.Examples/GroupCommunication/BroadcastReduceDriverAndTasks/MasterTask.cs
----------------------------------------------------------------------
diff --git a/lang/cs/Org.Apache.REEF.Network.Examples/GroupCommunication/BroadcastReduceDriverAndTasks/MasterTask.cs b/lang/cs/Org.Apache.REEF.Network.Examples/GroupCommunication/BroadcastReduceDriverAndTasks/MasterTask.cs
index 1ca2353..21d23f8 100644
--- a/lang/cs/Org.Apache.REEF.Network.Examples/GroupCommunication/BroadcastReduceDriverAndTasks/MasterTask.cs
+++ b/lang/cs/Org.Apache.REEF.Network.Examples/GroupCommunication/BroadcastReduceDriverAndTasks/MasterTask.cs
@@ -34,7 +34,7 @@ namespace Org.Apache.REEF.Network.Examples.GroupCommunication.BroadcastReduceDri
private readonly int _numIters;
private readonly int _numReduceSenders;
- private readonly IMpiClient _mpiClient;
+ private readonly IGroupCommClient _groupCommClient;
private readonly ICommunicationGroupClient _commGroup;
private readonly IBroadcastSender<int> _broadcastSender;
private readonly IReduceReceiver<int> _sumReducer;
@@ -43,14 +43,14 @@ namespace Org.Apache.REEF.Network.Examples.GroupCommunication.BroadcastReduceDri
public MasterTask(
[Parameter(typeof(GroupTestConfig.NumIterations))] int numIters,
[Parameter(typeof(GroupTestConfig.NumEvaluators))] int numEvaluators,
- IMpiClient mpiClient)
+ IGroupCommClient groupCommClient)
{
_logger.Log(Level.Info, "Hello from master task");
_numIters = numIters;
_numReduceSenders = numEvaluators - 1;
- _mpiClient = mpiClient;
+ _groupCommClient = groupCommClient;
- _commGroup = mpiClient.GetCommunicationGroup(GroupTestConstants.GroupName);
+ _commGroup = groupCommClient.GetCommunicationGroup(GroupTestConstants.GroupName);
_broadcastSender = _commGroup.GetBroadcastSender<int>(GroupTestConstants.BroadcastOperatorName);
_sumReducer = _commGroup.GetReduceReceiver<int>(GroupTestConstants.ReduceOperatorName);
}
@@ -78,7 +78,7 @@ namespace Org.Apache.REEF.Network.Examples.GroupCommunication.BroadcastReduceDri
public void Dispose()
{
- _mpiClient.Dispose();
+ _groupCommClient.Dispose();
}
private int TriangleNumber(int n)
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/c330dcff/lang/cs/Org.Apache.REEF.Network.Examples/GroupCommunication/BroadcastReduceDriverAndTasks/SlaveTask.cs
----------------------------------------------------------------------
diff --git a/lang/cs/Org.Apache.REEF.Network.Examples/GroupCommunication/BroadcastReduceDriverAndTasks/SlaveTask.cs b/lang/cs/Org.Apache.REEF.Network.Examples/GroupCommunication/BroadcastReduceDriverAndTasks/SlaveTask.cs
index c09bb80..6db3a5c 100644
--- a/lang/cs/Org.Apache.REEF.Network.Examples/GroupCommunication/BroadcastReduceDriverAndTasks/SlaveTask.cs
+++ b/lang/cs/Org.Apache.REEF.Network.Examples/GroupCommunication/BroadcastReduceDriverAndTasks/SlaveTask.cs
@@ -31,7 +31,7 @@ namespace Org.Apache.REEF.Network.Examples.GroupCommunication.BroadcastReduceDri
private static readonly Logger _logger = Logger.GetLogger(typeof(SlaveTask));
private readonly int _numIterations;
- private readonly IMpiClient _mpiClient;
+ private readonly IGroupCommClient _groupCommClient;
private readonly ICommunicationGroupClient _commGroup;
private readonly IBroadcastReceiver<int> _broadcastReceiver;
private readonly IReduceSender<int> _triangleNumberSender;
@@ -39,13 +39,13 @@ namespace Org.Apache.REEF.Network.Examples.GroupCommunication.BroadcastReduceDri
[Inject]
public SlaveTask(
[Parameter(typeof(GroupTestConfig.NumIterations))] int numIters,
- IMpiClient mpiClient)
+ IGroupCommClient groupCommClient)
{
_logger.Log(Level.Info, "Hello from slave task");
_numIterations = numIters;
- _mpiClient = mpiClient;
- _commGroup = _mpiClient.GetCommunicationGroup(GroupTestConstants.GroupName);
+ _groupCommClient = groupCommClient;
+ _commGroup = _groupCommClient.GetCommunicationGroup(GroupTestConstants.GroupName);
_broadcastReceiver = _commGroup.GetBroadcastReceiver<int>(GroupTestConstants.BroadcastOperatorName);
_triangleNumberSender = _commGroup.GetReduceSender<int>(GroupTestConstants.ReduceOperatorName);
}
@@ -69,7 +69,7 @@ namespace Org.Apache.REEF.Network.Examples.GroupCommunication.BroadcastReduceDri
public void Dispose()
{
- _mpiClient.Dispose();
+ _groupCommClient.Dispose();
}
private int TriangleNumber(int n)
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/c330dcff/lang/cs/Org.Apache.REEF.Network.Examples/GroupCommunication/PipelineBroadcastReduceDriverAndTasks/PipelinedBroadcastReduceDriver.cs
----------------------------------------------------------------------
diff --git a/lang/cs/Org.Apache.REEF.Network.Examples/GroupCommunication/PipelineBroadcastReduceDriverAndTasks/PipelinedBroadcastReduceDriver.cs b/lang/cs/Org.Apache.REEF.Network.Examples/GroupCommunication/PipelineBroadcastReduceDriverAndTasks/PipelinedBroadcastReduceDriver.cs
index d108d68..5d00fb1 100644
--- a/lang/cs/Org.Apache.REEF.Network.Examples/GroupCommunication/PipelineBroadcastReduceDriverAndTasks/PipelinedBroadcastReduceDriver.cs
+++ b/lang/cs/Org.Apache.REEF.Network.Examples/GroupCommunication/PipelineBroadcastReduceDriverAndTasks/PipelinedBroadcastReduceDriver.cs
@@ -50,16 +50,16 @@ namespace Org.Apache.REEF.Network.Examples.GroupCommunication.PipelineBroadcastR
private readonly int _numIterations;
private readonly int _chunkSize;
- private readonly IMpiDriver _mpiDriver;
+ private readonly IGroupCommDriver _groupCommDriver;
private readonly ICommunicationGroupDriver _commGroup;
- private readonly TaskStarter _mpiTaskStarter;
+ private readonly TaskStarter _groupCommTaskStarter;
[Inject]
public PipelinedBroadcastReduceDriver(
[Parameter(typeof (GroupTestConfig.NumEvaluators))] int numEvaluators,
[Parameter(typeof(GroupTestConfig.NumIterations))] int numIterations,
[Parameter(typeof(GroupTestConfig.ChunkSize))] int chunkSize,
- MpiDriver mpiDriver)
+ GroupCommDriver groupCommDriver)
{
Logger.Log(Level.Info, "*******entering the driver code " + chunkSize);
@@ -68,9 +68,9 @@ namespace Org.Apache.REEF.Network.Examples.GroupCommunication.PipelineBroadcastR
_numIterations = numIterations;
_chunkSize = chunkSize;
- _mpiDriver = mpiDriver;
+ _groupCommDriver = groupCommDriver;
- _commGroup = _mpiDriver.DefaultGroup
+ _commGroup = _groupCommDriver.DefaultGroup
.AddBroadcast<int[], IntArrayCodec>(
GroupTestConstants.BroadcastOperatorName,
GroupTestConstants.MasterTaskId,
@@ -84,7 +84,7 @@ namespace Org.Apache.REEF.Network.Examples.GroupCommunication.PipelineBroadcastR
new PipelineIntDataConverter(_chunkSize))
.Build();
- _mpiTaskStarter = new TaskStarter(_mpiDriver, numEvaluators);
+ _groupCommTaskStarter = new TaskStarter(_groupCommDriver, numEvaluators);
CreateClassHierarchy();
}
@@ -99,14 +99,14 @@ namespace Org.Apache.REEF.Network.Examples.GroupCommunication.PipelineBroadcastR
public void OnNext(IAllocatedEvaluator allocatedEvaluator)
{
- IConfiguration contextConf = _mpiDriver.GetContextConfiguration();
- IConfiguration serviceConf = _mpiDriver.GetServiceConfiguration();
+ IConfiguration contextConf = _groupCommDriver.GetContextConfiguration();
+ IConfiguration serviceConf = _groupCommDriver.GetServiceConfiguration();
allocatedEvaluator.SubmitContextAndService(contextConf, serviceConf);
}
public void OnNext(IActiveContext activeContext)
{
- if (_mpiDriver.IsMasterTaskContext(activeContext))
+ if (_groupCommDriver.IsMasterTaskContext(activeContext))
{
Logger.Log(Level.Info, "******* Master ID " + activeContext.Id );
@@ -128,7 +128,7 @@ namespace Org.Apache.REEF.Network.Examples.GroupCommunication.PipelineBroadcastR
.Build();
_commGroup.AddTask(GroupTestConstants.MasterTaskId);
- _mpiTaskStarter.QueueTask(partialTaskConf, activeContext);
+ _groupCommTaskStarter.QueueTask(partialTaskConf, activeContext);
}
else
{
@@ -151,7 +151,7 @@ namespace Org.Apache.REEF.Network.Examples.GroupCommunication.PipelineBroadcastR
.Build();
_commGroup.AddTask(slaveTaskId);
- _mpiTaskStarter.QueueTask(partialTaskConf, activeContext);
+ _groupCommTaskStarter.QueueTask(partialTaskConf, activeContext);
}
}
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/c330dcff/lang/cs/Org.Apache.REEF.Network.Examples/GroupCommunication/PipelineBroadcastReduceDriverAndTasks/PipelinedMasterTask.cs
----------------------------------------------------------------------
diff --git a/lang/cs/Org.Apache.REEF.Network.Examples/GroupCommunication/PipelineBroadcastReduceDriverAndTasks/PipelinedMasterTask.cs b/lang/cs/Org.Apache.REEF.Network.Examples/GroupCommunication/PipelineBroadcastReduceDriverAndTasks/PipelinedMasterTask.cs
index 0e4ccee..416fe41 100644
--- a/lang/cs/Org.Apache.REEF.Network.Examples/GroupCommunication/PipelineBroadcastReduceDriverAndTasks/PipelinedMasterTask.cs
+++ b/lang/cs/Org.Apache.REEF.Network.Examples/GroupCommunication/PipelineBroadcastReduceDriverAndTasks/PipelinedMasterTask.cs
@@ -35,7 +35,7 @@ namespace Org.Apache.REEF.Network.Examples.GroupCommunication.PipelineBroadcastR
private readonly int _numReduceSenders;
private readonly int _arraySize;
- private readonly IMpiClient _mpiClient;
+ private readonly IGroupCommClient _groupCommClient;
private readonly ICommunicationGroupClient _commGroup;
private readonly IBroadcastSender<int[]> _broadcastSender;
private readonly IReduceReceiver<int[]> _sumReducer;
@@ -45,15 +45,15 @@ namespace Org.Apache.REEF.Network.Examples.GroupCommunication.PipelineBroadcastR
[Parameter(typeof(GroupTestConfig.NumIterations))] int numIters,
[Parameter(typeof(GroupTestConfig.NumEvaluators))] int numEvaluators,
[Parameter(typeof(GroupTestConfig.ArraySize))] int arraySize,
- IMpiClient mpiClient)
+ IGroupCommClient groupCommClient)
{
Logger.Log(Level.Info, "Hello from master task");
_numIters = numIters;
_numReduceSenders = numEvaluators - 1;
_arraySize = arraySize;
- _mpiClient = mpiClient;
+ _groupCommClient = groupCommClient;
- _commGroup = mpiClient.GetCommunicationGroup(GroupTestConstants.GroupName);
+ _commGroup = groupCommClient.GetCommunicationGroup(GroupTestConstants.GroupName);
_broadcastSender = _commGroup.GetBroadcastSender<int[]>(GroupTestConstants.BroadcastOperatorName);
_sumReducer = _commGroup.GetReduceReceiver<int[]>(GroupTestConstants.ReduceOperatorName);
Logger.Log(Level.Info, "finished master task constructor");
@@ -91,7 +91,7 @@ namespace Org.Apache.REEF.Network.Examples.GroupCommunication.PipelineBroadcastR
public void Dispose()
{
- _mpiClient.Dispose();
+ _groupCommClient.Dispose();
}
private int TriangleNumber(int n)
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/c330dcff/lang/cs/Org.Apache.REEF.Network.Examples/GroupCommunication/PipelineBroadcastReduceDriverAndTasks/PipelinedSlaveTask.cs
----------------------------------------------------------------------
diff --git a/lang/cs/Org.Apache.REEF.Network.Examples/GroupCommunication/PipelineBroadcastReduceDriverAndTasks/PipelinedSlaveTask.cs b/lang/cs/Org.Apache.REEF.Network.Examples/GroupCommunication/PipelineBroadcastReduceDriverAndTasks/PipelinedSlaveTask.cs
index 503e6e3..547df28 100644
--- a/lang/cs/Org.Apache.REEF.Network.Examples/GroupCommunication/PipelineBroadcastReduceDriverAndTasks/PipelinedSlaveTask.cs
+++ b/lang/cs/Org.Apache.REEF.Network.Examples/GroupCommunication/PipelineBroadcastReduceDriverAndTasks/PipelinedSlaveTask.cs
@@ -31,7 +31,7 @@ namespace Org.Apache.REEF.Network.Examples.GroupCommunication.PipelineBroadcastR
private static readonly Logger Logger = Logger.GetLogger(typeof(PipelinedSlaveTask));
private readonly int _numIterations;
- private readonly IMpiClient _mpiClient;
+ private readonly IGroupCommClient _groupCommClient;
private readonly ICommunicationGroupClient _commGroup;
private readonly IBroadcastReceiver<int[]> _broadcastReceiver;
private readonly IReduceSender<int[]> _triangleNumberSender;
@@ -39,13 +39,13 @@ namespace Org.Apache.REEF.Network.Examples.GroupCommunication.PipelineBroadcastR
[Inject]
public PipelinedSlaveTask(
[Parameter(typeof(GroupTestConfig.NumIterations))] int numIters,
- IMpiClient mpiClient)
+ IGroupCommClient groupCommClient)
{
Logger.Log(Level.Info, "Hello from slave task");
_numIterations = numIters;
- _mpiClient = mpiClient;
- _commGroup = _mpiClient.GetCommunicationGroup(GroupTestConstants.GroupName);
+ _groupCommClient = groupCommClient;
+ _commGroup = _groupCommClient.GetCommunicationGroup(GroupTestConstants.GroupName);
_broadcastReceiver = _commGroup.GetBroadcastReceiver<int[]>(GroupTestConstants.BroadcastOperatorName);
_triangleNumberSender = _commGroup.GetReduceSender<int[]>(GroupTestConstants.ReduceOperatorName);
}
@@ -78,7 +78,7 @@ namespace Org.Apache.REEF.Network.Examples.GroupCommunication.PipelineBroadcastR
public void Dispose()
{
- _mpiClient.Dispose();
+ _groupCommClient.Dispose();
}
private int TriangleNumber(int n)
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/c330dcff/lang/cs/Org.Apache.REEF.Network.Examples/GroupCommunication/ScatterReduceDriverAndTasks/MasterTask.cs
----------------------------------------------------------------------
diff --git a/lang/cs/Org.Apache.REEF.Network.Examples/GroupCommunication/ScatterReduceDriverAndTasks/MasterTask.cs b/lang/cs/Org.Apache.REEF.Network.Examples/GroupCommunication/ScatterReduceDriverAndTasks/MasterTask.cs
index a3d17d7..b949d56 100644
--- a/lang/cs/Org.Apache.REEF.Network.Examples/GroupCommunication/ScatterReduceDriverAndTasks/MasterTask.cs
+++ b/lang/cs/Org.Apache.REEF.Network.Examples/GroupCommunication/ScatterReduceDriverAndTasks/MasterTask.cs
@@ -31,18 +31,18 @@ namespace Org.Apache.REEF.Network.Examples.GroupCommunication.ScatterReduceDrive
{
private static readonly Logger _logger = Logger.GetLogger(typeof(MasterTask));
- private readonly IMpiClient _mpiClient;
+ private readonly IGroupCommClient _groupCommClient;
private readonly ICommunicationGroupClient _commGroup;
private readonly IScatterSender<int> _scatterSender;
private readonly IReduceReceiver<int> _sumReducer;
[Inject]
- public MasterTask(IMpiClient mpiClient)
+ public MasterTask(IGroupCommClient groupCommClient)
{
_logger.Log(Level.Info, "Hello from master task");
- _mpiClient = mpiClient;
+ _groupCommClient = groupCommClient;
- _commGroup = mpiClient.GetCommunicationGroup(GroupTestConstants.GroupName);
+ _commGroup = groupCommClient.GetCommunicationGroup(GroupTestConstants.GroupName);
_scatterSender = _commGroup.GetScatterSender<int>(GroupTestConstants.ScatterOperatorName);
_sumReducer = _commGroup.GetReduceReceiver<int>(GroupTestConstants.ReduceOperatorName);
}
@@ -60,7 +60,7 @@ namespace Org.Apache.REEF.Network.Examples.GroupCommunication.ScatterReduceDrive
public void Dispose()
{
- _mpiClient.Dispose();
+ _groupCommClient.Dispose();
}
private List<string> GetScatterOrder()
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/c330dcff/lang/cs/Org.Apache.REEF.Network.Examples/GroupCommunication/ScatterReduceDriverAndTasks/ScatterReduceDriver.cs
----------------------------------------------------------------------
diff --git a/lang/cs/Org.Apache.REEF.Network.Examples/GroupCommunication/ScatterReduceDriverAndTasks/ScatterReduceDriver.cs b/lang/cs/Org.Apache.REEF.Network.Examples/GroupCommunication/ScatterReduceDriverAndTasks/ScatterReduceDriver.cs
index fab12e5..88dac3d 100644
--- a/lang/cs/Org.Apache.REEF.Network.Examples/GroupCommunication/ScatterReduceDriverAndTasks/ScatterReduceDriver.cs
+++ b/lang/cs/Org.Apache.REEF.Network.Examples/GroupCommunication/ScatterReduceDriverAndTasks/ScatterReduceDriver.cs
@@ -45,19 +45,19 @@ namespace Org.Apache.REEF.Network.Examples.GroupCommunication.ScatterReduceDrive
private readonly int _numEvaluators;
- private readonly IMpiDriver _mpiDriver;
+ private readonly IGroupCommDriver _groupCommDriver;
private readonly ICommunicationGroupDriver _commGroup;
- private readonly TaskStarter _mpiTaskStarter;
+ private readonly TaskStarter _groupCommTaskStarter;
[Inject]
public ScatterReduceDriver(
[Parameter(typeof(GroupTestConfig.NumEvaluators))] int numEvaluators,
- MpiDriver mpiDriver)
+ GroupCommDriver groupCommDriver)
{
Identifier = "BroadcastStartHandler";
_numEvaluators = numEvaluators;
- _mpiDriver = mpiDriver;
- _commGroup = _mpiDriver.DefaultGroup
+ _groupCommDriver = groupCommDriver;
+ _commGroup = _groupCommDriver.DefaultGroup
.AddScatter<int, IntCodec>(
GroupTestConstants.ScatterOperatorName,
GroupTestConstants.MasterTaskId,
@@ -68,7 +68,7 @@ namespace Org.Apache.REEF.Network.Examples.GroupCommunication.ScatterReduceDrive
new SumFunction())
.Build();
- _mpiTaskStarter = new TaskStarter(_mpiDriver, numEvaluators);
+ _groupCommTaskStarter = new TaskStarter(_groupCommDriver, numEvaluators);
CreateClassHierarchy();
}
@@ -83,14 +83,14 @@ namespace Org.Apache.REEF.Network.Examples.GroupCommunication.ScatterReduceDrive
public void OnNext(IAllocatedEvaluator allocatedEvaluator)
{
- IConfiguration contextConf = _mpiDriver.GetContextConfiguration();
- IConfiguration serviceConf = _mpiDriver.GetServiceConfiguration();
+ IConfiguration contextConf = _groupCommDriver.GetContextConfiguration();
+ IConfiguration serviceConf = _groupCommDriver.GetServiceConfiguration();
allocatedEvaluator.SubmitContextAndService(contextConf, serviceConf);
}
public void OnNext(IActiveContext activeContext)
{
- if (_mpiDriver.IsMasterTaskContext(activeContext))
+ if (_groupCommDriver.IsMasterTaskContext(activeContext))
{
// Configure Master Task
IConfiguration partialTaskConf = TaskConfiguration.ConfigurationModule
@@ -99,13 +99,13 @@ namespace Org.Apache.REEF.Network.Examples.GroupCommunication.ScatterReduceDrive
.Build();
_commGroup.AddTask(GroupTestConstants.MasterTaskId);
- _mpiTaskStarter.QueueTask(partialTaskConf, activeContext);
+ _groupCommTaskStarter.QueueTask(partialTaskConf, activeContext);
}
else
{
// Configure Slave Task
string slaveTaskId = GroupTestConstants.SlaveTaskId +
- _mpiDriver.GetContextNum(activeContext);
+ _groupCommDriver.GetContextNum(activeContext);
IConfiguration partialTaskConf = TaskConfiguration.ConfigurationModule
.Set(TaskConfiguration.Identifier, slaveTaskId)
@@ -113,7 +113,7 @@ namespace Org.Apache.REEF.Network.Examples.GroupCommunication.ScatterReduceDrive
.Build();
_commGroup.AddTask(slaveTaskId);
- _mpiTaskStarter.QueueTask(partialTaskConf, activeContext);
+ _groupCommTaskStarter.QueueTask(partialTaskConf, activeContext);
}
}
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/c330dcff/lang/cs/Org.Apache.REEF.Network.Examples/GroupCommunication/ScatterReduceDriverAndTasks/SlaveTask.cs
----------------------------------------------------------------------
diff --git a/lang/cs/Org.Apache.REEF.Network.Examples/GroupCommunication/ScatterReduceDriverAndTasks/SlaveTask.cs b/lang/cs/Org.Apache.REEF.Network.Examples/GroupCommunication/ScatterReduceDriverAndTasks/SlaveTask.cs
index f2b1c4a..27906b4 100644
--- a/lang/cs/Org.Apache.REEF.Network.Examples/GroupCommunication/ScatterReduceDriverAndTasks/SlaveTask.cs
+++ b/lang/cs/Org.Apache.REEF.Network.Examples/GroupCommunication/ScatterReduceDriverAndTasks/SlaveTask.cs
@@ -31,18 +31,18 @@ namespace Org.Apache.REEF.Network.Examples.GroupCommunication.ScatterReduceDrive
{
private static readonly Logger _logger = Logger.GetLogger(typeof(SlaveTask));
- private readonly IMpiClient _mpiClient;
+ private readonly IGroupCommClient _groupCommClient;
private readonly ICommunicationGroupClient _commGroup;
private readonly IScatterReceiver<int> _scatterReceiver;
private readonly IReduceSender<int> _sumSender;
[Inject]
- public SlaveTask(IMpiClient mpiClient)
+ public SlaveTask(IGroupCommClient groupCommClient)
{
_logger.Log(Level.Info, "Hello from slave task");
- _mpiClient = mpiClient;
- _commGroup = _mpiClient.GetCommunicationGroup(GroupTestConstants.GroupName);
+ _groupCommClient = groupCommClient;
+ _commGroup = _groupCommClient.GetCommunicationGroup(GroupTestConstants.GroupName);
_scatterReceiver = _commGroup.GetScatterReceiver<int>(GroupTestConstants.ScatterOperatorName);
_sumSender = _commGroup.GetReduceSender<int>(GroupTestConstants.ReduceOperatorName);
}
@@ -61,7 +61,7 @@ namespace Org.Apache.REEF.Network.Examples.GroupCommunication.ScatterReduceDrive
public void Dispose()
{
- _mpiClient.Dispose();
+ _groupCommClient.Dispose();
}
}
}
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/c330dcff/lang/cs/Org.Apache.REEF.Network.Tests/GroupCommunication/GroupCommunicationTests.cs
----------------------------------------------------------------------
diff --git a/lang/cs/Org.Apache.REEF.Network.Tests/GroupCommunication/GroupCommunicationTests.cs b/lang/cs/Org.Apache.REEF.Network.Tests/GroupCommunication/GroupCommunicationTests.cs
index c0808d8..cb73cd6 100644
--- a/lang/cs/Org.Apache.REEF.Network.Tests/GroupCommunication/GroupCommunicationTests.cs
+++ b/lang/cs/Org.Apache.REEF.Network.Tests/GroupCommunication/GroupCommunicationTests.cs
@@ -99,9 +99,9 @@ namespace Org.Apache.REEF.Network.Tests.GroupCommunication
int numTasks = 3;
int fanOut = 2;
- var mpiDriver = GetInstanceOfMpiDriver(driverId, masterTaskId, groupName, fanOut, numTasks);
+ var groupCommunicationDriver = GetInstanceOfGroupCommDriver(driverId, masterTaskId, groupName, fanOut, numTasks);
- ICommunicationGroupDriver commGroup = mpiDriver.DefaultGroup
+ ICommunicationGroupDriver commGroup = groupCommunicationDriver.DefaultGroup
.AddBroadcast<int, IntCodec>(
broadcastOperatorName,
masterTaskId)
@@ -111,7 +111,7 @@ namespace Org.Apache.REEF.Network.Tests.GroupCommunication
new SumFunction())
.Build();
- var commGroups = CommGroupClients(groupName, numTasks, mpiDriver, commGroup);
+ var commGroups = CommGroupClients(groupName, numTasks, groupCommunicationDriver, commGroup);
//for master task
IBroadcastSender<int> broadcastSender = commGroups[0].GetBroadcastSender<int>(broadcastOperatorName);
@@ -154,9 +154,9 @@ namespace Org.Apache.REEF.Network.Tests.GroupCommunication
int numTasks = 5;
int fanOut = 2;
- IMpiDriver mpiDriver = GetInstanceOfMpiDriver(driverId, masterTaskId, groupName, fanOut, numTasks);
+ IGroupCommDriver groupCommDriver = GetInstanceOfGroupCommDriver(driverId, masterTaskId, groupName, fanOut, numTasks);
- ICommunicationGroupDriver commGroup = mpiDriver.DefaultGroup
+ ICommunicationGroupDriver commGroup = groupCommDriver.DefaultGroup
.AddScatter<int, IntCodec>(
scatterOperatorName,
masterTaskId)
@@ -166,7 +166,7 @@ namespace Org.Apache.REEF.Network.Tests.GroupCommunication
new SumFunction())
.Build();
- List<ICommunicationGroupClient> commGroups = CommGroupClients(groupName, numTasks, mpiDriver, commGroup);
+ List<ICommunicationGroupClient> commGroups = CommGroupClients(groupName, numTasks, groupCommDriver, commGroup);
IScatterSender<int> sender = commGroups[0].GetScatterSender<int>(scatterOperatorName);
IReduceReceiver<int> sumReducer = commGroups[0].GetReduceReceiver<int>(reduceOperatorName);
@@ -217,13 +217,13 @@ namespace Org.Apache.REEF.Network.Tests.GroupCommunication
int value = 1337;
int fanOut = 3;
- IMpiDriver mpiDriver = GetInstanceOfMpiDriver(driverId, masterTaskId, groupName, fanOut, numTasks);
+ IGroupCommDriver groupCommDriver = GetInstanceOfGroupCommDriver(driverId, masterTaskId, groupName, fanOut, numTasks);
- var commGroup = mpiDriver.DefaultGroup
+ var commGroup = groupCommDriver.DefaultGroup
.AddBroadcast<int, IntCodec>(operatorName, masterTaskId)
.Build();
- List<ICommunicationGroupClient> commGroups = CommGroupClients(groupName, numTasks, mpiDriver, commGroup);
+ List<ICommunicationGroupClient> commGroups = CommGroupClients(groupName, numTasks, groupCommDriver, commGroup);
IBroadcastSender<int> sender = commGroups[0].GetBroadcastSender<int>(operatorName);
IBroadcastReceiver<int> receiver1 = commGroups[1].GetBroadcastReceiver<int>(operatorName);
@@ -251,13 +251,13 @@ namespace Org.Apache.REEF.Network.Tests.GroupCommunication
int value = 1337;
int fanOut = 3;
- IMpiDriver mpiDriver = GetInstanceOfMpiDriver(driverId, masterTaskId, groupName, fanOut, numTasks);
+ IGroupCommDriver groupCommDriver = GetInstanceOfGroupCommDriver(driverId, masterTaskId, groupName, fanOut, numTasks);
- var commGroup = mpiDriver.DefaultGroup
+ var commGroup = groupCommDriver.DefaultGroup
.AddBroadcast(operatorName, masterTaskId)
.Build();
- List<ICommunicationGroupClient> commGroups = CommGroupClients(groupName, numTasks, mpiDriver, commGroup);
+ List<ICommunicationGroupClient> commGroups = CommGroupClients(groupName, numTasks, groupCommDriver, commGroup);
IBroadcastSender<int> sender = commGroups[0].GetBroadcastSender<int>(operatorName);
IBroadcastReceiver<int> receiver1 = commGroups[1].GetBroadcastReceiver<int>(operatorName);
@@ -285,13 +285,13 @@ namespace Org.Apache.REEF.Network.Tests.GroupCommunication
int value3 = 99;
int fanOut = 2;
- IMpiDriver mpiDriver = GetInstanceOfMpiDriver(driverId, masterTaskId, groupName, fanOut, numTasks);
+ IGroupCommDriver groupCommDriver = GetInstanceOfGroupCommDriver(driverId, masterTaskId, groupName, fanOut, numTasks);
- var commGroup = mpiDriver.DefaultGroup
+ var commGroup = groupCommDriver.DefaultGroup
.AddBroadcast<int, IntCodec>(operatorName, masterTaskId)
.Build();
- List<ICommunicationGroupClient> commGroups = CommGroupClients(groupName, numTasks, mpiDriver, commGroup);
+ List<ICommunicationGroupClient> commGroups = CommGroupClients(groupName, numTasks, groupCommDriver, commGroup);
IBroadcastSender<int> sender = commGroups[0].GetBroadcastSender<int>(operatorName);
IBroadcastReceiver<int> receiver1 = commGroups[1].GetBroadcastReceiver<int>(operatorName);
@@ -324,13 +324,13 @@ namespace Org.Apache.REEF.Network.Tests.GroupCommunication
string masterTaskId = "task0";
int fanOut = 2;
- IMpiDriver mpiDriver = GetInstanceOfMpiDriver(driverId, masterTaskId, groupName, fanOut, numTasks);
+ IGroupCommDriver groupCommDriver = GetInstanceOfGroupCommDriver(driverId, masterTaskId, groupName, fanOut, numTasks);
- var commGroup = mpiDriver.DefaultGroup
+ var commGroup = groupCommDriver.DefaultGroup
.AddReduce<int, IntCodec>(operatorName, "task0", new SumFunction())
.Build();
- List<ICommunicationGroupClient> commGroups = CommGroupClients(groupName, numTasks, mpiDriver, commGroup);
+ List<ICommunicationGroupClient> commGroups = CommGroupClients(groupName, numTasks, groupCommDriver, commGroup);
IReduceReceiver<int> receiver = commGroups[0].GetReduceReceiver<int>(operatorName);
IReduceSender<int> sender1 = commGroups[1].GetReduceSender<int>(operatorName);
@@ -359,13 +359,13 @@ namespace Org.Apache.REEF.Network.Tests.GroupCommunication
string masterTaskId = "task0";
int fanOut = 2;
- IMpiDriver mpiDriver = GetInstanceOfMpiDriver(driverId, masterTaskId, groupName, fanOut, numTasks);
+ IGroupCommDriver groupCommDriver = GetInstanceOfGroupCommDriver(driverId, masterTaskId, groupName, fanOut, numTasks);
- var commGroup = mpiDriver.DefaultGroup
+ var commGroup = groupCommDriver.DefaultGroup
.AddReduce<int, IntCodec>(operatorName, "task0", new SumFunction())
.Build();
- List<ICommunicationGroupClient> commGroups = CommGroupClients(groupName, numTasks, mpiDriver, commGroup);
+ List<ICommunicationGroupClient> commGroups = CommGroupClients(groupName, numTasks, groupCommDriver, commGroup);
IReduceReceiver<int> receiver = commGroups[0].GetReduceReceiver<int>(operatorName);
IReduceSender<int> sender1 = commGroups[1].GetReduceSender<int>(operatorName);
@@ -403,13 +403,13 @@ namespace Org.Apache.REEF.Network.Tests.GroupCommunication
int numTasks = 5;
int fanOut = 2;
- IMpiDriver mpiDriver = GetInstanceOfMpiDriver(driverId, masterTaskId, groupName, fanOut, numTasks);
+ IGroupCommDriver groupCommDriver = GetInstanceOfGroupCommDriver(driverId, masterTaskId, groupName, fanOut, numTasks);
- var commGroup = mpiDriver.DefaultGroup
+ var commGroup = groupCommDriver.DefaultGroup
.AddScatter(operatorName, masterTaskId)
.Build();
- List<ICommunicationGroupClient> commGroups = CommGroupClients(groupName, numTasks, mpiDriver, commGroup);
+ List<ICommunicationGroupClient> commGroups = CommGroupClients(groupName, numTasks, groupCommDriver, commGroup);
IScatterSender<int> sender = commGroups[0].GetScatterSender<int>(operatorName);
IScatterReceiver<int> receiver1 = commGroups[1].GetScatterReceiver<int>(operatorName);
@@ -442,13 +442,13 @@ namespace Org.Apache.REEF.Network.Tests.GroupCommunication
int numTasks = 5;
int fanOut = 2;
- IMpiDriver mpiDriver = GetInstanceOfMpiDriver(driverId, masterTaskId, groupName, fanOut, numTasks);
+ IGroupCommDriver groupCommDriver = GetInstanceOfGroupCommDriver(driverId, masterTaskId, groupName, fanOut, numTasks);
- var commGroup = mpiDriver.DefaultGroup
+ var commGroup = groupCommDriver.DefaultGroup
.AddScatter(operatorName, masterTaskId)
.Build();
- List<ICommunicationGroupClient> commGroups = CommGroupClients(groupName, numTasks, mpiDriver, commGroup);
+ List<ICommunicationGroupClient> commGroups = CommGroupClients(groupName, numTasks, groupCommDriver, commGroup);
IScatterSender<int> sender = commGroups[0].GetScatterSender<int>(operatorName);
IScatterReceiver<int> receiver1 = commGroups[1].GetScatterReceiver<int>(operatorName);
@@ -481,13 +481,13 @@ namespace Org.Apache.REEF.Network.Tests.GroupCommunication
int numTasks = 5;
int fanOut = 2;
- IMpiDriver mpiDriver = GetInstanceOfMpiDriver(driverId, masterTaskId, groupName, fanOut, numTasks);
+ IGroupCommDriver groupCommDriver = GetInstanceOfGroupCommDriver(driverId, masterTaskId, groupName, fanOut, numTasks);
- var commGroup = mpiDriver.DefaultGroup
+ var commGroup = groupCommDriver.DefaultGroup
.AddScatter<int, IntCodec>(operatorName, masterTaskId)
.Build();
- List<ICommunicationGroupClient> commGroups = CommGroupClients(groupName, numTasks, mpiDriver, commGroup);
+ List<ICommunicationGroupClient> commGroups = CommGroupClients(groupName, numTasks, groupCommDriver, commGroup);
IScatterSender<int> sender = commGroups[0].GetScatterSender<int>(operatorName);
IScatterReceiver<int> receiver1 = commGroups[1].GetScatterReceiver<int>(operatorName);
@@ -531,13 +531,13 @@ namespace Org.Apache.REEF.Network.Tests.GroupCommunication
int numTasks = 4;
int fanOut = 2;
- IMpiDriver mpiDriver = GetInstanceOfMpiDriver(driverId, masterTaskId, groupName, fanOut, numTasks);
+ IGroupCommDriver groupCommDriver = GetInstanceOfGroupCommDriver(driverId, masterTaskId, groupName, fanOut, numTasks);
- var commGroup = mpiDriver.DefaultGroup
+ var commGroup = groupCommDriver.DefaultGroup
.AddScatter<int, IntCodec>(operatorName, masterTaskId)
.Build();
- List<ICommunicationGroupClient> commGroups = CommGroupClients(groupName, numTasks, mpiDriver, commGroup);
+ List<ICommunicationGroupClient> commGroups = CommGroupClients(groupName, numTasks, groupCommDriver, commGroup);
IScatterSender<int> sender = commGroups[0].GetScatterSender<int>(operatorName);
IScatterReceiver<int> receiver1 = commGroups[1].GetScatterReceiver<int>(operatorName);
@@ -578,13 +578,13 @@ namespace Org.Apache.REEF.Network.Tests.GroupCommunication
int numTasks = 4;
int fanOut = 2;
- IMpiDriver mpiDriver = GetInstanceOfMpiDriver(driverId, masterTaskId, groupName, fanOut, numTasks);
+ IGroupCommDriver groupCommDriver = GetInstanceOfGroupCommDriver(driverId, masterTaskId, groupName, fanOut, numTasks);
- var commGroup = mpiDriver.DefaultGroup
+ var commGroup = groupCommDriver.DefaultGroup
.AddScatter<int, IntCodec>(operatorName, masterTaskId)
.Build();
- List<ICommunicationGroupClient> commGroups = CommGroupClients(groupName, numTasks, mpiDriver, commGroup);
+ List<ICommunicationGroupClient> commGroups = CommGroupClients(groupName, numTasks, groupCommDriver, commGroup);
IScatterSender<int> sender = commGroups[0].GetScatterSender<int>(operatorName);
IScatterReceiver<int> receiver1 = commGroups[1].GetScatterReceiver<int>(operatorName);
IScatterReceiver<int> receiver2 = commGroups[2].GetScatterReceiver<int>(operatorName);
@@ -641,25 +641,25 @@ namespace Org.Apache.REEF.Network.Tests.GroupCommunication
Assert.AreEqual(10, reduceFunction.Reduce(new int[] { 1, 2, 3, 4 }));
}
- public static IMpiDriver GetInstanceOfMpiDriver(string driverId, string masterTaskId, string groupName, int fanOut, int numTasks)
+ public static IGroupCommDriver GetInstanceOfGroupCommDriver(string driverId, string masterTaskId, string groupName, int fanOut, int numTasks)
{
var c = TangFactory.GetTang().NewConfigurationBuilder()
- .BindStringNamedParam<MpiConfigurationOptions.DriverId>(driverId)
- .BindStringNamedParam<MpiConfigurationOptions.MasterTaskId>(masterTaskId)
- .BindStringNamedParam<MpiConfigurationOptions.GroupName>(groupName)
- .BindIntNamedParam<MpiConfigurationOptions.FanOut>(fanOut.ToString())
- .BindIntNamedParam<MpiConfigurationOptions.NumberOfTasks>(numTasks.ToString())
+ .BindStringNamedParam<GroupCommConfigurationOptions.DriverId>(driverId)
+ .BindStringNamedParam<GroupCommConfigurationOptions.MasterTaskId>(masterTaskId)
+ .BindStringNamedParam<GroupCommConfigurationOptions.GroupName>(groupName)
+ .BindIntNamedParam<GroupCommConfigurationOptions.FanOut>(fanOut.ToString())
+ .BindIntNamedParam<GroupCommConfigurationOptions.NumberOfTasks>(numTasks.ToString())
.BindImplementation(GenericType<IConfigurationSerializer>.Class, GenericType<AvroConfigurationSerializer>.Class)
.Build();
- IMpiDriver mpiDriver = TangFactory.GetTang().NewInjector(c).GetInstance<MpiDriver>();
- return mpiDriver;
+ IGroupCommDriver groupCommDriver = TangFactory.GetTang().NewInjector(c).GetInstance<GroupCommDriver>();
+ return groupCommDriver;
}
- public static List<ICommunicationGroupClient> CommGroupClients(string groupName, int numTasks, IMpiDriver mpiDriver, ICommunicationGroupDriver commGroup)
+ public static List<ICommunicationGroupClient> CommGroupClients(string groupName, int numTasks, IGroupCommDriver groupCommDriver, ICommunicationGroupDriver commGroup)
{
List<ICommunicationGroupClient> commGroups = new List<ICommunicationGroupClient>();
- IConfiguration serviceConfig = mpiDriver.GetServiceConfiguration();
+ IConfiguration serviceConfig = groupCommDriver.GetServiceConfiguration();
List<IConfiguration> partialConfigs = new List<IConfiguration>();
for (int i = 0; i < numTasks; i++)
@@ -678,12 +678,12 @@ namespace Org.Apache.REEF.Network.Tests.GroupCommunication
for (int i = 0; i < numTasks; i++)
{
string taskId = "task" + i;
- IConfiguration mpiTaskConfig = mpiDriver.GetMpiTaskConfiguration(taskId);
- IConfiguration mergedConf = Configurations.Merge(mpiTaskConfig, partialConfigs[i], serviceConfig);
+ IConfiguration groupCommTaskConfig = groupCommDriver.GetGroupCommTaskConfiguration(taskId);
+ IConfiguration mergedConf = Configurations.Merge(groupCommTaskConfig, partialConfigs[i], serviceConfig);
IInjector injector = TangFactory.GetTang().NewInjector(mergedConf);
- IMpiClient mpiClient = injector.GetInstance<IMpiClient>();
- commGroups.Add(mpiClient.GetCommunicationGroup(groupName));
+ IGroupCommClient groupCommClient = injector.GetInstance<IGroupCommClient>();
+ commGroups.Add(groupCommClient.GetCommunicationGroup(groupName));
}
return commGroups;
}
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/c330dcff/lang/cs/Org.Apache.REEF.Network.Tests/GroupCommunication/GroupCommunicationTreeTopologyTests.cs
----------------------------------------------------------------------
diff --git a/lang/cs/Org.Apache.REEF.Network.Tests/GroupCommunication/GroupCommunicationTreeTopologyTests.cs b/lang/cs/Org.Apache.REEF.Network.Tests/GroupCommunication/GroupCommunicationTreeTopologyTests.cs
index b244f33..b159153 100644
--- a/lang/cs/Org.Apache.REEF.Network.Tests/GroupCommunication/GroupCommunicationTreeTopologyTests.cs
+++ b/lang/cs/Org.Apache.REEF.Network.Tests/GroupCommunication/GroupCommunicationTreeTopologyTests.cs
@@ -58,13 +58,13 @@ namespace Org.Apache.REEF.Network.Tests.GroupCommunication
string masterTaskId = "task0";
int fanOut = 3;
- var mpiDriver = GroupCommunicationTests.GetInstanceOfMpiDriver(driverId, masterTaskId, groupName, fanOut, numTasks);
+ var groupCommDriver = GroupCommunicationTests.GetInstanceOfGroupCommDriver(driverId, masterTaskId, groupName, fanOut, numTasks);
- ICommunicationGroupDriver commGroup = mpiDriver.DefaultGroup
+ ICommunicationGroupDriver commGroup = groupCommDriver.DefaultGroup
.AddReduce<int, IntCodec>(operatorName, masterTaskId, new SumFunction(), TopologyTypes.Tree)
.Build();
- var commGroups = GroupCommunicationTests.CommGroupClients(groupName, numTasks, mpiDriver, commGroup);
+ var commGroups = GroupCommunicationTests.CommGroupClients(groupName, numTasks, groupCommDriver, commGroup);
IReduceReceiver<int> receiver = commGroups[0].GetReduceReceiver<int>(operatorName);
IReduceSender<int> sender1 = commGroups[1].GetReduceSender<int>(operatorName);
@@ -114,13 +114,13 @@ namespace Org.Apache.REEF.Network.Tests.GroupCommunication
int value3 = 99;
int fanOut = 3;
- var mpiDriver = GroupCommunicationTests.GetInstanceOfMpiDriver(driverId, masterTaskId, groupName, fanOut, numTasks);
+ var groupCommDriver = GroupCommunicationTests.GetInstanceOfGroupCommDriver(driverId, masterTaskId, groupName, fanOut, numTasks);
- ICommunicationGroupDriver commGroup = mpiDriver.DefaultGroup
+ ICommunicationGroupDriver commGroup = groupCommDriver.DefaultGroup
.AddBroadcast<int, IntCodec>(operatorName, masterTaskId, TopologyTypes.Tree)
.Build();
- var commGroups = GroupCommunicationTests.CommGroupClients(groupName, numTasks, mpiDriver, commGroup);
+ var commGroups = GroupCommunicationTests.CommGroupClients(groupName, numTasks, groupCommDriver, commGroup);
IBroadcastSender<int> sender = commGroups[0].GetBroadcastSender<int>(operatorName);
IBroadcastReceiver<int> receiver1 = commGroups[1].GetBroadcastReceiver<int>(operatorName);
@@ -190,9 +190,9 @@ namespace Org.Apache.REEF.Network.Tests.GroupCommunication
int numTasks = 10;
int fanOut = 3;
- var mpiDriver = GroupCommunicationTests.GetInstanceOfMpiDriver(driverId, masterTaskId, groupName, fanOut, numTasks);
+ var groupCommDriver = GroupCommunicationTests.GetInstanceOfGroupCommDriver(driverId, masterTaskId, groupName, fanOut, numTasks);
- ICommunicationGroupDriver commGroup = mpiDriver.DefaultGroup
+ ICommunicationGroupDriver commGroup = groupCommDriver.DefaultGroup
.AddBroadcast<int, IntCodec>(
broadcastOperatorName,
masterTaskId,
@@ -204,7 +204,7 @@ namespace Org.Apache.REEF.Network.Tests.GroupCommunication
TopologyTypes.Tree)
.Build();
- var commGroups = GroupCommunicationTests.CommGroupClients(groupName, numTasks, mpiDriver, commGroup);
+ var commGroups = GroupCommunicationTests.CommGroupClients(groupName, numTasks, groupCommDriver, commGroup);
//for master task
IBroadcastSender<int> broadcastSender = commGroups[0].GetBroadcastSender<int>(broadcastOperatorName);
@@ -303,13 +303,13 @@ namespace Org.Apache.REEF.Network.Tests.GroupCommunication
int numTasks = 5;
int fanOut = 2;
- var mpiDriver = GroupCommunicationTests.GetInstanceOfMpiDriver(driverId, masterTaskId, groupName, fanOut, numTasks);
+ var groupCommDriver = GroupCommunicationTests.GetInstanceOfGroupCommDriver(driverId, masterTaskId, groupName, fanOut, numTasks);
- ICommunicationGroupDriver commGroup = mpiDriver.DefaultGroup
+ ICommunicationGroupDriver commGroup = groupCommDriver.DefaultGroup
.AddScatter<int, IntCodec>(operatorName, masterTaskId, TopologyTypes.Tree)
.Build();
- var commGroups = GroupCommunicationTests.CommGroupClients(groupName, numTasks, mpiDriver, commGroup);
+ var commGroups = GroupCommunicationTests.CommGroupClients(groupName, numTasks, groupCommDriver, commGroup);
IScatterSender<int> sender = commGroups[0].GetScatterSender<int>(operatorName);
IScatterReceiver<int> receiver1 = commGroups[1].GetScatterReceiver<int>(operatorName);
@@ -348,13 +348,13 @@ namespace Org.Apache.REEF.Network.Tests.GroupCommunication
int numTasks = 5;
int fanOut = 2;
- var mpiDriver = GroupCommunicationTests.GetInstanceOfMpiDriver(driverId, masterTaskId, groupName, fanOut, numTasks);
+ var groupCommDriver = GroupCommunicationTests.GetInstanceOfGroupCommDriver(driverId, masterTaskId, groupName, fanOut, numTasks);
- ICommunicationGroupDriver commGroup = mpiDriver.DefaultGroup
+ ICommunicationGroupDriver commGroup = groupCommDriver.DefaultGroup
.AddScatter<int, IntCodec>(operatorName, masterTaskId, TopologyTypes.Tree)
.Build();
- var commGroups = GroupCommunicationTests.CommGroupClients(groupName, numTasks, mpiDriver, commGroup);
+ var commGroups = GroupCommunicationTests.CommGroupClients(groupName, numTasks, groupCommDriver, commGroup);
IScatterSender<int> sender = commGroups[0].GetScatterSender<int>(operatorName);
IScatterReceiver<int> receiver1 = commGroups[1].GetScatterReceiver<int>(operatorName);
@@ -402,13 +402,13 @@ namespace Org.Apache.REEF.Network.Tests.GroupCommunication
int numTasks = 4;
int fanOut = 2;
- var mpiDriver = GroupCommunicationTests.GetInstanceOfMpiDriver(driverId, masterTaskId, groupName, fanOut, numTasks);
+ var groupCommDriver = GroupCommunicationTests.GetInstanceOfGroupCommDriver(driverId, masterTaskId, groupName, fanOut, numTasks);
- ICommunicationGroupDriver commGroup = mpiDriver.DefaultGroup
+ ICommunicationGroupDriver commGroup = groupCommDriver.DefaultGroup
.AddScatter<int, IntCodec>(operatorName, masterTaskId, TopologyTypes.Tree)
.Build();
- var commGroups = GroupCommunicationTests.CommGroupClients(groupName, numTasks, mpiDriver, commGroup);
+ var commGroups = GroupCommunicationTests.CommGroupClients(groupName, numTasks, groupCommDriver, commGroup);
IScatterSender<int> sender = commGroups[0].GetScatterSender<int>(operatorName);
IScatterReceiver<int> receiver1 = commGroups[1].GetScatterReceiver<int>(operatorName);
@@ -453,13 +453,13 @@ namespace Org.Apache.REEF.Network.Tests.GroupCommunication
int numTasks = 4;
int fanOut = 2;
- var mpiDriver = GroupCommunicationTests.GetInstanceOfMpiDriver(driverId, masterTaskId, groupName, fanOut, numTasks);
+ var groupCommDriver = GroupCommunicationTests.GetInstanceOfGroupCommDriver(driverId, masterTaskId, groupName, fanOut, numTasks);
- ICommunicationGroupDriver commGroup = mpiDriver.DefaultGroup
+ ICommunicationGroupDriver commGroup = groupCommDriver.DefaultGroup
.AddScatter<int, IntCodec>(operatorName, masterTaskId, TopologyTypes.Tree)
.Build();
- var commGroups = GroupCommunicationTests.CommGroupClients(groupName, numTasks, mpiDriver, commGroup);
+ var commGroups = GroupCommunicationTests.CommGroupClients(groupName, numTasks, groupCommDriver, commGroup);
IScatterSender<int> sender = commGroups[0].GetScatterSender<int>(operatorName);
IScatterReceiver<int> receiver1 = commGroups[1].GetScatterReceiver<int>(operatorName);
@@ -505,13 +505,13 @@ namespace Org.Apache.REEF.Network.Tests.GroupCommunication
int numTasks = 6;
int fanOut = 2;
- var mpiDriver = GroupCommunicationTests.GetInstanceOfMpiDriver(driverId, masterTaskId, groupName, fanOut, numTasks);
+ var groupCommDriver = GroupCommunicationTests.GetInstanceOfGroupCommDriver(driverId, masterTaskId, groupName, fanOut, numTasks);
- ICommunicationGroupDriver commGroup = mpiDriver.DefaultGroup
+ ICommunicationGroupDriver commGroup = groupCommDriver.DefaultGroup
.AddScatter<int, IntCodec>(operatorName, masterTaskId, TopologyTypes.Tree)
.Build();
- var commGroups = GroupCommunicationTests.CommGroupClients(groupName, numTasks, mpiDriver, commGroup);
+ var commGroups = GroupCommunicationTests.CommGroupClients(groupName, numTasks, groupCommDriver, commGroup);
IScatterSender<int> sender = commGroups[0].GetScatterSender<int>(operatorName);
IScatterReceiver<int> receiver1 = commGroups[1].GetScatterReceiver<int>(operatorName);
@@ -569,9 +569,9 @@ namespace Org.Apache.REEF.Network.Tests.GroupCommunication
int numTasks = 5;
int fanOut = 2;
- var mpiDriver = GroupCommunicationTests.GetInstanceOfMpiDriver(driverId, masterTaskId, groupName, fanOut, numTasks);
+ var groupCommDriver = GroupCommunicationTests.GetInstanceOfGroupCommDriver(driverId, masterTaskId, groupName, fanOut, numTasks);
- ICommunicationGroupDriver commGroup = mpiDriver.DefaultGroup
+ ICommunicationGroupDriver commGroup = groupCommDriver.DefaultGroup
.AddScatter<int, IntCodec>(
scatterOperatorName,
masterTaskId,
@@ -582,7 +582,7 @@ namespace Org.Apache.REEF.Network.Tests.GroupCommunication
new SumFunction(),
TopologyTypes.Tree).Build();
- var commGroups = GroupCommunicationTests.CommGroupClients(groupName, numTasks, mpiDriver, commGroup);
+ var commGroups = GroupCommunicationTests.CommGroupClients(groupName, numTasks, groupCommDriver, commGroup);
IScatterSender<int> sender = commGroups[0].GetScatterSender<int>(scatterOperatorName);
IReduceReceiver<int> sumReducer = commGroups[0].GetReduceReceiver<int>(reduceOperatorName);
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/c330dcff/lang/cs/Org.Apache.REEF.Network/Group/Config/GroupCommConfigurationOptions.cs
----------------------------------------------------------------------
diff --git a/lang/cs/Org.Apache.REEF.Network/Group/Config/GroupCommConfigurationOptions.cs b/lang/cs/Org.Apache.REEF.Network/Group/Config/GroupCommConfigurationOptions.cs
new file mode 100644
index 0000000..f86c546
--- /dev/null
+++ b/lang/cs/Org.Apache.REEF.Network/Group/Config/GroupCommConfigurationOptions.cs
@@ -0,0 +1,102 @@
+/**
+ * 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.Tang.Annotations;
+
+namespace Org.Apache.REEF.Network.Group.Config
+{
+ public class GroupCommConfigurationOptions
+ {
+ [NamedParameter("Name of the communication group")]
+ public class CommunicationGroupName : Name<string>
+ {
+ }
+
+ [NamedParameter("Name of the Group Communication operator")]
+ public class OperatorName : Name<string>
+ {
+ }
+
+ [NamedParameter("Driver identifier")]
+ public class DriverId : Name<string>
+ {
+ }
+
+ [NamedParameter("Timeout for receiving data", defaultValue: "50000")]
+ public class Timeout : Name<int>
+ {
+ }
+
+ [NamedParameter("Retry times", defaultValue: "5")]
+ public class RetryCount : Name<int>
+ {
+ }
+
+ [NamedParameter("sleep time to wait for handlers to be registered", defaultValue: "500")]
+ public class SleepTimeWaitingForHandler : Name<int>
+ {
+ }
+
+ [NamedParameter("Retry times to wait for handlers to be registered", defaultValue: "5")]
+ public class RetryCountWaitingForHanler : Name<int>
+ {
+ }
+
+ [NamedParameter("Master task identifier")]
+ public class MasterTaskId : Name<string>
+ {
+ }
+
+ [NamedParameter("Group name", defaultValue: "Group1")]
+ public class GroupName : Name<string>
+ {
+ }
+
+ [NamedParameter("Number of tasks", defaultValue: "5")]
+ public class NumberOfTasks : Name<int>
+ {
+ }
+
+ [NamedParameter("with of the tree in topology", defaultValue:"2")]
+ public class FanOut : Name<int>
+ {
+ }
+
+ [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/c330dcff/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
deleted file mode 100644
index 7ef575b..0000000
--- a/lang/cs/Org.Apache.REEF.Network/Group/Config/MpiConfigurationOptions.cs
+++ /dev/null
@@ -1,102 +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.Collections.Generic;
-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("Timeout for receiving data", defaultValue: "50000")]
- public class Timeout : Name<int>
- {
- }
-
- [NamedParameter("Retry times", defaultValue: "5")]
- public class RetryCount : Name<int>
- {
- }
-
- [NamedParameter("sleep time to wait for handlers to be registered", defaultValue: "500")]
- public class SleepTimeWaitingForHandler : Name<int>
- {
- }
-
- [NamedParameter("Retry times to wait for handlers to be registered", defaultValue: "5")]
- public class RetryCountWaitingForHanler : Name<int>
- {
- }
-
- [NamedParameter("Master task identifier")]
- public class MasterTaskId : Name<string>
- {
- }
-
- [NamedParameter("Group name", defaultValue: "Group1")]
- public class GroupName : Name<string>
- {
- }
-
- [NamedParameter("Number of tasks", defaultValue: "5")]
- public class NumberOfTasks : Name<int>
- {
- }
-
- [NamedParameter("with of the tree in topology", defaultValue:"2")]
- public class FanOut : Name<int>
- {
- }
-
- [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>>
- {
- }
- }
-}