How a LINQ to HPC Job Runs

LINQ to HPC uses a graph metaphor to describe the control flow and dataflow dependencies among the individual tasks that are involved in the execution of a distributed query.

Converting jobs into graphs

The graph manager reads the execution plan that was created by the LINQ to HPC provider. This plan is represented by a directed acyclic graph (DAG), which, in the context of a LINQ to HPC job, is referred to as a LINQ to HPC graph.

Each node of the graph represents a unit of work that will be performed by a single DSC node of the cluster using specific inputs and producing specific outputs. Each unit of work is called a vertex. (For more information about vertices, see the Vertices section.) The graph manager transforms and optimizes the execution plan. This means that the final plan may not be the same as the plan that the LINQ to HPC provider created.

The job’s vertices are grouped into stages (as shown in Figure 5), each of which carries out a particular phase of a query that contains multiple subqueries. For each stage, there will generally be one vertex for each DSC file in the file set being processed by that stage (some stages may use a single vertex). If there are more vertices than DSC nodes, the graph manager queues execution of some vertices. Note also that the number of DSC files in the output of a stage does not necessarily equal the number of DSC files in the input to that stage.

LINQ to HPC uses a generalization of the UNIX piping mechanism to connect the vertices that comprise a job. The pipes, or channels, are the arcs of the graph. (For more information, see Channels.) With standard pipes, a job consists of a chain of processes, where each process pipes its output to the next process in the chain. LINQ to HPC extends the standard pipe model to handle distributed applications that run on a cluster.

Figure 4 illustrates how a query runs on a single computer, and compares it to how that query might be distributed over the nodes of an HPC cluster. The left side illustrates a simple job with a single stage that runs on a single computer. The right side illustrates how that same job would run on a cluster. The letter "S" indicates that each of the vertices is performing the same data parallel operation over a distributed data set that has been partitioned into multiple DSC files.

Figure 4. A simple distributed job

Typically, LINQ to HPC jobs are much more complex than the one illustrated in Figure 4. Figure 5 illustrates a LINQ to HPC graph for a more typical LINQ to HPC job. The basic execution plan on the left uses the standard pipe model, where each stage of the job pipes its data to the next stage. The LINQ to HPC graph on the right shows how LINQ to HPC might implement the operation as a distributed job that runs on a cluster.

Figure 5. A LINQ to HPC graph

A LINQ to HPC graph is composed of input file sets, stages, vertices, and channels. In Figure 5, the vertices show how operations R, X, and M can be divided across the cluster. The channels, shown in the illustration as arrows that connect the vertices of the graph, represent data flow. Vertices can execute whenever their inputs become available, which means that it is possible that the stages may overlap. If a vertex in stage 2 has all of its required inputs, it may begin execution even though other vertices in stage 1 are still running.

The stages of a LINQ to HPC job correspond to the elements of the basic execution plan. Each stage consists of one or more identical vertices. Different stages can have different numbers of vertices. In some cases, the LINQ to HPC runtime determines the number of vertices in order to optimize performance. If a stage processes static data, the number of vertices is dictated by the number of data files. (If there are more data files than DSC nodes, LINQ to HPC runs the vertices in turn, until all the data is processed.)

Vertices

A vertex is an independent instance of the data processing code for a stage. Each vertex processes a subset of data in a stage. The graph manager allows a DSC node to run only one vertex at a time. The scheduler allocates whole nodes for running vertices so that no other jobs can execute on the same node. Typically, LINQ to HPC jobs are I/O bound, so running multiple vertices concurrently on the same node is unlikely to result in improved performance.

A vertex runs methods from a Microsoft .NET Framework assembly that was dynamically generated by the LINQ to HPC provider that runs on the client workstation. The code in the .NET Framework assembly can use the P/Invoke mechanism to call unmanaged code, or it can invoke a completely external program. For example, it can use the .NET Process class in the System.Diagnostics namespace to launch new operating system processes.

The same vertex type can be used in multiple stages. This is illustrated in Figure 5, where two different stages have vertices of type M. A vertex type refers to the portion of processing logic that the vertex runs. All vertices in the same stage are of the same type.

Channels

Vertices use channels to pass the processed data from one stage of the job to the next stage. Channels are point-to-point. They serialize the output of one vertex and transmit it to the input of another vertex. A vertex can have channels to more than one vertex in the next stage.

Just as it chooses the number of vertices to optimize performance, LINQ to HPC also chooses the way that the vertices are connected by channels. In addition, the LINQ to HPC graph manager can sometimes take advantage of run-time information to dynamically modify the later stages of the graph to improve performance. This means that the original graph that was created for the job may not be the graph that is executed, or that successive runs of the same job may not use the same graph.

Steps used to execute a LINQ to HPC job

A LINQ to HPC job includes the following steps.

1. When a LINQ to HPC application runs, it creates a LINQ to HPC query and uses the LINQ to HPC provider to evaluate it.

2. The LINQ to HPC provider performs the following actions:

   1. It generates an execution plan. LINQ applications, including LINQ to HPC applications, operate on data sets rather than on individual items. This means that the provider has considerable flexibility in how it translates a query into an execution plan.

   2. It generates the data processing code for the vertices. The data processing code for each stage is compiled into a .NET assembly, and then dispatched to the cluster’s computers at the appropriate stage of the operation.

   3. It collates and serializes any side information that is necessary for the computation, particularly local variables that have been referenced in the query with a closure.

3. The LINQ to HPC provider contacts the HPC job scheduler. The scheduler queues a job for the query according to the number of nodes that are specified in the HpcLinqConfiguration object associated with the query. The scheduler allocates on a per-node basis. This means that no other jobs can run on the nodes scheduled for the query. The scheduler creates an instance of the LINQ to HPC graph manager on the compute node that has been reserved for this purpose.

   Note

   The graph manager does not need to run on a compute node that is designated as a DSC node.

4. The graph manager reads the execution graph, based on the plan that was generated in Step 2.

5. The graph manager contacts the HPC Dsc service on the head node to locate the files that make up the input file set.

6. The graph manager assigns vertices to DSC nodes that contain the files. Unless there are conflicts with components such as the graph manager, a vertex will run on the DSC node that stores the DSC file that is associated with the vertex. (If there is a conflict, the system may copy a DSC file to another DSC node before running the associated vertex.)

7. The graph manager starts the vertices for the job's first stage.

   1. The graph manager schedules vertices as DSC nodes become available.

   2. Each vertex runs code to process a file, and to prepare the results for subsequent stages or for the final output.

   3. The input to a vertex can be either a DSC file or data from the preceding stage.

   4. The graph manager can apply run-time policies and optimizations to the execution graph to optimize performance.

8. Every vertex in the first stage simultaneously processes the file on its computer (unless there are more vertices than DSC nodes, in which case some vertices may be queued until others complete).

9. Subsequent stages operate on dynamically produced data, which is data that was generated by the previous stage.

10. At the end of the job, the last file set is made available to the client application as an enumerable object or as a single value. The final output is also written to the DSC.