Typed Group Communication

Group communication is a crucial feature for high-performance and Grid computing. While previous works and libraries proposed such a characteristic (e.g. MPI, or object-oriented frameworks), the use of groups imposed specific constraints on programmers, for instance the use of dedicated interfaces to trigger group communications. We aim at a more flexible mechanism by proposing a scheme where, given a Java class, one can initiate group communications using the standard public methods of the class together with the classical dot notation; in this way, group communications remains typed.

In order to ease the use of the group communication, we provide a set of static methods on the PAGroup class and a set of methods on the Group interface. ProActive also provides typed group communication, meaning that only methods defined on classes or interfaces implemented by members of the group can be called.

There are several ways to create groups of active objects. Similar to active objects, we have instantiation based creation and object based creation. Instantiation based creation is done through newGroup(..) and newGroupInParallel while object based creation is done through turnActiveAsGroup(...). Group creation takes several parameters similar to active object creation. To understand the parameters that are not explained here read Chapter 9, Active Objects: Creation And Advanced Concepts as it contains a detailed explanation of every active object creation parameter. The following diagrams contain all the methods in the PAGroup class and Group interface.

Figure 10.1. PAGroup class and Group interface

In this chapter we will show how to create a group, send a request to the members of the group, and retrieve the results. For the examples we will be using the following class that can be instantiated as an active object:

public class Worker implements Serializable{
	public Worker(){}//empty no-arg constructor needed by ProActive
	public Worker(Long age, String Name){};
	public void doNothingGracefully(){};
}

Any object that is reifiable can be included in a group. Groups are created using the static method PAGroup.newGroup(...) or PAGroup.newGroupInParallel(...). The common superclass for all the group members has to be specified, thus giving the group a minimal type. We will be using the Worker class specified above.

try{
	//create the active object on the first node on
	//the first virtual node available
	Worker workersGroup =  (Worker) PAGroup.newActiveAsGroup(
		Worker.class.getName(), null, someVirtualNode);
}
catch (NodeException nodeExcep){
	System.err.println(nodeExcep.getMessage());
}
catch (ActiveObjectCreationException aoExcep){
	System.err.println(aoExcep.getMessage());
}

We can also specify parameters for the creation. If we have a class A we can specify an array of Object to hold the parameters:

       node0 = NodeFactory.getDefaultNode();
       node1 = super.getANode();
       node2 = super.getANode();

       Object[][] params = { { "Agent0" }, { "Agent1" }, { "Agent2" } };
       Node[] nodes = { node0, node1, node2 };

       this.typedGroup = (A) PAGroup.newGroup(A.class.getName(), params, nodes);

We can access the individual members of the group using get(index)

A agent0 = (A) agentGroup.get(0);
A agent1 = (A) agentGroup.get(1);
A agent2 = (A) agentGroup.get(2);

Elements can be included into a typed group only if their class equals or extends the class specified at the group creation. For example, an object of class B (B extending A) can be included to a group of type A. However based on Java typing, only the methods defined in the class A can be invoked on the group.

The typed group representation we have presented corresponds to the functional view of groups of objects. In order to provide a dynamic management of groups, a second and complementary representation of a group has been designed. In order to manage a group, this second representation must be used instead. This second representation, the management representation, follows a more standard pattern for grouping objects: the Group interface.

We are careful to have a strong coherence between both representations of the same group, which implies that modifications executed through one representation are immediately reported on the other one. In order to switch from one representation to the other, two methods have been defined : the static method named PAGroup.getGroup, returns the Group form associated to the given group object; the method getGroupByType defined in the Group interface does the opposite.

Below is an example of when and how to use each representation of a group:

   // definition of one standard Java object
   // and two active objects
A a1 = new A();
A a2 = (A) PAActiveObject.newActive('A', paramsA[], node);
B b  = (B) PAActiveObject.newActive('B', paramsB[], node);
   // Note that B extends A
   // For management purposes, get the representation
   // as a group given a typed group, created with
   // code on the left column:
Group gA = PAGroup.getGroup(ag1);
   // Now, add objects to the group:
   // Note that active and non-active objects 
   // may be mixed in groups
gA.add(a1);
gA.add(a2);
gA.add(b); 
   // The addition of members to a group immediately
   // reflects on the typed group form,  so a method
   // can be invoked on the typed group  and will
   // reach all its current members
ag1.foo(); // the caller of ag1.foo() may not belong to ag1
   // A new reference to the typed group  
   // can also be built as follows
A ag1new = (A) gA.getGroupByType();

       Object[][] params = { { "Agent0" }, { "Agent1" }, { "Agent2" } };
       Node[] nodes = { NodeFactory.getDefaultNode(), super.getANode(), super.getANode() };
       this.typedGroup = (A) PAGroup.newGroup(A.class.getName(), params, nodes);

       Group<A> g = PAGroup.getGroup(this.typedGroup);

       g.add(new A("Agent3"));
       g.add(new A("Agent4"));
       g.add(new A("Agent5"));

