ProActive is an open source Java library aiming to simplify the programming of multithreaded, parallel, and distributed applications for Grids, multi-cores, clusters, and data-centers. It allows concurrent and parallel programming and offers distributed and asynchronous communications, mobility, and a deployment framework. With a small set of primitives, ProActive provides an API allowing the development of parallel applications, which may be deployed on distributed systems and on Grids. ProActive does not require any modifications to Java or to the Java Virtual Machine (JVM), therefore allowing the deployment of applications using the ProActive API on any operating system that provides a compatible JVM.

Overall, ProActive promotes a few basic and simple principles:

Figure 1.1. ProActive Features

Figure 1.1, “ProActive Features” exposes the main ProActive features:

ProActive is based on the concept of active object, which is an entity with its own configurable activity. A distributed or concurrent application using ProActive is composed of a number of entities called active objects. Each active object has one distinguished element, the root, which is the only entry point to the active object. Each active object has its own thread of control and is granted the ability to decide in which order to serve the incoming method calls that are automatically stored in a queue of pending requests. Method calls sent to active objects are asynchronous with transparent future objects and synchronization is handled by a mechanism known as wait-by-necessity. There is a short rendezvous at the beginning of each asynchronous remote call, which blocks the caller until the call has reached the context of the callee. More detailed information on active objects will be presented in "ProActiveBasis".

The ProActive solution relies on asynchronous method calls mastering both complexity and efficiency. It also proposes advanced features such as groups, mobility, and components. In the framework of formal calculus, Asynchronous Sequential Processes (ASP), confluence, and determinism have been proven for this programming model: CH05 and CHS04.

Grids gather large amounts of heterogeneous resources, different processor architectures and operating systems. In this context, using a language which relies on a virtual machine allows maximum portability. ProActive is developed in Java, a cross-platform language and the compiled application may run on any operating system providing a compatible virtual machine. Moreover, ProActive only relies on standard APIs and does not use any operating-system specific routines, other than to run daemons or to interact with legacy applications. There are no modifications to the JVM nor to the semantics of the Java language, and the bytecode of the application classes is never modified.

In ProActive, the distribution is transparent: invoking methods on remote objects does not require the developer to design remote objects with an explicit remote mechanism. Therefore, the developer can focus on the business logic since the distribution is automatically handled and transparent. Moreover, the ProActive library preserves polymorphism on remote objects (through the reference stub, which is a subclass of the remote root object).

Communication between active objects is realized through method invocations, which are reified and passed as messages. These messages are serializable Java objects which may be compared to TCP packets. Indeed, one part of the message contains routing information towards the different elements of the library, and the other part contains the data to be communicated to the called object. Although all communications proceed through method invocations, the communication semantics depends upon the signature of the method, and the resulting communication may not always be asynchronous. Three cases are possible: synchronous invocation, one-way asynchronous invocation, and asynchronous invocation with future result.

If an invocation from an object A on an active object B triggers another invocation on another active object C, the future result received by A may be updated with another future object. In that case, when the result is available from C, the future of B is automatically updated, and the future object in A is also update with this result value, through a mechanism called automatic continuation.

ProActive programming also features a general purpose component framework allowing software architects to design their parallel applications using a hierarchical component framework: the Grid Component Model (GCM).

The GCM is a high level programming framework which relies on Active Object and therefore inherits usual ProActive properties like asynchronous call, automatic continuation, deployment interoperability...

The GCM extends the Fractal component model to cover distribution and parallel application's needs. Using the GCM, you can easily express at design time the dependencies between software blocks composing your application. In addition, GCM components provide non-functional features allowing developers to benefits extra capabilities such as monitoring, autonomic management...

The GCM provides software architects with a comprehensive framework to express at design time the parallelism and the distribution of an application. Thus, the architecture of the system itself captures the parallel/distributed aspects, acting as a powerful specification and documentation. Furthermore, developers do not need to spend extensive time to learn distribute programming or implement collective communication, but rather concentrate on the business code and leverage the GCM framework.

ProActive uses the "MOP" to provide flexibility and configurability: it allows the addition of meta-objects for managing new required features. Moreover, the library also proposes a deployment framework, which allows the deployment of active objects on various infrastructures. The MOP architecture is represented in the next figure.

Figure 1.2. MOP architecture

The library may be represented in four layers: IDE, programming model, deployment facilities, and non-functional features.

ProActive offers IC2D, an interactive graphical enviroment for remote monitoring and control of distributed applications. Due to the fact that it uses the standard (starting with the J2SE platform 5.0) Java Management Extensions (JMX) technology in order to monitor distribution, IC2D can be deployed without requiring changes to the JVM or Java.

The programming model consists of the active objects model which offers asynchronous communication, typed group communication, and the object-oriented SPMD programming paradigm. In addition to the standard object oriented programming paradigm, ProActive also proposes a component-based programming paradigm, by providing an implementation of the Fractal component model geared at Grid computing.

The deployment layer includes a deployment framework (detailed in "ProActiveDeployment") and it allows the creation of remote active objects on various infrastructures. A scheduler is also proposed to manage the deployment of jobs. The load-balancing framework uses the migration capabilities to optimize the placement of the distributed active objects.

Non-functional features include a transparent fault-tolerance mechanism based on a communication-induced check-pointing protocol, a security framework for communications between remote active objects, migration capabilities for the mobility of active objects, a mechanism for the management of exceptions, a complete distributed garbage collector for active objects, and a mechanism for wrapping legacy code, notably as a way to control and interact with MPI applications.

In the communication layer, several protocols are provided for the communication between active objects: Java RMI as the default protocol, SSH, HTTP, tunneled RMI. It is also possible to export active objects as Web Services, which can then be accessed using the standard SOAP protocol. A file transfer mechanism is also implemented. It allows the transfer of files between active objects, for instance to send large data input files or to retrieve results files.

The next sections details these different steps to install ProActive.

You can download the archive file (a zip or a tar.gz file) containing ProActive from the download section of the ProActive home page. You will be asked to accept the license agreement and provide a few details including your email address. You will be then redirected towards the download page.

Unzip this archive using your favorite program for uncompressing, such as 7Zip under Windows or the unzip and tar command-line utilities under most Linux/Unix systems. Uncompressing archive creates a ProActive directory and all the files contained in this archive go into this directory and its subdirectories. As already said, ProActive should be compiled. If it is not the case, you have to compile it using the build[.bat] deploy.all in the compile directory.

Here is a quick overview of the directory structure of the archive:

In order to use ProActive to write your own application, you need to place the dist/lib/ProActive.jar file in your CLASSPATH. As said previously, it makes references to other JAR files which have to be in the same directory as ProActive.jar. If it not the case, you have to include the following JAR files to your CLASSPATH:

	Note
	You do not need to modify your CLASSPATH permanently if you include the entries above using a Java IDE or a shell script.

Under Linux/Unix:

Open a Linux/Unix terminal and go to your ProActive directory using the cd command. Then, set the CLASSPATH as follows:

cd [Your ProActive install directory]
PA_PATH=$(pwd)
export CLASSPATH=.:${PA_PATH}/dist/lib/ProActive_examples.jar:${PA_PATH}/dist/lib/ProActive.jar

Under Windows:

Open a Windows command terminal by executing either cmd or command in Start-> Run

Go to the ProActive installation folder by using cd and set the CLASSPATH by executing the following commands:

cd [Your ProActive install directory]
set CLASSPATH=.;%CD%\dist\lib\ProActive_examples.jar;%CD%\dist\lib\ProActive.jar;

	Note
	Keep in mind that some JAR files are needed by ProActive.jar and so, they have to be in the same directory. If not, add them to the CLASSPATH.

In addition to the jar files above, you may want to add the following jar files. None of them are used directly by the core functionalities of ProActive but only in part of the library. They are not needed at runtime if those specific functionalities are not used but they are needed to compile all the code.

The jar files can be found under Proactive/lib and under ProaActive/dist/lib.

See Permissions in the JavaTM 2 SDK to learn more about Java permissions. The option -Djava.security.policy=pathToFile will specify which security policy file to use within ProActive. As a first approximation, you can create a simple security policy file granting all permissions:

grant {
  permission java.security.AllPermission;

  // Reflect access private Members
  // Used by:
  //         -Unicore Process
  permission java.lang.RuntimePermission "accessDeclaredMembers";
  permission java.lang.reflect.ReflectPermission "suppressAccessChecks";
};

	Note
	If you use the scripts provided with the distribution to run examples, an existing policy file named proactive.java.policy will be used by default. It is exactly the same as described above.

See for logging using the Apache log4j. The option -Dlog4j.configuration=file:pathToFile will specify which logging policy file to use within ProActive. As a first approximation, you can create a simple logging policy file:

# This file represents the log4j config file for non regression tests.
# It is given as property (in the proactive.xml) when starting nonregression tests

# The default logging level is INFO
# The root logger logs in the test.log file

log4j.rootLogger=INFO,CONSOLE
 
# If INFO is enabled Jetty is too verbose at startup
log4j.logger.org.mortbay = WARN


########### The following are existing categories in ProActive
#log4j.logger.proactive = INFO
#log4j.logger.proactive.pnp = TRACE
#log4j.logger.proactive.classloader = DEBUG
#log4j.logger.proactive.events = DEBUG
#log4j.logger.proactive.runtime = DEBUG
#log4j.logger.proactive.body = DEBUG
#log4j.logger.proactive.mop = DEBUG
#log4j.logger.proactive.groups = DEBUG
#log4j.logger.proactive.sync_call = DEBUG
#log4j.logger.proactive.deployment = DEBUG
#log4j.logger.proactive.deployment.log = DEBUG
#log4j.logger.proactive.deployment.process = DEBUG
#log4j.logger.proactive.deployment.filetransfer = DEBUG,CONSOLE
#log4j.logger.proactive.filetransfer = DEBUG
#log4j.logger.proactive.nfe = FATAL
#log4j.logger.proactive.gc = DEBUG
#log4j.logger.proactive.ft = DEBUG
#log4j.logger.proactive.ft.cic = DEBUG
#log4j.logger.proactive.ft.pml = DEBUG
#log4j.logger.proactive.communication.transport.http = DEBUG
#log4j.logger.proactive.communication.rmi = DEBUG
#log4j.logger.proactive.communication.ssh = DEBUG
#log4j.logger.proactive.communication.transport.http = DEBUG
#log4j.logger.proactive.migration = DEBUG
#log4j.logger.proactive.communication.requests = DEBUG
#log4j.logger.proactive.examples = DEBUG

#log4j.logger.proactive.components = DEBUG
#log4j.logger.proactive.components.requests = DEBUG
#log4j.logger.proactive.components.activity = DEBUG
#log4j.logger.proactive.components.bytecodegeneration = DEBUG
#log4j.logger.proactive.components.adl = DEBUG
#log4j.logger.proactive.components.gui = DEBUG

#log4j.logger.proactive.security = DEBUG
#log4j.logger.proactive.security.node = DEBUG
#log4j.logger.proactive.security.session = DEBUG
#log4j.logger.proactive.security.body = DEBUG
#log4j.logger.proactive.security.manager = DEBUG
#log4j.logger.proactive.security.request = DEBUG
#log4j.logger.proactive.security.runtime = DEBUG
#log4j.logger.proactive.security.policy = DEBUG
#log4j.logger.proactive.security.policyserver = DEBUG
#log4j.logger.proactive.security.crypto = DEBUG
#log4j.logger.proactive.security.psm = DEBUG

#log4j.logger.proactive.skeletons = DEBUG
#log4j.logger.proactive.skeletons.taskpool = DEBUG
#log4j.logger.proactive.skeletons.structure = DEBUG
#log4j.logger.proactive.skeletons.environment = DEBUG
#log4j.logger.proactive.skeletons.application = DEBUG
#log4j.logger.proactive.skeletons.diagnosis     = DEBUG
#log4j.logger.proactive.skeletons.system = DEBUG

#log4j.logger.proactive.masterworker = DEBUG
#log4j.logger.proactive.masterworker.workermanager = DEBUG
#log4j.logger.proactive.masterworker.pinger = DEBUG
#log4j.logger.proactive.masterworker.repository = DEBUG
#log4j.logger.proactive.masterworker.workers = DEBUG


