JPPF Overview

A JPPF grid is made of three different types of components that communicate together:

• clients are entry points to the grid and enable developers to submit work via the client APIs
• servers are the components that receive work from the clients, dispatch it to the nodes, receive the results fom the nodes, and send these results back to the clients
• nodes perform the actual work execution

The figure below shows how all the components are organized together:

From this picture, we can see that the server plays a central role, and its interactions with the nodes define a master / worker architecture, where the server (i.e. master) distributes the work to the nodes (i.e. workers).

This also represents the most common topology in a JPPF grid, where each client is connected to a single server, and many nodes are attached to the same server. As with any such architecture, this one is facing the risk of a single point of failure. To mitigate this risk, JPPF provides the ability to connect multiple servers together in a peer-to-peer network and additional connectivity options for clients and nodes, as illustrated in this figure:

Note how some of the clients are connected to multiple servers, providing failover as well as load balancing capabilities. In addition, and not visible in the previous figure, the nodes have a failover mechanism that will enable them to attach to a different server, should the one they are attached to fail or die.

The connection between two servers is directional: if server A is connected to server B then A will see B as a client, and B will see A as a node. This relationship can be made bi-directional by also connecting B to A. Note that in this scenario, each server taken locally still acts as a master in a master/worker paradigm.

In short, we can say that the single point of failure issue is addressed by a combination of redundancy and dynamic reconfiguration of the grid topology.

2 Work distribution

To understand how the work is distributed in a JPPF grid, and what role is played by each component, we will start by defining the two units of work that JPPF handles.

A task is the smallest unit of work that can be handled in the grid. From the JPPF perspective, it is considered atomic.

A job is a logical grouping of tasks that are submitted together, and may define a common service level agreement (SLA) with the JPPF grid. The SLA can have a significant influence on how the job's work will be distributed in the grid, by specifying a number of behavioral characteristics:

• rule-based filtering of nodes, specifying which nodes the work can be distributed to (aka execution policies)
• maximum number of nodes the work can be distributed to
• job priority
• start and expiration schedule
• user-defined metadata which can be used by the load balancer

To illustrate the most common flow of a job's execution, let's take a look at the following flow chart:

This chart shows the different steps involved in the execution of a job, and where each of them takes place with regards to the grid component boundaries.

It also shows that the main source of parallelism is provided by the load balancer, whose role is to split each job into multiple subsets that can be executed on multiple nodes in parallel. There are other sources of parallelism at different levels, and we will describe them in the next sections.

3 Jobs and tasks granularity

The granuarity of the jobs and tasks, that is, the way you divide the workload into small independant units of work, has a sginificant impact on performance. By design, JPPF is particularly well adapted for workloads that can be divided into many small tasks independant from each other, also known as the class of embarassingly parallel problems.

There are, however, limits to how much performance can be gained by dividing a work load. In particular, if the tasks are too small, the overhead of executing them on a grid may largely overweigh the benefits of parallelization, resulting in overall performance loss. On the other hand, if the tasks are too coarse, the execution time may be equivalent to what you'd get with a sequential execution. The granalarity of the tasks must therefore be carefully considered.

To illustrate this notion, let's take an example: the multiplication of two square matrices. Let's say we have 2 square matrices A and B of size n. We define the matrix C as the result of their multiplication:

We can see that each element c_ij of matrix C is the result of n multiplications and (n - 1) additions. The matrix multiplication is therefore the result of (2n - 1) n² arithmetic operations and the computational complexity is in O(n³).

Let's consider multiple ways to divide this into independent tasks:

• at the coarsest level, we could have a task that performs the entire matrix multiplication. There is no work division and therefore no performance gain when compared to the sequential way of doing it, unless we consider performing many matrices multiplcations in parallel
• at the finest level, we could have each task perform the computation of a single c_ij element, that is, (2n -1) artihmetic operations, with a complexity in O(n). These operations are extremely fast and will be measured in microseconds at worst. In this case, and unless n is very large, the overhead of parallelizing the tasks and distributing them over a grid will cost more than the actual computation
• a more appropriate level of granularity would be to have each task compute an entire column of the resulting matric C:
  
  Each task will perform (2n - 1) n arithmectic operations and its complexity is in O(n²). For sufficiently large values of n, executing the tasks in parallel will provide a noticeable performance gain. The greater the value of n, the higher the gain.

Conclusion: chosing the granularity of the tasks is a very important part of the design of a grid-enabled application. As a rule of thumb, if the execution of a task takes less than 10 milliseconds, you should consider coarser tasks or sequential execution. At the same time, remember that JPPF is good at executing workloads with many tasks. You have therefore to find a balance between the granularity of the tasks and the level of parallelization that the division of the workload provides.

4 Networking considerations

4.1 Two channels per connection

