A a = (A) ProActive.newActive('A', params, node);
 // Create an active Object of type A in the JVM specified by Node

a.foo (param); 
 // A one way typed asynchronous communication towards the (remote) AO a
 // A request is sent to a,
v = a.bar (param);
 // A typed asynchronous communication with result.
 // v is first an awaited Future, to be transparently filled up after
 // service of the request, and reply
...
v.gee (param);
 // Use of the result of an asynchronous call.

 // If v is still an awaited future, it triggers an automatic 
 // wait: Wait-by-necessity

 

 boolean isAwaited(Object);
 // Returns True if the object is still an awaited Future 

 void  waitFor(Object);
 // Blocks until the object is no longer awaited

 // A request is sent to a,

 void  waitForAll(Vector);
 // Blocks until all the objects in Vector are no longer awaited 


 int waitForAny(Vector);
 // Blocks until one of the objects in Vector is no longer awaited.

 // Returns the index of the available future.

 

When an AO must implement an activity that is not FIFO, the RunActive interface has to be implemented: it specifies the AO behavior in the method named runActivity():

Interface RunActive 
void runActivity(Body body) 
 // The activity of the active object instance of the current class

 

Example:

public class A implements RunActive { 
 // Implements RunActive for programming a specific behavior 

 // runActivity() is automatically called when such an AO is created 
 public void  runActivity(Body body) { 
 	Service service = new Service(body); 
 	while ( terminate ) { 
 		... // Do some activity on its own 

		... 
 		... // Do some services, e.g. a FIFO service on method named foo 
 		service.serveOldest('foo'); 
 		... 
 		} 
 	} 
}

 

Two other interfaces can also be specified:

Interface InitActive 
void  initActivity(Body body) 
 // Initializes the activity of the active object. 
 // not called in case of restart after migration 

 // Called before runActivity() method, and only once: 

Interface EndActive 
void  endActivity(Body body) 
 // Finalizes the active object after the activity stops by itself. 
 // Called after the execution of runActivity() method, and only once: 
 // not called before a migration 

 

Even when an AO is busy doing its own work, it can remain reactive to external events (method calls). One just has to program non-blocking services to take into account external inputs. 

public class BusyButReactive implements RunActive { 

 public void  runActivity(Body body) { 
 	Service service = new Service(body); 
 	while ( ! hasToTerminate ) { 
 		... 
 		// Do some activity on its own ... 

 		... 
 		// Non blocking service ... 
 		service.serveOldest('changeParameters', 'terminate'); 
 		... 
 		} 
 	}  
 
 public void  changeParameters () {
 	...... 
 	// change computation parameters
 	} 

 public void  terminate (){ 
 	hasToTerminate=true;
 	} 
 }

It also allows one to specify explicit termination of AOs (there is currently no Distributed Garbage Collector). Of course, the reactivity is up to the length of going around the loop. Similar techniques can be used to start, suspend, restart, and stop AOs.

 

Non-blocking services: returns immediately if no matching request is pending

void serveOldest(); 
// Serves the oldest request in the request queue 

void  serveOldest(String methodName) 

// Serves the oldest request aimed at a method of name methodName 

void serveOldest(RequestFilter requestFilter) 
// Serves the oldest request matching the criteria given be the filter

 

Blocking services: waits until a matching request can be served

void blockingServeOldest(); 
// Serves the oldest request in the request queue 

void  blockingServeOldest(String methodName) 

// Serves the oldest request aimed at a method of name methodName 

void blockingServeOldest(RequestFilter requestFilter) 
// Serves the oldest request matching the criteria given be the filter

 

Blocking timed services: wait a matching request at most a time given in ms

void blockingServeOldest (long timeout) 

// Serves the oldest request in the request queue. 
// Returns after timeout (in ms) if no request is available 
 
void  blockingServeOldest(String methodName, long timeout) 
// Serves the oldest request aimed at a method of name methodName 
// Returns after timeout (in ms) if no request is available 

void blockingServeOldest(RequestFilter requestFilter) 
// Serves the oldest request matching the criteria given be the filter

 

Waiting primitives:

void waitForRequest(); 
// Wait until a request is available or until the body terminates 

void waitForRequest(String methodName); 

// Wait until a request is available on the given method name, 
// or until the body terminates

 

Others:

void fifoServing(); 
// Start a FIFO service policy. Call does not return. In case of 
// a migration, a new runActivity() will be started on the new site 

void  lifoServing() 

// Invoke a LIFO policy. Call does not return. In case of 
// a migration, a new runActivity() will be started on the new site 

void  serveYoungest() 
// Serves the youngest request in the request queue 