#log4j.logger.proactive.configuration = DEBUG
#log4j.logger.proactive.remoteobject = DEBUG

#log4j.logger.proactive.jmx = DEBUG
#log4j.logger.proactive.jmx.mbean = DEBUG
#log4j.logger.proactive.jmx.notification = DEBUG

#log4j.logger.proactive.webservices = DEBUG

################ Appenders ####################
#
# Appender output can be configured by using a pattern layout
# See: http://logging.apache.org/log4j/1.2/apidocs/org/apache/log4j/PatternLayout.html
#
#   - %c the category of the logging event
#   - %d the date
#   - %m he application supplied message
#   - %n the platform dependent line separator character or characters
#   - %p the priority of the logging event
#   - %t the name of the thread that generated the logging event
#   - %X{hostname}         the hostname 
#   - %X{id@hostname}      the VMID and the hostname 
#   - %X{shortid@hostname} the short VMID and the hostname (a collision can occur between two shortids, you should use id@hostname) 
#   - %X{runtime}          the ProActive runtime url (does not work very well since a different MDC is associated to each thread)


# CONSOLE appender is used by default
log4j.appender.CONSOLE=org.apache.log4j.ConsoleAppender
log4j.appender.CONSOLE.layout=org.apache.log4j.PatternLayout
log4j.appender.CONSOLE.layout.ConversionPattern=%X{shortid@hostname} - [%d %p %20.20c{2}] %m%n

# Appender FILE writes to the file "tests.log".
# This file is recreated a file for each run
log4j.appender.FILE=org.apache.log4j.FileAppender
log4j.appender.FILE.File=tests.log
log4j.appender.FILE.Append=false
log4j.appender.FILE.layout=org.apache.log4j.PatternLayout
log4j.appender.FILE.layout.ConversionPattern=%5p [%t]: %m%n

	Note
	If you use the scripts provided with the distribution to run the examples, an existing log4j file named proactive-log4j will be used by default.

You can try to run the test applications provided with ProActive. Each example comes with a script to launch the application. This script is located in ProActive/examples . All the example code sources can be found in the directory ProActive/src/Examples/org/objectweb/proactive/examples/.

The C3D application uses three scripts that can be found under the ProActive/examples/c3d directory:

To use the application first start the dispatcher using either c3d_one_user.[sh, bat] or c3d_no_user.[sh, bat] and then add users with c3d_add_user.[sh, bat].

To start application, use examples/readers.

ProActive starts a node (on an already existing JVM or on a new one) on the current machine and creates 3 Writers, 3 Readers, a ReaderWrite (the application controller) and a ReaderDisplay, which are all active objects.

The examples has three synchronization modes "Priority to Writers", "Priority to Readers", and "Even Policy". The application is able to assign different priority to readers or writers without blocking on reading or writing.

Figure 5.2. Started GUI illustrating the activities of Reader and Writer objects.

To start the application use the examples/philosophers/philosophers.sh or examples\philosophers\philosophers.bat script depending on your operating system.

ProActive creates a new node and instantiates the active objects of the application: DinnerLayout, Table, and Philosopher.

Figure 5.3. The Dining Philosophers Example

The pictures represent the state of the philosophers. They can be:

The forks can have two states:

You can either run the application in autopilot mode running it on its own without encountering a deadlock or in manual mode where you click on the philosophers' heads to switch their modes. With the second mode, you can, for instance, make one of the philosophers starve, making his two closer neighbours alternatively eat and think.

Use the examples/penguin/penguin.sh or examples/penguin/penguin.bat script to start the penguin controller. The controller allows you to add a new agent (penguin), and control its route.

The active object is moving between machines (specified in the deployment descriptor) and the penguin window disappears and reappears on the screen associated with the new JVM.

After selecting them, use buttons to:

To start a chat client, run chat.sh or chat.bat from the examples/chat/ folder. The syntax is chat.[sh|bat] user_name host_to_connect_to user_to_connect_to .

When you start the first client the last argument can have any value. However if you want to connect to another chat you have to supply a valid user name and host. The communication is not dependent on any particular user. The first user that created the chat can leave and the other users can still communicate.

Start the first user by running chat.[sh|bat] toto localhost no_user. The following window appears:

To connect the user "tutu", run chat.[sh|bat] tutu localhost toto.

You can later connect other users in the same manner. Run chat.[sh|bat] tata localhost tutu. You can connect to any user in the chat and you will have access to all the users.

The chat application can be migrated to another computer. All we have to do in order to fully migrate one chat is to start a node on the remote computer using startNode.sh or startNode.bat from the ProActive/bin/ folder. We have to use a parameter in the form of rmi://hostname:port/node_name to start the node. Port number is not mandatory. If you do not precise it, the node will be started on the default port, that is 1099. After starting the node, we can migrate the chat application by putting the URL //hostname:port/node_name in the "migrate to" textbox and clicking on the Migrate button. For the migration, port is number is mandatory. The application will be recreated on the remote computer with its state intact and ready to communicate to the other users.

In this chapter, we are going to see a simple example of an MPI written program ported to ProActive.

First let's introduce what we are going to compute.

The MPI PI program approximates pi by computing:

Which is approximated by:

The only input data required is therefore N, the number of iterations.

Involved files :

In examples/integralPi, run integralpi.[sh|bat]. You can specify the number of workers from the command line. Feel free to edit scripts to specify another deployment descriptor.

bash-3.00$ ./integralpi.sh

--- IntegralPi --------------------------------------------------
The number of workers is 4
 --> This ClassFileServer is reading resources from classpath 2011
Created a new registry on port 1099
ProActive Security Policy (proactive.runtime.security) not set. Runtime Security disabled
************* Reading deployment descriptor: file:./../../descriptors/Matrix.xml ********************
created VirtualNode name=matrixNode
**** Starting jvm on amda.inria.fr
**** Starting jvm on amda.inria.fr
**** Starting jvm on amda.inria.fr
ProActive Security Policy (proactive.runtime.security) not set. Runtime Security disabled
 --> This ClassFileServer is reading resources from classpath 2012
ProActive Security Policy (proactive.runtime.security) not set. Runtime Security disabled
ProActive Security Policy (proactive.runtime.security) not set. Runtime Security disabled
 --> This ClassFileServer is reading resources from classpath 2013
 --> This ClassFileServer is reading resources from classpath 2014
**** Starting jvm on amda.inria.fr
Detected an existing RMI Registry on port 1099
Detected an existing RMI Registry on port 1099
Detected an existing RMI Registry on port 1099
ProActive Security Policy (proactive.runtime.security) not set. Runtime Security disabled
 --> This ClassFileServer is reading resources from classpath 2015
//amda.inria.fr/matrixNode2048238867 successfully bound in registry at //amda.inria.fr/matrixNode2048238867
**** Mapping VirtualNode matrixNode with Node: //amda.inria.fr/matrixNode2048238867 done
//amda.inria.fr/matrixNode690267632 successfully bound in registry at //amda.inria.fr/matrixNode690267632
**** Mapping VirtualNode matrixNode with Node: //amda.inria.fr/matrixNode690267632 done
//amda.inria.fr/matrixNode1157915128 successfully bound in registry at //amda.inria.fr/matrixNode1157915128
**** Mapping VirtualNode matrixNode with Node: //amda.inria.fr/matrixNode1157915128 done
Detected an existing RMI Registry on port 1099
//amda.inria.fr/matrixNode-814241328 successfully bound in registry at //amda.inria.fr/matrixNode-814241328
**** Mapping VirtualNode matrixNode with Node: //amda.inria.fr/matrixNode-814241328 done
4 nodes found
Generating class : pa.stub.org.objectweb.proactive.examples.integralpi.Stub_Worker

Enter the number of iterations (0 to exit) : 100000
Generating class : pa.stub.org.objectweb.proactive.examples.integralpi.Stub_Worker
Generating class : pa.stub.org.objectweb.proactive.examples.integralpi.Stub_Worker
Generating class : pa.stub.org.objectweb.proactive.examples.integralpi.Stub_Worker
Generating class : pa.stub.org.objectweb.proactive.examples.integralpi.Stub_Worker

         Worker 2 Calculated x = 0.7853956634245252 in 43 ms


         Worker 3 Calculated x = 0.7853906633745299 in 30 ms


         Worker 1 Calculated x = 0.7854006634245316 in 99 ms


         Worker 0 Calculated x = 3.141592653598117 in 12 ms


Calculated PI is 3.141592653598117 error is 8.324008149429574E-12

Enter the number of iterations (0 to exit) :  

In the folder ProActive/examples/, run:

nbody.[bat|sh] [-nodisplay | -displayft | -3d | -3dft] totalNbBodies maxIter

Right after starting the application, users have to choose one algorithm for computing amongst the following ones:

Mouse controls with the 3D GUI:

This way to construct the nbody simulation is based on a very different algorithm. This is described here to show how one can express this algorithm in ProActive, but having a different approach to solve the problem. Here's how it works:

To avoid broadcasting the new position of every particle to every active object, a tree implementation can simplify the problem by agglomerating sets of particles as a single particle, with a mass equal to the sum of masses of all the particles. This is the core of the Barnes-Hut algorithm. This method allows us to have a complexity brought down to O(N log N).

In our parallel implementation, we have defined an Active Object called Domain , which represents a volume in space, and which contains Planets . It is either subdivided into smaller Domains , or is a leaf of the total tree, and then only contains Planets . A Planet is still an Object with mass, velocity and position, but is no longer on a one-to-one connection with a Domain . We have cut down communications to the biggest Domains possible: when a Planet is distant enough, its interactions are not computed, but it is grouped with its local neighbours to a bigger particle. Here is an example of the Domains which would be known by the Domain drawn in red:

The Domain in the lower right hand-corner, drawn in blue, is also divided into sub-Domains, but that does not need to be known by the Domain in red: it assumes all the particles in the blue Domain are only one big one, centered at the center of mass of all the particles within the blue.

In this version, Domains communicate with a reduced set of other Domains , spanning on volumes of different sizes. Synchronization is achieved by sending explicitly iteration numbers, and returning when needed older positions. You may notice that some Domains seem desynchronized with other ones, having several iterations in-between. That is not a problem because if they need to be synchronized and send each other information, a mechanism saving the older positions permits to send them when needed.

This is a snapshot of the ProActive n-body example running on 3 hosts with 8 bodies:

And here is a n-body screenshot, with the application GUI and Java3D installed:

For our Monitoring agent, we use three classes. The first two classes are regular Java objects. The first class is State which we use to get some information on the JVM the object is located on. This object will be use as return value for the getCurrentState() method in the CMAgent class.

package org.objectweb.proactive.examples.userguide.cmagent.simple;

import java.io.Serializable;
import java.lang.management.ManagementFactory;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.Date;


//TODO 4. remove Serializable and run the agent
public class State implements Serializable {
    private long commitedMemory = ManagementFactory.getMemoryMXBean().getHeapMemoryUsage().getCommitted();
    private long initMemory = ManagementFactory.getMemoryMXBean().getHeapMemoryUsage().getInit();
    private long maxMemory = ManagementFactory.getMemoryMXBean().getHeapMemoryUsage().getMax();
    private long usedMemory = ManagementFactory.getMemoryMXBean().getHeapMemoryUsage().getUsed();
    private String osArch = ManagementFactory.getOperatingSystemMXBean().getArch();
    private String osName = ManagementFactory.getOperatingSystemMXBean().getName();
    private String osVersion = ManagementFactory.getOperatingSystemMXBean().getVersion();
    private int osProcs = ManagementFactory.getOperatingSystemMXBean().getAvailableProcessors();
    private int liveThreads = ManagementFactory.getThreadMXBean().getThreadCount();
    private long startedThreads = ManagementFactory.getThreadMXBean().getTotalStartedThreadCount();
    private int peakThreads = ManagementFactory.getThreadMXBean().getPeakThreadCount();
    private int deamonThreads = ManagementFactory.getThreadMXBean().getDaemonThreadCount();
    private Date timePoint = new Date();
    private String hostname;
    {
        try {
            hostname = InetAddress.getLocalHost().toString();
        } catch (UnknownHostException e) {
            e.printStackTrace();
        }
    }