The particularity of our group communication mechanism is that the result of a typed group communication is also a group. The result group is transparently built at invocation time, with a future for each elementary reply. It will be dynamically updated with the incoming results, thus gathering results. Nevertheless, the result group can be immediately used to execute another method call, even if all the results are not available. In that case the wait-by-necessity mechanism implemented by ProActive is used.

   // A method call on a group, returning a result
V vg = ag1.bar();
   // vg is a typed group of 'V': operation 
   // below is also a collective operation
   // triggered on results
vg.f1();

As said in the Group creation section, groups whose type is based on final classes or primitive types cannot be built. So, the construction of a dynamic group as a result of a group method call is also limited. Consequently, only methods whose return type is either void or is a 'reifiable type', in the sense of the Meta Object Protocol of ProActive, may be called on a group of objects; otherwise, they will raise an exception at run-time, because the transparent construction of a group of futures of non-reifiable types fails.

To take advantage with the asynchronous remote method call model of ProActive, some new synchronization mechanisms have been added. Static methods defined in the PAGroup class enable to execute various forms of synchronisation. For instance: waitOne, waitN, waitAll, waitTheNth, waitAndGet. Here is an exemple:

   // A method call on a typed group
V vg = ag1.bar(); 
   // To wait and capture the first returned 
   // member of vg
V v = (V) PAGroup.waitAndGetOne(vg);
   // To wait all the members of vg are arrived
PAGroup.waitAll(vg);

Regarding the parameters of a method call towards a group of objects, the default behaviour is to broadcast them to all members. But sometimes, only a specific portion of the parameters, usually dependent of the rank of the member in the group, may be really useful for the method execution, and so, parts of the parameter transmissions are useless. In other words, in some cases, there is a need to transmit different parameters to the various members.

A common way to achieve the scattering of a global parameter is to use the rank of each member of the group, in order to select the appropriate part that it should get in order to execute the method. There is a natural traduction of this idea inside our group communication mechanism:the use of a group of objects in order to represent a parameter of a group method call that must be scattered to its members.

The default behaviour regarding parameters passing for method call on a group, is to pass a deep copy of the group of type P to all members. Thus, in order to scatter this group of elements of type P instead, the programmer must apply the static method setScatterGroup of the PAGroup class to the group. In order to switch back to the default behaviour, the static method unsetScatterGroup is available.

We create a new group and a new group of parameters:

Object[][] params = { { "Agent0" }, { "Agent1" }, { "Agent2" } };
Node[] nodes = { NodeFactory.getDefaultNode(), super.getANode(), super.getANode() };
this.typedGroup = (A) PAGroup.newGroup(A.class.getName(), params, nodes);
Object[][] paramsParameter = { { "AgentA" }, { "AgentB" }, { "AgentC" } };
Node[] nodesParameter = { super.getANode(), NodeFactory.getDefaultNode(), super.getANode() };
this.parameterGroup = (A) PAGroup.newGroup(A.class.getName(), paramsParameter, nodesParameter);

And then set the scattering, dispatch and set unscatter

PAGroup.setScatterGroup(this.parameterGroup);
this.resultTypedGroup = this.typedGroup.asynchronousCall(this.parameterGroup);
PAGroup.unsetScatterGroup(this.parameterGroup);

Group members can be accessed by name if on addition we specify the name. To use the named members we create the group and add members with the addNamedElement(String key,E value) method from the org.objectweb.proactive.core.group.Group interface.

       typedGroup = (A) PAGroup.newGroup(A.class.getName());

       Group<A> group = PAGroup.getGroup(typedGroup);
       group.addNamedElement("number0", new A("Agent0"));
       group.add(new A("Agent1"));
       group.addNamedElement("number2", new A("Agent2"));

To retrieve the named members we use the getNamedElement(String) from the same interface.

A agent0_indexed = (A) group.get(0);
A agent0_named = (A) group.getNamedElement("number0");
A agent1_indexed = (A) group.get(1);
A agent2_named = (A) group.getNamedElement("number2");
A agent2_indexed = (A) group.get(2);

Unique serialization is an optimization option that allows performing the serialization of the arguments before streaming them. If unique serialization is activated the arguments will be transformed into a byte array once, and afterward streamed. The one time serialization is useful if the size of the arguments is significant and we consider the bandwidth saved more important than ability to stream the serialized objects as soon as possible. This behaviour can be toggled using setUniqueSerialization(Object) and unsetUniqueSerialization(Object) from the static class org.objectweb.proactive.api.PAGroup

PAGroup.setUniqueSerialization(this.typedGroup);
this.typedGroup.onewayCall();
PAGroup.unsetUniqueSerialization(this.typedGroup);

To learn more, see the JavaDoc and the paper [BBC02].