void  flushAll() 
// Removes all requests in the pending queue

 

Object newActive(String classname, Object[] constructorParameters,Node node);
 // Creates a new AO of type classname. The AO is located on the given node,
 // or on a default node in the local JVM if the given node is nul

Object newActive(String classname,Object[] constructorParameters,VirtualNode virtualnode);
 // Creates a new set of AO of type classname.

 // The AO are located on each JVMs the Virtual Node is mapped onto

Object turnActive(Object, Node node);
 // Copy an existing Java object and turns it into an AO.
 // The AO is located on the given node, or on a default node in 

 

A ga = (A) ProActiveGroup.newGroup( 'A', params, nodes);
 // Created at once a group of AO of type 'A' in the JVMs specified 

 // by nodes. ga is a Typed Group of type 'A'. 
 // The number of AO being created matches the number of param arrays.
 // Nodes can be a Virtual Node defined in an XML descriptor */
ga.foo(...);
 // A general group communication without result.
 // A request to foo is sent in parallel to AO in group ga */
V gv = ga.bar(...);
 // A general group communication with a result.

 // gv is a typed group of 'V', which is first a group
 // of awaited Futures, to be filled up asynchronously 
gv.gee (...);
 // Use of the result of an asynchronous group call. It is also a 
 // collective operation: gee method is called in parallel on each object\
 in group. 
 // Wait-by-necessity occurs when results are awaited */
Group ag = ProActiveGroup.getGroup(ga);
 // Get the group representation of a typed group

ag.add(o);
 // Add object in the group ag. o can be a standard Java object or an AO,
 // and in any case must be of a compatible type 
ag.remove(index)
 // Removes the object at the specified index
A ga2 = (A) ag.getGroupByType();
 // Returns to the typed view of a group
 
void setScatterGroup(g);
 // By default, a group used as a parameter of a group communication

 // is sent to all as it is (deep copy of the group).
 // When set to scatter, upon a group call (ga.foo(g)) such a scatter 
 // parameter is dispatched in a round robing fashion to AOs in the
 // target group, e.g. upon ga.foo(g) */ 
 
void  unsetScatterGroup(g);
 // Get back to the default: entire group transmission in all group

 // communications, e.g. upon ga.foo(g) */

 

Methods both in Interface Group, and static in class ProActiveGroup

boolean  ProActiveGroup.allAwaited (Object); 
// Returns True if object is a group and all members are still awaited 

boolean ProActiveGroup.allArrived (Object); 

// Returns False only if at least one member is still awaited 

void  ProActiveGroup.waitAll (Object); 
// Wait for all the members in group to arrive (all no longer awaited) 

void  ProActiveGroup.waitN (Object, int nb); 

// Wait for at least nb members in group to arrive

int  ProActiveGroup.waitOneAndGetIndex (Object); 
// Waits for at least one member to arrived, and returns its index

 

A spmdGroup = (A) ProSPMD.newSPMDGroup('A', params, nodes);
 // Creates an SPMD group and creates all members with params on the nodes.
 // An SPMD group is a typed group in which every member has a reference to

 // the others (the SPMD group itself).

A mySpmdGroup = (A) ProSPMD.getSPMDGroup();
 // Returns the SPMD group of the activity.

int rank = ProSPMD.getMyRank();
 // Returns the rank of the activity in its SPMD group.

ProSPMD.barrier('barrierID');
 // Blocks the activity (after the end of the current service) until all

 // other members of the SPMD group invoke the same barrier. 
 // Three barriers are available: total barrier, neighbors based barrier
 // and method based barrier. 

 

Methods both in Interface Group, and static in class ProActiveGroup

void  migrateTo(Object o); 

// Migrate the current AO to the same JVM as the AO 

void  void migrateTo(String nodeURL); 
// Migrate the current AO to JVM given by the node URL 

int  void migrateTo(Node node); 

// Migrate the current AO to JVM given by the node

 

To initiate the migration of an object from outside, define a public method, that upon service will call the static migrateTo primitive:

public void moveTo(Object) { 
	try{ ProActive.migrateTo(t); } 
 	catch (Exception e) { 
 		e.printStackTrace(); 
 		logger.info('Cannot migrate.'); 
 	} 
} 

void onDeparture(String MethodName); 
// Specification of a method to execute before migration 

void  onArrival(String MethodName); 

// Specification of a method to execute after migration, upon the 
// arrival in a new JVM 

void setMigrationStrategy(MigrationStrategy); 
// Specifies a migration itinerary 

void migrationStrategy.add(Destination); 
// Adds a JVM destination to an itinerary 

void migrationStrategy.remove(Destination d) ; 
// Remove a JVM destination in an itinerary

 