    public State() {
    }

    public String toString() {

        return new String("\n======= [" + "State at " + timePoint + " on " + hostname + "] =======" +
            "\nCommited memory: " + commitedMemory + " bytes\nInitial memory requested: " + initMemory +
            " bytes\nMaximum memory available: " + maxMemory + " bytes\nUsed memory: " + usedMemory +
            " bytes\nOperating System: " + osName + " " + osVersion + " " + osArch + "\nProcessors: " +
            osProcs + "\nCurrent live threads: " + liveThreads + "\nTotal started threads: " +
            startedThreads + "\nPeak number of live threads: " + peakThreads + "\nCurrent daemon threads: " +
            deamonThreads +
            "\n===============================================================================\n");

    }
}

The Monitoring agent class is a regular Java class with only one method:

package org.objectweb.proactive.examples.userguide.cmagent.simple;

import org.objectweb.proactive.extensions.annotation.ActiveObject;


@ActiveObject
public class CMAgent {
    // empty constructor is required by Proactive
    public CMAgent() {
    }

    public State getCurrentState() {
        return new State();
    }

}

For this simple exercise, we only need to add ProActive code in the Main class where we instantiate the active object.

package org.objectweb.proactive.examples.userguide.cmagent.simple;

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


public class Main {
    public static void main(String args[]) {
        try {
            String currentState = new String();
            //TODO 1. Create the active object
            //TODO 2. Get the current state
            //TODO 3. Print the state
            //TODO 4. Stop the active object and
            //        terminate the application
        } catch (NodeException nodeExcep) {
            System.err.println(nodeExcep.getMessage());
        } catch (ActiveObjectCreationException aoExcep) {
            System.err.println(aoExcep.getMessage());
        }
    }
}

You can find the skeleton codes for this exercise into the tutorials/src/Examples/org/objectweb/proactive/examples/userguide/cmagent/simple/ directory. Full code is also available into the directory ProActive/src/Examples/org/objectweb/proactive/examples/userguide/cmagent/simple/.

Once filled in, go to the tutorials/compile directory and type build[.bat] cmagent.simple to compile your code. Then, after a successful compilation, go to the scripts/CMAgent directory and launch the simpleCMA.[sh|bat] script to execute your code.

The full code of the Main class is the following:

package org.objectweb.proactive.examples.userguide.cmagent.simple;

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


public class Main {
    public static void main(String args[]) {
        try {
            String currentState = new String();
            //TODO 1. Create the active object
            CMAgent ao = (CMAgent) PAActiveObject.newActive(CMAgent.class.getName(), null);
            //TODO 2. Get the current state
            currentState = ao.getCurrentState().toString();
            //TODO 3. Print the state
            System.out.println(currentState);
            //TODO 4. Stop the active object and
            //        terminate the application
            PAActiveObject.terminateActiveObject(ao, true);
            PALifeCycle.exitSuccess();
        } catch (NodeException nodeExcep) {
            System.err.println(nodeExcep.getMessage());
        } catch (ActiveObjectCreationException aoExcep) {
            System.err.println(aoExcep.getMessage());
        }
    }
}

The CMAgentInitialized class extends the CMAgent class from the previous example, and implements the interfaces InitActive and EndActive. It acts as a server for the Main class.

To implement the application we will create a class that inherits from the CMAgent class and implements the InitActive, RunActive, and EndActive interfaces.

package org.objectweb.proactive.examples.userguide.cmagent.initialized;

import org.objectweb.proactive.Body;
import org.objectweb.proactive.EndActive;
import org.objectweb.proactive.InitActive;
import org.objectweb.proactive.RunActive;
import org.objectweb.proactive.Service;
import org.objectweb.proactive.core.util.wrapper.LongWrapper;
import org.objectweb.proactive.examples.userguide.cmagent.simple.CMAgent;
import org.objectweb.proactive.extensions.annotation.ActiveObject;


@ActiveObject
public class CMAgentInitialized extends CMAgent implements InitActive, RunActive, EndActive {
    private long lastRequestDuration;
    private long startTime;
    private long requestsServed = 0;

    public void initActivity(Body body) {
        //TODO 1. Print start information

        //TODO 2. Record start time
    }

    public void runActivity(Body body) {
        Service service = new Service(body);
        long currentRequestDuration = 0;
        while (body.isActive()) {
            //TODO 3. wait for a request

            //TODO 4. Record time

            //TODO 5. Serve request

            //TODO 6. Calculate request duration

            //TODO 7. Increment the number of requests served
        }
    }

    public void endActivity(Body body) {
        //TODO 8. Calculate the running time of the active object using the start time recorded in initActivity()

        //TODO 9. Print various stop information
    }

    public LongWrapper getLastRequestServeTime() {
        //TODO 10. Use wrappers for primitive types so the calls are asynchronous
    }
}

By default an active object has a never-ending thread that should be stopped when the object is not needed anymore. The method terminate() serves the purpose of destroying the object. However, if an explicit call to terminate the object is not made, ProActive has its own distributed garbage collection system that is able to decide when an active object can be destroyed.

The Main is similar to the one in the previous example. We will change the object created from CMAAgent to CMAAgentInitialized and also we will call the terminate() method to destroy the object.

You can find the skeleton codes for this exercise into the tutorials/src/Examples/org/objectweb/proactive/examples/userguide/cmagent/initialized/ directory. Full code is also available into the directory ProActive/src/Examples/org/objectweb/proactive/examples/userguide/cmagent/initialized/.

Once filled in, go to the tutorials/compile directory and type build[.bat] cmagent.initialized to compile your code. Then, after a successful compilation, go to the scripts/CMAgent directory and launch the initializedCMA.[sh|bat] script to execute your code.

We only need to extend the CMAgent class and implement the interfaces. In initActivity(Body body), we use body.getName() to get the name of the active object and body.getNodeUrl() to get the location. In endActivity(Body body), we also use body.getNodeUrl() to get the location.

public void initActivity(Body body) {
    //TODO 1. Print start information
    System.out.println("### Started Active object " + body.getMBean().getName() + " on " +
        body.getMBean().getNodeUrl());

    //TODO 2. Record start time
    startTime = System.currentTimeMillis();
}

public void runActivity(Body body) {
    Service service = new Service(body);
    long currentRequestDuration = 0;
    while (body.isActive()) {
        //TODO 3. wait for a request
        service.waitForRequest(); // block until a request is received

        //TODO 4. Record time
        currentRequestDuration = System.currentTimeMillis();

        //TODO 5. Serve request
        service.serveOldest(); //server the requests in a FIFO manner

        //TODO 6. Calculate request duration
        currentRequestDuration = System.currentTimeMillis() - currentRequestDuration;

        // an intermediary variable is used so
        // when calling getLastRequestServeTime()
        // we get the first value before the last request
        // i.e when calling getLastRequestServeTime
        // the lastRequestDuration is update with the
        // value of the getLastRequestServeTime call
        // AFTER the previous calculated value has been returned
        lastRequestDuration = currentRequestDuration;

        //TODO 7. Increment the number of requests served
        requestsServed++;
    }
}

public void endActivity(Body body) {
    //TODO 8. Calculate the running time of the active object using the start time recorded in initActivity()
    long runningTime = System.currentTimeMillis() - startTime;

    //TODO 9. Print various stop information
    System.out.println("### You have killed the active object. The final" + " resting place is on " +
        body.getNodeURL() + "\n### It has faithfully served " + requestsServed + " requests " +
        "and has been an upstanding active object for " + runningTime + " ms ");
}

public LongWrapper getLastRequestServeTime() {
    //TODO 10. Use wrappers for primitive types so the calls are asynchronous
    return new LongWrapper(lastRequestDuration);
}

We will change the Main class to declare and load the deployment descriptors to be used. For this, we will use a deploy() method that returns a Virtual Node which has several nodes (as specified in the deployment file) that we can deploy on. First, the method creates an object representation of the deployment file, then activates all the nodes, and then returns the first available node. We also have to change the deployment descriptor files to fit our local settings.

package org.objectweb.proactive.examples.userguide.cmagent.deployed;

import java.io.File;

import org.objectweb.proactive.ActiveObjectCreationException;
import org.objectweb.proactive.api.PAActiveObject;
import org.objectweb.proactive.api.PALifeCycle;
import org.objectweb.proactive.core.ProActiveException;
import org.objectweb.proactive.core.node.NodeException;
import org.objectweb.proactive.examples.userguide.cmagent.initialized.CMAgentInitialized;
import org.objectweb.proactive.extensions.gcmdeployment.PAGCMDeployment;
import org.objectweb.proactive.gcmdeployment.GCMApplication;
import org.objectweb.proactive.gcmdeployment.GCMVirtualNode;


public class Main {
    private static GCMApplication pad;

    //deployment method
    private static GCMVirtualNode deploy(String descriptor) throws NodeException, ProActiveException {
        //TODO 1. Create object representation of the deployment file
        //TODO 2. Activate all Virtual Nodes
        //TODO 3. Wait for all the virtual nodes to become ready
        //TODO 4. Get the first Virtual Node specified in the descriptor file
        //TODO 5. Return the virtual node
    }


    public static void main(String args[]) {
        try {
            //TODO 6. Get the virtual node through the deploy method
            //TODO 7. Create the active object using a node on the virtual node
            //TODO 8. Get the current state from the active object
            //TODO 9. Print the state
            //TODO 10. Stop the active object

        } catch (NodeException nodeExcep) {
            System.err.println(nodeExcep.getMessage());
        } catch (ActiveObjectCreationException aoExcep) {
            System.err.println(aoExcep.getMessage());
        } catch (ProActiveException poExcep) {
            System.err.println(poExcep.getMessage());
        } finally {
            //TODO 11. Stop the virtual node
        }
    }
}

You can find the skeleton codes for this exercise into the tutorials/src/Examples/org/objectweb/proactive/examples/userguide/cmagent/deployed/ directory. Full code is also available into the directory ProActive/src/Examples/org/objectweb/proactive/examples/userguide/cmagent/deployed/.

Once filled in, go to the tutorials/compile directory and type build[.bat] cmagent.deployed to compile your code. Then, after a successful compilation, go to the scripts/CMAgent directory and launch the deployedCMA.[sh|bat] script to execute your code.

In the deploy() method, we return the first virtual node found in the deployment descriptor:

//deployment method
private static GCMVirtualNode deploy(String descriptor) throws NodeException, ProActiveException {
    //TODO 1. Create object representation of the deployment file
    pad = PAGCMDeployment.loadApplicationDescriptor(new File(descriptor));
    //TODO 2. Activate all Virtual Nodes
    pad.startDeployment();
    //TODO 3. Wait for all the virtual nodes to become ready
    pad.waitReady();
    //TODO 4. Get the first Virtual Node specified in the descriptor file
    GCMVirtualNode vn = pad.getVirtualNodes().values().iterator().next();
    //TODO 5. Return the virtual node
    return vn;
}

The active object is created by using the first node on the virtual node:

//TODO 7. Create the active object using a node on the virtual node
CMAgentInitialized ao = (CMAgentInitialized) PAActiveObject.newActive(CMAgentInitialized.class
        .getName(), new Object[] {}, vn.getANode());

The full Main class:

package org.objectweb.proactive.examples.userguide.cmagent.deployed;

import java.io.File;

import org.objectweb.proactive.ActiveObjectCreationException;
import org.objectweb.proactive.api.PAActiveObject;
import org.objectweb.proactive.api.PALifeCycle;
import org.objectweb.proactive.core.ProActiveException;
import org.objectweb.proactive.core.node.NodeException;
import org.objectweb.proactive.examples.userguide.cmagent.initialized.CMAgentInitialized;
import org.objectweb.proactive.extensions.gcmdeployment.PAGCMDeployment;
import org.objectweb.proactive.gcmdeployment.GCMApplication;
import org.objectweb.proactive.gcmdeployment.GCMVirtualNode;


public class Main {
    private static GCMApplication pad;

