Job dependencies and job graphs

1 Introduction

JPPF jobs can depend on other jobs, meaning that their execution will start only when all the jobs they depend on have completed. In turn, the jobs they depend on may have their own dependencies, and the same behavior is applied recursively.

1.1 Definitions

Together, a set of jobs and their dependencies form a job dependency graph, which is a directed acyclic graph (DAG). In the rest of this documentation such graphs may be shortened to job graphs. In JPPF terminology, each job in a dependency graph is called a node or graph node.

A job dependency graph distinguishes between two kinds of nodes:

• graph roots represent jobs no other job depends on. Upon completion of a job graph root, it and all its dependencies are removed from the graph
• regular nodes represent jobs such that at least one other job depends on them

Nodes in a job dependency graph are identified with a dependency id. This id is an arbitrary identifier string, distinct from the job's UUID, which fulfills two major roles:

• when specified, it marks the job as being part of a job dependency graph. Otherwise, it is just a regular job
• it uniquely identifies a job within a job graph and each dependency of a job is specified with its own dependency id.

According to this, a job dependency graph can thus be represented as a graph of dependency ids.

1.2 Features and limitations

The choice of an arbitrary dependency id implies that you can define job graphs even when not all the jobs in the graph already exist. All that's needed are the dependency ids of the jobs. In particular, you are not constrained to using job UUIDs, which are automatically created by JPPF, and thus cannot be known in advance.

Thanks to this level of freedom, a job graph can be submitted in multiple parts from multiple JPPF clients, including clients running on separate physical or virtual machines.

Furthermore, there is no constraint on the order of submission of the jobs in a dependency graph. The specification of each job always contains enough information to know what to do with it: either wait until all its dependencies have completed, or execute the job immediately.

The default behavior, when a job in a dependency graph is cancelled, is to cascade the cancellation to all the jobs that depend on it, recursively, until all the paths that lead to this job from the graph roots are cancelled. This behavior can be overriden by setting the cascadeCancellation attribute of each job's dependency specification.

There is one major limitiation, which applies to all multi-server grid topologies: all the jobs in a dependency graph must be submitted to the same JPPF driver. If the jobs of a graph are distributed over multiple drivers, then the graph may never complete in any of the drivers. In multi-server topologies, individual jobs in a graph may be offloaded to other drivers by the job scheduler, but only as regular jobs, that is, not a spart of a graph. In other words, a job graph has a single, non-redundant representation held by a single driver at any given time. One way to address this is to assign an execution policy to the client-side SLA of each job, which enforces submission to the same driver.

2 Specifying job dependencies

2.1 With the SLA's dependency specification

We have seen, in section job dependencies specification, that a job dependency graph can be specified using the jobDependencySpec attribute of the server-side SLA in each job. Let's take the example of a simple graph with two jobs, where job_1 depends on job_2:

// define the dependency graph
job_1.getSLA().getDependencySpec()
  .setId(job_1.getName())
  .setGraphRoot(true)                // add job_1 to the graph as a root
  .addDependencies(job_2.getName()); // add job_2 as a dependency of job_1
job_2.getSLA().getDependencySpec()
  .setId(job_2.getName());

Here, we have defined a job graph where the dependency ids are equal to the job names, job_1 is a graph root and job_2 is a dependency of job_1. It is also implied that the cancellation of job_2 will cascade to the cancellation of job_1, since it is the default behavior.

2.2 With the JPPFJob API

Definiing a job graph with the JobDependencySpec API can be quite cumbersome, especially when defining complex job graphs. To alleviate this burden, the JPPFJob class provides a number of methods that make the graph defintion easier, shorter and more readable:

public class JPPFJob extends AbstractJPPFJob<JPPFJob> 
  implements Iterable<Task<?>>, Future<List<Task<?>>> {

  // Set this job's dependency id from the specified id supplier
  public JPPFJob setDependencyId(Supplier<String> idSupplier)

  // Set this job's dependency id from a dependency id supplier
  public JPPFJob setDependencyId(JobDependencyIdSupplier idSupplier)

  // Set this job's dependency id
  public JPPFJob setDependencyId(String id)

  // Convenience method to set this job's uuid as dependency id
  public JPPFJob setUuidAsDependencyId()

  // Convenience method to set this job's name as dpeendency id
  public JPPFJob setNameAsDependencyId()

  // Add the specified array of jobs as dependencies to this job
  public JPPFJob addDependencies(JPPFJob...dependencies)

  // Add the specified collection of jobs as dependencies to this job
  public JPPFJob addDependencies(Collection<JPPFJob> dependencies)

  // Set whether this job is a root in a job dependency graph
  public JPPFJob setGraphRoot(boolean graphRoot)

  // Set whether cancellation of this job should trigger the cancellation of its dependents
  public JPPFJob setCascadeCancellation(boolean cascadeCancellation)
}

Notice the setDependencyId() methods that take either a generic Supplier<String> or JobDependencyIdSupplier, which provide a lot of flexibility as to how the dependency ids are computed and supplied to the jobs. In the following code example, each line achieves the same goal, to set the job's name as its dependency id:

myJob.getSLA().getDependencySpec().setId(myJob.getName());
myJob.setDependencyId(() -> myJob.getName());
myJob.setDependencyId(job -> job.getName());
myJob.setDependencyId(myJob.getName());
myJob.setNameAsDependencyId();

With this API, the example in the previous section can be rewritten much more concisely:

