Connection pools

All server connections in a JPPF client are organized into connection pools, whose number is determined by the client configuration properties, and whose size is based on the configuration as well and can also be changed dynamically via the JPPF APIs. In the next sections, we will see how connection pools can be configured, explored and programmatically accessed.

1 The JPPFConnectionPool class

A connection pool is represented by the JPPFConnectionPool class, defined as follows:

public class JPPFConnectionPool extends AbstractConnectionPool<JPPFClientConnection>
  implements Comparable<JPPFConnectionPool> {
  // Get the name of this pool
  public String getName()
  // Get the id of this pool, unique within a JPPFClient instance
  public int getId()
  // Get the priority associated with this pool
  public int getPriority()
  // Check whether this pool is for SSL connections
  public boolean isSslEnabled()
  // Get the uuid of the driver to which connections in this pool are connected
  public String getDriverUuid()
  // Get the host name or IP address of the remote driver
  public String getDriverHost()
  // Get the port number to which the client connects on the remote driver
  public int getDriverPort()
  // Set the size of this pool, starting or stopping connections as needed
  public int setSize(int size)
  // Get a list of connections in this pool whose status is one of those specified
  public List<JPPFClientConnection> getConnections(JPPFClientConnectionStatus...statuses)
  // Get the number of connections in this pool whose status is one of those specified
  public int connectionCount(JPPFClientConnectionStatus...statuses)
}

As we can see, JPPFConnectionPool extends the class AbstractConnectionPool, which in turn implements the interface ConnectionPool, defined as follows:

public interface ConnectionPool<E extends AutoCloseable>
  extends Iterable<E>, AutoCloseable {
  // Get the next connection that is connected and available
  E getConnection();
  // Determine whether this pool is empty
  boolean isEmpty();
  // Get the current size of this pool
  int connectionCount();
  // Get the size of this connection pool
  int getSize();
  // Set the size of this pool, starting or stopping connections as needed
  int setSize(int size);
  // Get a list of connections held by this pool
  List<E> getConnections();
}

The pool name is based on the client configuration properties and defined as follows:

With server discovery disabled:

jppf.discovery.enabled = false
jppf.drivers = <driver_name_1> ... <driver_name_N>

Each driver name specified in “jppf.drivers” will be the name of the corresponding connection pool.

With discovery enabled: for each discovered server, the JPPF client will create a connection pool named “jppf_discovery-n” where n is a sequence number automatically assigned by the client.

The id attribute of a pool is a sequence number that is guaranteed to be unique within a JPPF client. It is used to distinguish pools that may have the same driver uuid, priority and size. The name may be used similarly, however JPPF does not do any checking on pool names, so they should be used with caution.

The pool's priority is as defined in the configuration. For instance if we have the following:

jppf.drivers = myPool
myPool.jppf.server.host = www.myhost.com
myPool.jppf.server.port = 11111
myPool.jppf.priority = 10
# pool size
myPool.jppf.pool.size = 5

The corresponding JPPFConnectionPool object's getPriority() method will return 10.

The pool's actual size can be grown or shrunk dynamically, using the setSize(int) method. The JPPF client will create or close connections accordingly. An attempt to set a size equal to the current size will have no effect whatsoever.

In some cases, when trying to reduce the connection pool's max size, there may be too many connections in the pool busy executing jobs and the client will not be able to close all the requested connections. In this case, setSize() will return the new actual size, which will be smaller than the requested size.

The pool size can be defined in the configuration, either with the <poolName>.jppf.pool.size property for manually configured pools, or with jppf.pool.size for auto-discovered pools.

The two getConnections() methods allow you to explore the connections currently in the pool. The overloaded version of this method permits filtering of the connections by their status, represented by one or more JPPFClientConnectionStatus enum values.

2 Client connections

2.1 The JPPFClientConnection interface

As we have seen, connection pools contain and manage a set of connections from a client to a driver. These connections are represented by the JPPFClientConnection interface, defined as follows:

public interface JPPFClientConnection
       extends ClientConnectionStatusHandler, AutoCloseable {
  // Get the priority assigned to this connection
  int getPriority();
  // Shutdown this connection and release all the resources it is using
  void close();
  // Determine whether this connection was closed
  boolean isClosed();
  // Get the name assigned to this client connection
  String getName();
  // Determines if this connection is over SSL
  boolean isSSLEnabled();
  // Get the driver's host name or ip address
  String getHost();
  // Get the port number on which the dirver is listeneing for connections
  int getPort();
  // Get the unique identifier of the remote driver
  String getDriverUuid();
  // Get the system information for the remote driver this connection refers to
  JPPFSystemInformation getSystemInfo();
  // Get the unique ID for this connection and its two channels
  String getConnectionUuid();
  // Get the pool this connection belongs to
  JPPFConnectionPool getConnectionPool();
}