    //deployment method
    private static GCMVirtualNode deploy(String descriptor) throws NodeException, ProActiveException {
        //TODO 1. Create object representation of the deployment file
        pad = PAGCMDeployment.loadApplicationDescriptor(new File(descriptor));
        //TODO 2. Activate all Virtual Nodes
        pad.startDeployment();
        //TODO 3. Wait for all the virtual nodes to become ready
        pad.waitReady();
        //TODO 4. Get the first Virtual Node specified in the descriptor file
        GCMVirtualNode vn = pad.getVirtualNodes().values().iterator().next();
        //TODO 5. Return the virtual node
        return vn;
    }


    public static void main(String args[]) {
        try {
            //TODO 6. Get the virtual node through the deploy method
            GCMVirtualNode vn = deploy(args[0]);
            //TODO 7. Create the active object using a node on the virtual node
            CMAgentInitialized ao = (CMAgentInitialized) PAActiveObject.newActive(CMAgentInitialized.class
                    .getName(), new Object[] {}, vn.getANode());
            //TODO 8. Get the current state from the active object
            String currentState = ao.getCurrentState().toString();
            //TODO 9. Print the state
            System.out.println(currentState);
            //TODO 10. Stop the active object
            PAActiveObject.terminateActiveObject(ao, false);

        } catch (NodeException nodeExcep) {
            System.err.println(nodeExcep.getMessage());
        } catch (ActiveObjectCreationException aoExcep) {
            System.err.println(aoExcep.getMessage());
        } catch (ProActiveException poExcep) {
            System.err.println(poExcep.getMessage());
        } finally {
            //TODO 11. Stop the virtual node
            if (pad != null)
                pad.kill();
            PALifeCycle.exitSuccess();
        }
    }
}

In order to answer to our last question, we follow the given hint. To do that, the simplest way is to remove the local host from the host list in the nodeProvider tag. In our case, the local host is kisscool. Try to do same thing and launch the application with the two different variable initializations.

	Note
	the remote machine needs to have the rsh service running since we are using rsh deployment.

If you do not change the State class, you will get a display looking like the following one:

         ======= [State at Wed May 27 14:45:05 CEST 2009 on paquito.inria.fr/138.96.218.106] =======
         Commited memory: 63111168 bytes
         Initial memory requested: 63961152 bytes
         Maximum memory available: 985530368 bytes
         Used memory: 7939936 bytes
         Operating System: Linux 2.6.23.1-21.fc7 i386
         Processors: 2
         Current live threads: 27
         Total started threads: 27
         Peak number of live threads: 27
         Current daemon threads: 21
         ===============================================================================
      

Whereas with the variable initializations done in the toString() method, you will get something like this:

         ======= [State at Wed May 27 14:46:14 CEST 2009 on kisscool/138.96.218.126] =======
         Commited memory: 161021952 bytes
         Initial memory requested: 130110784 bytes
         Maximum memory available: 1034027008 bytes
         Used memory: 45900640 bytes
         Operating System: Linux 2.6.27.9-159.fc10.x86_64 i386
         Processors: 4
         Current live threads: 41
         Total started threads: 41
         Peak number of live threads: 41
         Current daemon threads: 37
         ===============================================================================
      

You can therefore realise that information returned in the second display are information about the local host but not about the host where the active object was running on. This is totally normal since when you call the toString() method, you will implicitly get the active object from the remote host and call this method next. So, if variable initializations are done in the toString() method, returned information will be information on the local host, whereas if variable initializations are done at the active object creation, returned information will be information on the host where the active object has been created that is to say, the remote host.

There are no new classes used in this exercise. Instead we will show how to use thePAActiveObject.getStubOnThis() method.

There are no new concepts present in the Main class. We just create the active objects and make the method calls on one of the active objects.

public static void main(String args[]) {
    try {
        GCMVirtualNode vn = deploy(args[0]);
        Vector<CMAgentChained> agents = new Vector<CMAgentChained>();
        //create the active objects
        //create a collection of active objects
        for (Node node : vn.getCurrentNodes()) {
            CMAgentChained ao = (CMAgentChained) PAActiveObject.newActive(CMAgentChained.class.getName(),
                    null, node);
            agents.add(ao);

            //connect to the neighbour
            int size = agents.size();
            if (size > 1) {
                CMAgentChained lastAgent = agents.get(size - 1);
                CMAgentChained previousAgent = agents.get(size - 2);
                lastAgent.setPreviousNeighbour(previousAgent);
            }
        }
        //start chained call                
        Vector<State> states = agents.get(agents.size() / 2).getAllPreviousStates();
        for (State s : states) {
            System.out.println(s.toString());
        }

        states = agents.get(agents.size() / 2).getAllNextStates();
        for (State s : states) {
            System.out.println(s.toString());
        }
    } catch (ActiveObjectCreationException e) {
        System.err.println(e.getMessage());
    } catch (NodeException nodeExcep) {
        System.err.println(nodeExcep.getMessage());
    } catch (ProActiveException e) {
        System.err.println(e.getMessage());
    } finally {
        //stopping all the objects and JVMS
        if (pad != null)
            pad.kill();
        PALifeCycle.exitSuccess();
    }
}

The agent class inherits from CMAgentInitialized:

package org.objectweb.proactive.examples.userguide.cmagent.synch;

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

import org.objectweb.proactive.api.PAActiveObject;
import org.objectweb.proactive.examples.userguide.cmagent.initialized.CMAgentInitialized;
import org.objectweb.proactive.examples.userguide.cmagent.simple.State;
import org.objectweb.proactive.extensions.annotation.ActiveObject;


@ActiveObject
public class CMAgentChained extends CMAgentInitialized implements Serializable {
    private CMAgentChained previousNeighbour;
    private CMAgentChained nextNeighbour;

    public void setPreviousNeighbour(CMAgentChained neighbour) {
        this.previousNeighbour = neighbour;
        //TODO 1. Pass a remote reference of this object to the neighbour
        // Hint: This object is "nextNeighbour" for previous neighbour if not null
    }

    public void setNextNeighbour(CMAgentChained neighbour) {
        this.nextNeighbour = neighbour;
        //TODO 2. Pass a remote reference of this object to the neighbour
        // Hint: This object is "previousNeighbour" for next neighbour if not null
    }


    public CMAgentChained getPreviousNeigbour() {
        return previousNeighbour;
    }

    public CMAgentChained getNextNeigbour() {
        return nextNeighbour;
    }

    public Vector<State> getAllPreviousStates() {
        System.out.println(PAActiveObject.getStubOnThis());

        if (this.previousNeighbour != null) {
            System.out.println("Passing the call to the previous neighbour...");
            // wait-by-necessity
            Vector<State> states = this.previousNeighbour.getAllPreviousStates();
            // states is a future

            // TODO 3. Is this explicit synchronization mandatory ? (NO the wait was removed)

            return states;
        } else {
            System.out.println("No more previous neighbours..");
            Vector<State> states = new Vector<State>();
            states.add(this.getCurrentState());
            return states;
        }
    }

    public Vector<State> getAllNextStates() {
        System.out.println(PAActiveObject.getStubOnThis());
        if (this.nextNeighbour != null) {
            // wait-by-necessity
            System.out.println("Passing the call to the next neighbour..");
            Vector<State> states = this.nextNeighbour.getAllNextStates();
            // states is a future

            // TODO 4. Is this explicit synchronization mandatory ?    (NO the wait was removed)

            return states;
        } else {
            System.out.println("No more next neighbours");
            Vector<State> states = new Vector<State>();
            states.add(this.getCurrentState());
            return states;
        }
    }
}

You can find the skeleton codes for this exercise into the tutorials/src/Examples/org/objectweb/proactive/examples/userguide/cmagent/synch/ directory. Full code is also available into the directory ProActive/src/Examples/org/objectweb/proactive/examples/userguide/cmagent/synch/.

Once filled in, go to the tutorials/compile directory and type build[.bat] cmagent.synch to compile your code. Then, after a successful compilation, go to the scripts/CMAgent directory and launch the synchCMA.[sh|bat] script to execute your code.

To pass a remote reference to the active object we are currently in, we use PAActiveObject.getStubOnThis(). A simple use of the this keyword will return a reference to the passive object components of the active object and not a reference to the stub.

public void setPreviousNeighbour(CMAgentChained neighbour) {
    this.previousNeighbour = neighbour;
    //TODO 1. Pass a remote reference of this object to the neighbour
    // Hint: This object is "nextNeighbour" for previous neighbour if not null
    if (neighbour.getNextNeigbour() == null)
        neighbour.setNextNeighbour((CMAgentChained) PAActiveObject.getStubOnThis());
}

public void setNextNeighbour(CMAgentChained neighbour) {
    this.nextNeighbour = neighbour;
    //TODO 2. Pass a remote reference of this object to the neighbour
    // Hint: This object is "previousNeighbour" for next neighbour if not null
    if (neighbour.getPreviousNeigbour() == null)
        neighbour.setPreviousNeighbour((CMAgentChained) PAActiveObject.getStubOnThis());
}

The full code for the CMAAgentChained class is:

package org.objectweb.proactive.examples.userguide.cmagent.synch;

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

import org.objectweb.proactive.api.PAActiveObject;
import org.objectweb.proactive.examples.userguide.cmagent.initialized.CMAgentInitialized;
import org.objectweb.proactive.examples.userguide.cmagent.simple.State;
import org.objectweb.proactive.extensions.annotation.ActiveObject;


@ActiveObject
public class CMAgentChained extends CMAgentInitialized implements Serializable {
    private CMAgentChained previousNeighbour;
    private CMAgentChained nextNeighbour;

    public void setPreviousNeighbour(CMAgentChained neighbour) {
        this.previousNeighbour = neighbour;
        //TODO 1. Pass a remote reference of this object to the neighbour
        // Hint: This object is "nextNeighbour" for previous neighbour if not null
        if (neighbour.getNextNeigbour() == null)
            neighbour.setNextNeighbour((CMAgentChained) PAActiveObject.getStubOnThis());
    }

    public void setNextNeighbour(CMAgentChained neighbour) {
        this.nextNeighbour = neighbour;
        //TODO 2. Pass a remote reference of this object to the neighbour
        // Hint: This object is "previousNeighbour" for next neighbour if not null
        if (neighbour.getPreviousNeigbour() == null)
            neighbour.setPreviousNeighbour((CMAgentChained) PAActiveObject.getStubOnThis());
    }


    public CMAgentChained getPreviousNeigbour() {
        return previousNeighbour;
    }

    public CMAgentChained getNextNeigbour() {
        return nextNeighbour;
    }

    public Vector<State> getAllPreviousStates() {
        System.out.println(PAActiveObject.getStubOnThis());

        if (this.previousNeighbour != null) {
            System.out.println("Passing the call to the previous neighbour...");
            // wait-by-necessity
            Vector<State> states = this.previousNeighbour.getAllPreviousStates();
            // states is a future

            // TODO 3. Is this explicit synchronization mandatory ? (NO the wait was removed)
            states.add(this.getCurrentState());

            return states;
        } else {
            System.out.println("No more previous neighbours..");
            Vector<State> states = new Vector<State>();
            states.add(this.getCurrentState());
            return states;
        }
    }

    public Vector<State> getAllNextStates() {
        System.out.println(PAActiveObject.getStubOnThis());
        if (this.nextNeighbour != null) {
            // wait-by-necessity
            System.out.println("Passing the call to the next neighbour..");
            Vector<State> states = this.nextNeighbour.getAllNextStates();
            // states is a future

            // TODO 4. Is this explicit synchronization mandatory ?    (NO the wait was removed)
            states.add(this.getCurrentState());

            return states;
        } else {
            System.out.println("No more next neighbours");
            Vector<State> states = new Vector<State>();
            states.add(this.getCurrentState());
            return states;
        }
    }
}

In the previous example, we have some parts of the code that may trigger a wait-by-necessity. A wait-by-necessity happens when we try to use a future that has not been updated yet. When a future in this state is used, the thread trying to use the future blocks until the value of the future is updated.

We create a CMAgentMigrator class, that inherits from CMAInitialized. This class will implement all the non-functional behavior concerning the migration, for which this example is created.

The migration has to be initiated by the active object itself. We will have to write the migrate method in the code of CMAgentMigrator - i.e. a method that contains an explicit call to the migration primitive.

CMAgentMigrator skeleton class:

package org.objectweb.proactive.examples.userguide.cmagent.migration;

import java.io.Serializable;