Each connection between a server and any other component is in fact a grouping of two network channels:

• one channel is used to transport job data
• the other channel is used by the JPPF distributed class loader, that allows Java classes to be deployed on-demand where they are needed, completely transparently from a developer's perspective.

4.2 Synchronous networking

In JPPF, all network communications are asynchronous and follow a protocol based on a request/response paradigm, where multiple requests can be handled concurrently by each connection. The attribution of requester vs. responder role depends on which components communicate and through which channel.

We illustrate this in the following picture:

This communication model has a number of important implications:

• nodes can process multile job concurrently; moreover, they can execute multiple tasks in parallel, which can be part of one or more jobs
• in the same way, a single client / server connection can process multiple job at a time; furthermore, each client can be connected multiple times to the same server, or multiple times to many servers
• in the case of a server-to-server communication, multiple jobs can be processed in parallel as well, since a server attaches to another server in exactly the same way as a node.

4.3 Protocol

JPPF components communicate by exchanging messages. As described in the previous section, each JPPF transaction will be made of a request message, followed by a response message.

Messages all have the same structure, and are made of one or more blocks of data (in fact blocks of bytes), each preceded by its block size. Each block of data represents a serialized object graph. Thus, each message can be represented generically as follows:

The actual message format is different for each type of communication channel, and may also differ depending on whether it is a request or response message:

Job data channel

A job data request is composed of the following elements:

• a header, which is an object representing information about the job, including the number of tasks in the job, the job SLA, job metadata, and additional information required internally by the JPPF components.
• a data provider, which is a read-only container for data shared among all the tasks
• the tasks themselves

It can be represented as follows:

To read the full message, JPPF has to first read the header and obtain the number of tasks in the job.

The response will be in a very similar format, except that it doesn't have a data provider: being read-only, no change to its content is expected, which removes the need to send it in the response. Thus the response can be represented as:

Class loader channel

A class loader message, either request or response, is always made of a single serialized object. Therefore, the message structure is always as follows:

4.4 Connection handshaking

When a connection to the JPPF server is established, the connecting component first sends a single integer value to the server, which allows the server to know what type of channel is connecting: node vs. client vs. other server, as well as job data vs. class loader vs. JMX vs. heartbeating channel. All known channel types are enumerated in the JPPFIdentifier class, which is internal and thus not exposed in the public API.

This technique allows all types of channel to connect to the same server port, with the benefit of simplifying both the JPPF grid configuration and the network security settings (for instance by opening a single server-side port in a firewall).

5 Sources of parallelism

5.1 At the client level

There are three ways JPPF clients can provide parallel processing, which may be used individually or in any combination:

Single client, multiple concurrent jobs

A single client may submit multiple jobs in parallel. This differs from the single client/single job scenario in that the jobs must be submitted in non-blocking mode, and their results are retrieved asynchronously. Multple jobs can be processed concurrently with a single connection, over multiple connections to the same driver (I/O parellelism), over multiple connections to different drivers, or any combination of these.

Multiple clients

In this configuration, the parallelism occurs naturally, by letting the different clients work concurrently.

Mixed local and remote execution

Clients have the ability to execute jobs locally, within the same process, rather than remotely on the server. They may also use both capabilities at the same time, in which case a load-balancing mechanism will provide an additional source of parallelism.

5.2 At the server level

The server has a number of factors that determine what can be parallelized and how much:

Number of attached nodes

The number of nodes in a JPPF grid is the most important factor of performance speed up. It determines how much of the overall workload can be effectively performed in parallel.

Load balancing

This is the mechanism that splits the jobs into multiple subsets of their tasks, and distributes these subsets over the available nodes. Given the synchronous nature of the server to node connectins, a node is available only when it is not already executing a job subset. The load balancing also computes how many tasks will be sent to each node, in a way that can be static, dynamic, or even user-defined.

Job SLA

The job Service Level Agreement is used to filter out nodes in which the user does not want to see a job executed. This can be done by specifying an execution policy (rules-based filtering) for the job, or by configuring the maximum number of nodes a job can run on (grid partitioning).

Parallel I/O

Each server maintains internally a pool of threads dedicated to network I/O. This pool grows and shrinks dynamically, based on the number of nodes connected to the server, and their activity. Furthermore, as communication with the nodes is non-blocking, this pool of I/O threads is part of a mechanism that achieves a preemptive multitasking of the network I/O. This means that, even if you have a limited number of I/O threads, the overall result will be as if the server were communicating with all nodes in parallel.

5.3 At the node level

To execute tasks, each node uses a pool of threads that are called "processing threads". The size of the pool determines the maximum number of tasks a single node can execute in parallel.

The pool size may be adjusted either statically or dynamically to account for the actual number of processors available to the node, and for the tasks' resource usage profile (i.e. I/O bound tasks versus CPU bound tasks).