// define the dependency graph
job_1.setNameAsDependencyId()
  .setGraphRoot(true)      // add job_1 to the graph as a root
  .addDependencies(job_2); // add job_2 as a dependency of job_1
job_2.setNameAsDependencyId();

3 Cycles in the job dependency graph

In a job graph, cycles, or circular dependencies, can only be detected in the server where the graph is submitted. There are two reasons for this:

• since jobs in a graph can be submitted by multiple JPPF clients, only the server is expected to hold the full graph
• because of that, job dependencies cannot be fullly identified until they are sent to the server. Only then can they be associated with an existing job UUID.

When a cycle is detected, the server will first throw and log a JPPFJobDependencyCycleException whose message describes the cycle path, for example:

org.jppf.node.protocol.graph.JPPFJobDependencyCycleException: job-3 ==> job-1 ==> job-2 ==> job-3

After this, the JPPF server will cancel all the jobs in the cycle path, following the specified cancellation behavior.

4 Job cancellation behavior

By default, when a job in a graph is canclled, all the jobs that depend on it will also be cancelled. This process is repeated recursively until we reach all the graph roots whose dependency path leads to the originally cancelled job. This behavior is called cancellation cascading.

This can be overriden for each indvidual job, by setting the cascadeCancellation attribute of thier dependency specification to false:

JPPFJob job = ...;
job.getSLA().getJobDependencySpec().setCascadeCancellation(false);

When cancellation cascading is turned of on a job, the jobs that depend on it will simply run to completion, instead being cacelled and temrinating early.

To illustrate this, consider the following job graph:

By default, if we cancel Job C, this will trigger the cancellation of jobs D, E and F, but not of jobs A and B which will run to completion. However, if cancellation cascade is turned off for job E, then only jobs D and E will be cancelled as a consequence of cancelling job C.

5 Job graphs monitoring

Job graphs in the JPPF server can be monitored with a specialized management MBean which implements the JobDependencyManagerMBean interface:

public interface JobDependencyManagerMBean {
  // Object name of this server-side MBean
  String MBEAN_NAME = "org.jppf:name=jobDependencyManager,type=driver";

  // Get the size (number of nodes or vertices that represent jobs) of the job dependency grpah
  int getGraphSize();

  // Get the ids of all the nodes currently in the graph
  Set<String> getNodeIds();

  // Get all the nodes in the job dependency graph
  Collection<JobDependencyNode> getAllNodes();

  // Get the nodes in the job dependency graph whose corresponding job is queued in the server
  Collection<JobDependencyNode> getQueuedNodes();

  // Get the nodes in the job dependency graph whose corresponding job is queued in the server,
  // filtered by the specified job selector
  Collection<JobDependencyNode> getQueuedNodes(JobSelector selector);

  // Get the node with the specified dependency id
  JobDependencyNode getNode(String id);

  // Get the grap of job dependencies
  JobDependencyGraph getGraph();
}

With this MBean, you can retrieve the full job dependency graph with the getGraph() method. This method returns an instance of the JobDependencyGraph interface, defined as follows:

public interface JobDependencyGraph extends Serializable {
  // Get the node for the specified dependency id
  JobDependencyNode getNode(String id);
  // Get the size of this graph, that is, the number of nodes or vertices
  int getSize();
  // Get the dependency ids of all the nodes currently in this graph
  Set<String> getNodeIds();
  // Get the node whose corresponding job has the specified uuid
  JobDependencyNode getNodeByJobUuid(String jobUuid);
  // Get all the nodes currently in this graph
  Collection<JobDependencyNode> getAllNodes();
  // Get the nodes in the job dependency graph whose corresponding job is queued in the server
  Collection<JobDependencyNode> getQueuedNodes();
  // Get the nodes that depend on the node with the specified dependency id
  Collection<JobDependencyNode> getDependedOn(String id);
  // Determine whether the job dependency graph is empty
  boolean isEmpty();
}

Each node in the graph is represented as an instance of JobDependencyNode:

public class JobDependencyNode implements Serializable {
  // Get the dependency id of this node
  public String getId()
  // Determine whether this node has any pending (not completed) dependency
  public boolean hasPendingDependency()
  // Get the dependency with the specified id
  public JobDependencyNode getDependency(String id)
  // Get the dependencies for this node
  public Collection<JobDependencyNode> getDependencies()
  // Get the ids of the dependencies for this node
  public Set<String> getDependenciesIds()
  // Get the nodes that depend on this node
  public Collection<JobDependencyNode> getDependedOn()
  // Determine whether the job represented by this node has completed
  public boolean isCompleted()
  // Get the JPPF uuid of the associated job
  public String getJobUuid()
  // Whether this node is a graph root
  public boolean isGraphRoot()
  // Determine whether the job represented by this dependency node has been cancelled
  // or should be cancelled when it arrives in the server queue
  public boolean isCancelled()
}

Here is an example usage of the MBean and related classes:

// get the job dependency manager MBean from a JPPF client
JPPFClient client = new JPPFClient();
JMXDriverConnectionWrapper jmx = client.awaitWorkingConnectionPool().awaitWorkingJMXConnection();
JobDependencyManagerMBean manager = jmx.getJobDependencyManager();

// get the job dependency graph and use it
JobDependencyGraph graph = manager.getGraph();
if (!graph.isEmpty()) {
  JobDependencyNode node = graph.getNode("my_dependency_id");
  if ((node != null) && !node.isCompleted()) {
    for (JobDependencyNode dependency: node.getDependencies()) {
     // do soemthing with the job's dependencies ...
    }
  }
}