Note that most of these methods, except for close(), isClosed(), getConnectionUuid() and getName(), actually delegate to the JPPFConnectionPool to which the connection belongs.

2.2 Status notifications for existing connections

Each server connection has a status that depends on the state of its network connection to the server and whether it is executing a job request. A connection status is represented by the enum JPPFClientConnectionStatus, and has the following possible values: NEW, DISCONNECTED, CONNECTING, ACTIVE, EXECUTING, FAILED or CLOSED.

JPPFClientConnection extends the interface ClientConnectionStatusHandler, which provides the following methods to handle the connection status and register or remove listeners:

public interface ClientConnectionStatusHandler {
  // Get the status of this connection
  JPPFClientConnectionStatus getStatus();
  // Set the status of this connection
  void setStatus(JPPFClientConnectionStatus status);
  // Register a connection status listener with this connection
  void addClientConnectionStatusListener(ClientConnectionStatusListener listener);
  // Remove a connection status listener from the registered listeners
  void removeClientConnectionStatusListener(ClientConnectionStatusListener listener);
}

Here is a sample status listener implementation:

public class MyStatusListener extends ClientConnectionStatusListener {
  @Override
  public void statusChanged(ClientConnectionStatusEvent event) {
    // obtain the client connection from the event
    JPPFClientConnection connection =
      (JPPFClientConnection) event.getClientConnectionStatusHandler();
    // get the new status
    JPPFClientConnectionStatus status = connection.getStatus();
    System.out.println("Connection " + connection.getName() + " status changed to "
      + status);
  }
}

3 Exploring the connections in a pool

JPPFConnectionPool also provides a set of methods to wait for if needed, and obtain, a set of connections that satisfy a particular condition:

public class JPPFConnectionPool extends AbstractConnectionPool<JPPFClientConnection>
  implements Comparable<JPPFConnectionPool> {

  // Wait for the specified number of connections to be in the ACTIVE status
  public List<JPPFClientConnection> awaitActiveConnections(
    Operator operator, int nbConnections)

  // Wait for the specified number of connections to be in ACTIVE or EXECUTING} status
  public List<JPPFClientConnection> awaitWorkingConnections(
    Operator operator, int nbConnections)

  // Wait for the specified number of connections to be in one of the specified statuses
  public List<JPPFClientConnection> awaitConnections(
    Operator operator, int nbConnections, JPPFClientConnectionStatus...statuses)

  // Wait for the specified number of connections to be in one of the specified statuses
  // or the specified timeout to expire, whichever happens first
  public List<JPPFClientConnection> awaitConnections(Operator operator,
    int nbConnections, long timeout, JPPFClientConnectionStatus...statuses)
}

Examples:

JPPFConnectionPool pool = ...;
// wait until more than 2 connections in the pool are active
List<JPPFClientConnection> list = pool.awaitActiveConnections(Operator.GREATER, 2);
// wait for no more than 5 seconds or until at least one connection is closed
List<JPPFClientConnection> list2 = pool.awaitConnections(
  Operator.AT_LEAST, 1, 5000L, JPPFClientConnectionStatus.CLOSED);

The Operator parameter defines the possible conditions that the number of connections must satisfy:

public enum Operator {
  // The number of connections is equal to the expected number
  EQUAL,
  // The number of connections is different from the expected number
  NOT_EQUAL,
  // The number of connections is at least to the expected number
  AT_LEAST,
   // The number of connections is at most the expected number
  AT_MOST,
  // The number of connections is strictly greater than the expected number
  GREATER,
  // The number of connections is strictly less than the expected number
  LESS
}

4 Associated JMX connection pool

Each connection pool has an associated pool of JMX connections to the same remote driver.. To access and manipulate this JMX pool, the JPPFConnectionPool class provides the following API:

public class JPPFConnectionPool extends AbstractConnectionPool<JPPFClientConnection>
  implements Comparable<JPPFConnectionPool> {
  // Get a connected JMX connection among those in the JMX pool
  public JMXDriverConnectionWrapper getJmxConnection()
  // Get a JMX connection in the specified state from the JMX pool
  public JMXDriverConnectionWrapper getJmxConnection(boolean connectedOnly)
  // Get the jmx port to use on the remote driver
  public int getJmxPort()
  // Get a connected JMX connection among those in the JMX pool
  public JMXDriverConnectionWrapper getJmxConnection()
  // Get a JMX connection with the specified state among those in the JMX pool
  public JMXDriverConnectionWrapper getJmxConnection(final boolean connectedOnly)
  // Get the core size of the associated JMX connection pool
  public int getJMXPoolCoreSize()
  // Get the current maximum size of the associated JMX connection pool
  public int getJMXPoolMaxSize()
  // Set a new maximum size for the associated pool of JMX connections,
  // adding new or closing existing connections as needed
  public int setJMXPoolMaxSize(int maxSize)
  // Get the list of connections currently in the JMX pool
  public List<JMXDriverConnectionWrapper> getJMXConnections()
  // Wait for the specified number of JMX connections to be in the specified state
  public List<JMXDriverConnectionWrapper> awaitJMXConnections(
    Operator operator, int nbConnections, final boolean connectedOnly)
  // Wait for the specified number of JMX connections to be in the specified state,
  // or the specified timeout to expire, whichever happens first
  public List<JMXDriverConnectionWrapper> awaitJMXConnections(
    Operator operator, int nbConnections, long timeout, boolean connectedOnly)
}

Note that the JMX pool core size, when left unspecified, defaults to 1. Otherwise, it is defined in the configuration as:

When discovery is enabled:

jppf.jmx.pool.size = 5

When discovery is disabled:

driver-1.jppf.jmx.pool.size = 5

Also note that the driver host for a JMX connection is the same as JPPFConnectionPool.getDriverHost(). In the same way, to determine whether a JMX connection is secure, JPPFConnectionPool.isSSLEnabled() should be used.

5 Exploring the connection pools

The JPPFClient class, or more exactly its super-super class AbstractJPPFClient, provides a number of methods to discover and explore the connection pools currently handled by the client:

public class JPPFClient extends AbstractGenericClient { ... }

public abstract class AbstractGenericClient extends AbstractJPPFClient { ... }

public abstract class AbstractJPPFClient
  implements ClientConnectionStatusListener, AutoCloseable {
  // Find the connection pool with the specified priority and id
  public JPPFConnectionPool findConnectionPool(int priority, int poolId)
  // Find the connection pool with the specified id
  public JPPFConnectionPool findConnectionPool(int poolId)
  // Find the connection pool with the specified name
  public JPPFConnectionPool findConnectionPool(String name)
  // Find the connection pools whose name matches the specified regular expression
  public List<JPPFConnectionPool> findConnectionPools(String name)
  // Find the connection pools that have at least one connection matching
  // one of the specified statuses
  public List<JPPFConnectionPool> findConnectionPools(
    JPPFClientConnectionStatus...statuses)
  // Get a set of existing connection pools with the specified priority
  public List<JPPFConnectionPool> getConnectionPools(int priority)
  // Get a list of all priorities for the currently existing pools in descending order
  public List<Integer> getPoolPriorities()
  // Get a list of existing connection pools, ordered by descending priority
  public List<JPPFConnectionPool> getConnectionPools()
  // Get a pool with the highest possible priority that has at least 1 active connection
  public JPPFConnectionPool getConnectionPool()
  // Get the connection pools that pass the specified filter
  public List<JPPFConnectionPool> findConnectionPools(
    ConnectionPoolFilter<JPPFConnectionPool> filter)
}

Note that the connection pools are held in a multimap-like data structure, where the key is the pool priority sorted in descending order (highest priority first). Consequently, all getXXX() and findXXX() methods which return a list of connection pools are guaranteed to have the resulting elements od the list sorted by order of descending priority.

The last findConnectionPools() method provides a generic way of filtering the existing connection pools, by making use of a ConnectionPoolFilter, defined as follows:

public interface ConnectionPoolFilter<E extends ConnectionPool> {
  // Determine whether this filter accepts the specified connection pool
  boolean accepts(E pool);
}

In addition to this, JPPFClient provides a set of methods which wait until one or more conenction pools fulfill a specified condition, and return a list of pools which satisfy the condition:

