Active Objects: Creation And Advanced Concepts

To create a single active object from the class Worker in the local JVM we use the following code. 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.

Worker charlie;
//set the constructor values to be used
Object[] params = new Object[] { new IntegerWrapper (26), "Charlie" };
try {
	charlie = (Worker)PAActiveObject.newActive(Worker.class.getName(),
				params);
} 
catch (ActiveObjectCreationException aoExcep) {
	// 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 each node contained in the virtual node someVirtualNode . In this case the Worker constructor doesn't take any arguments. However, we have to create an Object array equal in length with 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.

Worker[] workers;
Node[] nodes;
try{
	nodes =  someVirtualNode.getNodes();
	Worker[] workers = (Worker[]) PAActiveObject.newActiveInParallel(Worker.class.getName(), 
						new Object [nodes.length][],nodes);
}						
catch (ActiveObjectCreationException aoExcep){
	System.err.println(aoExcep.getMessage());
}
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(...) must 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 .

//class for which passing null as 
//an argument for the constructor 
//leads to the generation of an exception
class DontPassNullForTheConstructor{
	public DontPassNullForTheConstructor (String s){
	}
	public DontPassNullForTheConstructor (Vector v){
	}
}

If we use PAActiveObject.newActiveInParallel(..., Nodes[] nodes) we have the make sure 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 of the array contains the constructor arguments for each deployed active object. Different active objects can have different constructor arguments. If we use PAActiveObject.newActiveInParallel(..., VirtualNode virtualNode) the virtual node will be activated if it is not active and one active object will be started on each node contained in the virtual node. In this case all the active objects take the same constructor arguments as we use a one dimensional array java.lang.Object[] .

It is possible to pass an argument to the call to newActive in order to create the new active object on a specific JVM, possibly remote. The JVM is identified using a Node object, an array of Node or a String that contains an URL pointing to the node. If the parameter is not given, the active object is created in the current JVM and is attached to a default Node . The node argument and the URL string are used with PAActiveObject.newActive(...) and PAActiveObject.turnActive(...). The array of Node and the virtual node argument is used with PAActiveObject.newActiveInParallel(...) . The virtual node will be activated if it is not active and one active object will be started on each node contained in the virtual node.

Worker charlie;
Node node;
//set the constructor values to be used
Object[] params = new Object[] { new Long(26), "Charlie" };
try {
	//get the node form a virtual node
	node = someVirtualNode.getNode();
	charlie = (Worker)PAActiveObject.newActive(Worker.class.getName(),
				params, node);
} 
catch (ActiveObjectCreationException aoExcep) {
	// creation of ActiveObject failed
	System.err.println(aoExcep.getMessage());
}
catch (NodeException nodeExcep){
	System.err.println(nodeExcep.getMessage());
}

To deploy using the node's URL we just have the use the URL instead of the Node as argument:

Worker charlie;
String nodeURL;
Node node;
//set the constructor values to be used
Object[] params = new Object[] { new IntegerWrapper (26), "Charlie" };
try {
	//get the node form a virtual node
	node = someVirtualNode.getNode();
	//get the node URL
	nodeURL = new String(node.getNodeInformation().getURL());
	charlie = (Worker)PAActiveObject.newActive(Worker.class.getName(),
				params, nodeURL);
} 
catch (ActiveObjectCreationException aoExcep) {
	// creation of ActiveObject failed
	System.err.println(aoExcep.getMessage());
}
catch (NodeException nodeExcep){
	System.err.println(nodeExcep.getMessage());
}

The usage is similar for PAActiveObject.turnActive(...) as in the examples above.

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

Worker[] workers;
//the virtual nodes needs to be initialized before using it
try{
	Worker[] workers = (Worker[]) PAActiveObject.newActiveInParallel(Worker.class.getName(), 
						new Object [], someVirtualNode);
}						
catch (ActiveObjectCreationException aoExcep){
	System.err.println(aoExcep.getMessage());
}
catch (ClassNotFoundException classExcep) {
	System.err.println(classExcep.getMessage());
}

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

Two ways are possible to define each of the three phases of an active object.

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

9.5.4.1.  Algorithms deciding which activity to invoke

The algorithms for each running phase are the following ( activity is the object passed as a parameter to newActive or turnActive ):

Figure 9.2. Activity algorithm

import org.objectweb.proactive.Service;
import org.objectweb.proactive.RunActive;
import org.objectweb.proactive.InitActive;
import org.objectweb.proactive.Body;
import org.objectweb.proactive.api.PAActiveObject;

public class CustomActivity implements InitActive, RunActive {
	private String myName;
	public String getName() {
		return myName;
	}
	// -- implements InitActive
	public void initActivity(Body body) {
		myName = body.getName();
	}
	// -- implements RunActive for serving request in a LIFO fashion
	public void runActivity(Body body) {
		Service service = new Service(body);
		while (body.isActive()) {
			//LIFO order
			service.blockingServeYoungest();
		}
	}
	public static void main(String[] args) throws Exception {
		CustomActivity a = (CustomActivity) 
							PAActiveObject.newActive(CustomActivity.class.getName(),
							null);
		String call1 = a.getName();
	}
  }

import org.objectweb.proactive.ProActive;
import org.objectweb.proactive.Service;
import org.objectweb.proactive.RunActive;
import org.objectweb.proactive.Body;
import org.objectweb.proactive.api.PAActiveObject;


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();
					ProActive.exitSuccess();
				}
		}
	}
}

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 fashion
	public void runActivity(Body body) {
		Service service = new Service(body);
		while (body.isActive()) {
			System.out.println("Serving...");
			service.blockingServeYoungest();
		}
	}
}

