Create a new class:

Note: We will start by performing the computation on several JVMs on localhost. Then, if you have some available machines, you can use those ones in order to perform the computation.

I assume you have downloaded the last version of the ProActive Scheduler in the SCHEDULER_HOME folder. (for me this folder is “/home/ProActiveScheduler” )

Note: we will go on with the implementation and start the scheduler server later.

Add ProActive jars

Modify the class DivisorCounterBean

First, we have to establish a connection to the Scheduler. Note that the Scheduler will accept one connection per Thread (this is due to the ProActive ActiveObjects programming pattern). As JBoss will reuse the threads we will have to keep a static variable that contain a mapping between the threads and the connections to the Scheduler. We also add three static variables containing the url of the scheduler, username and password. Lets add the next lines to the DivisorCounterBean class:

public final static String schedulerURL ="//192.168.1.10";
public final static String schedulerUserName = "user1";
public final static String userPassword="pwd1";
private static int numberOfTasks = 4;
public UserSchedulerInterface schedulerUI=null;
public SchedulerAuthenticationInterface schedulerAuth=null;
public static HashMap<Long, UserSchedulerInterface> connections=new HashMap<Long, UserSchedulerInterface>();

We will use the schedulerUI and schedulerAuth objects in order to connect to the Scheduler and submit a job.

Connect to the scheduler

We will start by writing a private method to connect to the scheduler.

Important Note: This is is the trickiest thing in this tutorial. First, we need to keep in mind that we need to connect to the ProActive Scheduler and therefore we are now in a ProActive programming model context. the ProActive Scheduler accepts only one connection per caller Thread (this is because the Scheduler accepts one connection per ActiveObject caller, plase see the ProActive programming model if you want to have a deeper understanding of all this). On the other hand, JBoss uses a thread pool in order to serve user requests. So, if at one point, I want to know if my session bean is connected to the Scheduler, I actually need to check if the current Thread has a connection to the Scheduler. I know it sounds complicated but it isn’t ..

As we need to keep one connection per Thread we will start by verifying if the current thread already has a connection attached, and, if not, create one and ad an entry in the connections HashMap. The method we obtain is:

private void connectToScheduler() throws SchedulerException, LoginException
{
      	long threadId = Thread.currentThread().getId();
	schedulerUI = DivisorCounterBean.connections.get(new Long(threadId));
	if (schedulerUI!=null)
	{
		System.out.println("already logged in");
		return;
	}
	else
	{
	 schedulerAuth = SchedulerConnection.join(schedulerURL);
	 schedulerUI = schedulerAuth.logAsUser(schedulerUserName, userPassword);
	 DivisorCounterBean.connections.put(new Long(threadId), schedulerUI);
	 System.out.println("new Connection to the schdeuler has been created. Nb of connections: "+connections.size());
	}
}

Add ejbCreate() method

We want to connect to the scheduler when the bean is created. We’ll just add a method called ejbCreate() that we annotate with @javax.ejb.Init in which we call the connectToScheduler() method. The method looks like this:

@javax.ejb.Init
      public void ejbCreate()
	{
		System.out.println("Creation of DivicorCounterBean. ");
		try{
			connectToScheduler();
		   }
		catch (Exception e)
		{
		  e.printStackTrace();
		}
	}

We have connected our bean to the scheduler, we only need now to split
the data, create the Tasks which count the number of divisors, send
them to the Scheduler and addition the results.

Implement the Task class

Lets start by implementing the Task code. A task will take in argument the number n for which we are counting divisors and interval in which we are looking for divisors, given by its lower and upper bounds. A Scheduler Task will have its input data in a HashMap of arguments. The execute method can take as input a list of results of the Tasks on which this task depends on, but, as our tasks don’t have dependencies, this list will always be empty. Furthermore, the code for the Task class it is easy to understand. Here it is:

package com.activeeon.demos.ejb.divisors;
  import java.math.BigInteger;
  import org.objectweb.proactive.core.util.wrapper.StringWrapper;
  import org.ow2.proactive.scheduler.common.task.TaskResult;
  public class DivisorsTask extends org.ow2.proactive.scheduler.common.task.executable.JavaExecutable{
  @Override
  public Serializable execute(TaskResult... arg0) throws Throwable {
	String lowerBoundString = this.getArgs().get("LowerBound");
	BigInteger lowerBound = new BigInteger(lowerBoundString);
	//BigInteger fromNumbber = new BigInteger(from);
	String upperBoundString = this.getArgs().get("UpperBound");
	BigInteger upperBound = new BigInteger(upperBoundString);
	String n = this.getArgs().get("n");
	BigInteger nNumber = new BigInteger(n);
	long divisors = 0;
	for (BigInteger i = lowerBound; i.compareTo(upperBound)<=0; i=i.add(new BigInteger("1")))
	{
		if (nNumber.remainder(i).equals(BigInteger.ZERO))
			divisors++;
	}
	String s = (new Long(divisors)).toString();
	StringWrapper sw = new StringWrapper(s);
	return sw;
      }
 }

Our bean is almost finished. We only need to create a job (which is formed of tasks) and submit it to the Scheduler.

Lets first add to our com.activeeon.demos.ejb.divisors.DivisorCounter interface a method

public long distributedCountDivisors(BigInteger n);

We will create a TaskFlowJob (which is a job that can contain one or more task(s) ), will add tasks to this job and then submit it to the scheduler.

Implement the distributedCountDivisors method

• job creation

TaskFlowJob job = new TaskFlowJob();
job.setName("DivisorCounter");
      //if error occur on one task, should we cancel the job?
      job.setCancelOnError(true);
     job.setDescription("Job sent from an EJB application deployed on JBoss v4.2 AS");

• classpath

As we need to execute the code in the DivisorTask class that we
previously wrote, we need to have this class on the remote machines (on
the nodes). We need so, either to provide a classpath containing the
DivisorTask class when starting the nodes, either to provide this
classpath to the job by using a JobEnvironment object. This is done by
these lines:

		String appClassPath = "";
		try {
			File appMainFolder = new File(DivisorCounterBean.class
					.getProtectionDomain().getCodeSource().getLocation()
					.toURI());
			appClassPath = appMainFolder.getAbsolutePath();
		} catch (URISyntaxException e1) {
			System.out
					.println("An exception has occured while setting the job classpath.");
			e1.printStackTrace();
		}
		JobEnvironment je = new JobEnvironment();
		je.setJobClasspath(new String[] { appClassPath });
		job.setEnv(je);

• creation of the tasks

We will create m intervals in which we will be counting the divisors of the number n. To do so, we will split the interval (2, n/2+1) into m intervals (where m = numberOfTasks, statically defined), and provide one interval to each Task. An interval is defined by its lower bound and its upper bound. Our tasks will receive then 3 arguments: lower bound, upper bound and the number n. For two values lower bound and upper bound already computed, the code which creates the task is:

JavaTask jt = new JavaTask();
jt.setName("Divisors Task "+i);
jt.setDescription("Count divisors of number "+n.toString()+" within the interval ("+lowerBound.toString()+" , "+upperBound.toString()+")");
jt.addArgument("LowerBound",lowerBound.toString());
jt.addArgument("UpperBound", upperBound.toString());
jt.addArgument("n", n.toString());
jt.setPreciousResult(true);
jt.setExecutableClassName(DivisorsTask.class.getName());
job.addTask(jt);

• submit the job to the scheduler and wait for the results

Once the job is submitted, we only have to wait for the results, addition them when arrived and send the final result to the client.
Note: In this implementation we will just keep waiting (perform a Thread sleep) and ask the scheduler every n milliseconds if the result is available. A better solution would be to subscribe a listener to the Scheduler in order to be notified, by the Scheduler, when the result is available. See the Advanced section at the end of this article.