Components are formed from AOs, a component is linked and communicates with other remote components. A component can be composite, made of other components, and as such itself distributed over several machines. Component systems are defined in XML files (ADL: Architecture Description Language); these files describe the definition, the assembly, and the bindings of components.

Components follow the Fractal hierarchical component model specification and API, see http://fractal.objectweb.org

The following methods are specific to ProActive.

In the class org.objectweb.proactive.ProActive:

Component newActiveComponent('A', params, VirtualNode, ComponentParameters); 
// Creates a new ProActive component from the specified class A. 
// The component is distributed on JVMs specified by the Virtual Node 
// The ComponentParameters defines the configuration of a component: 

// name of component, interfaces (server and client), etc. 
// Returns a reference to a component, as defined in the Fractal API

 

In the class org.objectweb.proactive.core.component.Fractive:

ProActiveInterface createCollectiveClientInterface(String itfName, String itfSignature); 
// This method is used in primitive components. 
// It generates a client collective interface named itfName, and typed as itfSignature. 
// This collective interface is a typed ProActive group.

 

An X.509 Public Key Infrastructure (PKI) allowing communication Authentication, Integrity, and Confidentiality (AIC) to be configured in an XML security file, at deployment, outside any source code. Security is compatible with mobility, allows for hierarchical domain specificationand dynamically negotiated policies.

Example of specification:

<Rule> 
 <From>
 <Entity type='VN' name='VN1'/> 
 </From>
 <To> 
 <Entity type='VN' name='VN2'/> 
 </To> 
 <Communication> 
 <Request value='authorized'> 
 <Attributes authentication='required' integrity='required' confidentiality='optional'/> 
 </Request> 
 </Communication> 
 <Migration>denied</Migration> 
 <AOCreation>denied</AOCreation> 

</Rule>

This rule specifies that: from Virual Node 'VN1' to the VN 'VN2', the communications (requests) are authorized, provided authentication and integrity are being used, while confidentiality is optional. Migration and AO creation are not authorized.

 

Virtual Nodes (VN) allow one to specify the location where to create AOs. A VN is uniquely identified as a String, is defined in an XML Deployment Descriptor where it is mapped onto JVMs. JVMs are themselves mapped onto physical machines: VN --> JVMs --> Machine. Various protocols can be specified to create JVMs onto machines (ssh, Globus, LSF, PBS, rsh, rlogin, Web Services, etc.). After activation, a VN contains a set of nodes, living in a set of JVMs. Overall, VNs and deployment descriptors allow to abstract away from source code: machines, creation, lookup and registry protocols.

Descriptor example: creates one jvm on the local machine

<ProActiveDescriptor xmlns:xsi='http://www.w3.org/2001/XMLSchema-instance' 
 xsi:noNamespaceSchemaLocation='DescriptorSchema.xsd'>

 <virtualNodesDefinition>
 <virtualNode name='Dispatcher'/> <!-- Name of the Virtual Node that will be used in
 program source -->
 </virtualNodesDefinition>
 <componentDefinition/>

 <deployment>
 <mapping>
 <!-- This part contains the mapping VNs -- JVMs -->
 <map virtualNode='Dispatcher'>
 <jvmSet>

 <vmName value='Jvm1'/> <!-- Virtual Node Dispatcher is mapped onto
 Jvm1 -->
 </jvmSet>
 </map>

 </mapping>
 <jvms>
 <jvm name='Jvm1'>
 <!-- This part defines how the jvm will be obtained: creation or
 acquisition: creation in this example -->

 <creation>
 <processReference refid='creationProcess'/>
 <!-- Jvm1 will be created using creationProcess defined below -->
 </creation>

 </jvm>
 </jvms>
 </deployment>
 <infrastructure>
 <processes>

 <processDefinition id='creationProcess'>
 <!-- Definition of creationProcess referenced above -->
 <jvmProcess class='org.objectweb.proactive.core.process.JVMNodeProcess'/>
 <!-- creationProcess is a jvmProcess. The jvm will be created on
 the local machine using default settings (classpath, java path,...) -->

 </processDefinition>
 </processes>
 </infrastructure>
 <componentDefinition>
</ProActiveDescriptor>

 

Deployment API

ProActiveDescriptor pad = ProActive.getProActiveDescriptor(String File); 
// Returns a ProActiveDescriptor object from the xml 
// descriptor file name 

pad.activateMapping(String VN); 
// Activates the given Virtual Node: launches or acquires 
// all the JVMs the VN is mapped onto

pad.activateMappings(); 
// Activates all VNs defined in the ProActiveDescriptor 

