To create a single active object from the class Worker in the local JVM, we use the code hereafter. If the invocation of the constructor of class Worker throws an exception, it is placed inside an exception of type ActiveObjectCreationException. When the call to newActive returns, the active object has been created and its active thread is started.

// Set the constructor parameters
Object[] params = new Object[] { new IntWrapper(26), "Charlie" };

Worker charlie;
try {
    charlie = PAActiveObject.newActive(Worker.class, params);
} catch (ActiveObjectCreationException aoExcep) {
    // the creation of ActiveObject failed
    System.err.println(aoExcep.getMessage());
} catch (NodeException nodeExcep) {
    System.err.println(nodeExcep.getMessage());
}

The following code deploys an active object on nodes contained in the virtual node VN (current nodes). In this case, the Worker constructor does not take any arguments. However, we have to create an Object array whose length is equal to the number of nodes. The creation of active objects is optimized by a thread pool. When the call to newActiveInParallel returns, the active objects have been created and their threads have been started.

/***** GCM Deployment *****/
File applicationDescriptor = new File(gcmaPath);

GCMApplication gcmad;
try {
    gcmad = PAGCMDeployment.loadApplicationDescriptor(applicationDescriptor);
} catch (ProActiveException e) {
    e.printStackTrace();
    return;
}
gcmad.startDeployment();

// Take 2 nodes from the available nodes of VN
GCMVirtualNode vn = gcmad.getVirtualNode("VN");
vn.waitReady();
Node[] nodes = vn.getCurrentNodes().toArray(new Node[(int) vn.getNbCurrentNodes()]);
/**************************/

try {
    Worker[] workers = (Worker[]) PAActiveObject.newActiveInParallel(Worker.class.getName(),
            new Object[nodes.length][], nodes);
} catch (ClassNotFoundException classExcep) {
    System.err.println(classExcep.getMessage());
}

PAActiveObject.newActive(...) and PAActiveObject.newActiveInParallel(...) always take as a first argument the class name from which the active object will be instantiated. The classname argument must be of type java.lang.String and its value is usually obtained by calling ClassToInstantiateFrom.class.getName(). PAActiveObject.turnActive(...) does not create an active object from a class but from an existing object, therefore it takes as a first argument the object to be turned active.

In order to create the active object, PAActiveObject.newActive(...) and PAActiveObject.newActiveInParallel(...) have to take a list of constructor arguments to be passed to the constructor of the class to be instantiated as an active object. Arguments to the constructor of the class have to be passed as an array of Object. Also, we have to make sure that the constructor arguments are Serializable since they are passed by a serialized copy to the object. The ProActive runtime determines which constructor of the class to call according to the type of the elements of this array. Nevertheless, there is still room for some ambiguity in resolving the constructor because as the arguments of the constructor are stored in an array of type Object[] or Object[][]. If one argument is null the runtime can obviously not determine its type. In this case a exception is thrown specifying that ProActive cannot determine the constructor. In the example below, an ambiguity exists between the two constructors if the corresponding element of the Object array is null

// Passing null as
// an argument for the constructor
// leads to the generation of an exception
public Worker(IntWrapper age) {
    this.setAge(age);
}

public Worker(String name) {
    this.setName(name);
}

If we use PAActiveObject.newActiveInParallel(..., Nodes[] nodes) we have the make sure that the length of the first dimension of java.lang.Object[][] is equal to the number of nodes since an active object will be deployed on each node. The second dimension contains the number of the constructor arguments for each deployed active object. Different active objects can have different constructor arguments.

If you want to instantiate a generic object as an active object, you have to specify its type parameters into the constructor methods. For instance, let's consider the generic class Pair hereafter:

package org.objectweb.proactive.examples.documentation.classes;

public class Pair<X, Y> {
    private X first;
    private Y second;

    public Pair() {
    }

    public Pair(X a1, Y a2) {
        first = a1;
        second = a2;
    }

    public X getFirst() {
        return first;
    }

    public Y getSecond() {
        return second;
    }

    public void setFirst(X arg) {
        first = arg;
    }

    public void setSecond(Y arg) {
        second = arg;
    }
}

Pair is a generic class which requires two type parameters. When we create an active object using this class, we have to specify these two types. This is done through the second argument of the PAActiveObject.newActive() method which has to be an array of classes. In that case, the array of constructor arguments is therefore moved to the third position of the arguments of PAActiveObject.newActive().

Thus, instantiation of a Pair active object will look like this:

Worker charlie = new Worker(new IntWrapper(30), "Charlie");
try {
    // Declaration of the array of Class representing
    // type parameters
    Class<?>[] typeParameters = new Class<?>[] { Worker.class, String.class };

    // Declaration of the constructor parameters
    Object[] constructorParameters = new Object[] { charlie, "Researcher" };

    // Instantiation of the active object
    Pair<Worker, String> pair = (Pair<Worker, String>) PAActiveObject.newActive(Pair.class.getName(),
            typeParameters, constructorParameters);

} catch (ActiveObjectCreationException e) {
    e.printStackTrace();
} catch (NodeException e) {
    e.printStackTrace();
}

As you can see, we have created an active object of Pair whose type parameters are Worker and String. As the constructor used for this creation takes the two type parameters as parameters, we gave a Worker (first type parameter) and a String to the PAActiveObject.newActive() method.

We have presented this particularity using the PAActiveObject.newActive() method but obviously, it can also be applied to the PAActiveObject.newActiveInParallel() method as well as to the PAActiveObject.turnActive() method.

In order to create the new active object on a specific (remote) JVM, we can also give a node as parameter to newActive method. The JVM is identified using a Node object or a String that contains an URL pointing to the node. If this parameter is not given, the active object is created in the current JVM and is attached to a default Node. The node argument or the URL string can be used with PAActiveObject.newActive(...) and PAActiveObject.turnActive(...). As for the PAActiveObject.newActiveInParallel(..) method, we can specify nodes using an array of Node since we create several active objects. In that case, an active object will be created on each node contained in this array.

/***** GCM Deployment *****/
File applicationDescriptor = new File(gcmaPath);

GCMApplication gcmad;
try {
    gcmad = PAGCMDeployment.loadApplicationDescriptor(applicationDescriptor);
} catch (ProActiveException e) {
    e.printStackTrace();
    return;
}
gcmad.startDeployment();

// Take a node from the available nodes of VN
GCMVirtualNode vn = gcmad.getVirtualNode("VN");
vn.waitReady();
Node node = vn.getANode();
/**************************/

// Set the constructor parameters
Object[] params = new Object[] { new IntWrapper(26), "Charlie" };

Worker charlie;
try {
    charlie = PAActiveObject.newActive(Worker.class, params, node);
} catch (ActiveObjectCreationException aoExcep) {
    // the creation of ActiveObject failed
    System.err.println(aoExcep.getMessage());
} catch (NodeException nodeExcep) {
    System.err.println(nodeExcep.getMessage());
}

To deploy using the URL of a node, we just have to replace the node previously used by its URL:

String nodeURL = node.getNodeInformation().getURL();
charlie = PAActiveObject.newActive(Worker.class, params, nodeURL);

If you want to use the PAActiveObject.turnActive(...) method instead of the PAActiveObject.newActive(...) one, you can proceed exactly in the same way as previously.

In order to deploy several active objects with PAActiveObject.newActiveInParallel(...) , we can only pass an array of Node (see the example in Section 2.3.2, “ Using PAActiveObject.newActiveInParallel(...) ”).

Customizing the activity of the active object is at the core of ProActive because it allows to specify fully the behavior of an active object. By default, an object turned into an active object serves its incoming requests in a FIFO manner. In order to specify another policy for serving the requests or to specify any other behaviors, we can implement interfaces defining methods that will be automatically called by ProActive.

It is possible to specify what to do before the activity starts, what the activity is and what to do after it ends. The three steps are:

Three interfaces are used to define and implement each step:

In case of a migration, an active object stops and restarts its activity automatically without invoking the initialization or ending phases. Only the activity itself is restarted.

They are two ways to define each of the three phases of an active object:

Note that methods defined by those 3 interfaces are guaranteed to be called by the active thread of the active object.

2.5.5.1. Algorithms deciding which activity to invoke

Activity invocations follow the algorithm hereafter (activity is the object passed as a parameter to newActive or turnActive):

Figure 2.1. Activity algorithm

When the active object is about to be instantiated, if an activity object has been given to the creation method and if it implements the initActivity method, then this method is executed. Else, if the object we want to activate implements this method, then this one is executed. Otherwise, no method is executed before the active object creation.

In the same way, the activity itself used for the object is first the one defined in the activity object (implementing the runActivity method). Then, if no runActivity method has been defined in an activity object, the one defined in the object itself is executed. Otherwise, the standard activity phase will be used (which serves requests in a FIFO order).

Finally, when terminating the active object, the same scenario occurs. In that case, as it was for the initialising phase, if no endActivity method has been defined anywhere, no method will be executed. In other word, there is no default behaviour for ending.

Note

If an activity method (initActivity, runActivity or endActivity) is implemented both in an activity object passed as parameter to the creation method and also in the object we want to activate, then it is the method defined in the activity object that will be executed. This behaviour enables in particular to override activity phases of an object that we do not have access to its code (or that we do not want to modify).

package org.objectweb.proactive.examples.documentation.classes;

import java.io.Serializable;
import org.objectweb.proactive.Body;
import org.objectweb.proactive.core.util.wrapper.IntWrapper;

import org.objectweb.proactive.InitActive;



/**
 * @author ProActive Team
 *
 * Class used for documentation example, in particular
 * for Chapter 9. Active Objects: Creation And Advanced Concepts.
 */
public class Worker implements Serializable, InitActive {

    /**
     * 
     */
    private static final long serialVersionUID = 51L;
    private IntWrapper age = new IntWrapper(0);
    private String name = "Anonymous";

    /**
     * Empty no-arg constructor needed by ProActiv
     */
    public Worker() {
    }

    /**
     * Constructor with arguments
     *
     * @param age
     * @param name
     */
    public Worker(IntWrapper age, String name) {
        this.setAge(age);
        this.setName(name);
    }