public class JPPFClient extends AbstractGenericClient
  // Wait for at least one connection pool with at least one connection in ACTIVE status
  public JPPFConnectionPool awaitActiveConnectionPool()
  // Wait for at least one connection pool with at least one connection
  // in ACTIVE or EXECUTING status
  public JPPFConnectionPool awaitWorkingConnectionPool()
  // Wait for at least one connection pool with at least one connection
  // in one of the specified statuses
  public JPPFConnectionPool awaitConnectionPool(JPPFClientConnectionStatus...statuses)
  // Wait for at least one connection pool with at least one connection in one of the
  // specified statuses or the specified timeout (in ms) expires, whichever happens first
  public JPPFConnectionPool awaitConnectionPool(
    long timeout, JPPFClientConnectionStatus...statuses)
  // Wait for at least one connection pool with at least one connection in one of the
  // specified statuses or the specified timeout (in ms) expires, whichever happens first
  public List<JPPFConnectionPool> awaitConnectionPools(
    long timeout, JPPFClientConnectionStatus...statuses)
}

6 Connection Pool Events

The JPPF client API allows the registration or unregistration of listeners to connection pool events: connection pools added to or removed from the client, or connections added to or removed from a connection pool. This can be done in two ways:

1) From a JPPFClient constructor:

public class JPPFClient extends AbstractGenericClient {

  // Initialize this client with an automatically generated application UUID
  public JPPFClient(final ConnectionPoolListener... listeners)

  // Initialize this client with the specified UUID and listeners
  public JPPFClient(final String uuid, final ConnectionPoolListener... listeners) {
}

2) Using the related add and remove methods in the grand parent of JPPFClient:, AbstractJPPFClient:

public abstract class AbstractJPPFClient
    implements ClientConnectionStatusListener, AutoCloseable {

  // Add a listener to the list of listeners to this client
  public void addConnectionPoolListener(final ConnectionPoolListener listener)

  // Remove a listener from the list of listeners to this client
  public void removeConnectionPoolListener(final ConnectionPoolListener listener)
}

As we can see in the methods signatures, the listeners implement the interface ConnectionPoolListener, is defined as:

public interface ConnectionPoolListener extends EventListener {

  // Called when a new connection pool is created
  void connectionPoolAdded(ConnectionPoolEvent event);

  // Called when a connection pool removed
  void connectionPoolRemoved(ConnectionPoolEvent event);

  // Called when a new connection is created
  void connectionAdded(ConnectionPoolEvent event);

  // Called when a connection pool is removed
  void connectionRemoved(ConnectionPoolEvent event);
}

Note that, if you do not wish to implement all the methods in ConnectionPoolListener, you can instead extend the adapter class ConnectionPoolListenerAdapter, which implments each method of the interface as an empty method.

All notification methods receive an event of type ConnectionPoolEvent, defined as follows:

public class ConnectionPoolEvent extends EventObject {

  // Get the source of this event
  public JPPFConnectionPool getConnectionPool()

  // Get the connection that triggered this event, if any
  public JPPFClientConnection getConnection()
}

Please note that, in the case of a connectionPoolAdded() or connectionPoolAdded() notification, the method getConnection() will return nulll, since no connection is involved.

7 Putting it all together

We will illustrate how client, connection pool events and connection events fit together, with an example which prints the status of all connections created by the client:

// this status listener prints the old and new status of each connection
final ClientConnectionStatusListener myStatusListener =
    new ClientConnectionStatusListener {
  @Override
  public void statusChanged(ClientConnectionStatusEvent event) {
    // obtain the client connection from the event
    JPPFClientConnection connection =
      (JPPFClientConnection) event.getClientConnectionStatusHandler();
    // get the new and old status
    JPPFClientConnectionStatus newStatus = connection.getStatus();
    JPPFClientConnectionStatus oldStatus = event.getOldStatus();
    // print the connection name, old status and new status
    System.out.println("Connection " + connection.getName() + " status changed from "
      + oldStatus + " to " + newStatus);
  }
}

// create a connection pool listener which registers
// a status listener on new connections
ConnectionPoolListener myPoolListener = new ConnectionPoolListenerAdapter() {
  @Override
  public void connectionAdded(final ConnectionPoolEvent event) {
    // obtain the connection pool and client connection from the event
    JPPFConnectionPool pool = event.getConnectionPool();
    JPPFClientConnection connection = event.getConnection();
    System.out.println("connection " + connection + " added to pool " + pool);
    // add the status listener to the connection
    connection.addClientConnectionStatusListener(myListener);
  }

  @Override
  public void connectionRemoved(final ConnectionPoolEvent event) {
    // obtain the connection pool and client connection from the event
    JPPFConnectionPool pool = event.getConnectionPool();
    JPPFClientConnection connection = event.getConnection();
    System.out.println("connection " + connection + " removed from pool " + pool);
    // remove the status listener from the connection
    connection.removeClientConnectionStatusListener(myListener);
  }
}

// create a JPPFClient with our connection pool listener
JPPFClient client = new JPPFClient(myPoolListener);