Custom load-balancer state persistence

Many of the existing load-balancing algorithms are adaptive and go through a convergence phase before reaching a state of optimal effciency. Unfortunately, this state disappears whenever a node, driver or client involved in load-balancing is stopped.

The load-balancer persistence allows storing the state of load-balancer instances and restoring them whenever a node reconnects to a driver (server-side load-balancing) or a client reconnects to a driver (client-side load-balancing).

1 Identification of persisted load-balancer states

Persistence and restoration of a load-balancer state requires an identifier that is both unique and repeatable accross reconnection and restart of the peers involved (node + driver or driver + client). Using combinations of the components' UUIDs will not work here, since UUIDs only exist as long as each component's JVM is alive and are generated anew at each restart.

Instead, JPPF computes a resilient identifier which includes the IP address of both parties, the related driver's port and other properties found in the JPPF component's configuration. A load-balancer state is further idenfied by the name of the of its algorithm. For normalization purposes, both parts of the identfier are transformed using a hash function. This function defaults to SHA-1 and can be changed using the following configuration property:

jppf.load.balancing.persistence.hash = SHA-1

2 PersistentState interface

Any load-balancer implementation that wishes to have its state persisted must implement the PersistentState interface, defined as follows:

public interface PersistentState {
 // Get this load-balancer's state
  Object getState();

 // Set this load-balancer's state
  void setState(Object state);

 // Set this load-balancer's state
  Lock getStateLock();
}

Any Bundler implementation which also implements this interface will see its state, as provided by the getState() method, persisted when load-balancer peristence is enabled. The setState() method is be called by a JPPF driver or client upon the load-balancer's initialization, when the load-balancer's state is restored from the persistence store.

State synchronization:

Care must be taken to synchronize access to the bundler's state. The persistence facility will first serialize it, before storing it. Keep in mind that the serialization process will traverse any data structure owned by the bundler's state. At the same time, especially when the persistence is asynchronous, the bundler may still receive updates from another thread via its feedback() method and modify its state.

This means that bundler persistence introduces a potential race condition. When the bundler state uses collections, this will typically lead to ConcurrentModificationException errors and prevent the persistence from working, or even corrupt it.

To avoid this, both the load-balancer and the persistence implementation must synchronize on the state. This the very reason why PersistentState has a getStateLock() method, to provide a lock that both load-balancer and persistence can use for safe access tot he state.

For instance, all the built-in JPPF load-balancers that support persistence implement a pattern similar to the one below:

public class MyBundler extends AbstractAdaptiveBundler<MyProfile> implements PersistentState {
  // The state of this bundler
  private final MyBundlerState state;
  //The laod-balancer state lock
  private final Lock lock = new ReentrantLock();

  public MyBundler(MyProfile profile) {
    super(profile);
    state = new MyBundlerState();
  }

  @Override public void feedback(fint size, double time) {
    lock.lock();    
    try {
      // ... compute the new bundle size ...
    } finally {
      lock.unlock()
    }
  }

  @Override public Object getState() {
    lock.lock();    
    try {
      return state;
    } finally {
      lock.unlock()
    }
  }

  @Override public void setState(Object o) {
    BundlerState other = (BundlerState) o;
    lock.lock();    
    try {
      state.bundleSize = other.bundleSize;
      state.performanceCache = other.performanceCache;
    } finally {
      lock.unlock()
    }
  }

  @Override public Lock getStateLock() {
    return lock;
  }

  // Holds the state of this bundler for persistence
  private static class BundlerState implements Serializable {
    private int bundleSize = 1;
    private PerformanceCache performanceCache = new PerformanceCache();
  }
}

3 LoadBalancerPersistence interface

A custom load-balancer persistence must implement the LoadBalancerPersistence interface, defined a s follows:

public interface LoadBalancerPersistence {
  // Load the state of a load balancer from the persistence sstore
  Object load(LoadBalancerPersistenceInfo info) throws LoadBalancerPersistenceException;

  // Store a load balancer state to the persistence sstore
  void store(LoadBalancerPersistenceInfo info) throws LoadBalancerPersistenceException;

  // Delete the specified load-balancer state(s) from the persistence store
  void delete(LoadBalancerPersistenceInfo info) throws LoadBalancerPersistenceException;

  // Retrieve the specifiedchannel or algorithm IDs from the persistence store
  List<String> list(LoadBalancerPersistenceInfo info) throws LoadBalancerPersistenceException;
}

All the methods of this interface are called by the JPPF load-balancer persistence facility, during the load-balancers life cycle, but also during user-initiated management requests. As we also can see, all the methods take a single parameter of type LoadBalancerPersistenceInfo, which identifies the artifact(s) to store, retrieve or delete. It is defined as follows:

public class LoadBalancerPersistenceInfo {
  // Get a readable identifier for the channel, resilient over
  // restarts of the processes involved in the load-balancing
  public String getChannelString()
  // Get the channel identifier, a hash of the string returned by getChannelString()
  public String getChannelID()
  // Get thereadablename of the related load-balancing algorithm
  public String getAlgorithm()
  // Get a hash of the load-balancing algorithm name obtained via getAlgorithm()
  public String getAlgorithmID()
  // Get the load-balancer state
  public Object getState()
  // Get a lock used to synchronize access to the state
  public Lock getStateLock()
  // Serialize the state into an array of bytes
  public byte[] getStateAsBytes() throws Exception
  // Serialize the state into an output stream
  public void serializeToStream(final OutputStream stream) throws Exception
}

The methods getChannelString() and getAlgorithm() provide human-readable versions of the identfiers for the channel and algorithm. They are here for logging and debugging purposes only and it is not recommended to use them to actually identify load-balancer states in the persistence store.

The getStateLock() method returns the same Lock object as the one provided by the getStateLock() method of PersistentState. In principle, it should only by used in the store() and load() methods of LoadBalancerPersistence, and it will return null when passed to the list() and delete() methods of LoadBalancerPersistence.

The getStateAsBytes() and serializeToStream() methods are convenience methods that serialize the load-balancer state to a byte array or a stream, respectively. They also abstract two characteristics of the serialization:

• internally, they synchronize on the Lock object provided by getStateLock()
• they also perform the serialization according to the serialization scheme configured in JPPF

Back to the LoadBalancerPersistence interface, the specification for the list() and delete() methods is tightly coupled to the content of their LoadBalancerPersistenceInfo parameter.

For list(LoadBalancerPersistenceInfo info):

• if info is null or both info.getChannelID() and info.getAlgorithmID() are null, then all channel IDs in the persistence store are returned
• if only info.getAlgorithmID() is null, then all the algorithm IDs for the specified channel are returned
• if only info.getChannelID() is null, then the the IDs of the channels that have a persisted state for the algorithm are returned
• if neither info.getChannelID() nor info.getAlgorithmID() are null, then the specified algorithm ID is returned (list with a single entry) if the channel has an entry for it, otherwise an empty list must be returned

For delete(LoadBalancerPersistenceInfo info):

• if info is null or both info.getChannelID() and info.getAlgorithmID() are null, then all entries in the persistence store are deleted
• if only info.getAlgorithmID() is null, then the states of all algorithm for the specified channel are deleted
• if only info.getChannelID() is null, then the states of the specified algorithm are deleted for all the channels
• if neither info.getChannelID() nor info.getAlgorithmID() are null, then only the state of the specified algorithm for the channel is deleted

Typically, a load-balancer persistence implementation will have the following structure:

public class MyPersistence implements LoadBalancerPersistence {
  public MyPersistence(String...params) {
    // process the parameters
  }

  @Override public Object load(LoadBalancerPersistenceInfo info)
    throws LoadBalancerPersistenceException {
    return ...;
  }

  @Override public void store(LoadBalancerPersistenceInfo info)
    throws LoadBalancerPersistenceException {
    // store the load-balancer state
  }

  @Override public void delete(LoadBalancerPersistenceInfo info)
    throws LoadBalancerPersistenceException {
    if ((info == null) ||
      ((info.getChannelID() == null) && (info.getAlgorithmID() == null))) {
      // delete all entries in the store
    } else if (info.getAlgorithmID() == null) {
      // delete all entries for the channel with info.getChannelID()
    } else if (info.getChannelID() == null) {
      // delete all info.getAlgorithmID() entries for the channels that have it
    } else {
      // delete the [info.getChannelID(), info.getAlgorithmID()] entry
    }
  }

  @Override public List<String> list(LoadBalancerPersistenceInfo info)
    throws LoadBalancerPersistenceException {
    List<String> results = new ArrayList<>();
    if ((info == null) ||
      ((info.getChannelID() == null) && (info.getAlgorithmID() == null))) {
      // retrieve all channelIDs in the store
    } else if (info.getAlgorithmID() == null) {
      // retrieve all algorithmIDs for the channnel with info.getChannelID()
    } else if (info.getChannelID() == null) {
      // delete all channelIDs that have an entry for info.getAlgorithmID()
    } else {
      // return info.getAlgorithmID() if the channel with info.getChannelID() has it
    }
    return results;
  }
}

4 Configuration

Load-balancer persistence is setup via the JPPF configuration, in the following format:

jppf.load.balancing.persistence = <persistence_class_name> [param1 ... paramN]

As we can see, optional parameters can be passed on to an implementation from the configuration. To receive them, the implementation class must declare either a constructor that takes a vararg String... parameter, or a public void setParameters(String...params) method. The space-separated parameters can be used to specify the root directory for a file-based implementation, or JDBC connection parameters, but are not limited to these.

An implementation that does not declare a consttuctor with a String... argument must declare a no-args constructor. On the other hand, if a persistence implementation declares both a constructor with a String... argument and a setParameters() method, then the constructor will always be prefered.

When the "jppf.load.balancing.persistence" property is unspecified, then load-balancer persistence is disabled.

5 Reference

We invite you to read the reference section on load-balancer state persistence, including the details on the JPPF built-in implementations.