    /* (non-Javadoc)
     * @see org.objectweb.proactive.InitActive#initActivity(org.objectweb.proactive.Body)
     */
    public void initActivity(Body body) {
        System.out.println("===> Started Active object: ");
        System.out.println(body.getMBean().getName() + " on " + body.getMBean().getNodeUrl());
    }

    /* (non-Javadoc)
     * @see org.objectweb.proactive.EndActive#endActivity(org.objectweb.proactive.Body)
     */
    public void endActivity(Body body) {
        System.out.println("===> You have killed the active object");
    }
}

package org.objectweb.proactive.examples.documentation.classes;

import org.objectweb.proactive.Body;
import org.objectweb.proactive.RunActive;
import org.objectweb.proactive.Service;
import org.objectweb.proactive.api.PALifeCycle;


public class Simulation implements RunActive {
    private boolean stoppedSimulation = false;
    private boolean startedSimulation = false;
    private boolean suspendedSimulation = false;
    private boolean notStarted = true;

    public void startSimulation() {
        // Simulation starts
        System.out.println("Simulation started...");
        notStarted = false;
        startedSimulation = true;
    }

    public void restartSimulation() {
        // Simulation is restarted
        System.out.println("Simulation restarted...");
        startedSimulation = true;
        suspendedSimulation = false;
    }

    public void suspendSimulation() {
        // Simulation is suspended
        System.out.println("Simulation suspended...");
        suspendedSimulation = true;
        startedSimulation = false;
    }

    public void stopSimulation() {
        // Simulation is stopped
        System.out.println("Simulation stopped...");
        stoppedSimulation = true;
    }

    public void runActivity(Body body) {
        Service service = new Service(body);
        while (body.isActive()) {
            // If the simulation is not yet started wait until startSimulation
            // method
            if (notStarted)
                service.blockingServeOldest("startSimulation");
            // If the simulation is started serve request with FIFO
            if (startedSimulation)
                service.blockingServeOldest();
            // If simulation is suspended wait until restartSimulation method
            if (suspendedSimulation)
                service.blockingServeOldest("restartSimulation");
            // If simulation is stopped, exit
            if (stoppedSimulation) {
                body.terminate();
                PALifeCycle.exitSuccess();
            }
        }
    }
}

package org.objectweb.proactive.examples.documentation.classes;

import org.objectweb.proactive.Body;
import org.objectweb.proactive.RunActive;
import org.objectweb.proactive.Service;


public class LIFOActivity implements RunActive {

    // -- implements RunActive for serving request in a LIFO order
    public void runActivity(Body body) {
        Service service = new Service(body);
        while (body.isActive()) {
            System.out.println("Serving...");
            service.blockingServeYoungest();
        }
    }
}

// Set the constructor parameters
Object[] params = new Object[] { new IntWrapper(26), "Charlie" };

// Instantiate a LIFOActivity object
LIFOActivity activity = new LIFOActivity();

Worker charlie;
try {
    charlie = PAActiveObject.newActive(Worker.class, null, params, null, activity, null);

    System.out.println(charlie.getName() + " is " + charlie.getAge());
    charlie.setAge(new IntWrapper(25));
    charlie.setName("Anonymous");
    System.out.println(charlie.getName() + " is " + charlie.getAge());

    PAActiveObject.terminateActiveObject(charlie, true);
} catch (ActiveObjectCreationException aoExcep) {
    // the creation of ActiveObject failed
    System.err.println(aoExcep.getMessage());
} catch (NodeException nodeExcep) {
    System.err.println(nodeExcep.getMessage());
}

Creating an active object using ProActive might be a little bit cumbersome and requires more lines of code that for creating a regular object. A nice solution to this problem is through the use of the factory pattern. This mainly applies to class based creation. It consists in creating a WorkerFactory class with a static method to class that takes care of instantiating the active object and returns it.

package org.objectweb.proactive.examples.documentation.classes;

import org.objectweb.proactive.ActiveObjectCreationException;
import org.objectweb.proactive.api.PAActiveObject;
import org.objectweb.proactive.core.node.Node;
import org.objectweb.proactive.core.node.NodeException;
import org.objectweb.proactive.core.util.wrapper.IntWrapper;


/**
 * @author ProActive Team
 *
 * Factory used to instantiate Worker active objects
 */
public class WorkerFactory {

    /**
     * Create a new Worker active object
     *
     * @param age
     * @param name
     * @param node
     * @return the Worker active object
     */
    public static Worker createActiveWorker(int age, String name, Node node) {
        Object[] params = new Object[] { new IntWrapper(age), name };
        try {
            return (Worker) PAActiveObject.newActive(Worker.class.getName(), params, node);
        } catch (ActiveObjectCreationException e) {
            e.printStackTrace();
            return null;
        } catch (NodeException e) {
            e.printStackTrace();
            return null;
        }
    }
}

The static method in the factory class is then used to create active objects:

// Creates locally (Node=null) a Worker whose name is Charlie and who is 19
Worker charlie = WorkerFactory.createActiveWorker(19, "Charlie", null);