VirtualNode vn = pad.getVirtualNode(String) 
// Created at once a group of AO of type 'A' in the JVMs specified 
// by the given vn. The Virtual Node is automatically activated if not 
// explicitly done before 

Node[] n = vn.getNodes(); 
// Returns all nodes mapped to the target Virtual Node 

Object[] n[0].getActiveObjects(); 
// Returns a reference to all AOs deployed on the target Node 

ProActiveRuntime part = n[0].getProActiveRuntime(); 
// Returns a reference to the ProActive Runtime (the JVM) where the 
// node has been created 

pad.killall(boolean softly); 
// Kills all the JVMs deployed with the descriptor 
// not softly: all JVMs are killed abruptely 

// softly: all JVMs that originated the creation of a rmi registry 
// wait until registry is empty before dying

 

Functional exceptions with asynchrony

ProActive.tryWithCatch(MyException.class);
// Just before the try 
try { 
	// Some asynchronous calls with exceptions 

 	// One can use ProActive.throwArrivedException() and 
 	// ProActive.waitForPotentialException() here 
 	ProActive.endTryWithCatch();
 	// At the end of the try 
} catch (MyException e) { 
 	// ... 

} finally { 
 	ProActive.removeTryWithCatch();
 	// At the beginning of the finally }

 

Non-Functional Exceptions

Adding a handler to an active object on its side:

ProActive.addNFEListenerOnAO(myAO, new NFEListener() { 

 public boolean handleNFE(NonFunctionalException nfe) { 
 	// Do something with the exception... 
 	// Return true if we were able to handle it
 	return true; 
 	} 
 });

Handlers can also be added to the client side of an active object with

ProActive.addNFEListenerOnProxy(ao, handler)

or to a JVM with

ProActive.addNFEListenerOnJVM(handler)

These handlers can also be removed with

ProActive.removeNFEListenerOnAO(ao, handler),
ProActive.removeNFEListenerOnProxy(ao, handler),
ProActive.removeNFEListenerOnJVM(handler)

It's possible to define an handler only for some exception types, for example:

ProActive.addNFEListenerOnJVM(new TypedNFEListener(
 	SendRequestCommunicationException.class,
 	new NFEListener() { 
 		public boolean handleNFE(NonFunctionalException e) { 
 			// Do something with the SendRequestCommunicationException... 

 			// Return true if we were able to handle it 
 			return true; 
 		} }
 ));

The behaviour of the default handler (if none could handle the exception) is to throw the exception if it's on the proxy side, or log it if it's on the body side.

 

ProActive allows active objects exportation as web services. The service is deployed onto a Jakarta Tomcat web server with a given url. It is identified by its urn, an unique id of the service. It is also possible to choose the exported methods of the object.

The WSDL file matching the service will be accesible at http://localhost:8080/servlet/wsdl?id=a for a service which name is 'a' and which id deployed on a web server which location is http://localhost:8080.

A a = (A) ProActive.newActive('A', new Object []{}); 
 // Constructs an active object 

String [] methods = new String [] {'foo', 'bar'}; 
//A String array containing the exported methods

ProActive.exposeAsWebService(a,'http://localhost:8080','a',methods);
//Export the active object as a web service

ProActive.unExposeAsWebService('a', 'http://localhost:8080'); 
//Undeploy the service 'a' on the web server located at http://localhost:8080

 

ProActive can provide fault-tolerance capabilities through two differents protocols: a Communication-Induced Checkpointing protocol (CIC) or a pessimistic message logging protocol (PML). Making a ProActive application fault-tolerant is fully transparent; active objects are turned fault-tolerant using Java properties that can be set in the deployment descriptor. The programmer can select at deployment time the most adapted protocol regarding the application and the execution environment.

A Fault-tolerant deployment descriptor

<ProActiveDescriptor> 
 ...
 <virtualNodesDefinition> 
 <virtualNode name='NonFT-Workers' property='multiple'/> 
 <virtualNode name='FT-Workers' property='multiple' ftServiceId='appli'/>

 </virtualNodesDefinition> 
 ... 
 <serviceDefinition id='appli'>
 <faultTolerance> 
 
 <!-- Protocol selection: cic or pml -->
 <protocol type='cic' /> 

 <!-- URL of the fault-tolerance server --> 
 <globalServer url='rmi://localhost:1100/FTServer'/>

 <!-- URL of the resource server; all the nodes mapped on this virtual 
 node will be registred in as resource nodes for recovery --> 
 <resourceServer url='rmi://localhost:1100/FTServer'/>

 <!-- Average time in seconds between two consecutive checkpoints for each object -->
 <ttc value='5'/> 
 </faultTolerance> 
 </serviceDefinition> 
 </services> 
 ...

</ProActiveDescriptor>

 

Starting the fault-tolerance server

The global fault-tolerance server can be launched using the ProActive/scripts/[unix|windows]/FT/startGlobalFTServer.[sh|bat] script, with 5 optional parameters:

This aims to help you to create a P2P infrastructure over your desktop workstations network. It is self-organized and configurable. The infrastructure maintains a dynamic JVMs network for deploying computational applications.

Deploying the Infrastructure:

Firstly, you have to start P2P Services on each shared machine:

$ cd ProActive/scripts/unix/p2p

$ ./startP2PService [-acq acquisitionMethod] [-port portNumber] [-s Peer ...]

With that parameters (all are optionals):

A simple example:

first.peer.host$ ./startP2PService.sh

second.peer.host$ ./startP2PService.sh -s //first.peer.host

third.peer.host$ ./startP2PService.sh -s //second.peer.host

Acquiring Nodes:

Now you have a P2P Infrastructure running, you might want to deploy your ProActive application on it. That is simple, just modify the XML deployment descriptor:

... 
<jvms> 
 <jvm name='Jvm1'> 
 <acquisition>
 <serviceReference refid='p2plookup'/> 
 </acquisition>

 </jvm> 
 ... 
</jvms> 
... 
<infrastructure> 
 ...
 <services> 
 <serviceDefinition id='p2plookup'> 
 <P2PService nodesAsked='2' acq='rmi' port='6666'> 
 <peerSet>

 <peer>//second.peer.host</peer> 
 </peerSet>
 </P2PService> 
 </serviceDefinition> 
 ... 
 </services> 
 ...

</infrastructure> 
...

 

In the nodesAsked argument, a special value MAX is allowed. When it is used, the P2P infrastructure returns the maximun number of nodes avilable, and continue while the application running to return new nodes to the application. To use all the benefit of that feature, you might add a nodes creation event listener to your application.

Usage Example:

// getting the p2p virtual node 
VirtualNode vn = pad.getVirtualNode('p2pvn'); 

// adding 'this' as a listener

((VirtualNodeImpl) vn).addNodeCreationEventListener(this); 

// then activate the virtual node 
vn.activate();

 

'this' has to implement the NodeCreationEventListener interface:

public void nodeCreated(NodeCreationEvent event) { 
 	// get the node Node 
 	newNode = event.getNode(); 
 	// now you can create an active object on your node.
 }

 

Firstly, create your own task:

import org.objectweb.proactive.branchnbound.core.Task; 

public class YourTask extends Task { 
 

 public Result execute() { 
 	// Your code here for computing a solution 
 } 

 public Vector split() { 
 	// Your code for generating sub-tasks 
 } 
 
 public Result gather(Result[] results) { 
 	// Override optional 

 	// Default behavior based on the smallest gave by the compareTo 
 } 

 public void initLowerBound() { 
 	// Your code here for computing a lower bound 
 } 

 public void initUpperBound() { 
 	// Your code here for computing a lower bound 
 } 

 public int compareTo(Object arg) { 
 	// Strongly recommended to override this method 

 	// with your behavior 
 } 

}

 

How to interact with the framework from inside a task:

Secondly, choose your task queue:

Finally, start the compution:

Task task = new YourTask(someArguments); 
Manager manager = ProActiveBranchNBound.newBnB(task, nodes, LargerQueueImpl.class.getName()); 

Result futureResult = manager.start();
 // this call is asynchronous ...

 

Keep in mind that is only 'initLower/UpperBound' and 'split' methods are called on the root task. The 'execute' method is called on the root task's splitted task. Here the methods order execution:

File Transfer Deployment is a tool for transfering files at deployment time. This files are specified using the ProActive XML Deployment Descriptor in the following way: 

<VirtualNode name='exampleVNode' FileTransferDeploy='example'/>

....
</deployment>
<FileTransferDefinitions>
 <FileTransfer id='example'>
 <file src='hello.dat' dest='world.dat'/>

 <dir src='exampledir' dest='exampledir'/>
 </FileTransfer>
 ...
</FileTransferDefinitions>
<infrastructure>
....

<processDefinition id='xyz'>
 <sshProcess>...
 <FileTransferDeploy='implicit'> 
<!-- referenceID or keyword 'implicit' (inherit)-->
 <copyProtocol>processDefault, scp, rcp</copyProtocol>

 <sourceInfo prefix='/home/user'/>
 <destinationInfo prefix='/tmp' hostname='foo.org' username='smith' />
 </FileTransferDeploy>
 </sshProcess>

</processDefinition>
...