Print

[cm]

Hi,

I am working with JPPF 4.0 on windows xp.
I try to implement your Receiving_notifications_from_the_tasks example from the documentation (see JPPF_3.zip).
For the Listener I build a jar (myListener.jar), as described in http://www.jppf.org/doc/v4/index.php?title=Receiving_notifications_from_the_tasks
I put the jar in the lib directories of the driver and the node.
The Example runs without errors, but I cant find anywhere the output on System.out from the listener. Si I think the listener is never called. Wher is my mistake?
The files JPPF_3.zip and myListener.jar you can find at http://www.wi-bw.tfh-wildau.de/~drmue/JPPF_0/

Thanks Christian

[lolo]

Hello Christian,

There is nothing incorrect in your code. This is in fact a JPPF bug, which I registered as JPPF-218 Calling fireNotification() from a task does not send notifications to local listeners. Clearly, we did not properly test this feature and I apologize for the inconvenience.

I am working on a fix (and unit test) and will publish a patch as soon as it is working, normally tomorrow or the day after at the latest. I will update you in this thread.

Sincerely,
-Laurent

[lolo]

Hello Christian,

This is to let you know that the bug has been fixed and a patch is available here.

Sincerely,
-Laurent

[cm]

Hello Laurent,

thanks for your patch, but with the patch the client runs in a endless loop with a resultcollector state execute.

I have seen, your patch is from Feb 06 and the new jar file in the patch is from Jan, 30
Is this the right jar file?

thanks for jour help.
Christian

[lolo]

Hi Christian,

You're right, I somehow put the wrong jar files in the zip. I uploaded a new zip file with the correct version of the jars. You can get it from the same link.
Sorry about this.

-Laurent

[cm]

Hi Laurent,

thanks for your update. Now it works fine.

In this context I have two additional questions:
1. Why is the taskId recived by the listener null?
2. What is the best way to make the notifications accesiable for the client. I think on pluggable MBeans. Is this the right way and what means this for the performance?

Thanks
Christian

[lolo]

Hi Christian,

Regarding you first question, can you detail what exactly is null? On my side if in the listener I call event.getTask().getId(), I see the same task id that was set in the remote client. Isn't this what you observe?

For the second question, the wiring for sending task notifications as JMX notifications already exists, this is the whole point of the seond parameter in AbstractTask.fireNotification(Object, boolean). If the poolean parameter is true, then the notification will also be sent via JMX. So you do don't need to create a custom MBean for this. See the related documentation here. Additionally, you might want to register a notification listener which listens on all the nodes for these notifications, or any subset of nodes you specify, as explained in the related documentation. You can found a code smaple for this in one of our unit tests here (plus related classes).

Sincerely,
-Laurent

[cm]

Hi Laurent,

About question 1:
When I set a taskId, then the taskId is not null. I was thinking JPPF generates a unique taskId.