It is up to the programmer to decide whether this method has to throw exceptions or not. We recommend that this method only throw exceptions that appear in the signature of the reified constructor (none here as the constructor of Worker that we call doesn't throw any exception). However, the non functional exceptions induced by the creation of the active object have to be dealt with somewhere in the code.

There are many cases where you may want to customize the body used when creating an active object. For instance, one may want to add some debug messages or some timing behavior when sending or receiving requests. The body is a non changeable object that delegates most of its tasks to helper objects called MetaObjects. Standard MetaObjects are already used by default in ProActive but one can easily replace any of those MetaObjects by a custom one.

We have defined a MetaObjectFactory interface that is able to create factories for each of those MetaObjects. This interface is implemented by ProActiveMetaObjectFactory which provides all the default factories used in ProActive.

When creating an active object, it is possible to specify which MetaObjectFactory to use for that particular instance of active object being created.

First you have to write a new MetaObject factory that inherits from ProActiveMetaObjectFactory or directly implements the MetaObjectFactory interface. Inheriting from ProActiveMetaObjectFactory is a great time saver as you only redefine what you really need to as opposed to redefining everything when inheriting from MetaObjectFactory . Here is an example:

package org.objectweb.proactive.examples.documentation.classes;

import java.io.Serializable;

import org.objectweb.proactive.core.body.MetaObjectFactory;
import org.objectweb.proactive.core.body.ProActiveMetaObjectFactory;
import org.objectweb.proactive.core.body.UniversalBody;
import org.objectweb.proactive.core.body.request.Request;
import org.objectweb.proactive.core.body.request.RequestFactory;
import org.objectweb.proactive.core.body.request.RequestImpl;
import org.objectweb.proactive.core.mop.MethodCall;


/**
 * @author ProActive Team
 *
 * Customized Meta-Object Factory
 */
public class CustomMetaObjectFactory extends ProActiveMetaObjectFactory {

    /**
     * 
     */
    private static final long serialVersionUID = 51L;
    private static final MetaObjectFactory instance = new CustomMetaObjectFactory();

    //return a new factory instance
    public static MetaObjectFactory newInstance() {
        return instance;
    }

    private CustomMetaObjectFactory() {
        super();
    }

    protected RequestFactory newRequestFactorySingleton() {
        System.out.println("Creating the  custom metaobject factory...");
        return new CustomRequestFactory();
    }

    protected class CustomRequestFactory extends RequestFactoryImpl implements Serializable {

        /**
         * 
         */
        private static final long serialVersionUID = 51L;

        public Request newRequest(MethodCall methodCall, UniversalBody sourceBody, boolean isOneWay,
                long sequenceID) {
            System.out.println("Received a new request...");
            return new CustomRequest(methodCall, sourceBody, isOneWay, sequenceID);
        }

        protected class CustomRequest extends RequestImpl {
            /**
             * 
             */
            private static final long serialVersionUID = 51L;

            public CustomRequest(MethodCall methodCall, UniversalBody sourceBody, boolean isOneWay,
                    long sequenceID) {
                super(methodCall, sourceBody, isOneWay, sequenceID);
                System.out.println("I am a custom request handler");
            }
        }
    }
}

The factory above simply redefines the RequestFactory in order to make the body use a new type of request. The method protected RequestFactory newRequestFactorySingleton() is one convenience method that ProActiveMetaObjectFactory provides to simplify the creation of factories as singleton.

public class ProActiveMetaObjectFactory implements MetaObjectFactory, java.io.Serializable, Cloneable {
    /**
     * 
     */
    private static final long serialVersionUID = 51L;
    public static final String COMPONENT_PARAMETERS_KEY = "component-parameters";
    public static final String SYNCHRONOUS_COMPOSITE_COMPONENT_KEY = "synchronous-composite";
    protected static Logger logger = ProActiveLogger.getLogger(Loggers.MOP);

    //
    // -- PRIVATE MEMBERS -----------------------------------------------
    //
    // private static final MetaObjectFactory instance = new ProActiveMetaObjectFactory();
    private static MetaObjectFactory instance = new ProActiveMetaObjectFactory();
    public Map<String, Object> parameters = new HashMap<String, Object>();

    //
    // -- PROTECTED MEMBERS -----------------------------------------------
    //
    protected RequestFactory requestFactoryInstance;
    protected ReplyReceiverFactory replyReceiverFactoryInstance;
    protected RequestReceiverFactory requestReceiverFactoryInstance;
    protected RequestQueueFactory requestQueueFactoryInstance;
    protected MigrationManagerFactory migrationManagerFactoryInstance;

    //    protected RemoteBodyFactory remoteBodyFactoryInstance;
    protected ThreadStoreFactory threadStoreFactoryInstance;
    protected ProActiveSPMDGroupManagerFactory proActiveSPMDGroupManagerFactoryInstance;
    protected PAComponentFactory componentFactoryInstance;
    protected ProActiveSecurityManager proActiveSecurityManager;
    protected FTManagerFactory ftmanagerFactoryInstance;
    protected DebuggerFactory debuggerFactoryInstance;
    protected MessageTagsFactory requestTagsFactoryInstance;
    protected Object timItReductor;

    //
    // -- CONSTRUCTORS -----------------------------------------------
    //
    protected ProActiveMetaObjectFactory() {
        this.requestFactoryInstance = newRequestFactorySingleton();
        this.replyReceiverFactoryInstance = newReplyReceiverFactorySingleton();
        this.requestReceiverFactoryInstance = newRequestReceiverFactorySingleton();
        this.requestQueueFactoryInstance = newRequestQueueFactorySingleton();
        this.migrationManagerFactoryInstance = newMigrationManagerFactorySingleton();
        //        this.remoteBodyFactoryInstance = newRemoteBodyFactorySingleton();
        this.threadStoreFactoryInstance = newThreadStoreFactorySingleton();
        this.proActiveSPMDGroupManagerFactoryInstance = newProActiveSPMDGroupManagerFactorySingleton();
        this.ftmanagerFactoryInstance = newFTManagerFactorySingleton();
        this.debuggerFactoryInstance = newDebuggerFactorySingleton();
        this.requestTagsFactoryInstance = newRequestTagsFactorySingleton();
    }

    /**
     * Constructor with parameters
     * It is used for per-active-object configurations of ProActive factories
     * @param parameters the parameters of the factories; these parameters can be of any type
     */
    public ProActiveMetaObjectFactory(Map<String, Object> parameters) {
        this.parameters = parameters;
        if (parameters.containsKey(COMPONENT_PARAMETERS_KEY)) {
            ComponentParameters initialComponentParameters = (ComponentParameters) parameters
                    .get(COMPONENT_PARAMETERS_KEY);
            this.componentFactoryInstance = newComponentFactorySingleton(initialComponentParameters);
            this.requestFactoryInstance = newRequestFactorySingleton();
            this.replyReceiverFactoryInstance = newReplyReceiverFactorySingleton();
            this.requestReceiverFactoryInstance = newRequestReceiverFactorySingleton();
            this.requestQueueFactoryInstance = newRequestQueueFactorySingleton();
            this.migrationManagerFactoryInstance = newMigrationManagerFactorySingleton();
            //            this.remoteBodyFactoryInstance = newRemoteBodyFactorySingleton();
            this.threadStoreFactoryInstance = newThreadStoreFactorySingleton();
            this.proActiveSPMDGroupManagerFactoryInstance = newProActiveSPMDGroupManagerFactorySingleton();
            this.ftmanagerFactoryInstance = newFTManagerFactorySingleton();
            this.debuggerFactoryInstance = newDebuggerFactorySingleton();
            this.requestTagsFactoryInstance = newRequestTagsFactorySingleton();
        }
    }

    //
    // -- PUBLICS METHODS -----------------------------------------------
    //
    public static MetaObjectFactory newInstance() {
        return instance;
    }

    public static void setNewInstance(MetaObjectFactory mo) {
        instance = mo;
    }

    /**
     * getter for the parameters of the factory (per-active-object config)
     * @return the parameters of the factory
     */
    public Map<String, Object> getParameters() {
        return this.parameters;
    }

    //
    // -- implements MetaObjectFactory -----------------------------------------------
    //
    public RequestFactory newRequestFactory() {
        return this.requestFactoryInstance;
    }

    public ReplyReceiverFactory newReplyReceiverFactory() {
        return this.replyReceiverFactoryInstance;
    }

    public RequestReceiverFactory newRequestReceiverFactory() {
        return this.requestReceiverFactoryInstance;
    }

    public RequestQueueFactory newRequestQueueFactory() {
        return this.requestQueueFactoryInstance;
    }

    public MigrationManagerFactory newMigrationManagerFactory() {
        return this.migrationManagerFactoryInstance;
    }

    //    public RemoteBodyFactory newRemoteBodyFactory() {
    //        return this.remoteBodyFactoryInstance;
    //    }
    public ThreadStoreFactory newThreadStoreFactory() {
        return this.threadStoreFactoryInstance;
    }

    public ProActiveSPMDGroupManagerFactory newProActiveSPMDGroupManagerFactory() {
        return this.proActiveSPMDGroupManagerFactoryInstance;
    }

    public PAComponentFactory newComponentFactory() {
        return this.componentFactoryInstance;
    }

    public FTManagerFactory newFTManagerFactory() {
        return this.ftmanagerFactoryInstance;
    }

    public DebuggerFactory newDebuggerFactory() {
        return this.debuggerFactoryInstance;
    }

    public MessageTagsFactory newRequestTagsFactory() {
        return this.requestTagsFactoryInstance;
    }

    //
    // -- PROTECTED METHODS -----------------------------------------------
    //
    protected RequestFactory newRequestFactorySingleton() {
        return new RequestFactoryImpl();
    }

    protected ReplyReceiverFactory newReplyReceiverFactorySingleton() {
        return new ReplyReceiverFactoryImpl();
    }

    protected RequestReceiverFactory newRequestReceiverFactorySingleton() {
        return new RequestReceiverFactoryImpl();
    }

    protected RequestQueueFactory newRequestQueueFactorySingleton() {
        return new RequestQueueFactoryImpl();
    }

    protected MigrationManagerFactory newMigrationManagerFactorySingleton() {
        return new MigrationManagerFactoryImpl();
    }

    //    protected RemoteBodyFactory newRemoteBodyFactorySingleton() {
    //        return new RemoteBodyFactoryImpl();
    //    }
    protected ThreadStoreFactory newThreadStoreFactorySingleton() {
        return new ThreadStoreFactoryImpl();
    }

    protected ProActiveSPMDGroupManagerFactory newProActiveSPMDGroupManagerFactorySingleton() {
        return new ProActiveSPMDGroupManagerFactoryImpl();
    }

    protected PAComponentFactory newComponentFactorySingleton(ComponentParameters initialComponentParameters) {
        return new ProActiveComponentFactoryImpl(initialComponentParameters);
    }

    protected FTManagerFactory newFTManagerFactorySingleton() {
        return new FTManagerFactoryImpl();
    }

    protected DebuggerFactory newDebuggerFactorySingleton() {
        return new DebuggerFactoryImpl();
    }

    protected MessageTagsFactory newRequestTagsFactorySingleton() {
        return new MessageTagsFactoryImpl();
    }

    //  //
    //  // -- INNER CLASSES -----------------------------------------------
    //  //
    protected static class RequestFactoryImpl implements RequestFactory, java.io.Serializable {
        /**
         * 
         */
        private static final long serialVersionUID = 51L;

        public Request newRequest(MethodCall methodCall, UniversalBody sourceBody, boolean isOneWay,
                long sequenceID, MessageTags tags) {
            //########### exemple de code pour les nouvelles factories
            //            if(System.getProperty("migration.stategy").equals("locationserver")){
            //                  return new RequestWithLocationServer(methodCall, sourceBody,
            //                isOneWay, sequenceID, LocationServerFactory.getLocationServer());
            //            }else{
            return new org.objectweb.proactive.core.body.request.RequestImpl(methodCall, sourceBody,
                isOneWay, sequenceID, tags);
            //}
        }
    }

    // end inner class RequestFactoryImpl
    protected static class ReplyReceiverFactoryImpl implements ReplyReceiverFactory, java.io.Serializable {
        /**
         * 
         */
        private static final long serialVersionUID = 51L;

        public ReplyReceiver newReplyReceiver() {
            return new org.objectweb.proactive.core.body.reply.ReplyReceiverImpl();
        }
    }

    // end inner class ReplyReceiverFactoryImpl
    protected class RequestReceiverFactoryImpl implements RequestReceiverFactory, java.io.Serializable {
        /**
         * 
         */
        private static final long serialVersionUID = 51L;

        public RequestReceiver newRequestReceiver() {
            if (ProActiveMetaObjectFactory.this.parameters.containsKey(SYNCHRONOUS_COMPOSITE_COMPONENT_KEY) &&
                ((Boolean) ProActiveMetaObjectFactory.this.parameters
                        .get(ProActiveMetaObjectFactory.SYNCHRONOUS_COMPOSITE_COMPONENT_KEY)).booleanValue()) {
                return new SynchronousComponentRequestReceiver();
            }
            return new org.objectweb.proactive.core.body.request.RequestReceiverImpl();
        }
    }

    // end inner class RequestReceiverFactoryImpl
    protected class RequestQueueFactoryImpl implements RequestQueueFactory, java.io.Serializable {
        /**
         * 
         */
        private static final long serialVersionUID = 51L;

        public BlockingRequestQueue newRequestQueue(UniqueID ownerID) {
            if ("true".equals(ProActiveMetaObjectFactory.this.parameters
                    .get(SYNCHRONOUS_COMPOSITE_COMPONENT_KEY))) {
                return null;
            }

            //if (componentFactoryInstance != null) {
            // COMPONENTS
            // we need a request queue for components
            //return new ComponentRequestQueueImpl(ownerID);
            //} else {
            return new org.objectweb.proactive.core.body.request.BlockingRequestQueueImpl(ownerID);
            //}
        }
    }

    // end inner class RequestQueueFactoryImpl
    protected static class MigrationManagerFactoryImpl implements MigrationManagerFactory,
            java.io.Serializable {
        /**
         * 
         */
        private static final long serialVersionUID = 51L;

        public MigrationManager newMigrationManager() {
            //########### example de code pour les nouvelles factories
            //            if(System.getProperty("migration.stategy").equals("locationserver")){
            //                return new MigrationManagerWithLocationServer(LocationServerFactory.getLocationServer());
            //            }else{
            return new org.objectweb.proactive.core.body.migration.MigrationManagerImpl();
            //}
        }
    }


    // end inner class RemoteBodyFactoryImpl
    protected static class ThreadStoreFactoryImpl implements ThreadStoreFactory, java.io.Serializable {
        /**
         * 
         */
        private static final long serialVersionUID = 51L;

        public ThreadStore newThreadStore() {
            return new org.objectweb.proactive.core.util.ThreadStoreImpl();
        }
    }

    // end inner class ThreadStoreFactoryImpl
    protected static class ProActiveSPMDGroupManagerFactoryImpl implements ProActiveSPMDGroupManagerFactory,
            java.io.Serializable {
        /**
         * 
         */
        private static final long serialVersionUID = 51L;

        public ProActiveSPMDGroupManager newProActiveSPMDGroupManager() {
            return new ProActiveSPMDGroupManager();
        }
    }

    // end inner class ProActiveGroupManagerFactoryImpl
    protected class ProActiveComponentFactoryImpl implements PAComponentFactory, java.io.Serializable {
        /**
         * 
         */
        private static final long serialVersionUID = 51L;
        // COMPONENTS
        private ComponentParameters componentParameters;

        public ProActiveComponentFactoryImpl(ComponentParameters initialComponentParameters) {
            this.componentParameters = initialComponentParameters;
        }

        public PAComponent newPAComponent(Body myBody) {
            return new PAComponentImpl(this.componentParameters, myBody);
        }
    }

    // FAULT-TOLERANCE
    protected class FTManagerFactoryImpl implements FTManagerFactory, Serializable {
        /**
         * 
         */
        private static final long serialVersionUID = 51L;

        public FTManager newFTManager(int protocolSelector) {
            switch (protocolSelector) {
                case FTManagerFactory.PROTO_CIC_ID:
                    return new FTManagerCIC();
                case FTManagerFactory.PROTO_PML_ID:
                    return new FTManagerPMLRB();
                default:
                    logger.error("Error while creating fault-tolerance manager : " +
                        "no protocol is associated to selector value " + protocolSelector);
                    return null;
            }
        }

        public FTManager newHalfFTManager(int protocolSelector) {
            switch (protocolSelector) {
                case FTManagerFactory.PROTO_CIC_ID:
                    return new HalfFTManagerCIC();
                case FTManagerFactory.PROTO_PML_ID:
                    return new HalfFTManagerPMLRB();
                default:
                    logger.error("Error while creating fault-tolerance manager : " +
                        "no protocol is associated to selector value " + protocolSelector);
                    return null;
            }
        }
    }

    protected static class DebuggerFactoryImpl implements DebuggerFactory, java.io.Serializable {
        /**
         * 
         */
        private static final long serialVersionUID = 51L;

        public Debugger newDebugger() {
            return new DebuggerImpl();
        }
    }

    // REQUEST-TAGS
    protected static class MessageTagsFactoryImpl implements MessageTagsFactory, Serializable {

        /**
         * 
         */
        private static final long serialVersionUID = 51L;

        /**
         * @see MessageTagsFactory#newMessageTags()
         */
        public MessageTags newMessageTags() {
            return new MessageTags();
        }
    }

    // SECURITY
    public void setProActiveSecurityManager(ProActiveSecurityManager psm) {
        this.proActiveSecurityManager = psm;
    }

    public ProActiveSecurityManager getProActiveSecurityManager() {
        return this.proActiveSecurityManager;
    }

    @Override
    public Object clone() throws CloneNotSupportedException {

        try {
            return MakeDeepCopy.WithObjectStream.makeDeepCopy(this);
        } catch (IOException e) {
            //TODO replace by CloneNotSupportedException(Throwable e) java 1.6
            throw (CloneNotSupportedException) new CloneNotSupportedException(e.getMessage()).initCause(e);
        } catch (ClassNotFoundException e) {
            throw (CloneNotSupportedException) new CloneNotSupportedException(e.getMessage()).initCause(e);
        }
    }

    public void setTimItReductor(Object timItReductor) {
        this.timItReductor = timItReductor;
    }

    public Object getTimItReductor() {
        return this.timItReductor;
    }

}

More explanations can be found in the javadoc of the org.objectweb.proactive.core.body.ProActiveMetaObjectFactory class. The use of that factory is fairly simple. All you have to do is to pass an instance of the factory when creating a new active object. If we take the same example as before we have:

Object[] params = new Object[] { new IntWrapper(26), "Charlie" };
try {
    Worker charlie = (Worker) PAActiveObject.newActive(Worker.class.getName(), null, params, null,
            null, CustomMetaObjectFactory.newInstance());
} catch (ActiveObjectCreationException e) {
    e.printStackTrace();
} catch (NodeException e) {
    e.printStackTrace();
}

The role of the class Stub_Worker is to reify all method calls that can be performed through a reference of type Worker. Reifying a call simply means constructing an object that represents the call (in our case, all reified calls are instance of class org.objectweb.proactive.core.mop.MethodCall), so it can be manipulated as any other object. The reified call is then processed by the other components of the active object in order to achieve the behavior we expect from an active object.

Using a standard object for representing elements of the language that are not normally objects (such as method calls, constructor calls, references, types,...) is at the core of metaobject programming. The ProActive metaobject protocol (MOP) is is described in Chapter 43. MOP: Metaobject Protocol. but it is not a prerequisite for understanding and using ProActive.

As one of our objectives is to provide transparent active objects, references to active objects of class Worker need to be of the same type as references to passive instances of Worker (this feature is called polymorphism between passive and active instances of the same class). This is why Stub_Worker is a subclass of class Worker by construction, therefore allowing instances of class Stub_Worker to be assigned to variables of type Worker.

Class Stub_Worker redefines each of the methods inherited from its superclasses. The code of each method of class Stub_Worker builds an instance of class org.objectweb.proactive.core.mop.MethodCall in order to represent the call to this method. This object is then passed to the BodyProxy, which returns an object that is returned as the result of the method call. From the caller's point of view, everything looks like if the call had been performed on an instance of Worker.

Now that we know how stubs work, we can understand some of the limitations of ProActive:

The role of the proxy is to handle asynchronism in calls to active object. More specifically, it creates future objects if possible and needed, forwards calls to bodies, and returns future objects to the stubs. As this class operates on MethodCall objects, it is absolutely generic and does not depend at all on the type of the stub that feeds calls in through its reify method.

The body is responsible for storing calls (actually, Request objects) in a queue of pending requests and processing these requests according to a given synchronization policy. The default behaviour for the synchronization policy is FIFO. The Body has its own thread which chooses requests from the queue of requests and executes the associated call.

There is also a standard instance of class Worker. It may contain some synchronized information in its live method, if any. As the body executes calls one by one, there cannot be any concurrent execution of two portions of code for this object by two different threads. This enables the use of pre and post-conditions and class invariants. As a consequence, the use of the keyword synchronized in class Worker should not be necessary. Any synchronization scheme that can be expressed through monitors and synchronized statements can be expressed using ProActive's high-level synchronization mechanism in a much more natural and user-friendly way.

Whenever it is possible, a method call on an active object is reified as an asynchronous request. In that case, it immediately returns a future object. If it is not possible, the call is synchronous and blocks until the reply is received.

This future object acts as a placeholder for the result of the not-yet-performed method invocation. As a consequence, the calling thread can go on with executing its code, as long as it does not need to invoke methods on the returned object. At the time when the real result is needed, the calling thread is automatically blocked if the result of the method invocation is not yet available. Below are shown the different cases that can lead to an asynchronous call. Note that this table does not apply when using the tryWithCatch annotators to deal with asynchronous exceptions. See Chapter 5, Exception Handling.

As we can see, the creation of a future depends not only on the caller type, but also on the return object type. Creating a future is only possible if the object is reifiable. Although a future has a similar structure to an active object, a future object is not active. It only has a Stub and a Proxy as shown in the figure below:

Figure 2.3. A future object

During its lifetime, an active object can create many future objects. There are all automatically kept in a FuturePool.

Each time a future is created, it is inserted in the future pool of the corresponding active object. When the result becomes available, the future object is removed from the pool. There are several methods in org.objectweb.proactive.api.ProFuture which provide control over the way futures are returned.

Any call to a future object is reified in order to be blocked if the future is not yet available and later executed on the result object. However, two methods don't follow this scheme: equals and hashCode. They are often called by other methods from the Java library, like HashTable.add() and so are most of the time out of control from the user. This can lead very easily to deadlocks if they are called on a not yet available object.

There are some drawbacks with this technique, the main one being the impossibility to have a user override of the default HashTable and equals() methods.

There are few things to remember with active objects, futures, and asynchronous method calls, in order to avoid annoying debugging sessions:

Waiting for the update of a future in a ProActive application could freeze the application if the node responsible for updating the future experiences a failure. To prevent this, ProActive has a comprehensive generic fault tolerance mechanism. However, it can be too expensive to checkpoint all the active objects, when fault detection is only needed for certain objects.

By default, ProActive continuously pings active objects awaited for updating futures. When such a ping fails, the associated awaited future is updated with a runtime exception. Accessing the future will throw an exception.

Worker ao = PAActiveObject.newActive(Worker.class, null);
IntWrapper future = ao.getAge();
String str;
try {
    str = future.toString();
} catch (FutureMonitoringPingFailureException fmpfe) {
    System.out.println("The active object 'ao' had a failure");
    fmpfe.printStackTrace();
}

The ping is started when the future is being awaited, but it is also possible to start it beforehand using the ProFuture.monitorFuture(java.lang.Object future) method. To find out more about the fault tolerance mechanism read Chapter 32. Fault-Tolerance.

An automatic continuation can occur either when sending a request (request parameter is or contains a future) or when sending a reply (the result is or contains a future). Outgoing futures are registered in the FuturePool of the active object which sends the future. For instance, In the example presented in the Section 2.8.3, “Illustration of an Automatic Continuation”, a FuturePool is created on the Caller side. This FuturePool is a pool of couples <Future, Destination> and is filled in when the future is serialized. Every request or reply are serialized before being sent, and the future is part of the request or the reply. A thread Thread sending the message keeps in a static table (FuturePool.bodiesDestination) a reference to the destination body. Hence, when a future Future is serialized by the same thread, this thread looks up in the static table the destination Destination registered for the thread Thread. If it finds a destination for the future, the future notifies the FuturePool that it is going to leave, which in turn registers the <Future, Destination> as an automatic continuation.

When the value is available for the future, it is is propagated to all objects that received the future Future. This type of update is realized by a thread located in the FuturePool.

When a message containing a future is received by an active object the active object registers the future in the FuturePool to be able to update it when the value will be available. After the future is deserialized, it is registered in a static table FuturePool.incomingFutures.

To illustrate how automatic continuation takes place, we will use three classes: Caller, Value and Worker. The Caller and Value classes are used to instantiate passive objects and the Worker class is used to instantiate an active object. The Worker class has a foo() method and the Value class has a bar(Value v, int nbYears) method.

Value v = new Value();
Worker worker = PAActiveObject.newActive(Worker.class, null);
Value v1 = worker.foo(); //v1 is a future
Value v2 = v.bar(v1, 1); //v1 is passed as parameter

where

public Value foo() {
    return new Value(this.age, this.name);
}

and

public Value bar(Value v, int nbYears) {
    v.setAge(new IntWrapper(v.getAge().getIntValue() + nbYears));
    return this;
}

We will first call foo() on the worker. The call goes through the worker stub and the body proxy and it is placed in the queue of the body. We are using a passive object as an example, however the process would be similar if the caller was an active object or if the objects were all on the same machine. In those cases, the call will also go through the stub and proxy as described in the sequence diagrams above. The following pictures represent the continuation process. Note that in this diagram, the caller is named A and the value is named B.

After sending the request, the thread of object A continues immediately (asynchronous call) and v1 points to a future that has not been updated yet.

Next, object A calls bar(v1, 1) on B.

The call on B leads to the creation of another future to which v2 will point to. Since v1 has not been updated yet, there will be another future created on object B.

At some point, the active object finishes executing foo() and the value of the future for v1 will be updated by the Body.

After the value on A is updated, A updates the value in the future on B.

Once B has the value for the future, it will execute bar() with the received value and update the value v2 in A.

The final state is with all the future values updated.

As seen above, as long as there is a future needed for a computation, it will be passed to the object doing the computation when it is not needed immediately. If the value is needed right away, the caller thread will block until the future is updated, just like in single threaded call.

We can use the methods onArrival(String) and onDeparture(String) to specify which methods will be run when the active objects migrates.

The arguments for onArrival(String) and onDeparture(String) are respectively the methods to be run before serialization and after deserialization. We instantiate a new MigrationStrategyManagerImpl for the body of the object to be migrated and assign the methods to be run. For this, we use the initActivity(Body) method since we only need to run the creation and assignment once.

Here is the code of our UnSerializableAgent class:

package org.objectweb.proactive.examples.documentation.classes;

import java.io.Serializable;
import java.math.BigInteger;
import java.net.InetAddress;
import java.util.Random;

import org.objectweb.proactive.Body;
import org.objectweb.proactive.InitActive;
import org.objectweb.proactive.api.PAActiveObject;
import org.objectweb.proactive.api.PAMobileAgent;
import org.objectweb.proactive.core.migration.MigrationStrategyManagerImpl;


/**
 * @author The ProActive Team
 *
 */
public class UnSerializableAgent implements InitActive, Serializable {

    /**
     * 
     */
    private static final long serialVersionUID = 51L;
    private byte[] saved; //will store the values 
    private transient BigInteger toBeSaved; //will be null after serialization

    /**
     * Empty, no-arg constructor required by ProActive
     */
    public UnSerializableAgent() {
    }

    public UnSerializableAgent(int length) {
        toBeSaved = new BigInteger(length, new Random());
    }

    public void displayArray() {
        System.out.println(toBeSaved.toString());
    }

    /**
     * Migrates the active object
     * 
     * @param url - URL of the node where you want to 
     *              migrate your active object to
     */
    public void moveTo(String url) {
        try {
            PAMobileAgent.migrateTo(url);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * Retrieves the name of the host where the active
     * object is.
     * 
     * @return The node url and the hostname where the active object is
     */
    public String whereAreYou() {
        try {
            return PAActiveObject.getActiveObjectNodeUrl(PAActiveObject.getStubOnThis()) + " on host " +
                InetAddress.getLocalHost().getHostName();
        } catch (Exception e) {
            return "Localhost lookup failed";
        }
    }

    public void initActivity(Body body) {
        MigrationStrategyManagerImpl myStrategyManager = new MigrationStrategyManagerImpl(
            (org.objectweb.proactive.core.body.migration.Migratable) body);
        myStrategyManager.onArrival("rebuild");
        myStrategyManager.onDeparture("leaveHost");
    }

    /**
     * Restore values of unserializable variable
     * that have been previously saved
     */
    public void rebuild() {
        toBeSaved = new BigInteger(saved);
        System.out.println("Unserializable variables has been restored");
    }

    /**
     * Saves the values of the unserializable variables
     */
    public void leaveHost() {
        saved = toBeSaved.toByteArray();
        System.out.println("Unserializable variables has been saved");
    }
}

This class is normally non-serializable since BigInteger is not serializable. If we tried to migrate the active object without implementing the initActivity() method as previously, we would get a NullPointerException when invoking a method on our toBeSaved variable.

The following method is in charge of instantiate our object and to migrate it:

public static void unserializableMigration(String[] args) {
    if (args.length != 1) {
        System.out
                .println("Usage: java org.objectweb.proactive.examples.documentation.migration.Migration GCMA.xml");
        System.exit(-1);
    }

    // Gets a node from a deployment and
    // creates the SimpleAgent on this node

    /***** GCM Deployment *****/
    File applicationDescriptor = new File(args[0]);

    GCMApplication gcmad;
    try {
        gcmad = PAGCMDeployment.loadApplicationDescriptor(applicationDescriptor);
    } catch (ProActiveException e) {
        e.printStackTrace();
        return;
    }
    gcmad.startDeployment();

    // Take a node from the available nodes of VN
    GCMVirtualNode vn = gcmad.getVirtualNode("VN");
    vn.waitReady();
    Node node = vn.getANode();
    /**************************/

    UnSerializableAgent unserializableAgent;
    try {
        // Creates the SimpleAgent in this JVM
        Object[] params = new Object[] { 100 };
        unserializableAgent = PAActiveObject.newActive(UnSerializableAgent.class, params, node);
    } catch (Exception e) {
        e.printStackTrace();
        return;
    }

    System.out.println("The Active Object has been created on " + unserializableAgent.whereAreYou());
    unserializableAgent.displayArray();

    // Migrates the SimpleAgent to another node
    node = vn.getANode();
    unserializableAgent.moveTo(node.getNodeInformation().getURL());
    System.out.println("The Active Object is now on " + unserializableAgent.whereAreYou());
    unserializableAgent.displayArray();
}

Another way to deal with unserializable attributes is using the methods from the Serializable interface:

private void writeObject(java.io.ObjectOutputStream out) throws IOException;
private void readObject(java.io.ObjectInputStream in) throws IOException, ClassNotFoundException;
private void readObjectNoData() throws ObjectStreamException;

We override these methods to provide a way to save and restore the non-serializable components of the active object:

package org.objectweb.proactive.examples.documentation.classes;

import java.io.Serializable;
import java.math.BigInteger;
import java.net.InetAddress;
import java.util.Random;

import org.objectweb.proactive.api.PAActiveObject;
import org.objectweb.proactive.api.PAMobileAgent;


/**
 * @author The ProActive Team
 *
 */
public class UnSerializableAgent2 implements Serializable {

    /**
     * 
     */
    private static final long serialVersionUID = 51L;
    private transient BigInteger toBeSaved; //will be null after serialization

    /**
     * Empty, no-arg constructor required by ProActive
     */
    public UnSerializableAgent2() {
    }

    public UnSerializableAgent2(int length) {
        toBeSaved = new BigInteger(length, new Random());
    }

    public void displayArray() {
        System.out.println(toBeSaved.toString());
    }

    /**
     * Migrates the active object
     * 
     * @param url - URL of the node where you want to 
     *              migrate your active object to
     */
    public void moveTo(String url) {
        try {
            PAMobileAgent.migrateTo(url);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * Retrieves the name of the host where the active
     * object is.
     * 
     * @return The node url and the hostname where the active object is
     */
    public String whereAreYou() {
        try {
            return PAActiveObject.getActiveObjectNodeUrl(PAActiveObject.getStubOnThis()) + " on host " +
                InetAddress.getLocalHost().getHostName();
        } catch (Exception e) {
            return "Localhost lookup failed";
        }
    }

    private void writeObject(java.io.ObjectOutputStream out) throws java.io.IOException {
        //do the default serialization
        out.defaultWriteObject();
        //do the custom serialization
        out.writeObject(toBeSaved);
    }

    private void readObject(java.io.ObjectInputStream in) throws java.io.IOException, ClassNotFoundException {
        //do the default deserialization
        in.defaultReadObject();
        //do the custom deserialization
        toBeSaved = (BigInteger) in.readObject();
    }
}

To learn how to use the methods, see the Serializable interface in the standard JavaDoc.

There are two ways to communicate with an active object which has migrated:

Both techniques have their drawbacks. Two problems arise when using a forwarding scheme, especially if the ambition is scalable mobile agents over WAN. First, the forwarders use resources on a site as long as they have not been garbage collected. Thus, if a chain exists between to objects, it will remains even if there is no new communications going by. Second, the longer the chain is, the more likely it will be cut because of a hardware or software failure. As a consequence, while forwarders are more efficient under some conditions, they do not appear to be scalable, nor reliable.

On the other hand, the server is a single point of failure and a potential bottleneck. If a server is to help communicating with a higher number of mobile agents, then it might not be able to serve requests quickly enough. Furthermore, in case of a crash, it is not possible to communicate with mobile active objects until the server is back. It is possible to avoid most of these issues by having redundant servers with load balancing at the cost of increasing complexity.

Based on these observations and taking into account the variability of the environment, we use a configurable communication protocol which offers the main benefits from both the forwarder and the server while avoiding their drawbacks. Configurable with discrete and continuous parameters, it can be tailored to the environment to offer both performance and reliability.

Forwarders and agents have several parameters which control their behaviour:

If we consider that both the agent and the forwarders can send updates to the server, the server must be able to make the difference between messages from the forwarders and from the agent which are always the most up to date. Also, since we do not have any constraint on the Time To Live (TTL) of the forwarders, it could be that a forwarder at the beginning of a chain dies after on at the end. If this happens and we are not careful when dealing with the requests, the server could erase a more up to date reference with an old one.

As a default, ProActive uses a "Forwarder based" strategy. It means that the forwarders have a unlimited lifetime and the agent never updates the location server.

To configure your own strategy, you have to edit the file ProActive/src/Core/org/objectweb/proactive/core/config/ProActiveConfiguration.xml and change the four values listed above.

<?xml version="1.0" encoding="UTF-8"?>
<ProActiveUserProperties>
    <properties>
        <prop key="proactive.locationserver"
            value="org.objectweb.proactive.ext.locationserver.LocationServer" />
        <prop key="proactive.locationserver.rmi" value="//localhost/LocationServer" />
        <prop key="proactive.securitymanager" value="true" />
        <prop key="proactive.mixedlocation.ttl" value="-1" />
        <prop key="proactive.mixedlocation.updatingForwarder"
            value="false" />
        <prop key="proactive.mixedlocation.maxMigrationNb" value="-1" />
        <prop key="proactive.mixedlocation.maxTimeOnSite" value="-1" />
         <prop key="proactive.implicitgetstubonthis" value="false"/>
    
        <prop key="proactive.net.disableIPv6" value="true" />
        
        <prop key="proactive.future.ac" value="true" />
        <prop key="proactive.stack_trace" value="false" /> <!-- complete stack traces in requests -->
        <prop key="proactive.dgc" value="false" />
        <prop key="proactive.exit_on_empty" value="false" /> <!-- destroy the JVM when there is no more active object -->
        <prop key="schema.validation" value="true" />
        <prop key="proactive.communication.protocol" value="rmi" />
        <prop key="proactive.rmi.port" value="1099" />
       
       <!--<prop key="proactive.java.policy" value="../../../../../../scripts/unix/proactive.java.policy"/>-->
        <prop key="gcm.provider" value="org.objectweb.proactive.core.component.Fractive" />
       <!-- <prop key="proactive.tunneling.connect_timeout" value="2000"/> -->
        <prop key="proactive.communication.rmissh.try_normal_first" value="false" />
        <!-- <prop key="proactive.communication.benchmark.class" value="org.objectweb.proactive.core.remoteobject.benchmark.LargeByteArrayBenchmark"/> -->
        <prop key="proactive.classloading.useHTTP" value="true"/>

        <!-- SSL cipher suites used for RMISSL communications.
    List of cipher suites used for RMISSL, separated by commas.
    default is SSL_DH_anon_WITH_RC4_128_MD5. This cipher suite is used only
    to have encrypted communications, without authentication, and works with default
    JVM's keyStore/TrustStore

    Many others cipher suites can be used. for implementing a certificate authentication...
    see http://java.sun.com/javase/6/docs/technotes/guides/security/jsse/JSSERefGuide.html -->
        <prop key="proactive.ssl.cipher.suites" value="SSL_DH_anon_WITH_RC4_128_MD5"/>


        <!-- <prop key="proactive.ssh.known_hosts" value="/user/mlacage/home/.ssh/known_hosts"/> -->
        <!-- <prop key="proactive.ssh.key_directory" value="/user/mlacage/home/.ssh/"/> -->
        <!-- <prop key="proactive.ssh.port" value="22"/> -->
        <!-- <prop key="proactive.ssh.username" value="mlacage"/> -->
        <prop key="components.creation.timeout" value="10000" /> <!-- timeout in seconds for parallel creation of components -->        
        <!-- ***************************************************** -->
        <!-- MASTER/WORKER PROPERTIES VALUE -->
        <!-- Period of the slave pinging (in ms) -->
        <prop key="proactive.masterworker.pingperiod" value="10000" />
        <!-- Compressing tasks in the repository -->
        <prop key="proactive.masterworker.compresstasks" value="false" />
        
        <!-- Message Tagging -->
        <prop key="proactive.tagmemory.lease.max" value="60" />
        <prop key="proactive.tagmemory.lease.period" value="21" />
        <prop key="proactive.tag.dsf" value="false" />

        <!-- Workaround to fix rmissh conflict -->
        <prop key="proactive.runtime.stayalive" value="true" />        

        <!-- File Transfer Properties -->
        <prop key="proactive.filetransfer.services_number" value="16" />
        <prop key="proactive.filetransfer.blocks_number" value="8" />
        <prop key="proactive.filetransfer.blocks_size_kb" value="512" />
        <prop key="proactive.filetransfer.buffer_size_kb" value="256" />

        <prop key="proactive.test.perf.duration" value="30000" />
        
        <prop key="proactive.log4j.appender.provider" value="org.objectweb.proactive.core.util.log.remote.ThrottlingProvider" />

        <prop key="proactive.gcmd.unix.shell" value="/bin/sh"/>

        <prop key="proactive.mop.writestubondisk" value="false"/>                                        

        <prop key="proactive.webservices.framework" value="axis2" />
        <prop key="proactive.webservices.elementformdefault" value="false" />
        
    </properties>
    <javaProperties>
        <!-- <prop key="java.rmi.server.RMIClassLoaderSpi" value="org.objectweb.proactive.core.classloading.protocols.ProActiveRMIClassLoader"/> -->
        <prop key="java.protocol.handler.pkgs" value="org.objectweb.proactive.core.ssh|org.objectweb.proactive.core.classloading.protocols" />
        <prop key="ibis.name_server.host" value="localhost" />
        <prop key="ibis.name_server.key" value="1" />
        <prop key="ibis.io.serialization.classloader" value="org.objectweb.proactive.core.mop.MOPClassLoader" />
        <prop key="ibis.serialization" value="ibis" />
       </javaProperties>
</ProActiveUserProperties>

A location server is available in the package org.objectweb.proactive.core.body.migration.MixedLocationServer. A simpler one is also available in the package org.objectweb.proactive.ext.util.SimpleLocationServer. The use of location is a little more cumbersome. If want to use this functionality, please refer to the functionalTests.activeobject.locationserver.TestLocationServer functional test. It gives an implementation of an active object migration using the SimpleLocationServer class as location server.

Limitation of this functionality: there can be only one LocationServer for the migration.

The ProActive solution to this problem is to put barriers around try/catch blocks. This way the control flow cannot exit the block, the exception can be handled in the appropriate catch block, and the call is asynchronous within the block.

With this configuration, the potential exception can be thrown at several points:

Let's reuse the previous example to see how to use these barriers:

System.out.println("vvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvv");
PAException.tryWithCatch(Exception.class);
try {
    A a = PAActiveObject.newActive(A.class, null);
    a.throwsException(true); // Asynchronous method call that can throw an exception
    System.out.println("Hello");
    //...
    PAException.endTryWithCatch();
} catch (Exception e) {
    e.printStackTrace();
} finally {
    PAException.removeTryWithCatch();
}
System.out.println("^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^");

With this code, the method call will be asynchronous, and the exception will be handled in the correct catch block. Even if this implies waiting at the end of the try block for the completion of the call. If you try this piece of code, you will see that, contrary to the previous example, Hello will appear and then you will see the caught exception.

Let's see in detail the needed modifications to the code:

These needed annotations can be seen as cumbersome, so we provide a tool to add them automatically to a given source file. It transforms the first example code into the second one. Here is a sample session with the tool:

$ ProActive/bin/trywithcatch.sh MyClass.java
--- ProActive TryWithCatch annotator -----------------------
$ diff -u MyClass.java~ MyClass.java
--- MyClass.java~
+++ MyClass.java
@@ -1,9 +1,13 @@
 public class MyClass {
   public MyClass someMethod(AnotherClass a) {
+     PAException.tryWithCatch(AnException.class);
         try {
            return a.aMethod();
+           PAException.endTryWithCatch();
         } catch (AnException ae) {
            return null;
+        } finally {
+           PAException.removeTryWithCatch();
         }
   }
 }

As we can see, ProActive method calls are added to make sure all calls within try/catch blocks are handled asynchronously. The tool can be found in ProActive/bin/ and is named trywithcatch.[sh|bat]. The default behaviour of trywithcatch.[sh|bat] is to add only PAException... without importing the PAException's package. If you do not want to bother with this manual importation, launch this script with the -fullname option. That will write the full name of PAException, that is to say, org.objectweb.proactive.api.PAException.

We have seen the 3 methods mandatory to perform asynchronous calls with exceptions. However the complete API includes two more calls. So far the blocks boundaries define the barriers. But, some control over the barrier is provided thanks to two additional methods.

The first method is PAException.throwArrivedException(). During a computation an exception may be raised but there is no point from where the exception can be thrown (a future or a barrier). The solution is to call the PAException.throwArrivedException() method which simply queries ProActive to see if an exception has arrived with no opportunity of being thrown back in the user code. In this case, the exception is thrown by this method.

System.out.println("vvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvv");
PAException.tryWithCatch(Exception.class);
try {
    A a = PAActiveObject.newActive(A.class, null);
    a.throwsException(true); // Asynchronous method call that can throw an exception
    //...
    // Throws exceptions which has been already raised by active object
    PAException.throwArrivedException();

    // Should appear since the exception did not have the time
    // to be thrown (due to the sleep(5000))
    System.out.println("Hello");
    //...
    try {
        Thread.sleep(5000);
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
    // Throws exceptions which has been already raised by active object
    PAException.throwArrivedException();

    // Should not appear since the exception had the time to
    // be thrown.
    System.out.println("Hello");
    System.out.println("^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^");
    PAException.endTryWithCatch();
} catch (Exception e) {
    e.printStackTrace();
} finally {
    PAException.removeTryWithCatch();
}

If you run the previous, you will see that the method behaviour is dependent on the timing since the first call does not throw exception whereas the second one throws one. Thus, calling this method may or may not result in an exception being thrown, depending on the time for an exception to come back. That is why another method is provided: PAException.waitForPotentialException(). Unlike the previous PAException.throwArrivedException(), this method is blocking. After calling this method, either an exception is thrown or it is assured that all previous calls in the block completed successfully, so no exception can be thrown from the previous calls.

System.out.println("vvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvv");
PAException.tryWithCatch(Exception.class);
try {
    A a = PAActiveObject.newActive(A.class, null);
    a.throwsException(true); // Asynchronous method call that can throw an exception
    //...
    // At that moment, we want to be sure that no exception has been
    // raised by the code before.
    PAException.waitForPotentialException();
    //...
    // Should not appear since waitForPotentialException
    // is blocking.
    System.out.println("Hello");
    System.out.println("^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^");
    PAException.endTryWithCatch();
} catch (Exception e) {
    e.printStackTrace();
} finally {
    PAException.removeTryWithCatch();
}

In order to create a local master, we use a constructor without parameters:

master = new ProActiveMaster<A, Integer>();

Using this constructor, a master will be created in the current JVM and it will share CPU usage and memory with the user JVM. This master will compute tasks of type A and will produce Integer objects as a results.

In order to create a remote master the following constructors can be used:

/**
 * Creates a remote master that will be created on top of the given Node <br>
 * Resources can be added to the master afterwards
 *
 * @param remoteNodeToUse this Node will be used to create the remote master
 */
public ProActiveMaster(Node remoteNodeToUse)

/**
 * Creates an empty remote master that will be created on top of the given Node with an initial worker memory
 *
 * @param remoteNodeToUse this Node will be used to create the remote master
 * @param memoryFactory factory which will create memory for each new workers
 */
public ProActiveMaster(Node remoteNodeToUse, MemoryFactory memoryFactory)

Using either of these constructors, a master will be created in the specified remote resource(JVM), the master will share CPU usage and memory with existing running applications on the remote host. The mechanism in use to deploy the master remotely is the ProActive deployment mechanism (see Chapter 15, XML Deployment Descriptors for further details).

The Worker memory structure is very simple: it consists of <key, value> associations. A Java object value is therefore saved in the memory with the given name, and this name will be needed to retrieve the value later on.

The Worker Memory API consists of three methods save, load, and erase. The interface to the worker memory is available when running a Task as a parameter of the run method. The user can use this interface to save, load or erase objects in the local worker's memory. Below is the detailed WorkerMemory interface:

/**
 * This interface gives access to the memory of a worker, a task can record data in this memory under a specific name. <br/>
 * This data could be loaded later on by another task <br/>
 * @author The ProActive Team
 *
 */
@PublicAPI
public interface WorkerMemory {

    /**
     * Save data under a specific name
     * @param name name of the data
     * @param data data to be saved
     */
    void save(String name, Object data);

    /**
     * Load some data previously saved
     * @param name the name under which the data was saved
     * @return the data
     */
    Object load(String name);

    /**
     * Erase some data previously saved
     * @param name the name of the data which need to be erased
     */
    void erase(String name);
}

A user can store data in the Workers' memory either when:

Usage of the first mechanism is done by providing a list of <key, value> pairs (Map) to the constructors of the ProActiveMaster class. Every constructors detailed above have a version including this extra parameter. The given list will be the initial memory of every Workers created by the master.

Usage of the second mechanism is done by using the WorkerMemory parameter in the Task interface's run method. In contradiction with the first method, only the Worker currently running the Task will store the given data.

Loading and using any object stored in a Worker's memory is simply done through the WorkerMemory parameter in the run method of the Task interface.

Calcium is part of the ProActive Grid Middleware for programming structured parallel and distributed applications. The framework provides a basic set of structured patterns (skeletons) that can be nested to represents more complex patterns. Skeletons are considered as a high level programming model because all the parallelisms details are hidden from the programmer. In Calcium, distributed programming is achieved by using ProActive deployment framework and active object model.

The Calcium implementation can be found in the following package of the ProActive Middleware distribution:

package org.objectweb.proactive.extensions.calcium;

The following steps must be performed for programming with this framework.

Problems inputed into the framework are treated as tasks. The tasks are interpreted by the remote skeleton interpreters as shown in the following figure:

Figure 7.1. Task Flow in Calcium

All the generation, the distribution of tasks and their resources are completely hidden from the programmer. In fact, the task concept is never used when programming in Calcium.

The approach consists of dividing the original search space into several smaller search spaces, and computing each sub search space in parallel. Therefore, the most suitable pattern corresponds to Divide and Conquer.

public Skeleton<Interval, Primes> root;

public FindPrimes() {
    root = new DaC<Interval, Primes>(new IntervalDivide(), new IntervalDivideCondition(),
        new SearchInterval(), new JoinPrimes());
}

We will call the problem an Interval and we will represent it using the following class.

package org.objectweb.proactive.extensions.calcium.examples.findprimes;

import java.io.Serializable;


class Interval implements Serializable {
    /**
     * 
     */
    private static final long serialVersionUID = 51L;
    public int min;
    public int max;
    public int solvableSize;

    /**
     * Creates a new interval to search for primes.
     * @param min  Beginning of interval
     * @param max  End of interval
     * @param solvableSize Acceptable size of search interval
     */
    public Interval(int min, int max, int solvableSize) {
        this.min = min;
        this.max = max;
        this.solvableSize = solvableSize;
    }

}

The primes that are found will be stored in a Primes class.

package org.objectweb.proactive.extensions.calcium.examples.findprimes;

import java.io.Serializable;
import java.util.Vector;


public class Primes implements Serializable {
    /**
     * 
     */
    private static final long serialVersionUID = 51L;
    public Vector<Integer> primes;

    public Primes() {
        primes = new Vector<Integer>();
    }
}

package org.objectweb.proactive.extensions.calcium.examples.findprimes;

import org.objectweb.proactive.extensions.calcium.muscle.Divide;
import org.objectweb.proactive.extensions.calcium.system.SkeletonSystem;


public class IntervalDivide implements Divide<Interval, Interval> {

    /**
     * 
     */
    private static final long serialVersionUID = 51L;

    public Interval[] divide(Interval param, SkeletonSystem system) {

        int middle = param.min + ((param.max - param.min) / 2);

        Interval top = new Interval(middle + 1, param.max, param.solvableSize);

        Interval bottom = new Interval(param.min, middle, param.solvableSize);

        return new Interval[] { top, bottom };
    }
}

package org.objectweb.proactive.extensions.calcium.examples.findprimes;

import org.objectweb.proactive.extensions.calcium.muscle.Condition;
import org.objectweb.proactive.extensions.calcium.system.SkeletonSystem;


public class IntervalDivideCondition implements Condition<Interval> {
    /**
     * 
     */
    private static final long serialVersionUID = 51L;

    public boolean condition(Interval params, SkeletonSystem system) {
        return (params.max - params.min) > params.solvableSize;
    }
}

package org.objectweb.proactive.extensions.calcium.examples.findprimes;

import org.objectweb.proactive.extensions.calcium.muscle.Execute;
import org.objectweb.proactive.extensions.calcium.system.SkeletonSystem;


public class SearchInterval implements Execute<Interval, Primes> {
    /**
     * 
     */
    private static final long serialVersionUID = 51L;

    public Primes execute(Interval param, SkeletonSystem system) {
        Primes primes = new Primes();

        for (int i = param.min; i <= param.max; i++) {
            if (isPrime(i)) {
                primes.primes.add(new Integer(i));
            }
        }

        return primes;
    }

    private boolean isPrime(int p) {
        for (int i = 2; i < p; i++) {
            if ((p % i) == 0) {
                return false;
            }
        }
        return true;
    }
}

package org.objectweb.proactive.extensions.calcium.examples.findprimes;

import java.util.Collections;

import org.objectweb.proactive.extensions.calcium.muscle.Conquer;
import org.objectweb.proactive.extensions.calcium.system.SkeletonSystem;


public class JoinPrimes implements Conquer<Primes, Primes> {
    /**
     * 
     */
    private static final long serialVersionUID = 51L;

    public Primes conquer(Primes[] p, SkeletonSystem system) {
        Primes conquered = new Primes();

        for (Primes param : p) {
            conquered.primes.addAll(param.primes);
        }

        Collections.sort(conquered.primes);
        return conquered;
    }
}

The instantiation of the framework is performed in the following way:

Environment environment = EnvironmentFactory.newMultiThreadedEnvironment(2);
Calcium calcium = new Calcium(environment);

Stream<Interval, Primes> stream = calcium.newStream(root);

Vector<CalFuture<Primes>> futures = new Vector<CalFuture<Primes>>(3);
futures.add(stream.input(new Interval(1, 6400, 300)));
futures.add(stream.input(new Interval(1, 100, 20)));
futures.add(stream.input(new Interval(1, 640, 64)));
calcium.boot(); //begin the evaluation

for (CalFuture<Primes> future : futures) {
    Primes res = future.get();
    for (Integer i : res.primes) {
        System.out.print(i + " ");
    }
}

Stats futureStats = future.getStats();
System.out.println(futureStats);

StatsGlobal stats = calcium.getStatsGlobal();
System.out.println(stats);

calcium.shutdown();

The MultithreadEnvironment is the simplest execution environment. It uses threads to execute tasks, and can thus be used efficiently on multiprocessor machines. It is also an easier environment to debug applications, before submitting them to a distributed environment.

Environment environment = EnvironmentFactory.newMultiThreadedEnvironment(2);

The ProActiveEnvironment is the current stable way of executing a skeleton program on a distributed, but controlled, execution environment. It is mostly suitable for short lived distributed applications, as it does not yet support a suitable fault-tolerance mechanism. The ProActiveEnvironment uses ProActive Deployment Descriptors to acquire computation nodes and it uses active objects to communicate and distribute the program to the computation nodes.

String descriptor = FindPrimes.class.getResource("LocalDescriptor.xml").getPath();
Environment environment = EnvironmentFactory.newProActiveEnvironment(descriptor);

To instantiate the environment, a path toward a descriptor deployment has to be specified. The ProActiveEnvironment requires that this descriptor file provides the following contractual variables:

<variables>
     <descriptorVariable name="SKELETON_FRAMEWORK_VN" value="framework" />         
    <descriptorVariable name="INTERPRETERS_VN" value="interpreters" />
     <programDefaultVariable name="MAX_CINTERPRETERS" value="3"/>
</variables>

Where the variables represent:

The new GCM Deployment mechanism is also supported through the following factory mechanism:

String descriptor = FindPrimes.class.getResource("LocalDescriptor.xml").getPath();
Environment environment = EnvironmentFactory.newProActiveEnviromentWithGCMDeployment(descriptor);

The ProActiveSchedulerEnvironment is suitable for executing long running applications, and uses the ProActive Scheduler at the lower level to handle the distribution and execution of tasks. Currently, tasks requiring file access and transfer are not supported using this environment, but will be supported in future releases.

Environment environment = ProActiveSchedulerEnvironment.factory("schedulerURL","user", "password");

To use the scheduler, a URL with its location, a username and a password must be provided.

The ProActiveSchedulerEnvironment is currently under development and as such represents an unstable version of the framework, thus it is located in the following package of the ProActive distribution:

package org.objectweb.proactive.extra.calcium.environment.proactivescheduler;

The workspace is an abstraction that can be used to create files from inside muscle functions. The framework guarantees that: 1. Any File created in the workspace will have read/write permissions; 2. If a File is passed as parameter to other muscle functions, the File will be locally available when another muscle function access it. Where File corresponds to the standard Java type java.io.File.

package org.objectweb.proactive.extensions.calcium.system;

import java.io.File;
import java.io.FileFilter;
import java.io.IOException;
import java.net.URL;

import org.apache.log4j.Logger;
import org.objectweb.proactive.annotation.PublicAPI;
import org.objectweb.proactive.core.util.log.Loggers;
import org.objectweb.proactive.core.util.log.ProActiveLogger;


/**
 * This class is the interface for creating files on the computation node.
 *
 * This class is not Serializable on purpose, since it is environment dependent.
 *
 * @author The ProActive Team
 */
@PublicAPI
public interface WSpace {
    static Logger logger = ProActiveLogger.getLogger(Loggers.SKELETONS_SYSTEM);

    /**
     * Copies a File into the workspace, and returns
     * a reference on the new File.
     *
     * @param src  The original location of the file.
     * @return A reference to the file inside the workspace.
     * @throws IOException
     */
    public File copyInto(File src) throws IOException;

    /**
     * Downloads the file specified by the URL and places a copy inside
     * the workspace.
     *
     * @param src The location of the original file
     * @return A reference to the file inside the workspace.
     * @throws IOException
     */
    public File copyInto(URL src) throws IOException;

    /**
     * This method is used to get a reference on a file inside the workspace.
     *
     * Note that this is only a reference, and can point to an unexistent File.
     * ie. no File is actually created by invoking this method.
     *
     * @param path The path to look inside the workspace.
     * @return A reference to the path inside the workspace.
     */
    public File newFile(String path);

    /**
     * This method is used to get a reference on a file inside the workspace.
     *
     * Note that this is only a reference, and can point to an unexistent File.
     * ie. no File is actually created by invoking this method.
     *
     * @param path The path to look inside the workspace.
     * @return A reference to the path inside the workspace.
     */
    public File newFile(File path);

    /**
     * This method returns true if a file with this name exists in the work space.
     *
     * @return true if the file exists, false otherwise.
     */
    public boolean exists(File path);

    /**
     * This method returns a list of the files currently available on the root
     * of the workspace.
     *
     * @return The list of files that are currently held in the workspace.
     */
    public File[] listFiles();

    /**
     * This method returns a list of the files currently available on the root
     * of the workspace that match the specified filter.
     *
     * @param filter The filter to use for listing the files.
     * @return The list of files thar are currently held in the workspace and match the specified filter.
     */
    public File[] listFiles(FileFilter filter);
}

Files are treated in a deep-copy fashion, analogous with parameters/results of muscle functions. That is to say, when a File reference is passed from one muscle function to the next, the File's data is copied. From this point modifications made on on the File by different muscle functions are made on copies of the File's data.

Muscle functions can be annotated to improve the file transfer performance. By annotating a muscle function, the Calcium framework will try to pre-fetch files matching the annotation, and passed as parameters to the function in advance, before the function is executed. The current supported annotation can fetch a file based on its name and size:

public @interface PrefetchFilesMatching {
    String name() default "[unassigned]";

    long sizeSmallerThan() default Long.MIN_VALUE;

    long sizeBiggerThan() default Long.MAX_VALUE;
}

It is important to note that annotations only represent an optimization and are not required for the File support to work.

The following example is taken from the BLAST skeleton program. The muscle function presented here conquers the results of several parallel BLASTs into a single file. First, the annotation is used to try and pre-fetch files that begging with the prefix "merged". Then, a new File is created in the workspace to hold the merged files. Then, a mergeFiles function is called to merge the results, and is not detailed here since it is specific to BLAST. Finally the new File holding the merged results is returned.

@PrefetchFilesMatching(name = "merged.*")
public class ConquerResults implements Conquer<File, File> {
    /**
     * 
     */
    private static final long serialVersionUID = 51L;
    static Logger logger = ProActiveLogger.getLogger(Loggers.SKELETONS_APPLICATION);

    public File conquer(File[] param, SkeletonSystem system) throws Exception {
        if (logger.isDebugEnabled()) {
            logger.debug("Conquering Results");
        }

        WSpace wspace = system.getWorkingSpace();

        //Create a reference on the result merged file
        File merged = wspace.newFile("merged.result" + ProActiveRandom.nextPosInt());

        //Merge the files
        mergeFiles(merged, param);

        //Return the result
        return merged;
    }

    /**
     * This method takes a list of files and copies their content into a new
     * file, merging them all together.
     *
     * @param merged
     *            The output file, which will hold the merged result.
     * @param files
     *            The list of files to merge.
     *
     * @throws IOException
     */
    private void mergeFiles(File merged, File[] files) throws IOException {
        if (logger.isDebugEnabled()) {
            logger.debug("Conquering results into File: " + merged);
        }

        BufferedWriter bw = new BufferedWriter(new FileWriter(merged));
        for (File f : files) {
            BufferedReader br = new BufferedReader(new FileReader(f));

            String line = br.readLine();
            while (line != null) {
                bw.write(line + System.getProperty("line.separator"));
                line = br.readLine();
            }
            br.close();
        }

        bw.close();
    }
}

The input and output are transparent as long as the File type class is used to reference files. All File type referenced from inside the input parameter are imported into the Calcium framework when submitting the parameter into the stream.

Stream<BlastParams, File> stream = calcium.newStream(root);
CalFuture<File> future = stream.input(parameters);

try {
    File res = future.get();
} catch (Exception e) {
    e.printStackTrace();
}

Then, when the result is available, all File typed referenced in the result object are copied into the local machine before the result is returned to the user.

These statistics refer to the global state of the framework by providing state information. The tasks can be in three different states: ready for execution, processing , waiting for other tasks to finish, and finished (ready to be collected by the user). The statistics corresponding to these states are:

Statistics for a specific moment can be directly retrieved from the Calcium instance:

StatsGlobal stats = calcium.getStatsGlobal();

An alternative is to create a monitor that can perform functions based on the statistics. In the following example, we activate a simple logger monitor that prints the statistics every second.

Monitor monitor = new SimpleLogMonitor(calcium, 1);
monitor.start();
// ...
monitor.stop();

This statistics are specific for each result obtained from the framework. They provide information on how the result was obtained:

for (CalFuture<Primes> future : futures) {
    Primes res = future.get();
    Stats futureStats = future.getStats();
    System.out.println(futureStats);
}

Total barrier directly involves the SPMD group. A call to

public static void totalBarrier(String barrierName)

will block until all the members in the SPMD group have reached and called the identical barrier primitive. Such a call communicates with all the members of the SPMD group. The barrier is released when the Active Object has received a barrier message from all other members of the SPMD group (including itself).

	Warning
	The string parameter is used as a unique identity name for the barrier. It is the programmer responsibility to ensure that two (or more) different barriers with the same id name are not invoked simultaneously.

Let us take a Java class that contains a method calling a total barrier, here the method foo:

myspmdgroup.foo();
PASPMD.totalBarrier("'1'");
myspmdgroup.bar();
PASPMD.totalBarrier("'2'");
myspmdgroup.gee();

Note that usually, strings used as unique ID are more complex; they can be based on the full name of the class or the package (org.objectweb.proactive.ClassName), for example. The SPMD group is built as follows:

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

When the method start() is called on different instances of the class containing it, the different instances will wait for myspmdgroup.foo() to be completed and the barrier call to be made before they start myspmdgroup.bar(). The instances will also synchronize after the call to myspmdgroup.bar() as there is also a second barrier PASPMD.totalBarrier("'2'").

The programmer has to ensure that all the members of an SPMD group call the barrier method otherwise the members of the group may indefinitely wait.

The Neighbour barrier is a kind of lightweight barrier, involving only the Active Objects specified in a given group.

neighbourBarrier(String,group) initiates a barrier only with the objects of the specified group. Those objects, that contribute to the end of the barrier state, are called neighbours as they are usually local to a given topology. An object that invokes the Neighbour barrier HAVE TO BE IN THE GROUP given as parameter. The barrier message is only sent to the group of neighbours.

The programmer has to explicitly build this group of neighbours. The topology API can be used to build such a group. Topologies are groups. They just give special access to their members or (sub)groups members. For instance, a matrix fits well with the topology Plan that provides methods to get the reference of neighbour members (left, right, up, down). See the javadoc of the topology package for more information (org.objectweb.proactive.core.group.topology).

Like for the Total barrier, the string parameter represents a unique identity name for the barrier. The second parameter is the group of neighbours built by the programmer. Here is an example:

// synchronization to be sure that all submatrix have exchanged borders
PASPMD.neighbourBarrier("SynchronizationWithNeighbors" + this.iterationsToStop, this.neighbors);

Refer to the Jacobi example to see a use-case of the Neighbour barrier. Each submatrix needs only to be synchronized with the submatrixes which are in its cardinal neighbours.

This barrier increases the asynchronism and reduces the amount of exchanged messages.

The Method barrier does not involve more extra messages to communicate (i.e. the barrier messages). Communications between objects to release a barrier are achieved by the standard method call and request reception of ProActive.

As a standard barrier, the method methodBarrier(String[]) will finish the current request served by the active object that calls it, but it then waits for a request on the specified methods to resume. The array of string contains the name of the awaited methods. The order of the methods does not matter. For example:

 PASPMD.methodBarrier({'foo', 'bar', 'gee'}); 

The caller will stop and wait for the three methods. "bar" or "gee" can came first, then foo. If one wants to wait for foo, then wait for bar, then wait for gee, three calls can be successively done:

PASPMD.methodBarrier({'foo'});
PASPMD.methodBarrier({'bar'});
PASPMD.methodBarrier({'gee'}); 

A method barrier is used without any group (SPMD or not). To learn more on Groups, please refer to Chapter 3, Typed Group Communication.