Print

[rzo]

Hi,

I am using JPPF in Version 5.0.3. to scale an application to a few virtual machines.

My setup is just the following:

Driver 1: 7 Nodes
Driver 2: 7 Nodes
Driver 3: 7 Nodes

The drivers are configured to work as peers. Each node executes one task, which will take a lot of time until it completes.
For this reason I've registred a ClientTaskListener on my executing client assuming, that it will catch all notifications from every node.

However it seems, that only the notifications from Driver 1 are retrieved on my client.

The relevant piece of code should be this one:

Code: [Select]

    try (JPPFClient jppfClient = new JPPFClient()) {
            JMXDriverConnectionWrapper jmxDriver = waitForJMXConnection(jppfClient);

            String listenerID = registerJMXListener(jmxDriver);

            executeNonBlockingJob(jppfClient);

            jmxDriver.unregisterForwardingNotificationListener(listenerID);

        } catch (Exception e) {
            throw new RuntimeException("Starting failed", e);
        }
    }


    private String registerJMXListener(JMXDriverConnectionWrapper jmxDriver) throws Exception {
        // this selector selects all nodes attached to the driver
        NodeSelector selector = new AllNodesSelector();
        // create a JMX notification listener
        NotificationListener clientTaskListener = new ClientTaskListener(clientTaskListenerHandler);
        // register the notification listener with the JPPFNodeTaskMonitorMBean
        // on the selected nodes
        return jmxDriver.registerForwardingNotificationListener(
                selector, JPPFNodeTaskMonitorMBean.MBEAN_NAME, clientTaskListener, null, null);
    }

    private JMXDriverConnectionWrapper waitForJMXConnection(JPPFClient jppfClient) throws InterruptedException {
        // wait until a JMX connection to the driver is established
        while (!jppfClient.hasAvailableConnection()) {
            sleep();
        }

        JMXDriverConnectionWrapper jmxDriver;
        while ((jmxDriver = jppfClient.getClientConnection().getConnectionPool().getJmxConnection()) == null) {
            sleep();
        }


        while (!jmxDriver.isConnected()) {
            sleep();
        }
        return jmxDriver;
    }

    private void sleep() throws InterruptedException {
        Thread.sleep(10L);
    }

How can I achieve, that notifications from all nodes (even if they are connected via Driver 2 or Driver 3) are received by my client?


//EDIT: Currently i manually connect to each driver to register my client task listener. Is this the intended way or is there any possibility to do this automatically?
Thanks in advance,
rzo

[lolo]

Hello,

The easiest way to monitor a multi-driver topology is to use the dedicated grid topology monitoring API. The idea is to build a TopologyManager based on your JPPF client, and register a TopologyListener that will itself register a ClientTaskListener with each new driver detected in the topology. The code would look like this:

Code: [Select]

public static void main(String... args) {
  try (JPPFClient client = new JPPFClient()) {
    // a map of drivers to registered listener id
    final Map<TopologyDriver, String> listenerMap = new ConcurrentHashMap<>();
    final ClientTaskListener clientTaskListener = new ClientTaskListener();

    // register a topology event listener that registers a notification listener for each new driver
    TopologyManager manager = new TopologyManager(client, new TopologyListenerAdapter() {
      @Override
      public void driverAdded(TopologyEvent event) {
        // get the new driver and its jmx connection
        TopologyDriver driver = event.getDriver();
        JMXDriverConnectionWrapper jmxDriver = driver.getJmx();
        try {
          String listenerId = jmxDriver.registerForwardingNotificationListener(
            // we use the driver as handback parameter, so we can identify from which driver notifications are sent
            NodeSelector.ALL_NODES, JPPFNodeTaskMonitorMBean.MBEAN_NAME, clientTaskListener, null, driver);
          // store the listener id, so we can unregister it later
          listenerMap.put(driver, listenerId);
        } catch(Exception e) {
          e.printStackTrace();
        }
      }
    });

    // execute a job
    executeNonBlockingJob(client);

    // unregister all listeners
    for (Map.Entry<TopologyDriver, String> entry: listenerMap.entrySet()) {
      JMXDriverConnectionWrapper jmxDriver = entry.getKey().getJmx();
      jmxDriver.unregisterForwardingNotificationListener(entry.getValue());
    }
  } catch (Exception e) {
    e.printStackTrace();
  }
}

public static class ClientTaskListener implements NotificationListener {
  @Override
  public void handleNotification(Notification notification, Object handback) {
    // handback parameter identifies the driver from which the notification is issued
    TopologyDriver driver = (TopologyDriver) handback;
    // ... process the notification
  }
}
As you can see, this code has many similarities with the code you posted, the main difference being that it will register a listener for each and every driver discovered by the client.

Another approach could be to use the connection pools API, however it will be less easy, especially with JPPF 5.0.x which doesn't have the connection pool events (this feature came in v5.1).

Sincerely,
-Laurent

[rzo]

Hi,

thanks for your detailed answer regarding the grid topology monitoring AP. It seems, that it will do exactly the thing, which I need

Will go for your suggested approach. Thanks!

Sincerly,
rzo