import org.objectweb.proactive.api.PAMobileAgent;
import org.objectweb.proactive.core.ProActiveException;
import org.objectweb.proactive.core.node.Node;
import org.objectweb.proactive.examples.userguide.cmagent.initialized.CMAgentInitialized;
import org.objectweb.proactive.extensions.annotation.ActiveObject;
import org.objectweb.proactive.extensions.annotation.MigrationSignal;


@ActiveObject
public class CMAgentMigrator extends CMAgentInitialized implements Serializable {

    @MigrationSignal
    public void migrateTo(Node whereTo) {
        try {
            //TODO 1. Migrate the active object to the Node received as parameter
        } catch (ProActiveException moveExcep) {
            System.err.println(moveExcep.getMessage());
        }
    }
}

In the Main class, we only have to change a few lines of code to migrate the agent. As you may notice there is more code for creating the textual menu than for controlling the agents.

package org.objectweb.proactive.examples.userguide.cmagent.migration;

import java.io.BufferedReader;
import java.io.File;
import java.io.IOException;
import java.io.InputStreamReader;

import org.objectweb.proactive.ActiveObjectCreationException;
import org.objectweb.proactive.api.PAActiveObject;
import org.objectweb.proactive.api.PALifeCycle;
import org.objectweb.proactive.core.ProActiveException;
import org.objectweb.proactive.core.node.Node;
import org.objectweb.proactive.core.node.NodeException;
import org.objectweb.proactive.extensions.gcmdeployment.PAGCMDeployment;
import org.objectweb.proactive.gcmdeployment.GCMApplication;
import org.objectweb.proactive.gcmdeployment.GCMVirtualNode;


public class Main {
    private static GCMApplication pad;

    private static GCMVirtualNode deploy(String descriptor) throws NodeException, ProActiveException {

        //Create object representation of the deployment file
        pad = PAGCMDeployment.loadApplicationDescriptor(new File(descriptor));
        //Activate all Virtual Nodes
        pad.startDeployment();
        //Wait for all the virtual nodes to become ready
        pad.waitReady();

        //Get the first Virtual Node specified in the descriptor file
        GCMVirtualNode vn = pad.getVirtualNodes().values().iterator().next();

        return vn;
    }

    public static void main(String args[]) {
        try {
            BufferedReader inputBuffer = new BufferedReader(new InputStreamReader(System.in));
            GCMVirtualNode vn = deploy(args[0]);

            //create the active object
            CMAgentMigrator ao = (CMAgentMigrator) PAActiveObject.newActive(CMAgentMigrator.class.getName(),
                    new Object[] {}, vn.getANode());

            int k = 1;
            int choice;
            while (k != 0) {

                //display the menu with the available nodes 
                k = 1;
                for (Node node : vn.getCurrentNodes()) {
                    //TODO 2. Add the node URL to the menu
                    k++;
                }
                System.out.println("0.  Exit");

                Node[] nodeArray = vn.getCurrentNodes().toArray(new Node[] {});

                //select a node
                do {
                    System.out.print("Choose a node :> ");
                    try {
                        // Read am option from keyboard
                        choice = Integer.parseInt(inputBuffer.readLine().trim());
                    } catch (NumberFormatException noExcep) {
                        choice = -1;
                    }
                } while (!(choice >= 1 && choice < k || choice == 0));
                if (choice == 0)
                    break;

                //TODO 3. Migrate the active object to the selected node:  choice-1
                //TODO 4. Get the state and the last request time and print them out
                //TODO 5. Print the execution time of the last request
            }
        } catch (NodeException nodeExcep) {
            System.err.println(nodeExcep.getMessage());
        } catch (ActiveObjectCreationException aoExcep) {
            System.err.println(aoExcep.getMessage());
        } catch (IOException e) {
            System.err.println(e.getMessage());
        } catch (ProActiveException e) {
            System.err.println(e.getMessage());
        } finally {
            //TODO 6. Stop all the objects and JVM
        }
    }
}

You can find the skeleton codes for this exercise into the tutorials/src/Examples/org/objectweb/proactive/examples/userguide/cmagent/migration/ directory. Full code is also available into the directory ProActive/src/Examples/org/objectweb/proactive/examples/userguide/cmagent/migration/.

Once filled in, go to the tutorials/compile directory and type build[.bat] cmagent.migration to compile your code. Then, after a successful compilation, go to the scripts/CMAgent directory and launch the migrateCMA.[sh|bat] script to execute your code.

This first question is straightforward since you just have to call the PAMobileAgent.migrateTo method. The CMAgentMigrator class is therefore as follows:

package org.objectweb.proactive.examples.userguide.cmagent.migration;

import java.io.Serializable;

import org.objectweb.proactive.api.PAMobileAgent;
import org.objectweb.proactive.core.ProActiveException;
import org.objectweb.proactive.core.node.Node;
import org.objectweb.proactive.examples.userguide.cmagent.initialized.CMAgentInitialized;
import org.objectweb.proactive.extensions.annotation.ActiveObject;
import org.objectweb.proactive.extensions.annotation.MigrationSignal;


@ActiveObject
public class CMAgentMigrator extends CMAgentInitialized implements Serializable {

    @MigrationSignal
    public void migrateTo(Node whereTo) {
        try {
            //TODO 1. Migrate the active object to the Node received as parameter
            //should be the last call in this method
            //instructions after a call to PAMobileAgent.migrateTo are NOT executed 
            PAMobileAgent.migrateTo(whereTo);
        } catch (ProActiveException moveExcep) {
            System.err.println(moveExcep.getMessage());
        }
    }
}

	Warning
	As it said in comments, codes after the migrateTo method will not be executed.

Other questions are not more difficult and here is the CMAAgentMigrator class:

package org.objectweb.proactive.examples.userguide.cmagent.migration;

import java.io.BufferedReader;
import java.io.File;
import java.io.IOException;
import java.io.InputStreamReader;

import org.objectweb.proactive.ActiveObjectCreationException;
import org.objectweb.proactive.api.PAActiveObject;
import org.objectweb.proactive.api.PALifeCycle;
import org.objectweb.proactive.core.ProActiveException;
import org.objectweb.proactive.core.node.Node;
import org.objectweb.proactive.core.node.NodeException;
import org.objectweb.proactive.extensions.gcmdeployment.PAGCMDeployment;
import org.objectweb.proactive.gcmdeployment.GCMApplication;
import org.objectweb.proactive.gcmdeployment.GCMVirtualNode;


public class Main {
    private static GCMApplication pad;

    private static GCMVirtualNode deploy(String descriptor) throws NodeException, ProActiveException {

        //Create object representation of the deployment file
        pad = PAGCMDeployment.loadApplicationDescriptor(new File(descriptor));
        //Activate all Virtual Nodes
        pad.startDeployment();
        //Wait for all the virtual nodes to become ready
        pad.waitReady();

        //Get the first Virtual Node specified in the descriptor file
        GCMVirtualNode vn = pad.getVirtualNodes().values().iterator().next();

        return vn;
    }

    public static void main(String args[]) {
        try {
            BufferedReader inputBuffer = new BufferedReader(new InputStreamReader(System.in));
            GCMVirtualNode vn = deploy(args[0]);

            //create the active object
            CMAgentMigrator ao = (CMAgentMigrator) PAActiveObject.newActive(CMAgentMigrator.class.getName(),
                    new Object[] {}, vn.getANode());

            int k = 1;
            int choice;
            while (k != 0) {

                //display the menu with the available nodes 
                k = 1;
                for (Node node : vn.getCurrentNodes()) {
                    //TODO 2. Add the node URL to the menu
                    System.out.println(k + ".  Statistics for node :" + node.getNodeInformation().getURL());
                    k++;
                }
                System.out.println("0.  Exit");

                Node[] nodeArray = vn.getCurrentNodes().toArray(new Node[] {});

                //select a node
                do {
                    System.out.print("Choose a node :> ");
                    try {
                        // Read am option from keyboard
                        choice = Integer.parseInt(inputBuffer.readLine().trim());
                    } catch (NumberFormatException noExcep) {
                        choice = -1;
                    }
                } while (!(choice >= 1 && choice < k || choice == 0));
                if (choice == 0)
                    break;

                //TODO 3. Migrate the active object to the selected node:  choice-1
                ao.migrateTo(nodeArray[choice - 1]); //migrate
                //TODO 4. Get the state and the last request time and print them out
                String currentState = ao.getCurrentState().toString(); //get the state
                System.out.println("\n" + currentState);
                //TODO 5. Print the execution time of the last request
                System.out.println("Calculating the statistics took " +
                    ao.getLastRequestServeTime().getLongValue() + "ms \n");
            }
        } catch (NodeException nodeExcep) {
            System.err.println(nodeExcep.getMessage());
        } catch (ActiveObjectCreationException aoExcep) {
            System.err.println(aoExcep.getMessage());
        } catch (IOException e) {
            System.err.println(e.getMessage());
        } catch (ProActiveException e) {
            System.err.println(e.getMessage());
        } finally {
            //TODO 6. Stop all the objects and JVM
            if (pad != null)
                pad.kill();
            PALifeCycle.exitSuccess();
        }
    }
}

In this example, we only need to modify the Main class to create the group of objects. To instantiate the active object, we will use the CMAgentInitialized class that we have defined previously.

Main skeleton class:

package org.objectweb.proactive.examples.userguide.cmagent.groups;

import java.io.BufferedReader;
import java.io.File;
import java.io.IOException;
import java.io.InputStreamReader;
import java.util.Vector;

import org.objectweb.proactive.ActiveObjectCreationException;
import org.objectweb.proactive.api.PAActiveObject;
import org.objectweb.proactive.api.PAGroup;
import org.objectweb.proactive.api.PALifeCycle;
import org.objectweb.proactive.core.ProActiveException;
import org.objectweb.proactive.core.group.Group;
import org.objectweb.proactive.core.mop.ClassNotReifiableException;
import org.objectweb.proactive.core.node.Node;
import org.objectweb.proactive.core.node.NodeException;
import org.objectweb.proactive.examples.userguide.cmagent.migration.CMAgentMigrator;
import org.objectweb.proactive.examples.userguide.cmagent.simple.State;
import org.objectweb.proactive.extensions.gcmdeployment.PAGCMDeployment;
import org.objectweb.proactive.gcmdeployment.GCMApplication;
import org.objectweb.proactive.gcmdeployment.GCMVirtualNode;


public class Main {
    private static GCMApplication pad;

    private static GCMVirtualNode deploy(String descriptor) throws NodeException, ProActiveException {

        //Create object representation of the deployment file
        pad = PAGCMDeployment.loadApplicationDescriptor(new File(descriptor));
        //Activate all Virtual Nodes
        pad.startDeployment();
        //Wait for all the virtual nodes to become ready
        pad.waitReady();

        //Get the first Virtual Node specified in the descriptor file
        GCMVirtualNode vn = pad.getVirtualNodes().values().iterator().next();

        return vn;
    }

    public static void main(String args[]) {
        try {
            Vector<CMAgentMigrator> agents = new Vector<CMAgentMigrator>();
            BufferedReader inputBuffer = new BufferedReader(new InputStreamReader(System.in));
            GCMVirtualNode vn = deploy(args[0]);
            //TODO 1. Create a new empty group
            //TODO 2. Create a collection of active objects with on object on each node
            //TODO 3. Get a management representation of the monitors group
            //ask for adding or removing nodes
            //get statistics
            int k = 1;
            int choice;
            while (k != 0) {
                //display the menu 
                k = 1;
                System.out.println("Toggle monitored nodes (*) or display statistics: ");
                for (CMAgentMigrator agent : agents) {
                    //TODO 4. Print the node URL
                    if (gA.contains(agent)) {
                    } else {
                    }
                    k++;
                }
                System.out.println("-1.  Display statistics for monitored nodes");
                System.out.println(" 0.  Exit");

                //select a node
                do {
                    System.out.print("Choose a node to add or remove  :> ");
                    try {
                        // Read am option from keyboard
                        choice = Integer.parseInt(inputBuffer.readLine().trim());
                    } catch (NumberFormatException noExcep) {
                        choice = -1;
                    }
                } while (!(choice >= 1 && choice < k || choice == 0 || choice == -1));
                if (choice == 0)
                    break;
                if (choice == -1) {

                    State resultsGroup = monitorsGroup.getCurrentState();
                    while (PAGroup.size(resultsGroup) > 0) {
                        //TODO 5. Use PAGroup.waitAndGetOneThenRemoveIt() to control the list of State futures
                        System.out.println(statistic.toString());
                    }
                } else {
                    //TODO 6. Use the management representation to add or remove
                    //        the agent (choice-1) to/from the group.
                }
            }

        } catch (NodeException nodeExcep) {
            System.err.println(nodeExcep.getMessage());
        } catch (ActiveObjectCreationException aoExcep) {
            System.err.println(aoExcep.getMessage());
        } catch (IOException e) {
            System.err.println(e.getMessage());
        } catch (ClassNotReifiableException e) {
            System.err.println(e.getMessage());
        } catch (ClassNotFoundException e) {
            System.err.println(e.getMessage());
        } catch (ProActiveException e) {
            System.err.println(e.getMessage());
        } finally {
            //stopping all the objects and JVMS
            if (pad != null)
                pad.kill();
            PALifeCycle.exitSuccess();
        }
    }
}