The code of the method is:

public long distributedCountDivisors(BigInteger n) {
	if (schedulerUI == null) {
		try {
			connectToScheduler();
		} catch (LoginException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
			return 0;
		} catch (SchedulerException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
			return 0;
		}
	}
	long nbOfDivisors = 0;
	try {
		TaskFlowJob job = new TaskFlowJob();
		job.setName("DivisorCounter");
		// if error occur on one task, should we cancel the job?
		job.setCancelOnError(true);
		job
				.setDescription("Job sent from an EJB application deployed on JBoss v4.2 AS");
		// handle classpath
		String appClassPath = "";
		try {
			File appMainFolder = new File(DivisorCounterBean.class
					.getProtectionDomain().getCodeSource().getLocation()
					.toURI());
			appClassPath = appMainFolder.getAbsolutePath();
		} catch (URISyntaxException e1) {
			System.out
					.println("An exception has occured while setting the job classpath.");
			e1.printStackTrace();
		}
		JobEnvironment je = new JobEnvironment();
		je.setJobClasspath(new String[] { appClassPath });
		job.setEnv(je);
		BigInteger halfN = n.divide(new BigInteger("2"))
				.add(BigInteger.ONE);
		BigInteger intervalSize = halfN.divide(new BigInteger(new Integer(
				DivisorCounterBean.numberOfTasks - 1).toString()));
		BigInteger lowerBound;
		BigInteger upperBound;
		for (int i = 0; i < DivisorCounterBean.numberOfTasks; i++) {
			lowerBound = intervalSize.multiply(
					new BigInteger(new Integer(i).toString())).add(
					BigInteger.ONE);
			upperBound = lowerBound.add(intervalSize
					.subtract(BigInteger.ONE));
			// for the last interval
			if (upperBound.compareTo(halfN) > 0) {
				upperBound = halfN;
			}
			// Creating the Divisors Task for the interval (upperBound,
			// lowerBound)
			JavaTask jt = new JavaTask();
			jt.setName("Divisors Task " + i);
			jt.setDescription("Count divisors of number " + n.toString()
					+ " within the interval (" + lowerBound.toString()
					+ " , " + upperBound.toString() + ")");
			jt.addArgument("LowerBound", lowerBound.toString());
			jt.addArgument("UpperBound", upperBound.toString());
			jt.addArgument("n", n.toString());
			jt.setPreciousResult(true);
			jt.setExecutableClassName(DivisorsTask.class.getName());
			job.addTask(jt);
		}
		JobId jobId = schedulerUI.submit(job);
		System.out.println("The Job has been submited to the scheduler.");
		JobResult jResult = null;
		boolean keepWaiting = true;
		while (keepWaiting && (jResult == null)) {
			try {
				Thread.sleep(500);
				jResult = schedulerUI.getJobResult(jobId);
			} catch (SchedulerException se) {
				keepWaiting = false;
				se.printStackTrace();
				return 0;
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
		System.out.println("The results of the job are: " + jResult);
		HashMap<String, TaskResult> results = jResult.getAllResults();
		for (TaskResult t : results.values()) {
			String rs = ((StringWrapper) t.value()).toString();
			int r = new Integer(rs).intValue();
			nbOfDivisors += r;
		}
		// we add n as divisor of n
		nbOfDivisors++;
	} catch (Exception e) {
		e.printStackTrace();
	} catch (Throwable e) {
		// TODO Auto-generated catch block
		e.printStackTrace();
	}
	return nbOfDivisors;
}

Call the distributed method from the client

Lets call, from the client, the 2 methods (the sequential one and distributed one, in order to compare the computation times) Add the next lines to the main method oof the DivisorCounterTester class in the DivisorCounterClient project:

	System.out.println("Begining computation in distributed version");
	beginTime = System.currentTimeMillis();
	d= calculator.distributedCountDivisors(bi);
	endTime = System.currentTimeMillis();
	System.out.println(d+" divisors have been found");
	System.out.println("Answer have been received after "+(endTime-beginTime)+ " milliseconds. ");
	System.out.println("*****************/n");

Note: your client might not compile now because it does not now the distributedCountDivisors method. Just export again the DivisorCounterEJB project as a jar and import it again in the client project.

Start the scheduler and run your application

Important Note: before you do this, you should change the Proactive default rmi port in order not to have a conflict with JBoss port.

In your root folder you should have a folder called “.proactive” . Go inside and edit the file called ProActiveConfiguration.xml. Look for the line which defines the property “proactive.rmi.port” and change its value (1099 by default). In my case I put it to 1605 (it’s my credit card pin code )
Proactive Scheduler relies on a Resource Manager server.

• Start the Resource manager:

Open a command console and navigate to SCHEDULER_HOME/bin/unix (../windows if you are working on a windows system). Type ./startRM.sh (startRM.bat for windows) The Resource Manager will start and will deploy, by default, 4 nodes (jvm’s) on the localhost.

• Start the scheduler

Open a new command console and navigate to SCHEDULER_HOME/bin/unix (../windows) Only for the first time your are launching the scheduler, you have to create its database (which contains information about jobs, their tasks, results, etc..) Type: startScheduler_newDB.sh (.. bat for windows) This will create the database and start the scheduler which will connect to the resource manager in order to obtain resources and deploy tasks.

• Start the JBoss server

Important Note add this property to the jboss launcher: -Dlog4j.configuration=JBOSS_HOME/default/conf/jboss-log4j.xml

./run.sh -Dlog4j.configuration=JBOSS_HOME/default/conf/jboss-log4j.xml

You have to do this in order to force ProActive use the log4j.configuration
file of JBoss (in order not to have logger conflicts in the JBoss JVM).

• Run your client application.

On my machine the output of the client is:

Beginning computation in sequential version
12 divisors have been found
Answer have been received after 35638 milliseconds.
*****************

Beginning computation in distributed version
12 divisors have been found
Answer have been received after 18980 milliseconds.
*****************

This is it, if you followed me until this point, you have your session bean which delegates the computation on a set of JVMs. I only have few words to add about adding remote resources (nodes), starting the Scheduler itself on a remote machine, using the GUI for administration of the Scheduler and Resource manager.

Additional Information

Scheduler and Resource Manager Administration

Add remote resources

Running the scheduler and resource manager on remote machines

Remarks

Fast Test

I have launched the application by using 4 nodes/JVMs (2 local JVMs and two JVMs on different machines). The output of my client application is:

Beginning computation in sequential version
12 divisors have been found
Answer have been received after 33484 milliseconds.
*****************
Beginning computation in distributed version
12 divisors have been found
Answer have been received after 9163 milliseconds.
*****************

So, by distributing the computation on 4 JVMs, I had an answering time from the server 3.65 times lower then the time for the sequential application.

Advanced

In the distributedCountDivisors method from the DivisorCounterBean class we have created a job, submitted it to the ProActive Scheduler, and then, within a loop, we have “asked” the Scheduler, every n milliseconds, if it our result is available. This solution goes well for our little example but a much better solution would be to subscribe a Listener to the Scheduler in order to be notified when the result is available.
In order to do so, you should create a Listener by implementing the org.ow2.proactive.scheduler.common.scheduler.SchedulerEventListener. You must subscribe your listener to the Scheduler, and let your listener play the role of proxy for the Scheduler. Moreover, your Listener should be an active object or, at least should belong to an Active Object.
But this requires the understanding of some ProActive concepts. I might come back on this advanced solution in some other article. Do not hesitate to let me know if you’re planning to implement this solution, I might have some useful tips for you (especially about how to configure Jboss and ProActive Scheduler in order to solve some ClassLoading conflicts that might appear)

Documentation