About question 2:
For forwarding the notification I say:
   public ThreadRunner(JPPFClient jppfClient, String jobName){
      try{
         AbstractJPPFClientConnection conn = (AbstractJPPFClientConnection) jppfClient.getClientConnection();
         if(conn == null) System.out.println("conn is null");
         JMXDriverConnectionWrapper driverJmx = conn.getJmxConnection();
                        ...
then the conn is null.
Is this that what you mean, or a mistake from me?
The complete code is in the attachment.

Thanks
Christian

[lolo]

Hi Christian,

Yes, the task id is user-defined, and not automatically generated.

Regarding the null connection, this happens because the JPPF client has not yet had time to establish a connection to the driver. This is what I do to handle this situation:

Code: [Select]

// wait until a connection to the driver is properly established
while (!client.hasAvailableConnection()) Thread.sleep(10L);
JMXDriverConnectionWrapper jmxDriver = null;
while ((jmxDriver = client.getClientConnection().getJmxConnection()) == null) Thread.sleep(10L);
// wait until a JMX connection to the driver is established
while (!jmxDriver.isConnected()) Thread.sleep(10L);
Sincerely,
-Laurent

[cm]

Hi Laurent,

the code runs fine now! Here is the code: http://www.wi-bw.tfh-wildau.de/~drmue/JPPF/JPPF_1/

But I have 3 final questions:

1. Question: From each task I get 2 Notifications:
1.st Notification:
received notification from nodeUuid=178095D5-FA4E-8FFD-807A-3476E685D2F3, mBeanName=org.jppf:name=task.monitor,type=node,  message=
received notification with user object = task 9 is finished. jmx
received notification with info object = TaskInformation[taskId=task_id_9, jobId=28031E94-ACFE-A58E-3060-6668AD5B855F, cpuTime=-1, elapsedTime=-1, error=false
received notification with handback object = null
2.th Notification
received notification from nodeUuid=178095D5-FA4E-8FFD-807A-3476E685D2F3, mBeanName=org.jppf:name=task.monitor,type=node,  message=
received notification with user object = null
received notification with info object = TaskInformation[taskId=task_id_9, jobId=28031E94-ACFE-A58E-3060-6668AD5B855F, cpuTime=1357, elapsedTime=1377, error=false
received notification with handback object = null

The first notification includes the user object that I have send and the second is generated by JPPF and has a TaskInformation with cpu time and elapsed time. Is there a easy formal criteria to differ between both notifications?

2. Question: Which role plays the handback object in the notification listener? In everytime ist null.

3: Question: In my real tasks, I get big result objects from the tasks. Maks it a difference in performance, when I send them with a notification via jmx and not as a result?

Thanks
Christian

[lolo]

Hi Christian,

Currently, the only way to disambiguate between JPPF- and user-defined notifications is via the cpuTime attribute of the TaskInformation object in the notification: for user-defined notifications, the cpuTime is always set to -1, whereas for notifications generated by JPPF it is always >= 0. Thank you for noticing and reporting this. I registered an enhancement request for this: JPPF-219 Unambiguous way to distinguish between JPPF and user-defined task notifications. Right now you can always use the cpuTime, for example:

Code: [Select]

public class MyNotificationListener implements NotificationListener {
  @Override
  public void handleNotification(Notification notification, Object handback) {
    TaskExecutionNotification taskNotif = (TaskExecutionNotification) notification;
    boolean userDefined = taskNotif.getTaskInformation().getCpuTime() < 0;
    // .....
  }
}
As stated in the Javadoc for NotificationListener.handleNotification():

handback - An opaque object which helps the listener to associate information regarding the MBean emitter. This object is passed to the addNotificationListener call and resent, without modification, to the listener

In my experience, I found the handback object useful when registering the same listener instance multiple times, for example when it is registered for multiple MBeans, or the same MBean multiple times with a different NotificationFilter. The handback is always used locally, i.e. it is never sent to a remote MBean server, and never used in any way by the JMX framework, other than handing it back to the registered listener.

Regarding your last question, yes I believe sending large objects in JMX notifications can incur a significant performance overhead. This is not the intended purpose of the notification mechanism, which is oriented towards application monitoring and thus tends to require small, short notifications whenever possible. Unless you absolutely need intermediate results from your tasks, I would recommend to send these results as part of the "normal" JPPF execution flow. In JPPF the communication between nodes and servers, and between servers and clients is designed for handling large objects. In particular, the JPPF server has a disk overflow mechanism which allows it to handle tasks whose size is much larger than the JVM heap. This mechanism is not available for objects sent/received via JMX.

I hope this clarifies,
-Laurent

[cm]

Hi Laurent,

thanks for your detailed answer.

About the last point with the performance comparisation between jmx and the classical result management.
I have the following situation:

         jppfClient.submitJob(job);
         // wait on results
         while(! resultCollector.getStatus().equals(SubmissionStatus.COMPLETE)){
            List<Task<?>> results= resultCollector.awaitResults(1000L);
            if(results != null){
               for(Task<?> task : results) {
                  Integer pos = task.getPosition();
                  if(! orderedResults.containsKey(pos)){
                     orderedResults.put(pos, task);
                     if(task.getThrowable() != null)    System.out.println(task.getId()+" is broken");
                     else if(task.getResult() != null)   System.out.println(task.getId()+" has result");
                     else                  System.out.println(task.getId()+" has no result");
                  }
               }
            }
            System.out.println(resultCollector.getStatus());
         }

After submiting a job, the client waits on the results.
In my experiments all results from a node arrive the client together. Because the tasks of the job are well balanced on the nodes, all results are coming aprox. on the same time. In consequence the client waits some times on the results and then he must archive all the results on the same time. I want a better arrangement of the client activity on the whole time.

For this I was thinking about:
- Submiting a lot of jobs with only one tasks. I think this makes for JPPF no sence.
- The task writes the results back via a special data channel, like ssh, nfs ore a database access. This may be problematic when I use JPPF in a grid and outside a cluster, because I need access on special ports beside the JPPF protocol.
- Using jmx to send notifications with the results, when a task is finished. When the results are big in MB, then I get a performance problem.

Laurent, what is your suggestion to solve this problem?

Thanks

Christian

[lolo]

Hi Christian,

Sorry for this late answer.

In fact, in most cases the results of a job are received by the client almost as soon as they are returned by each node. JPPFResultCollector.awaitResults(timeout) gives a different impression because it is a synchronization point which will return either all results, or no result at all if not all results were received yet.

You can catch these partial results in 2 ways:

1) By overriding the JPPFResultCollector.resultsReceived() method:

Code: [Select]

JPPFClient jppfClient = new JPPFClient();
JPPFJob job = new JPPFJob();
JPPFResultCollector collector = new JPPFResultCollector(job) {
  @Override public synchronized void resultsReceived(TaskResultEvent event) {
    super.resultsReceived(event);
    // Get the partial results from a single node
    List<Task<?>> myResults = event.getTasks();
    // ... work with the results
  }
};
job.setResultListener(collector);
jppfClient.submitJob(job);
Here, the job is still submitted synchronously, but partial results are received asynchronously in resultsReceived().

2) The recommended approach is to use a JobListener:

Code: [Select]

JobListener jobListener = new JobListenerAdapter() {
  @Override public void jobReturned(JobEvent event) {
    // synchronize over potentially parallel calls, e.g. when the client has multiple connections to the server
    synchronized(event.getJob()) {
      // get the partial results
      List<Task<?>> myResults = event.getJobTasks();
      // ... work with the results
    }
  }
};
job.addJobListener(jobListener);
jppfClient.submitJob(job);
This is the preferred approach because the TaskResultListener interface, which JPPFResultCollector implements, is heavily used for internal processing by the JPPF client. Furthermore, JobListener.jobReturned() is called every time TaskResultListener.resultsReceived() is called, so no result will be missed. You just need to ensure proper synchronization, since jobReturned() may be called by multiple concurrent threads.

Additionally, you need to be aware that the dispatching of the tasks in a job is governed by the load-balancing:

- the client-side load-balancer will distribute the tasks in a job to the available execution channels, where a channel is either a connection to a remote server, or the local executor. For instance, the default load-balancer in the client uses the "manual" algorithm, which sends a fixed number of tasks to each channel, with a fixed size of 1,000,000. Since most jobs have less than 1 million tasks, this means all tasks in a job will be sent at once to the server over a single connection/channel. If you change the setting to have a fixed size of 5, then at most 5 tasks will be sent at once over each available channel. An important thing to note is that, when the tasks of a job are sent over multiple remote channels in parallel, they will be aggregated back in the server queue.

- the server-side load-balancer determines how many tasks are sent to each node connection. When a node returns its results back to the server, these results are immediately returned via the client channel that submitted the initial tasks. This may result, on the client side, in mutliple calls to jobReturned() for the same job, if the tasks were sent via multiple channels in parallel. The default server load-balancer uses the "proportional" algorithm, which is an adaptive algorithm that provides an even-balanced task dispatching in most cases. Furthermore, the server load-balancer can be changed dynamically using the related management APIs, or the admin console.

All this to say that you have a lot of flexibility in how you can receive the results of submitted jobs, and I would consider an out-of-the-box approach before going with a custom design.

Sincerely,
-Laurent

[cm]

Hi Laurent,

thanks for your answer. Both versions are working fine.
A main point was to activate in the server config file :
jppf.load.balancing.algorithm = manual
jppf.load.balancing.profile = manual_profile
jppf.load.balancing.profile.manual_profile.size = 1

Thanks again
Christian