Print

[braveht]

Hi,

Our job requires environmental setting on each host where runs nodes. So we plan to run a "pre-job", which checks environment setting of all hosts, before launching the real job. The "pre-job" is supposed to have multiple tasks, each checks environment setting on one host.

I know JPPF is equipped with execution policies, which limit tasks to execute on some nodes. Are execution policies good for our use? If not, which approach is recommended?

Thanks,
braveht

[lolo]

Hello braveht,

Execution policies apply to the entire job only, so it will not be a good fit for what you want to achieve. I can see several alternatives:

1) the simplest way would be to make the pre-job a broadcast job. It would contain a single task that will be executed (and thus duplicated) on all the nodes (unless you specifiy an execution policy). This will work if the task to execute only depends on the node's environment.

2) another possiblity could be to implement a node life cycle listener, and in its jobStarting() notification perform the setup you need based on job information such as its uuid, name or metadata.

3) yet another way would be to implement a pluggable node MBean, which you could call remotely from the client using the JMX forwarding mechanism, with at least one MBean method that would perform the required setup.

I hope this helps.

Sincerely,
-Laurent

[braveht]

Thanks for your response.
I tried broadcast job. But since broadcast job does not return any value or exception, I can't get check results.

[lolo]

A workaround is to have the task send a jmx notification to provide feedback on its execution, and register a notification listener on the client side to handle the notifications from the tasks executed on all the nodes.

For instance, the task sent in the broadcast job would look lie this:

Code: [Select]

public class BroadcastTask extends AbstractTask<Object> {
  @Override
  public void run() {
    // the execution result sent as a notification user object
    Map<String, Object> result = new HashMap<>();
    try {
      // do whatever the task is supposed to do here
      // ...

      // it could be a more complex data structure instead, such as a Map
      result.put("success", true);
    } catch (Exception e) {
      result.put("success", false);
      // add the exception so the notification listener knows what happened
      result.put("exception", e);
    }
    // first param is a user object, here our result Map
    fireNotification(result, true);   
  }
}
And the code on the client side would look like this:

Code: [Select]

public class BroadcastRunner {
  public static void main(final String[] args) {
    try (JPPFClient client = new JPPFClient()) {
      JMXDriverConnectionWrapper jmx = client.awaitWorkingConnectionPool()
        .awaitJMXConnections(Operator.AT_LEAST, 1, true).get(0);
      final Map<String, Boolean> responseMap = new ConcurrentHashMap<>();
      // register a notification listener on all nodes
      NotificationListener listener = new NotificationListener() {
        @Override
        public void handleNotification(final Notification notification, final Object handback) {
          JPPFNodeForwardingNotification notif = (JPPFNodeForwardingNotification) notification;
          TaskExecutionNotification taskNotif = (TaskExecutionNotification) notif.getNotification();
          if (taskNotif.isUserNotification()) {
            @SuppressWarnings("unchecked")
            Map<String, Object> result = (Map<String, Object>) taskNotif.getUserData();
            boolean success = (Boolean) result.get("success");
            responseMap.put(notif.getNodeUuid(), success);
            if (!success) {
              Exception e = (Exception) result.get("exception");
              System.out.printf("Exception occurred in node '%s' : %s%n", notif.getNodeUuid(), ExceptionUtils.getStackTrace(e));
            }
          }
        }
      };
      String listeneerID = jmx.registerForwardingNotificationListener(NodeSelector.ALL_NODES, JPPFNodeTaskMonitorMBean.MBEAN_NAME, listener, null, null);
      JPPFJob job = new JPPFJob();
      job.setName("my broadcast job");
      job.getSLA().setBroadcastJob(true);
      job.add(new BroadcastTask()).setId("broadcast task id");
      client.submitJob(job);
      // count of nodes is our exit condition
      int nbNodes = jmx.nbNodes();
      // some notifications may arrive in the client after the job is done
      while (responseMap.size() < nbNodes) Thread.sleep(10L);
      // handle the results
      int successCount = 0, errorCount = 0;
      for (Map.Entry<String, Boolean> entry: responseMap.entrySet()) {
        boolean success = entry.getValue();
        System.out.printf("Execution on node '%s' is a %s%n", entry.getKey(), (success ? "success" : "failure"));
        if (success) successCount++;
        else errorCount++;
      }
      System.out.printf("got tresponses from all nodes: success count = %d, error count = %d%n", successCount, errorCount);
    } catch (Exception e) {
      e.printStackTrace();
    }
  }
}
Sincerely,
-Laurent