You can find the skeleton codes for this exercise into the tutorials/src/Examples/org/objectweb/proactive/examples/userguide/cmagent/groups/ directory. Full code is also available into the directory ProActive/src/Examples/org/objectweb/proactive/examples/userguide/cmagent/groups/.

Once filled in, go to the tutorials/compile directory and type build[.bat] cmagent.groups to compile your code. Then, after a successful compilation, go to the scripts/CMAgent directory and launch the groups.[sh|bat] script to execute your code.

To create the group, we need to specify the type of the group. In our case, we use the previously defined CMAgentMigrator. We first create the group and then get a management representation of the group through the Group interface. The Group interface has the necessary methods for adding and removing members of the group.

//TODO 1. Create a new empty group
CMAgentMigrator monitorsGroup = (CMAgentMigrator) PAGroup.newGroup(CMAgentMigrator.class
        .getName());
//TODO 2. Create a collection of active objects with on object on each node
for (Node node : vn.getCurrentNodes()) {
    CMAgentMigrator ao = (CMAgentMigrator) PAActiveObject.newActive(CMAgentMigrator.class
            .getName(), new Object[] {}, node);
    agents.add(ao);
}
//TODO 3. Get a management representation of the monitors group
Group<CMAgentMigrator> gA = PAGroup.getGroup(monitorsGroup);

We use synchronization to wait for all the agents to send the states. As an agent returns a State, we remove it from the list of futures.

//TODO 5. Use PAGroup.waitAndGetOneThenRemoveIt() to control the list of State futures
State statistic = (State) PAGroup.waitAndGetOneThenRemoveIt(resultsGroup);

To test if an agent is into the group and then to add or remove it, we use the same methods as for the classical containers, that is to say, the methods contains, add and remove:

if (gA.contains(agents.elementAt(choice - 1))) {
    gA.remove(agents.elementAt(choice - 1));
} else {
    gA.add(agents.elementAt(choice - 1));
}

The full Main method:

public static void main(String args[]) {
    try {
        Vector<CMAgentMigrator> agents = new Vector<CMAgentMigrator>();
        BufferedReader inputBuffer = new BufferedReader(new InputStreamReader(System.in));
        GCMVirtualNode vn = deploy(args[0]);
        //TODO 1. Create a new empty group
        CMAgentMigrator monitorsGroup = (CMAgentMigrator) PAGroup.newGroup(CMAgentMigrator.class
                .getName());
        //TODO 2. Create a collection of active objects with on object on each node
        for (Node node : vn.getCurrentNodes()) {
            CMAgentMigrator ao = (CMAgentMigrator) PAActiveObject.newActive(CMAgentMigrator.class
                    .getName(), new Object[] {}, node);
            agents.add(ao);
        }
        //TODO 3. Get a management representation of the monitors group
        Group<CMAgentMigrator> gA = PAGroup.getGroup(monitorsGroup);
        //ask for adding or removing nodes
        //get statistics
        int k = 1;
        int choice;
        while (k != 0) {
            //display the menu 
            k = 1;
            System.out.println("Toggle monitored nodes (*) or display statistics: ");
            for (CMAgentMigrator agent : agents) {
                //TODO 4. Print the node URL
                if (gA.contains(agent)) {
                    System.out.println(" " + k + ".* " + PAActiveObject.getActiveObjectNodeUrl(agent));
                } else {
                    System.out.println(" " + k + ".  " + PAActiveObject.getActiveObjectNodeUrl(agent));
                }
                k++;
            }
            System.out.println("-1.  Display statistics for monitored nodes");
            System.out.println(" 0.  Exit");

            //select a node
            do {
                System.out.print("Choose a node to add or remove  :> ");
                try {
                    // Read am option from keyboard
                    choice = Integer.parseInt(inputBuffer.readLine().trim());
                } catch (NumberFormatException noExcep) {
                    choice = -1;
                }
            } while (!(choice >= 1 && choice < k || choice == 0 || choice == -1));
            if (choice == 0)
                break;
            if (choice == -1) {

                State resultsGroup = monitorsGroup.getCurrentState();
                while (PAGroup.size(resultsGroup) > 0) {
                    //TODO 5. Use PAGroup.waitAndGetOneThenRemoveIt() to control the list of State futures
                    State statistic = (State) PAGroup.waitAndGetOneThenRemoveIt(resultsGroup);
                    System.out.println(statistic.toString());
                }
            } else {
                //TODO 6. Use the management representation to add or remove
                //        the agent (choice-1) to/from the group.
                if (gA.contains(agents.elementAt(choice - 1))) {
                    gA.remove(agents.elementAt(choice - 1));
                } else {
                    gA.add(agents.elementAt(choice - 1));
                }
            }
        }

    } catch (NodeException nodeExcep) {
        System.err.println(nodeExcep.getMessage());
    } catch (ActiveObjectCreationException aoExcep) {
        System.err.println(aoExcep.getMessage());
    } catch (IOException e) {
        System.err.println(e.getMessage());
    } catch (ClassNotReifiableException e) {
        System.err.println(e.getMessage());
    } catch (ClassNotFoundException e) {
        System.err.println(e.getMessage());
    } catch (ProActiveException e) {
        System.err.println(e.getMessage());
    } finally {
        //stopping all the objects and JVMS
        if (pad != null)
            pad.kill();
        PALifeCycle.exitSuccess();
    }
}

In this example, we extend CMAgentInitialized and create a CMAgentWebService that we expose through the main method.

Since the usage of the webservice is independent of the underlying implementation, we give here only a Java example of how to access the webservice. This client enables in particular to call either axis2 or cxf services. However, you can simply use your favorite browser.

public class CMAgentWebServiceClient {

