Concurrency: Managing Parallel Processes

Processes and Processors	Working with Evaluation IDs
Starting and Waiting for Processes	Examples

Processes and Processors

A process is simply a Wolfram Language expression being evaluated. A processor is a parallel kernel that performs such evaluations.

The command ParallelEvaluate discussed in the tutorial "Parallel Evaluation" will send an evaluation to an explicitly given processor, requiring you to keep track of available processors and processes yourself. The scheduling functions discussed in this tutorial perform these functions for you. You can create any number of processes, many more than the number of available processors. If more processes are created than there are processors, the remaining processes will be queued and serviced when a processor becomes available.

Starting and Waiting for Processes

The two basic commands are ParallelSubmit[expr] to line up an expression for evaluation on any available processor, and WaitAll[pid] to wait until a given process has finished.

Each process in the queue is identified by its unique evaluation ID, or eid.

ParallelSubmit[cmd]	submits cmd for evaluation on a remote kernel and returns the queued job's eid
ParallelSubmit[{vars …},cmd]	builds a closure for the local values of the given variables before sending cmd to a remote kernel
WaitAll[eid]	waits for the given process to finish and returns its result
WaitAll[{eid₁,eid₂,…}]	waits for all given processes and returns the list of results
WaitAll[expr]	waits for all evaluation IDs contained in expr to finish and replaces them by the respective process' result
WaitNext[{eid₁,eid₂,…}]	waits for one of the given processes to finish. It returns {res,id,ids}, where id is the eid of the finished process, res is its result, and ids is the list of remaining eids

Queuing processes.

WaitNext is nondeterministic. It returns an arbitrary process that has finished. If no process has finished, it waits until a result is available. The third element of the result of WaitNext is suitable as an argument of another call to WaitNext.

The functions ParallelSubmit and WaitAll implement concurrency. You can start arbitrarily many processes, and they will all be evaluated eventually on any available remote processors. When you need a particular result, you can wait for any particular eid, or you can wait for all results using repeated calls to WaitNext.

Basic Usage

To try the examples here, start a few parallel kernels.

Queue the evaluation 1+1 for processing on a remote kernel. Note that Queue has the attribute HoldAll to prevent evaluation of the expression before queuing. The value returned by Queue is the evaluation ID (eid) of the queued process.

After queuing a process, you can perform other calculations and eventually collect the result. If the result is not yet available, WaitAll will wait for it.

You can queue several processes. Here, the expression 1², 2², …, 5² is queued for evaluation.

Next, wait for any one process to finish.

Note the reassignment of pids to the list of remaining evaluation IDs. Repeating the previous evaluation until the pids list becomes empty allows you to drain the queue.

You can also wait for all of the remaining processes to finish.

A Note on the Use of Variables

If an expression e in ParallelSubmit[e] involves variables with assigned values, care must be taken to ensure that the remote kernels have the same variable values defined. Unless you use DistributeDefinitions or shared variables, locally defined variables will not be available to remote kernels. See Values of Variables in the tutorial "Parallel Evaluation" for more information.

Here are a few common cases where there may be problems.

This assigns the value 2 to the variable a in the local master kernel.

You want to evaluate the expression Head[a] on a remote kernel. The result is not Integer, as it is on the local kernel, because on the remote kernel the symbol a does not have a value.

You can use a local constant, and even name it a, to textually insert the value of a into the argument of ParallelSubmit.

To make this frequent case simpler, you can use an optional first argument in ParallelSubmit to declare the variables. The variables will then be inserted into the expression in the second argument.

The syntax ParallelSubmit[{vars …},expr] is by design similar to Function[{vars …},expr]. Both form closures where the variables are bound to their values.

Iterator variables behave in the same way. In the following two outputs, the parallel evaluation does not give the correct result.

Insert the iterator variable as a constant or declare a closure to ensure that you are getting correct results, as is done with the following command.

Note that ParallelTable[] treats the iterator variable correctly.

Lower-Level Functions

The Parallel`Developer` context contains functions to control the process queue.

$Queue	the list of evaluations submitted with ParallelSubmit but not yet assigned to an available kernel
$QueueLength	gives the length of the input queue $Queue
ResetQueues[]	waits for all running processes and abandons any queued processes
QueueRun[]	collects finished evaluations from all kernels and assign new ones from the queue

Evaluation queue control.

QueueRun[] returns True if at least one evaluation was submitted to a kernel or one result received from a kernel and False otherwise. Normally, you should not have to run QueueRun yourself. It is called at appropriate places inside WaitAll and other functions. You need it only if you implement your own main loop for a concurrent program.

Access the developer functions.

Submit a number of evaluations.

Each time you invoke QueueRun[], you can see a change in the state of the evaluations displayed above.

This is the number of evaluations still waiting to be serviced.

Finally, let them run to completion.

Working with Evaluation IDs

WaitAll[{pid₁,pid₂,…}] is merely a simple form of a general mechanism to parallelize computations. WaitAll can take any expression containing evaluation IDs in its arguments and will wait for all associated processes to finish. The pids will then be replaced by the results of their processes.

