Reference Card

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

// A one way typed asynchronous communication towards the (remote) active object a
// A request is sent to a
a.foo (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 = a.bar (param);

...

// Use of the result of an asynchronous call.
// If v is still an awaited future, it triggers an automatic 
// wait-by-necessity
v.gee (param);

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

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

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

// Blocks until one of the objects in Vector is no longer awaited
// Returns the index of the available future
int PAFuture.waitForAny(Vector);

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():

package org.objectweb.proactive;

import org.objectweb.proactive.annotation.PublicAPI;


/**
 * <P>
 * RunActive is related to the activity of an active object.
 * When an active object is started, which means that its
 * active thread starts and serves the requests being sent
 * to its request queue, it is possible to define exactly how
 * the activity (the serving of requests amongst others) will
 * be done.
 * </P><P>
 * An object implementing this interface is invoked to run the
 * activity until an event trigger its end. The object being
 * reified as an active object can directly implement this interface
 * or an external class can also be used.
 * </P>
 * <P>
 * It is the role of the body of the active object to perform the
 * call on the object implementing this interface. For an active object
 * to run an activity, the method <code>runActivity</code> must not end
 * before the end of the activity. When the method <code>runActivity</code>
 * ends, the activity ends too and the <code>endActivity</code> can be invoked.
 * </P>
 * <P>
 * Here is an example of a simple implementation of <code>runActivity</code> method
 * doing a FIFO service of the request queue :
 * </P>
 * <pre>
 * public void runActivity(Body body) {
 *   Service service = new Service(body);
 *   while (body.isActive()) {
 *     service.blockingServeOldest();
 *   }
 * }
 * </pre>
 *
 * @author The ProActive Team
 * @version 1.0,  2002/06
 * @since   ProActive 0.9.3
 */
@PublicAPI
public interface RunActive extends Active {

    /**
     * Runs the activity of the active object.
     * @param body the body of the active object being started
     */
    public void runActivity(Body body);
}

Example:

 // Implements RunActive for programming a specific behavior 
public class A implements RunActive { 
   // 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:

The method initActivity(Body body) initializes the activity of the active object. It is called once before the runActivity(Body body) method. It is not called again if the active object restarts after migration.

package org.objectweb.proactive;

import org.objectweb.proactive.annotation.PublicAPI;


/**
 * <P>
 * InitActive is related to the initialization of the activity of an
 * active object. The initialization of the activity is done only once.
 * In case of a migration, an active object restarts its activity
 * automatically without reinitializing.
 * </P><P>
 * An object implementing this interface can be invoked to perform the
 * initialization work before the activity is started. The object being
 * reified as an active object can implement this interface or an external
 * class can also be used.
 * </P>
 * <P>
 * It is generally the role of the body of the active object to perform the
 * call on the object implementing this interface.
 * </P>
 *
 * @author The ProActive Team
 * @version 1.0,  2002/06
 * @since   ProActive 0.9.3
 */
@PublicAPI
public interface InitActive extends Active {

    /**
     * Initializes the activity of the active object.
     * @param body the body of the active object being initialized
     */
    public void initActivity(Body body);
}

The method endActivity(Body body) finalizes the activity of the active object. It is called once after the execution of the runActivity(Body body) method.

package org.objectweb.proactive;

import org.objectweb.proactive.annotation.PublicAPI;


/**
 * <P>
 * EndActive is related to the finalization of an active object.
 * The finalization of the activity is done only once when the object
 * stops to be active and becomes unusable.
 * In case of a migration, an active object stops its activity
 * before restarting on a new VM automatically without finalization.
 * </P><P>
 * An object implementing this interface can be invoked to perform the
 * finalization work after the activity is ended. The object being
 * reified as an active object can implement this interface or an external
 * class can also be used.
 * </P>
 * <P>
 * It is generally the role of the body of the active object to perform the
 * call on the object implementing this interface.
 * </P>
 * <P>
 * It is hard to ensure that the <code>endActivity</code> method will indeed
 * be invoked at the end of the activity. <code>Error<code>, <code>Exception<code>,
 * customized activity that never ends or sudden death of the JVM can prevents
 * this method to be called by the body of the active object.
 * </P>
 *
 * @author The ProActive Team
 * @version 1.0,  2002/06
 * @since   ProActive 0.9.3
 */
@PublicAPI
public interface EndActive extends Active {

    /**
     * Finalized the active object after the activity has been stopped.
     * @param body the body of the active object being finalized.
     */
    public void endActivity(Body body);
}

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. 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.

The following method are in the Service class. They can be used to control how requests are served.

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

// Serves the oldest request in the request queue <emphasis role="bold">
void serveOldest(); 

// Serves the oldest request aimed at a method of name methodName 
void serveOldest(String methodName) 

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

Blocking services: waits until a matching request can be served

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

//Serves the oldest request aimed at a method of name methodName 
void blockingServeOldest(String methodName) 
        
// Serves the oldest request matching the criteria given be the filter
void blockingServeOldest(RequestFilter requestFilter) 

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

// Serves the oldest request in the request queue. 
// Returns after timeout (in ms) if no request is available 
void blockingServeOldest (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(String methodName, long timeout) 

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

Waiting primitives:

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

// Wait until a request is available on the given method name, 
// or until the body terminates
void  waitForRequest(String methodName); 

Others:

// 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 fifoServing(); 

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

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

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

// 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 null
Object PAActiveObject.newActive(String classname, Object[] constructorParameters, Node node);

// Creates a new set of AO of type classname.
// The AO are located on each JVMs the Virtual Node is mapped onto
Object PAActiveObject.newActive(String classname,Object[] constructorParameters,VirtualNode
 virtualnode);

// 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 
Object turnActive(Object, Node node);

// Created 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 */
A ga = (A) PAGroup.newGroup( 'A', params, nodes);

// A general group communication without result.
// A request to foo is sent in parallel to the active objects in the in group ga  */
ga.foo(...);

// 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 
V gv = ga.bar(...);

// 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  the group. 
// Wait-by-necessity occurs when the results are awaited
gv.gee (...);

// Get the group representation of a typed group
Group ag = PAGroup.getGroup(ga);

// Add object in the group ag.
// o can be a standard Java object or an AO,
// it must be of a compatible type 
ag.add(o);

// Removes the object at the specified index
ag.remove(index)

// Returns to the typed view of a group
A ga2 = (A) ag.getGroupByType();
   
// By default, a group used as a parameter of a group communication
// is sent as a deep copy of the group.
// On a group call (ga.foo(g)) the scatter 
// parameter is dispatched in a round robin fashion to the active objects in the
// target group, e.g. upon ga.foo(g)
void PAGroup.setScatterGroup(g);
     
// Get back to the default: entire group transmission in all group
// communications, e.g. upon ga.foo(g) 
void PAGroup.unsetScatterGroup(g);

Methods both in Interface Group, and static in class PAGroup

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

// Returns False only if at least one member is still awaited
boolean PAGroup.allArrived (Object); 

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

// Wait for at least nb members in group to arrive
void PAGroup.waitN (Object, int nb); 

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

// 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 spmdGroup  =  (A) PASPMD.newSPMDGroup('A', params, nodes);

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

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

// 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.
PASPMD.barrier('barrierID');

Methods both in the interface MigrationController, and in class PAMobileAgent

// Migrate the current AO to the same JVM as the AO 
void migrateTo(Object o); 

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

// Migrate the current AO to JVM given by the node
int void migrateTo(Node 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 {
     PAMobileAgent.migrateTo(t);
  }
  catch (Exception e) { 
     e.printStackTrace(); 
     logger.info('Cannot migrate.'); 
  } 
}

To define a migration strategy we implement the methods in in the interface MigrationStrategyManager:

// Specification of a method to execute before migration <emphasis role="bold">
void onDeparture(String MethodName); 

// Specification of a method to execute after migration, upon the 
// arrival in a new JVM
void onArrival(String MethodName); 

// Specifies a migration itinerary <emphasis role="bold">
void setMigrationStrategy(MigrationStrategy); 

// Adds a JVM destination to an itinerary <emphasis role="bold">
void add(Destination); 

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

Components are formed from AOs. A component can be linked and can communicate with other remote components. A component can be composite, made of other components, and 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.

The following methods are specific to ProActive.

In the class PAComponent:

// 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
Component PAComponent.newActiveComponent('A', params, VirtualNode, ComponentParameters); 

Fractive:

// 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.
ProActiveInterface createCollectiveClientInterface(String itfName, String itfSignature); 

ProActive has an X.509 Public Key Infrastructure (PKI) allowing communication Authentication, Integrity, and Confidentiality (AIC) to be configured in an XML security file at deployment and outside any source code. Security is compatible with mobility, allows for hierarchical domain specification and 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 communications (requests) are authorized from Virtual Node 'VN1' to the VN 'VN2', provided that 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 that creates three JVMs on the local machine:

null

We can also specify lists of hosts for deployment:

null

Deployment API

Deployment is done in several steps: load the XML descriptor file, activate the virtual nodes, and manipulate nodes and objects.

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

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

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

// 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 
VirtualNode vn = pad.getVirtualNode(String) 

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

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

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

// Kills all the JVMs deployed with the descriptor 
// not softly(false): all JVMs are killed abruptly 
// softly(true): all JVMs that originated the creation of a RMI registry 
// wait until the registry is empty before dying
pad.killall(boolean softly); 

Functional exceptions with asynchrony

ProActive.tryWithCatch(MyException.class);
// Just before the try
try { 
      // Asynchronous calls with exceptions 
      .....
      // One can use PAException.throwArrivedException() and 
      // PAException.waitForPotentialException() here 
      
      // At the end of the try
      PAException.endTryWithCatch();
   } catch (MyException e) { 
	   // ... 
   } finally { 
	   // At the beginning of the finally
	   PAException.removeTryWithCatch();
}

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.

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

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

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

//Undeploy the service 'a' on the web server located at http://localhost:8080
WebServices.unExposeAsWebService('a', '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> 
  .......
  <componentDefinition>
    <virtualNodesDefinition>
      <virtualNode name="Workers" property="multiple"
        ftServiceId="appli" />
      <virtualNode name="Failed" property="multiple"
        ftServiceId="resources" />
    </virtualNodesDefinition>
  </componentDefinition>
  <deployment>
    <mapping>
      <map virtualNode="Workers">
        <jvmSet>
          <vmName value="Jvm1" />
          <vmName value="Jvm2" />
          <vmName value="Jvm3" />
        </jvmSet>
      </map>
      <map virtualNode="Failed">
        <jvmSet>
          <vmName value="JvmS1" />
          <vmName value="JvmS2" />
        </jvmSet>
      </map>
    </mapping>
  .....
  </deployment>
  .....
    <services>
      <serviceDefinition id="appli">
        <faultTolerance>
             <!-- Protocol selection: cic or pml -->
          <protocol type="cic"></protocol>
          <!-- URL of the fault-tolerance server -->   
          <globalServer url="rmi://eon11:1100/FTServer"></globalServer>
               <!--  Average time in seconds between two consecutive checkpoints for each object
 -->
          <ttc value="15"></ttc>
        </faultTolerance>
      </serviceDefinition>
      <serviceDefinition id="resources">
        <faultTolerance>
          <protocol type="cic"></protocol>
          <globalServer url="rmi://eon11:1100/FTServer"></globalServer>
              <!-- 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://eon11:1100/FTServer" />
          <!--  Average time in seconds between two consecutive checkpoints for each object -->
          <ttc value="15"></ttc>
        </faultTolerance>
      </serviceDefinition>
    </services>
  .....
</ProActiveDescriptor>

Starting the fault-tolerance server

The global fault-tolerance server can be launched using the bin/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/bin

$ ./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 maximum number of nodes available, and continues to return new nodes to the application while the application is running. To use all the benefits of the feature, we need to add a nodes creation event listener to the application.

Usage Example:

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

// adding 'this' as a listener
((VirtualNodeImpl) vn).addNodeCreationEventListener(<emphasis role="bold">this</emphasis>); 

// 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.extra.branchnbound.core.Task; 
public class YourTask extends Task { 
   
  public Result execute() { 
     // Your code here for computing a solution 
   } 

  public Vector<Task> 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()); 

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

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

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='<emphasis
          role="bold">implicit'> 
<!-- referenceID or keyword 'implicit' (inherit)-->
      <copyProtocol>processDefault, scp, rcp</<emphasis
          role="bold">copyProtocol>
      <sourceInfo prefix='/home/user'/>
      <destinationInfo prefix='/tmp' hostname='foo.org' username='smith' />
    </FileTransferDeploy>
  </sshProcess>
</processDefinition>