import org.objectweb.proactive.ActiveObjectCreationException;
import org.objectweb.proactive.api.PAActiveObject;
import org.objectweb.proactive.core.node.NodeException;

public class Main  {
	public static void main(String[] args)  {
		Worker w = new Worker();
		try {
			w = (Worker) PAActiveObject.turnActive(w, 
						null,   new LIFOActivity(), null);
			w.doNothingGracefully();
			w.doNothingGracefully();
			w.doNothingGracefully();
			w.doNothingGracefully();

		} 
		catch (ActiveObjectCreationException aoExcep) {
			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 adding a static method to class WorkerFactory that takes care of instantiating the active object and returns it. The code is:

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;

public class WorkerFactory extends Worker {
	public static Worker createActiveWorker (int age, String name, Node node) {
		Object[] params = new Object[] {new IntWrapper(age), name};
		try {
			return (Worker) PAActiveObject.newActive(PrimeManager.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:

import org.objectweb.proactive.core.node.Node;

public class Main  {
	public static void main(String[] args)  {
		Worker w = new Worker();
		Node node = null; //deploy locally
		w = WorkerFactory.createActiveWorker(26, "Charlie", node);
		w.doNothingGracefully();
	}
}

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:

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;

public class CustomMetaObjectFactory extends ProActiveMetaObjectFactory {
	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 {
		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 {
			public CustomRequest(MethodCall methodCall,
					UniversalBody sourceBody, boolean isOneWay, long sequenceID) {
				super(methodCall, sourceBody, isOneWay, sequenceID);
				System.out.println("I am a custom request handler");
			}
		}//end inner class RequestHandler
	}//end inner class CustomRequestFactory
}

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 {
   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 ProActiveComponentFactory componentFactoryInstance;
   protected ProActiveSecurityManager proActiveSecurityManager;
   protected FTManagerFactory ftmanagerFactoryInstance;
   protected DebuggerFactory debuggerFactoryInstance;
   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();
   }

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

   //
   // -- 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 ProActiveComponentFactory newComponentFactory() {
       return this.componentFactoryInstance;
   }

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

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

   //
   // -- 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 ProActiveComponentFactory newComponentFactorySingleton(
           ComponentParameters initialComponentParameters) {
       return new ProActiveComponentFactoryImpl(initialComponentParameters);
   }

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

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

   //  //
   //  // -- INNER CLASSES -----------------------------------------------
   //  //
   protected static class RequestFactoryImpl implements RequestFactory, java.io.Serializable {
       public Request newRequest(MethodCall methodCall, UniversalBody sourceBody, boolean isOneWay,
               long sequenceID) {
           //########### 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);
           //}
       }
   }

   // end inner class RequestFactoryImpl
   protected static class ReplyReceiverFactoryImpl implements ReplyReceiverFactory,
 java.io.Serializable {
       public ReplyReceiver newReplyReceiver() {
           return new org.objectweb.proactive.core.body.reply.ReplyReceiverImpl();
       }
   }

   // end inner class ReplyReceiverFactoryImpl
   protected class RequestReceiverFactoryImpl implements RequestReceiverFactory,
 java.io.Serializable {
       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 {
       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 {
       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 MigrationManagerFactoryImpl
   //    protected static class RemoteBodyFactoryImpl implements RemoteBodyFactory,
   //        java.io.Serializable {
   //        public UniversalBody newRemoteBody(UniversalBody body) {
   //            try {
   //                if (Constants.IBIS_PROTOCOL_IDENTIFIER.equals(
   //                            ProActiveConfiguration.getInstance()
   //                                                     
 .getProperty(Constants.PROPERTY_PA_COMMUNICATION_PROTOCOL))) {
   //                    if (logger.isDebugEnabled()) {
   //                        logger.debug(
   //                            "Using ibis factory for creating remote body");
   //                    }
   //                    return new org.objectweb.proactive.core.body.ibis.IbisBodyAdapter(body);
   //                } else if (Constants.XMLHTTP_PROTOCOL_IDENTIFIER.equals(
   //                            ProActiveConfiguration.getInstance()
   //                                                     
 .getProperty(Constants.PROPERTY_PA_COMMUNICATION_PROTOCOL))) {
   //                    if (logger.isDebugEnabled()) {
   //                        logger.debug(
   //                            "Using http factory for creating remote body");
   //                    }
   //
   //                    return new org.objectweb.proactive.core.body.http.HttpBodyAdapter(body);
   //                } else if (Constants.RMISSH_PROTOCOL_IDENTIFIER.equals(
   //                            ProActiveConfiguration.getInstance()
   //                                                     
 .getProperty(Constants.PROPERTY_PA_COMMUNICATION_PROTOCOL))) {
   //                    if (logger.isDebugEnabled()) {
   //                        logger.debug(
   //                            "Using rmissh factory for creating remote body");
   //                    }
   //                    return new org.objectweb.proactive.core.body.rmi.SshRmiBodyAdapter(body);
   //                } else {
   //                    if (logger.isDebugEnabled()) {
   //                        logger.debug(
   //                            "Using rmi factory for creating remote body");
   //                    }
   //                    return new org.objectweb.proactive.core.body.rmi.RmiBodyAdapter(body);
   //                }
   //            } catch (ProActiveException e) {
   //                throw new ProActiveRuntimeException("Cannot create Remote body adapter ",
   //                    e);
   //            }
   //        }
   //    }

   // end inner class RemoteBodyFactoryImpl
   protected static class ThreadStoreFactoryImpl implements ThreadStoreFactory,
 java.io.Serializable {
       public ThreadStore newThreadStore() {
           return new org.objectweb.proactive.core.util.ThreadStoreImpl();
       }
   }

   // end inner class ThreadStoreFactoryImpl
   protected static class ProActiveSPMDGroupManagerFactoryImpl implements
 ProActiveSPMDGroupManagerFactory,
           java.io.Serializable {
       public ProActiveSPMDGroupManager newProActiveSPMDGroupManager() {
           return new ProActiveSPMDGroupManager();
       }
   }

   // end inner class ProActiveGroupManagerFactoryImpl
   protected class ProActiveComponentFactoryImpl implements ProActiveComponentFactory,
 java.io.Serializable {
       // COMPONENTS
       private ComponentParameters componentParameters;

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

       public ProActiveComponent newProActiveComponent(Body myBody) {
           return new ProActiveComponentImpl(this.componentParameters, myBody);
       }
   }

   // FAULT-TOLERANCE
   protected class FTManagerFactoryImpl implements FTManagerFactory, Serializable {
       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 {
       public Debugger newDebugger() {
           return new DebuggerImpl();
       }
   }

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

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

   @Override
   public Object clone() throws CloneNotSupportedException {
       ProActiveMetaObjectFactory clone = null;

       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 that 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"};
Worker w;
try {
	w = (Worker) PAActiveObject.newActive(Worker.class.getName(),
			null,null,null,null,CustomMetaObjectFactory.newInstance());
} 
catch (ActiveObjectCreationException e) {
	e.printStackTrace();
} 
catch (NodeException e) {
	e.printStackTrace();
}

In the case of a turnActive we would have:

Worker w = new Worker(26, "Charlie");

try {
	charlie = (Worker) ProActive.turnActive(Worker.class.getName(), null, null,
 CustomMetaObjectFactory.newInstance());
}
catch (ActiveObjectCreationException aoExcep) {
	// creation of ActiveObject failed
	System.err.println(aoExcep.getMessage());
}
catch (NodeException nodeExcep){
	System.err.println(nodeExcep.getMessage());
}

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 metaobject protocol (MOP) ProActive is built on 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 request 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 request 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 possible a method call on an active object is reified as an asynchronous request. If not possible the call is synchronous and blocks until the reply is received. In case the request is asynchronous, it immediately returns a future object.

This 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 doesn't need to invoke methods on the returned object, in which case 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 anotators to deal with asynchronous exceptions. See Chapter 12, 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 9.4. 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 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 expected to update awaited futures. When such a ping fails, the associated awaited future is updated with a runtime exception. Accessing the future will throw an exception.

public static void function() {
	F future = ao.asyncCall();
	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). To find out more about the fault tolerance mechanism read Chapter 32, Fault-Tolerance

An automatic continuation can occur when sending a request (parameter of the request is a future or contains a future) or when sending a reply (the result is a future or contains a future). Outgoing futures are registered in the FuturePool of the Active Object sending the future. Registration for the couple Future - destination Body as an automatic continuation occurs 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.bodyDestination) a reference to the destination body. Hence when a future Future is serialized by the same 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 couple Future - Destination Body 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: A, B, and Worker. The A and B 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 B class has a bar(Value v) method.

public class A {
 	.... 
	public static void main(String[] args){
		......
		B b = new B(); 
		Worker worker = (Worker) PAActiveObject.newActive(Worker.class.getName(), //class name
						null);//constructor arguments 
		Value v1 = worker.foo(); //v1 is a future
		Value v2 = b.bar(v1); //v1 is passed as parameter
		........
	}	//end of main
} //end of class A

where

public class Worker {
	... 
	public Result foo(){ 
		...
	}
	...
} //end of class Worker

and

public class B {
	...
	public Result bar (Value v) {
		...
	}
	...
} //end of class B
			

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 is similar if the caller was an active object or if the 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.

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) on B.

The call on B leads to the creation of another future to which v2 will point to. Also because v1 has not been updated yet there will be a 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 recieved value and update the value v2 in A.

The final state is with all the future values updated.

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