You can view WaitAll as the inverse of ParallelSubmit; that is, WaitAll[ParallelSubmit[expr]] gives expr, evaluated on a remote kernel, just as expr itself is evaluated locally. Further, WaitAll[… ParallelSubmit[e₁]… ParallelSubmit[e_n]…] is equivalent to …e₁…e_n…, where each e_i is evaluated in parallel. Here the ellipses represent an arbitrary surrounding computation.

The pids generated by an instance of ParallelSubmit should be left intact and should neither be destroyed nor duplicated before WaitAll performs its task. The reason is that each of them represents an ongoing parallel computation whose result should be collected exactly once.

Examples of expressions that leave pids intact follow.

Pids in a list are safe, because the list operation does not do anything to its arguments; it merely keeps them together. Nested lists are also safe for the same reason.

Pids are symbolic objects that are not affected by Plus. They may be reordered, which is irrelevant. Most arithmetic operations are safe.

Mapping a function involving ParallelSubmit onto a list is safe because the result will contain the list of the pids.

Table returns lists of pids and is, therefore, safe.

Examples of expressions that are not safe include the following.

The Head operation will destroy the symbolic pid, as will other structural operations such as Length, ByteCount, and so on.

Multiplying a pid by 0 will destroy it.

Do does not return anything, so all pids are lost. A similar case is Scan.

To recover from a computation where pids were destroyed or duplicated, use the command AbortKernels[].

ProcessID[pid]	a unique integer identifying the process
Process[pid]	the expression representing the process
Scheduling[pid]	the priority assigned to the process
ProcessState[pid]	the state of the process: queued, running, finished

Properties of process IDs (in the developer context).

Here, several processes are queued.

Because the scheduler has not yet been running, all of them are still queued for evaluation.

To demonstrate how it works, invoke the scheduler by hand.

Now some processes are running on the available processors; some may already have finished. Note that some of this information is also available in the default output form of evaluations.

WaitAll[] invokes the scheduler until all processes are finished and returns their results. Note that the priorities are not used with the default queue type.

Examples

Before evaluating these examples, make sure that you can run parallel kernels.

An Infinite Calculation

If you want to verify that the polynomials for are all irreducible (an open conjecture), you can test them with IrreduciblePolynomialQ.

This computation will continue forever. To stop it, abort the local evaluation by pressing . or choosing Kernel ▶ Abort Evaluation. After the abort, collect any waiting results as follows.

Here is the definition of the polynomial in

with degree

You then make this definition known to all remote kernels.

Now you can start the computation, requiring that it print each result as it goes on. To stop the computation, abort it by choosing Kernel ▶ Abort Evaluation or pressing .. The explicit call of QueueRun[] is necessary in such examples where you program your own scheduling details.

Do not forget to collect the orphaned processes after an interrupt (or use AbortKernels[].)

Automatic Process Generation

A general way to parallelize many kinds of computation is to replace a function g occurring in a functional operation by Composition[ParallelSubmit,g]. This new operation will cause all instances of calls of the function g to be queued for parallel evaluation. The result of this composition is a process ID (pid) that will appear inside the structure constructed by the outer computation where g occurred. To put back the results of the computation of g, wrap the whole expression in WaitAll. This will replace any pid inside its expression by the result returned by the corresponding process.

Here are a few examples of such functional compositions.

Parallel Mapping

A parallel version of Map is easy to develop. The sequential Map wraps a function f around all elements in a list.

Simply use Composition[ParallelSubmit,f] instead of f to schedule all mappings for parallel execution. The result is a list of pids.

Finally, simply wait for the processes to finish. Every pid will be replaced with the result of its associated process.

Parallel Inner Products

To see how this works for symbolic inner products, assume you want a generalized inner product where is the common last dimension of a and first dimension of b. Think of p as Plus and t as Times.

Here is an example with .

You can use Composition[ParallelSubmit,p] in place of p in the previous expression to queue all calculations of p for concurrent execution. The result is a tensor of process IDs.

Now, simply wait for all processes in this expression.

Parallel Tables and Sums

This code generates a 55 matrix of random numbers, where each row is computed in parallel.

Here is a sum whose elements are computed in parallel. Each element of the sum is a numerical integration.

Here is the corresponding table of parallel exact results. The value of is .

Comparison with Parallelize

Parallel mapping, tables, and inner products were already introduced in "Parallel Evaluation". Those functions divide the task into batches of several subproblems each, under control of the method option. The functions in this section generate one evaluation for each subproblem. This division is equivalent to the setting Method->"FinestGrained".

If all subproblems take the same amount of time, the functions such as ParallelMap[] and ParallelTable[] are faster. However, if the computation times of the subproblems are different, and not easy to estimate in advance, it can be better to use WaitAll[… ParallelSubmit[]…] as described in this section or use the equivalent method option setting. If the number of processes generated is larger than the number of remote kernels, this method performs automatic load balancing, because jobs are assigned to a kernel as soon as the previous job is done, and all kernels are kept busy all the time.

Tracing

To observe how processes are scheduled, you can use tracing. To use these features, you have to load the debugging package before loading the toolkit and starting several kernels.

Now you can enable Queueing tracing.

Here several processes are queued showing how the queue grows in size.