Job Service Level Agreement

A job service level agreement (SLA) defines the terms and conditions in which a job will be processed. A job carries two distinct SLAs, one which defines a contract between the job and the JPPF server, the other defining a different contract between the job and the JPPF client.

Server and client SLAs have common attributes, which specify:

• the characteristics of the nodes it can run on (server side), or of the channels it can be sent through (client side): the job execution policy
• the time at which a job is scheduled to start
• an expiration date for the job

The attributes specific to the server side SLA are:

• the priority of a job
• whether it is submitted in suspended state
• the maximum number of nodes it can run on
• whether the job is a standard or broadcast job
• whether the server should immediately cancel the job, if the client that submitted it is disconnected

The attributes specific to the client side SLA are:

• the maximum number of channels it can be sent through

A job SLA is represented by the interface JobSLA for the server side SLA, and by the interface JobClientSLA for the client side SLA. It can be accessed from a job using the related getters and setters:

 public class JPPFJob implements Serializable, JPPFDistributedJob {
   // The job's server-side SLA
   public JobSLA getSLA()
   public void setSLA(final JobSLA jobSLA)
 
   // The job's client-side SLA
   public JobClientSLA getClientSLA()
   public void setClientSLA(final JobClientSLA jobClientSLA)
 }

Example usage:

 JPPFJob myJob = new JPPFJob();
 myJob.getClientSLA().setMaxChannels(2);
 myJob.getSLA().setPriority(1000);

Also note that both interfaces extend the common interface JobCommonSLA. We will go into the details of these interfaces in the following sections.

1 Attributes common to server and client side SLAs

As seen previously, the common attributes for server and client side SLAs are defined by the JobCommonSLA interface:

 public interface JobCommonSLA extends Serializable {
   // The execution policy
   ExecutionPolicy getExecutionPolicy();
   void setExecutionPolicy(ExecutionPolicy executionPolicy);
 
   // The job start schedule
   JPPFSchedule getJobSchedule();
   void setJobSchedule(JPPFSchedule jobSchedule);
 
   // The job expiration schedule
   JPPFSchedule getJobExpirationSchedule();
   void setJobExpirationSchedule(JPPFSchedule jobExpirationSchedule);
 }

1.1 Execution policy

An execution policy is an object that determines whether a particular set of JPPF tasks can be executed on a JPPF node (for the server-side SLA) or if it can be sent via a communication channel (for the client-side). It does so by applying the set of rules (or tests) it is made of, against a set of properties associated with the node or channel.

These properties include:

• JPPF configuration properties
• System properties (including -D*=* properties specified on the JVM command line)
• Environment variables (e.g. PATH, JAVA_HOME, etc.)
• Networking: list of ipv4 and ipv6 addresses with corresponding host name when it can be resolved
• Runtime information such as maximum heap memory, number of available processors, etc...
• Disk space and storage information (JDK 1.6 or later only)
• A special boolean property named “jppf.channel.local” which indicates whether a node (server-side) or communication channel (client-side) is local to the JPPF server or client, respectively.

The kind of tests that can be performed apply to the value of a property, and include:

• Binary comparison operators: ==, <, <=, >, >= ; for instance: property_value <= 15
• Range operators (intervals): property_value in [a,b] , [a,b[ , ]a,b] , ]a,b[
• "One of" operator (discrete sets): property_value in { a1, ... , aN }
• "Contains string" operator: property_value contains "substring"
• Regular expressions: property_value matches 'regexp'
• Custom, user-defined tests

The tests can also be combined into complex expressions using the boolean operators NOT, AND, OR and XOR.

Using this mechanism, it is possible to write execution policies such as:

• "Execute on a node only if the node has at least 256 MB of memory and at least 2 CPUs available"
• “Execute the job only in the client's local executor”

In the context of a server-side SLA, an execution policy is sent along with the tasks to the JPPF driver, and evaluated by the driver. They do not need to be sent to the nodes.

For a detailed and complete description of all policy elements, operators and available properties, please refer to the chapter Appendix B: Execution policy reference.

1.1.1 Creating and using an execution policy

An execution policy is an object whose type is a subclass of ExecutionPolicy. It can be built in 2 ways:

By API, using the classes in the org.jppf.node.policy package.

Example:

 // define a policy allowing only nodes with 2 processing threads or more
 ExecutionPolicy atLeast2ThreadsPolicy = new AtLeast("processing.threads", 2);
 // define a policy allowing only nodes that are part of the "mydomain.com"
 // internet domain (case ignored)
 ExecutionPolicy myDomainPolicy =
   new Contains("ipv4.addresses", true, "mydomain.com");
 // define a policy that requires both of the above to be satisfied
 ExecutionPolicy myPolicy = atLeast2ThreadsPolicy.and(myDomainPolicy);

Alternatively, this could be written in a single statement:

 // define the same policy in one statement
 ExecutionPolicy myPolicy = new AtLeast("processing.threads", 2).and(
   new Contains("ipv4.addresses", true, "mydomain.com"));

Using an XML policy document:

Example XML policy:

 <ExecutionPolicy>
   <!-- define a policy that requires both rules to be satisfied -->
   <AND>
     <!-- define a policy allowing only nodes with 2 processing threads or more -->
     <AtLeast>
       <Property>processing.threads</Property>
       <Value>2</Value>
     </AtLeast>
     <!-- allow only nodes in the "mydomain.com" internet domain (case ignored) -->
     <Contains ignoreCase="true">
       <Property>ipv4.addresses</Property>
       <Value>mydomain.com</Value>
     </Contains>
   </AND>
 </ExecutionPolicy>

As you can see, this is the exact equivalent of the policy we constructed programmatically before.

To transform this XML policy into an ExecutionPolicy object, we will have to parse it using the PolicyParser API, by the means of one of the following methods:

 static ExecutionPolicy parsePolicy(String)      // parse from a string
 static ExecutionPolicy parsePolicyFile(String)  // parse from a file
 static ExecutionPolicy parsePolicy(File)        // parse from a file
 static ExecutionPolicy parsePolicy(Reader)      // parse from a Reader
 static ExecutionPolicy parsePolicy(InputStream) // parse from an InputStream

Example use:

 // parse the specified XML file into an ExecutionPolicy object
 ExecutionPolicy myPolicy = PolicyParser.parsePolicyFile("../policies/MyPolicy.xml");

It is also possible to validate an XML execution policy against the JPPF Execution Policy schema using one of the validatePolicy() methods of PolicyParser:

 static ExecutionPolicy validatePolicy(String)      // validate from a string
 static ExecutionPolicy validatePolicyFile(String)  // validate from a file
 static ExecutionPolicy validatePolicy(File)        // validate from a file
 static ExecutionPolicy validatePolicy(Reader)      // validate from a Reader
 static ExecutionPolicy validatePolicy(InputStream) // validate from an InputStream

To enable validation, the document's namespace must be specified in the root element:

 <jppf:ExecutionPolicy xmlns:jppf="https://www.jppf.org/schemas/ExecutionPolicy.xsd">
   ...
 </jppf:ExecutionPolicy>

Example use:

 public ExecutionPolicy createPolicy(String policyPath) {
   try {
     // validate the specified XML file
     PolicyParser.validatePolicyFile(policyPath);
   } catch(Exception e) {
     // the validation and parsing errors are in the exception message
     System.err.println("The execution policy " + policyPath +
       " is not valid: " + e.getMessage());
    return null;
   }
   // the policy is valid, we can parse it safely
   return PolicyParser.parsePolicyFile(policyPath);
 }

1.1.2 Creating custom policies

It is possible to apply user-defined policies. When you do so, a number of constraints must be respected:

• the custom policy class must extend CustomPolicy
• the custom policy class must be deployed in the JPPF server classpath as well as the client's

Here is a sample custom policy code:

 package my.package;
 import org.jppf.management.JPPFSystemInformation;
 import org.jppf.node.policy.CustomPolicy;
 
 // define a policy allowing only nodes with 2 processing threads or more
 public class MyCustomPolicy extends CustomPolicy {
   public boolean accepts(JPPFSystemInformation info) {
     // get the value of the "processing.threads" property
     String s = this.getProperty(info, "processing.threads");
     int n = -1;
     try { n = Integer.valueOf(s); }
     catch(NumberFormatException e) { // process the exception }
     // node is accepted only if number of threads >= 2
     return n >= 2;
   }
 }

Now, let's imagine that we want our policy to be more generic, and to accept nodes with at least a parametrized number of threads given as argument to the policy.

Our accepts() method becomes then:

 public boolean accepts(JPPFSystemInformation info) {
   // get the value to compare with, passed as the first argument to this policy
   String s1 = getArgs()[0];
   int param = -1;
   try { param = Integer.valueOf(s1); }
   catch(NumberFormatException e) { }
   String s2 = getProperty(info, "processing.thread");
   int n = -1;
   try { n = Integer.valueOf(s2); }
   catch(NumberFormatException e) { }
   // node is accepted only if number of threads >= param
   return n >= param;
 }

Here we use the getArgs() method which returns an array of strings, corresponding to the arguments passed in the XML representation of the policy.

To illustrate how to use a custom policy in an XML policy document, here is an example XML representation of the custom policy we created above:

 <CustomRule class="my.package.MyCustomPolicy">
   <Arg>3</Arg>
 </CustomRule>

The "class" attribute is the fully qualified name of the custom policy class. There can be any number of <Arg> elements, these are the parameters that will then be accessible through CustomPolicy.getArgs().

When the XML descriptor is parsed, an execution policy object will be created exactly as in this code snippet:

 MyCustomPolicy policy = new MyCustomPolicy();
 policy.setArgs( "3" );

Finally, to enable the use of this custom policy, you will need to deploy the corresponding class(es) to both the server's and the client's classpath.

1.2 Job start and expiration scheduling

It is possible to schedule a job for a later start, and also to set a job for expiration at a specified date/time. The job SLA allows this by providing the following methods:

 // job start schedule
 public JPPFSchedule getJobSchedule()
 public void setJobSchedule(JPPFSchedule schedule)
 
 // job expiration schedule
 public JPPFSchedule getJobExpirationSchedule()
 public void setJobExpirationSchedule(JPPFSchedule schedule)

As we can see, this is all about getting and setting an instance of JPPFSchedule. A schedule is normally defined through one of its constructors:

As a fixed length of time

 public JPPFSchedule(long duration)

The semantics is that the job will start duration milliseconds after the job is received by the server. Here is an example:

 JPPFJob myJob = new Job();
 // set the job to start 5 seconds after being received
 JPPFSchedule mySchedule = new JPPFSchedule(5000L);
 myJob.getSLA().setJobSchedule(mySchedule);

As a specific date/time

 public JPPFSchedule(String date, String dateFormat)

Here the date format is specified as a pattern for a SimpleDateFormat instance.

Here is an example use of this constructor:

 JPPFJob myJob = new Job();
 String dateFormat = "MM/dd/yyyy hh:mm a z";
 // set the job to expire on September 30, 2010 at 12:08 PM in the CEDT time zone
 JPPFSchedule schedule = new JPPFSchedule("09/30/2010 12:08 PM CEDT", dateFormat);
 myJob.getSLA().setJobExpirationSchedule(mySchedule);

2 Server side SLA attributes

A server-side SLA is described by the JobSLA interface, defined as:

 public interface JobSLA extends JobCommonSLA {
   // Job priority
   int getPriority();
   void setPriority(int priority);
 
   // Maximum number of nodes the job can run on
   int getMaxNodes();
   void setMaxNodes(int maxNodes);
 
   // Whether the job is initially suspended
   boolean isSuspended();
   void setSuspended(boolean suspended);
 
   // Whether the job is a broadcast job
   boolean isBroadcastJob();
   void setBroadcastJob(boolean broadcastJob);
 
   // Determine whether the job should be canceled by the server
   // if the client gets disconnected
   boolean isCancelUponClientDisconnect();
   void setCancelUponClientDisconnect(boolean cancelUponClientDisconnect);
 }

2.1 Job priority

The priority of a job determines the order in which the job will be executed by the server. It can be any integer value, such that if jobA.getPriority() > jobB.getPriority() then jobA will be executed before jobB. There are situations where both jobs may be executed at the same time, for instance if there remain any available nodes for jobB after jobA has been dispatched. Two jobs with the same priority will have an equal share (as much as is possible) of the available grid nodes.

The priority attribute is also manageable, which means that it can be dynamically updated, while the job is still executing, using the JPPF administration console or the related management APIs.

2.2 Maximum number of nodes

The maximum number of nodes attribute determines how many grid nodes a job can run on, at any given time. This is an upper bound limit, and does not guarantee that always this number of nodes will be used, only that no more than this number of nodes will be assigned to the job. This attribute is also non-distinctive, in that it does not specify which nodes the job will run on.

The resulting assignment of nodes to the job is influenced by other attributes, especially the job priority and an eventual execution policy.

The maximum number of nodes is a manageable attribute, which means it can be dynamically updated, while the job is still executing, using the JPPF administration console or the related management APIs.

2.3 Initial suspended state

A job can be initially suspended. In this case, it will remain in the server's queue until it is explicitly resumed or canceled, or if it expires (if a timeout was set), whichever happens first. A job can be resumed and suspended again any number of times via the JPPF administration console or the related management APIs.

2.4 Broadcast jobs

A broadcast job is a specific type of job, for which each task will be be executed on all the nodes currently present in the grid. This opens new possibilities for grid applications, such as performing maintenance operations on the nodes or drastically reducing the size of a job that performs identical tasks on each node.

With regards to the job SLA, a job is set in broadcast mode via a boolean indicator, for which the interface JobSLA provides the following accessors:

 public boolean isBroadcastJob()
 public void setBroadcastJob(boolean broadcastJob)

To set a job in broadcast mode:

 JPPFJob myJob = new JPPFJob();
 myJob.getSLA().setBroadcastJob(true);

With respect to the dynamic aspect of a JPPF grid, the following behavior is enforced:

• a broadcast job is executed on all the nodes connected to the driver, at the time the job is received by the JPPF driver. This includes nodes that are executing another job at that time
• if a node dies or disconnects while the job is executing on it, the job is canceled for this node
• if a new node connects while the job is executing, the broadcast job will not execute on it
• a broadcast job does not return any results, i.e. it returns the tasks in the same state as they were submitted

Additionally, if local execution of jobs is enabled for the JPPF client, a broadcast job will not be executed locally. In other words, a broadcast job is only executed on remote nodes.

2.5 Canceling a job upon client disconnection

By default, if the JPPF client is disconnected from the server while a job is executing, the server will automatically attempt to cancel the job's execution on all nodes it was dispatched to, and remove the job from the server queue. You may disable this behavior on a per-node basis, for example if you want to let the job execute until completion but do not need the execution results.

This property is set once for each job, and cannot be changed once the job has been submitted to the server, i.e. it is not dynamically manageable.

3 Client side SLA attributes

A client-side SLA is described by the interface JobClientSLA, defined as:

 public interface JobClientSLA extends JobCommonSLA {
   // The maximum number of channels the job can be sent through,
   // including the local executor if any is configured
   int getMaxChannels();
   void setMaxChannels(int maxChannels);
 }

Note: since JPPF clients do not have a management interface, none of the client-side SLA attributes are manageable.

3.1 Maximum number of execution channels

The maximum number of channels attribute determines how many server connections a job can be sent through, at any given time. This is an upper bound limit, and does not guarantee that this number of channels will always be used. This attribute is also non-specific, since it does not specify which channels will be used.

Note 1: when the JPPF client is configured with a single server connection, this attribute has no effect.

Note 2: when local execution is enabled in the JPPF client, the local executor counts as one (additional) channel.

Note 3: the resulting assignment of channels to the job is influenced by other attributes, especially the execution policy.