    public static void main(String[] args) {

        try {
            String url = "";
            String wsFrameWork = "";
            if (args.length == 1) {
                url = "http://localhost:8080/";
                wsFrameWork = args[0];
            } else if (args.length == 2) {
                url = args[0];
                wsFrameWork = args[1];
            } else {
                System.out.println("Wrong number of arguments:");
                System.out.println("Usage: java CMAgentWebServiceClient [url] wsFrameWork");
                System.out.println("where wsFrameWork is either 'axis2' or 'cxf'");
                return;
            }

            ClientFactory cf = AbstractClientFactory.getClientFactory(wsFrameWork);
            Client client = cf.getClient(url, "cmAgentService", CMAgentService.class);

            Object[] response = client.call("getCurrentState", null, State.class);

            System.out.println("Current state is:\n" + ((State) response[0]).toString());

            response = client.call("waitLastRequestServeTime", null, long.class);

            System.out.println("Last request serve time = " + response[0]);

        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

In the CMAgentWebService, we need to write 3 lines of code to use the active object as an webservice.

package org.objectweb.proactive.examples.userguide.cmagent.webservice;

import org.objectweb.proactive.ActiveObjectCreationException;
import org.objectweb.proactive.api.PAActiveObject;
import org.objectweb.proactive.core.node.NodeException;
import org.objectweb.proactive.examples.userguide.cmagent.initialized.CMAgentInitialized;
import org.objectweb.proactive.extensions.annotation.ActiveObject;
import org.objectweb.proactive.extensions.webservices.AbstractWebServicesFactory;
import org.objectweb.proactive.extensions.webservices.WebServices;
import org.objectweb.proactive.extensions.webservices.WebServicesFactory;
import org.objectweb.proactive.extensions.webservices.exceptions.WebServicesException;


@ActiveObject
public class CMAgentService extends CMAgentInitialized {

    // we cannot use the getLastRequestServeTime() method
    // directly since LongWrapper is not recognize
    // TODO: See why
    public long waitLastRequestServeTime() {
        return this.getLastRequestServeTime().getLongValue();
    }

    public static void main(String[] args) {

        String url = "";
        String wsFrameWork = "";
        if (args.length == 1) {
            url = AbstractWebServicesFactory.getLocalUrl();
            wsFrameWork = args[0];
        } else if (args.length == 2) {
            url = args[0];
            wsFrameWork = args[1];
        } else {
            System.out.println("Wrong number of arguments:");
            System.out.println("Usage: java CMAgentService [url] wsFrameWork");
            System.out.println("where wsFrameWork is either 'axis2' or 'cxf'");
            return;
        }

        System.out.println("Started a monitoring agent on : " + url);

        CMAgentService hw;
        try {
            hw = (CMAgentService) PAActiveObject.newActive(
                    "org.objectweb.proactive.examples.userguide.cmagent.webservice.CMAgentService",
                    new Object[] {});

            // TODO 1.Expose as web service (on URL 'url') the methods
            // "getLastRequestServeTime" and "getCurrentState" 
            // of 'hw' CMAgentService.
            // Name your service  "cmAgentService" and use the web service framework given
            // in argument.

        } catch (ActiveObjectCreationException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        } catch (NodeException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        } catch (WebServicesException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
    }
}

You can find the skeleton codes for this exercise into the tutorials/src/Examples/org/objectweb/proactive/examples/userguide/cmagent/webservice/ directory. Full code is also available into the directory ProActive/src/Examples/org/objectweb/proactive/examples/userguide/cmagent/webservice/.

Once filled in, go to the tutorials/compile directory and type build[.bat] cmagent.webservice to compile your code. Then, after a successful compilation, go to the scripts/CMAgent directory and launch the exposeCMA.[sh|bat] [url] [axis2|cxf] script to execute your code. The url is only needed when you want to deploy the agent on an external application server.

WebServicesFactory wsf = AbstractWebServicesFactory.getWebServicesFactory(wsFrameWork);
WebServices ws = wsf.getWebServices(url);
ws.exposeAsWebService(hw, "cmAgentService", new String[] { "waitLastRequestServeTime",
        "getCurrentState" });

Refer to Exporting active objects and components as Web Services to get more information about web services exposition.

In this section, we will look at a simple sequential version of the application to find if a candidate number is prime.

package org.objectweb.proactive.examples.userguide.primes.sequential;

/**
 * This class illustrates a sequential algorithm for primality test.
 * <p>
 * Some primes : 4398042316799l, 63018038201, 2147483647
 * 
 * @author The ProActive Team
 * 
 */
public class Main {

    public static void main(String[] args) {
        // The default value for the candidate to test (is prime)
        long candidate = 3093215881333057l;
        // Parse the number from args if there is some
        if (args.length > 0) {
            try {
                candidate = Long.parseLong(args[0]);
            } catch (NumberFormatException numberException) {
                System.err.println(numberException.getMessage());
                System.err.println("Usage: Main <candidate>");
            }
        }
        // We don't need to check numbers greater than the square-root of the
        // candidate in this algorithm
        long squareRootOfCandidate = (long) Math.ceil(Math.sqrt(candidate));
        // Begin from 2 the first known prime number
        long begin = 2;
        // Until the end of the range
        long end = squareRootOfCandidate;
        // Check the primality
        boolean isPrime = Main.isPrime(candidate, begin, end);
        // Display the result
        System.out.println("\n" + candidate + (isPrime ? " is prime." : " is not prime.") + "\n");
    }

    /**
     * Tests a primality of a specified number in a specified range.
     * 
     * @param candidate
     *            the candidate number to check
     * @param begin
     *            starts check from this value
     * @param end
     *            checks until this value
     * @return <code>true</code> if is prime; <code>false</code> otherwise
     */
    public static Boolean isPrime(long candidate, long begin, long end) {
        for (long divider = begin; divider < end; divider++) {
            if ((candidate % divider) == 0) {
                return false;
            }
        }
        return true;
    }
}

sequential.[sh|bat] 2147483647

In this section, we will distribute the computation using agents from the precedent tutorial.

6.8.2.2. Architecture of the Distributed Version

The test range from 2 to is divided into a number of intervals that will be sent asynchronously to workers by the manager using a round-robin algorithm. Once all intervals have been sent, the manager waits for any answers and checks the result.

Figure 6.1. Distribution of the Test Range (Example with 2 Workers)

Each worker serves requests in a FIFO order, meanwhile the order of result reception by the manager is indeterministic, it depends on various factors like workers load, network latency, etc.

CMAgentPrimeWorker and CMAgentPrimeManager classes have several methods implementing the mechanism described above. The manager class has a method for adding workers to its list (addWorker(CMAgentPrimeWorker)), and a method that distributes the workload by performing method calls on workers (isPrime(long number)). The worker class has a method that checks if a number is prime ( isPrime(long number, long begin, long end) ). A main class is used to deploy the manager and its workers on a specified deployment descriptor. For this example, the manager and worker classes without code look like hereafter. Try to fill in the code for the methods. An example implementation is also provided below.

package org.objectweb.proactive.examples.userguide.primes.distributed;

import java.util.Vector;

import org.objectweb.proactive.api.PAFuture;
import org.objectweb.proactive.core.util.wrapper.BooleanWrapper;
import org.objectweb.proactive.extensions.annotation.ActiveObject;


/**
 * @author The ProActive Team
 */
@ActiveObject
public class CMAgentPrimeManager {
    /**
     * A vector of references on workers
     */
    private Vector<CMAgentPrimeWorker> workers = new Vector<CMAgentPrimeWorker>();
    /**
     * Default interval size
     */
    public static final int INTERVAL_SIZE = 100;

    /**
     * Empty no-arg constructor needed by ProActive
     */
    public CMAgentPrimeManager() {
    }

    /**
     * Tests a primality of a specified number. Synchronous !
     * 
     * @param number
     *            The number to test
     * @return <code>true</code> if is prime; <code>false</code> otherwise
     */
    public boolean isPrime(long number) {
        // We don't need to check numbers greater than the square-root of the
        // candidate in this algorithm
        long squareRootOfCandidate = (long) Math.ceil(Math.sqrt(number));

        // Begin from 2 the first known prime number
        long begin = 2;

        // The number of intervals       
        long nbOfIntervals = (long) Math.ceil(squareRootOfCandidate / INTERVAL_SIZE);

        // Until the end of the first interval
        long end = INTERVAL_SIZE;

        // The vector of futures
        final Vector<BooleanWrapper> answers = new Vector<BooleanWrapper>();

        // Non blocking (asynchronous method call) 
        for (int i = 0; i <= nbOfIntervals; i++) {

            // Use round robin selection of worker
            int workerIndex = i % workers.size();
            CMAgentPrimeWorker worker = workers.get(workerIndex);

            //TODO 1. Send asynchronous method call to the worker 

            //TODO 2. Add the future result to the vector of answers 

            // Update the begin and the end of the interval
            begin = end + 1;
            end += INTERVAL_SIZE;
        }
        // Once all requests was sent
        boolean prime = true;
        int intervalNumber = 0;
        // Loop until a worker returns false or vector is empty (all results have been checked)
        while (!answers.isEmpty() && prime) {

            // TODO 3. Block until a new response is available 
            // by using a static method from org.objectweb.proactive.api.PAFuture

            // Check the answer
            prime = answers.get(intervalNumber).getBooleanValue();

            // Remove the actualized future            
            answers.remove(intervalNumber);

        }
        return prime;
    }

    /**
     * Adds a worker to the local vector
     * 
     * @param worker
     *            The worker to add to the vector
     */
    public void addWorker(CMAgentPrimeWorker worker) {
        this.workers.add(worker);
    }
}

package org.objectweb.proactive.examples.userguide.primes.distributed;

import org.objectweb.proactive.core.util.wrapper.BooleanWrapper;
import org.objectweb.proactive.examples.userguide.cmagent.simple.CMAgent;
import org.objectweb.proactive.extensions.annotation.ActiveObject;


/**
 * @author The ProActive Team
 */
@ActiveObject
public class CMAgentPrimeWorker extends CMAgent {

    /**
     * Tests a primality of a specified number in a specified range.
     * 
     * @param candidate
     *            the candidate number to check
     * @param begin
     *            starts check from this value
     * @param end
     *            checks until this value
     * @return <code>true</code> if is prime; <code>false</code> otherwise
     */
    public BooleanWrapper isPrime(final long candidate, final long begin, final long end) {
        try {
            //Used for slowing down the application for in order 
            //to let one stop it for checking fault tolerance behavior
            Thread.sleep(300);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        //TODO 4. Return a reifiable wrapper for the Boolean type
        //    for asynchronous calls. 
    }

}

package org.objectweb.proactive.examples.userguide.primes.distributed;

import java.io.File;
import java.util.Collection;
import java.util.Iterator;

import org.objectweb.proactive.api.PAActiveObject;
import org.objectweb.proactive.core.ProActiveException;
import org.objectweb.proactive.core.node.Node;
import org.objectweb.proactive.core.node.NodeException;
import org.objectweb.proactive.examples.userguide.cmagent.simple.CMAgent;
import org.objectweb.proactive.extensions.gcmdeployment.PAGCMDeployment;
import org.objectweb.proactive.gcmdeployment.GCMApplication;
import org.objectweb.proactive.gcmdeployment.GCMVirtualNode;


/**
 * This class illustrates a distributed version of the sequential algorithm for
 * primality test based on the {@link CMAgent}.
 * <p>
 * Some primes : 3093215881333057, 4398042316799, 63018038201, 2147483647 
 * 
 * @author The ProActive Team
 * 
 */
public class Main {

    private static GCMApplication pad;

    private static Collection<GCMVirtualNode> deploy(String descriptor) {
        try {
            pad = PAGCMDeployment.loadApplicationDescriptor(new File(descriptor));
            //active all Virtual Nodes
            pad.startDeployment();
            pad.waitReady();

            return pad.getVirtualNodes().values();
        } catch (NodeException nodeExcep) {
            System.err.println(nodeExcep.getMessage());
        } catch (ProActiveException proExcep) {
            System.err.println(proExcep.getMessage());
        }
        return null;
    }

    public static void main(String[] args) {
        // The default value for the candidate to test (is prime)
        long candidate = 2147483647l;
        // Parse the number from args if there is some
        if (args.length > 1) {
            try {
                candidate = Long.parseLong(args[1]);
            } catch (NumberFormatException numberException) {
                System.err.println("Usage: Main <candidate>");
                System.err.println(numberException.getMessage());
            }
        }

        try {
            Collection<GCMVirtualNode> vNodes = deploy(args[0]);
            GCMVirtualNode vNode = vNodes.iterator().next();

            // create the active object on the first node on
            // the first virtual node available
            // start the master
            CMAgentPrimeManager manager = (CMAgentPrimeManager) PAActiveObject.newActive(
                    CMAgentPrimeManager.class.getName(), new Object[] {}, vNode.getANode());

            //TODO 5:  iterate through all nodes, deploy
            // a worker per node and add it to the manager

            // Check the primality (Send a synchronous method call to the manager)
            boolean isPrime = manager.isPrime(candidate);
            // Display the result
            System.out.println("\n" + candidate + (isPrime ? " is prime." : " is not prime.") + "\n");
            // Free all resources
            pad.kill();
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            System.exit(0);
        }
    }
}

You can find the skeleton codes for this exercise into the tutorials/src/Examples/org/objectweb/proactive/examples/userguide/primes/distributed/ directory. Full code is also available into the directory ProActive/src/Examples/org/objectweb/proactive/examples/userguide/primes/distributed/.

Once the code filled and the application run successfully, let's simulate a crash on a worker JVM to know if this application is fault-tolerant.

For this purpose add a Thread.sleep() in the isPrime method of the CMAgentPrimeWorker class in order to have time to monitor this application with IC2D.

From IC2D right-click on the JVM that contains a worker and hit the "Kill This JVM" option. Then Expect The Unexpected !

Figure 6.2. Killing a JVM from IC2D

Typically, the user will need to catch a org.objectweb.proactive.core.body.exceptions.FutureMonitoringPingFailureException exception manually to handle such failures.

Moreover, the application will never end or return an incorrect result since the manager will wait infinitely for an answer that never comes from a crashed worker.

package org.objectweb.proactive.examples.userguide.primes.distributed;

import java.io.File;
import java.util.Collection;
import java.util.Iterator;

import org.objectweb.proactive.api.PAActiveObject;
import org.objectweb.proactive.core.ProActiveException;
import org.objectweb.proactive.core.node.Node;
import org.objectweb.proactive.core.node.NodeException;
import org.objectweb.proactive.examples.userguide.cmagent.simple.CMAgent;
import org.objectweb.proactive.extensions.gcmdeployment.PAGCMDeployment;
import org.objectweb.proactive.gcmdeployment.GCMApplication;
import org.objectweb.proactive.gcmdeployment.GCMVirtualNode;


/**
 * This class illustrates a distributed version of the sequential algorithm for
 * primality test based on the {@link CMAgent}.
 * <p>
 * Some primes : 3093215881333057, 4398042316799, 63018038201, 2147483647 
 * 
 * @author The ProActive Team
 * 
 */
public class Main {

    private static GCMApplication pad;

    private static Collection<GCMVirtualNode> deploy(String descriptor) {
        try {
            pad = PAGCMDeployment.loadApplicationDescriptor(new File(descriptor));
            //active all Virtual Nodes
            pad.startDeployment();
            pad.waitReady();

            return pad.getVirtualNodes().values();
        } catch (NodeException nodeExcep) {
            System.err.println(nodeExcep.getMessage());
        } catch (ProActiveException proExcep) {
            System.err.println(proExcep.getMessage());
        }
        return null;
    }

    public static void main(String[] args) {
        // The default value for the candidate to test (is prime)
        long candidate = 2147483647l;
        // Parse the number from args if there is some
        if (args.length > 1) {
            try {
                candidate = Long.parseLong(args[1]);
            } catch (NumberFormatException numberException) {
                System.err.println("Usage: Main <candidate>");
                System.err.println(numberException.getMessage());
            }
        }

        try {
            Collection<GCMVirtualNode> vNodes = deploy(args[0]);
            GCMVirtualNode vNode = vNodes.iterator().next();

            // create the active object on the first node on
            // the first virtual node available
            // start the master
            CMAgentPrimeManager manager = (CMAgentPrimeManager) PAActiveObject.newActive(
                    CMAgentPrimeManager.class.getName(), new Object[] {}, vNode.getANode());

            //TODO 5:  iterate through all nodes, deploy
            // a worker per node and add it to the manager
            Iterator<Node> nodesIt = vNode.getCurrentNodes().iterator();
            while (nodesIt.hasNext()) {
                Node node = nodesIt.next();
                CMAgentPrimeWorker worker = (CMAgentPrimeWorker) PAActiveObject.newActive(
                        CMAgentPrimeWorker.class.getName(), new Object[] {}, node);
                manager.addWorker(worker);
            }

            // Check the primality (Send a synchronous method call to the manager)
            boolean isPrime = manager.isPrime(candidate);
            // Display the result
            System.out.println("\n" + candidate + (isPrime ? " is prime." : " is not prime.") + "\n");
            // Free all resources
            pad.kill();
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            System.exit(0);
        }
    }
}

package org.objectweb.proactive.examples.userguide.primes.distributed;

import java.util.Vector;

import org.objectweb.proactive.api.PAFuture;
import org.objectweb.proactive.core.util.wrapper.BooleanWrapper;
import org.objectweb.proactive.extensions.annotation.ActiveObject;


/**
 * @author The ProActive Team
 */
@ActiveObject
public class CMAgentPrimeManager {
    /**
     * A vector of references on workers
     */
    private Vector<CMAgentPrimeWorker> workers = new Vector<CMAgentPrimeWorker>();
    /**
     * Default interval size
     */
    public static final int INTERVAL_SIZE = 100;

    /**
     * Empty no-arg constructor needed by ProActive
     */
    public CMAgentPrimeManager() {
    }

    /**
     * Tests a primality of a specified number. Synchronous !
     * 
     * @param number
     *            The number to test
     * @return <code>true</code> if is prime; <code>false</code> otherwise
     */
    public boolean isPrime(long number) {
        // We don't need to check numbers greater than the square-root of the
        // candidate in this algorithm
        long squareRootOfCandidate = (long) Math.ceil(Math.sqrt(number));

        // Begin from 2 the first known prime number
        long begin = 2;

        // The number of intervals       
        long nbOfIntervals = (long) Math.ceil(squareRootOfCandidate / INTERVAL_SIZE);

        // Until the end of the first interval
        long end = INTERVAL_SIZE;

        // The vector of futures
        final Vector<BooleanWrapper> answers = new Vector<BooleanWrapper>();

        // Non blocking (asynchronous method call) 
        for (int i = 0; i <= nbOfIntervals; i++) {

            // Use round robin selection of worker
            int workerIndex = i % workers.size();
            CMAgentPrimeWorker worker = workers.get(workerIndex);

            //TODO 1. Send asynchronous method call to the worker 
            // Send asynchronous method call to the worker
            BooleanWrapper res = worker.isPrime(number, begin, end);

            //TODO 2. Add the future result to the vector of answers 
            // Adds the future to the vector
            answers.add(res);

            // Update the begin and the end of the interval
            begin = end + 1;
            end += INTERVAL_SIZE;
        }
        // Once all requests was sent
        boolean prime = true;
        int intervalNumber = 0;
        // Loop until a worker returns false or vector is empty (all results have been checked)
        while (!answers.isEmpty() && prime) {

            // TODO 3. Block until a new response is available 
            // by using a static method from org.objectweb.proactive.api.PAFuture
            // Will block until a new response is available
            intervalNumber = PAFuture.waitForAny(answers);

            // Check the answer
            prime = answers.get(intervalNumber).getBooleanValue();

            // Remove the actualized future            
            answers.remove(intervalNumber);

        }
        return prime;
    }

    /**
     * Adds a worker to the local vector
     * 
     * @param worker
     *            The worker to add to the vector
     */
    public void addWorker(CMAgentPrimeWorker worker) {
        this.workers.add(worker);
    }
}

package org.objectweb.proactive.examples.userguide.primes.distributed;

import org.objectweb.proactive.core.util.wrapper.BooleanWrapper;
import org.objectweb.proactive.examples.userguide.cmagent.simple.CMAgent;
import org.objectweb.proactive.extensions.annotation.ActiveObject;


/**
 * @author The ProActive Team
 */
@ActiveObject
public class CMAgentPrimeWorker extends CMAgent {

    /**
     * Tests a primality of a specified number in a specified range.
     * 
     * @param candidate
     *            the candidate number to check
     * @param begin
     *            starts check from this value
     * @param end
     *            checks until this value
     * @return <code>true</code> if is prime; <code>false</code> otherwise
     */
    public BooleanWrapper isPrime(final long candidate, final long begin, final long end) {
        try {
            //Used for slowing down the application for in order 
            //to let one stop it for checking fault tolerance behavior
            Thread.sleep(300);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        //TODO 4. Return a reifiable wrapper for the Boolean type
        //    for asynchronous calls. 
        for (long divider = begin; divider < end; divider++) {
            if ((candidate % divider) == 0) {
                return new BooleanWrapper(false);
            }
        }
        return new BooleanWrapper(true);
    }

}

distributed.[sh|bat] 2147483647

In this section, we will distribute the computation using the Master-Worker API.

6.8.3.2. Architecture of the Master-Worker Version

For this example, the master and worker classes without code look like hereafter. Try to fill in the code for the methods. An example implementation is also provided below.

	Note
	With the Master-Worker API, there is no need to explicitly deal with primitive type wrappers, these kind of pitfalls are simply handled internally.

package org.objectweb.proactive.examples.userguide.primes.distributedmw;

import org.objectweb.proactive.extensions.masterworker.interfaces.Task;
import org.objectweb.proactive.extensions.masterworker.interfaces.WorkerMemory;


/**
 * Task to find if any number in a specified interval divides the given
 * candidate
 * 
 * @author The ProActive Team
 * 
 */
public class FindPrimeTask implements Task<Boolean> {

    private long begin;
    private long end;
    private long taskCandidate;

    //TODO 1. Write the constructor for this task
    public FindPrimeTask(long taskCandidate, long begin, long end) {
    }

    //TOOD 2. Fill the code that checks if the taskCandidate
    // is prime. Note that no wrappers are needed !
    public Boolean run(WorkerMemory memory) {
        return Boolean.valueOf(true);
    }
}

package org.objectweb.proactive.examples.userguide.primes.distributedmw;

import java.net.URL;
import java.util.ArrayList;
import java.util.List;

import org.objectweb.proactive.extensions.masterworker.ProActiveMaster;


/**
 * 
 * Some primes : 3093215881333057l, 4398042316799l, 63018038201, 2147483647
 * 
 * @author The ProActive Team
 * 
 */
public class PrimeExampleMW {
    /**
     * Default interval size
     */
    public static final int INTERVAL_SIZE = 100;

    public static void main(String[] args) {
        // The default value for the candidate to test (is prime)
        long candidate = 2147483647l;
        // Parse the number from args if there is some
        if (args.length > 1) {
            try {
                candidate = Long.parseLong(args[1]);
            } catch (NumberFormatException numberException) {
                System.err.println("Usage: PrimeExampleMW <candidate>");
                System.err.println(numberException.getMessage());
            }
        }
        try {
            // Create the Master
            ProActiveMaster<FindPrimeTask, Boolean> master = new ProActiveMaster<FindPrimeTask, Boolean>();
            // Deploy resources
            master.addResources(new URL(args[0]));
            // Create and submit the tasks
            master.solve(createTasks(candidate));

            //TODO 3. Wait all results from master

            // Test the primality
            boolean isPrime = true;
            for (Boolean result : results) {
                isPrime = isPrime && result;
            }
            // Display the result
            System.out.println("\n" + candidate + (isPrime ? " is prime." : " is not prime.") + "\n");
            // Terminate the master and free all resources
            master.terminate(true);
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            System.exit(0);
        }
    }

    /**
     * Creates the prime computation tasks to be solved
     * 
     * @return A list of prime computation tasks
     */
    public static List<FindPrimeTask> createTasks(long number) {
        List<FindPrimeTask> tasks = new ArrayList<FindPrimeTask>();

        // We don't need to check numbers greater than the square-root of the
        // candidate in this algorithm
        long squareRootOfCandidate = (long) Math.ceil(Math.sqrt(number));

        // Begin from 2 the first known prime number
        long begin = 2;

        // The number of intervals       
        long nbOfIntervals = (long) Math.ceil(squareRootOfCandidate / INTERVAL_SIZE);

        // Until the end of the first interval
        long end = INTERVAL_SIZE;

        for (int i = 0; i <= nbOfIntervals; i++) {

            //TODO 4. Create a new task for the current interval and 
            // add it to the list of tasks 

            // Update the begin and the end of the interval
            begin = end + 1;
            end += INTERVAL_SIZE;
        }

        return tasks;
    }
}

You can find the skeleton codes for this exercise into the tutorials/src/Examples/org/objectweb/proactive/examples/userguide/primes/distributedmw/ directory. Full code is also available into the directory ProActive/src/Examples/org/objectweb/proactive/examples/userguide/primes/distributedmw/.

Once the code filled in and the application successfully run, let's crash a worker to see the difference with the precedent application.

For this purpose add a Thread.sleep() in the run() method of the FindPrimeTask class.

Like in the precedent application, from IC2D, kill a JVM that contains a worker. Then expect no exceptions or errors!

The main difference with the precedent application is that the master does not send any tasks to the workers. In fact, it is the workers that ask the master for tasks. Such "work-stealing" pattern proves to be fault-tolerant and it is one the benefits from using a High-Level ProActive API.

package org.objectweb.proactive.examples.userguide.primes.distributedmw;

import org.objectweb.proactive.extensions.masterworker.interfaces.Task;
import org.objectweb.proactive.extensions.masterworker.interfaces.WorkerMemory;


/**
 * Task to find if any number in a specified interval divides the given
 * candidate
 * 
 * @author The ProActive Team
 * 
 */
public class FindPrimeTask implements Task<Boolean> {

    private long begin;
    private long end;
    private long taskCandidate;

    //TODO 1. Write the constructor for this task
    public FindPrimeTask(long taskCandidate, long begin, long end) {
        this.begin = begin;
        this.end = end;
        this.taskCandidate = taskCandidate;
    }

    //TOOD 2. Fill the code that checks if the taskCandidate
    // is prime. Note that no wrappers are needed !
    public Boolean run(WorkerMemory memory) {
        try {
            Thread.sleep(300);
        } catch (Exception e) {
            e.printStackTrace();
        }
        for (long divider = begin; divider < end; divider++) {
            if ((taskCandidate % divider) == 0) {
                return Boolean.valueOf(false);
            }
        }
        return Boolean.valueOf(true);
    }
}

package org.objectweb.proactive.examples.userguide.primes.distributedmw;

import java.net.URL;
import java.util.ArrayList;
import java.util.List;

import org.objectweb.proactive.extensions.masterworker.ProActiveMaster;


/**
 * 
 * Some primes : 3093215881333057l, 4398042316799l, 63018038201, 2147483647
 * 
 * @author The ProActive Team
 * 
 */
public class PrimeExampleMW {
    /**
     * Default interval size
     */
    public static final int INTERVAL_SIZE = 100;

    public static void main(String[] args) {
        // The default value for the candidate to test (is prime)
        long candidate = 2147483647l;
        // Parse the number from args if there is some
        if (args.length > 1) {
            try {
                candidate = Long.parseLong(args[1]);
            } catch (NumberFormatException numberException) {
                System.err.println("Usage: PrimeExampleMW <candidate>");
                System.err.println(numberException.getMessage());
            }
        }
        try {
            // Create the Master
            ProActiveMaster<FindPrimeTask, Boolean> master = new ProActiveMaster<FindPrimeTask, Boolean>();
            // Deploy resources
            master.addResources(new URL(args[0]));
            // Create and submit the tasks
            master.solve(createTasks(candidate));

            //TODO 3. Wait all results from master
            // Collect results            
            List<Boolean> results = master.waitAllResults();

            // Test the primality
            boolean isPrime = true;
            for (Boolean result : results) {
                isPrime = isPrime && result;
            }
            // Display the result
            System.out.println("\n" + candidate + (isPrime ? " is prime." : " is not prime.") + "\n");
            // Terminate the master and free all resources
            master.terminate(true);
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            System.exit(0);
        }
    }

    /**
     * Creates the prime computation tasks to be solved
     * 
     * @return A list of prime computation tasks
     */
    public static List<FindPrimeTask> createTasks(long number) {
        List<FindPrimeTask> tasks = new ArrayList<FindPrimeTask>();

        // We don't need to check numbers greater than the square-root of the
        // candidate in this algorithm
        long squareRootOfCandidate = (long) Math.ceil(Math.sqrt(number));

        // Begin from 2 the first known prime number
        long begin = 2;

        // The number of intervals       
        long nbOfIntervals = (long) Math.ceil(squareRootOfCandidate / INTERVAL_SIZE);

        // Until the end of the first interval
        long end = INTERVAL_SIZE;

        for (int i = 0; i <= nbOfIntervals; i++) {

            //TODO 4. Create a new task for the current interval and 
            // add it to the list of tasks 
            // Adds the task for the current interval to the list of tasks
            tasks.add(new FindPrimeTask(number, begin, end));

            // Update the begin and the end of the interval
            begin = end + 1;
            end += INTERVAL_SIZE;
        }

        return tasks;
    }
}

distributedMW.[sh|bat] 2147483647