Parallel Computing Toolkit, Version 2

Dr. Roman E. Mäder
MathConsult Dr. R. Mäder
Samstagernstraße 58a
CH-8832 Wollerau

mailto:maeder@mathconsult.ch
http://www.mathconsult.ch

© 2004 MathConsult Dr. R. Mäder. All rights reserved.

Introduction

What is Parallel Computing Toolkit?

Parallel Computing Toolkit (PCT) is a Mathematica AddOn that allows you to develop and run parallel computations.

Parallel Computing ( a System Programmer's View)

Parallel programming requires support for:

In this context:

Requirements

All scheduling happens in the master kernel. No extra software is required on the slave kernels. No support system for parallel computing, such as PVM or MPI is required.

Slave kernels are normally started once and are then used for any number of computations. The Parallel Computing Toolkit contains its own scheduler and load balancing tools.

Master and slave kernels can run on a multi-processor machine, a tightly coupled network of processors (a cluster), a network of workstations or any remote machines accessible through TCP/IP.

In this demonstration the frontend runs on my laptop computer, the master kernel runs on the main node of an Orion Multisystems DT-12, and the slave kernels on the remaining 11 nodes of this desktop cluster.

PCT 2

Major new features of Version 2:

Enhancements:

System Configuration

Local configuration, such as a list of commonly available machines and the operating-system commands needed to launch a kernel there, can be set up in a user's init.m file.

The template $RemoteCommand may contain four slots filled as follows
`1`	hostname	first argument of LaunchSlave[]
`2`	linkname	as returned by LinkCreate[]
`3`	remote user	$RemoteUserName, default is $UserName
`4`	protocol	link protocol argument

$RemoteCommand

This is the default ist of available remote machines.

$AvailableMachines

Startup

Needs["Parallel`Debug`"]
Needs["Parallel`"]

SetOptions[LaunchSlave, LinkHost -> "198.18.0.1"];

LaunchSlaves[]

TableForm[RemoteEvaluate[{$ProcessorID, $MachineName, $SystemID, $Version, 
              TimeUsed[],  MaxMemoryUsed[]}],\(TableHeadings -> {None, {"ID", "host", \ "OS", "Mathematica Version", \ "CPU Time", "Memory"}}\)]

Parallelizing Commands

Wait and Queue

Queue[expr] queues the remote evaluation of expr and returns a process ID (pid).

Wait[pid] waits for the given process to finish and returns its result.

Therefore, Wait[Queue[expr]] is just the evaluation function (or identity).

Wait[Queue[1 + 2]]

Wait[{pid1,pid2,...}] waits for several processes.

Wait[Queue /@ {1 + 1, 2 + 2, 3 + 3}]

This example was not very parallel. We need to be careful.

Wait[Queue /@ Unevaluated[{1 + 1, 2 + 2, 3 + 3}]]

To convince the skeptic.

Wait[Queue /@ Unevaluated[{$ProcessorID, $ProcessorID}]]

Wait[Queue /@ {$ProcessorID, $ProcessorID}]

Elegant, but Slow

A recipe for parallelizing functional operations.

Map[f, {a, b, c}]

Composition[q, f][x]

Map[Composition[q, f], {a, b, c}]

Map[Composition[Queue,f],{a, b, c}]

Wait[%]

Wait[Table[Queue[{i}, Prime[i]],{i, 1, 10}]]

A Marketing Problem

In V1, everybody seemed to try out something like this at first.

ParallelTable[Sin[i], {i, 0.0, 100000.0}] // Short

What is the underlying primitive of parallel evaluation of many Mathematica commands?

If we know how many remote processors we have, we do not have to send out 100,000 processes of the form Sin[i], but one process of the form Table[Sin[i],{i,(Subscript[i, 0]),(Subscript[i, 1])}] to each processor.

Types of Parallel Evaluations

{1 + 1, 2 + 2, 3 + 3, 4 + 4}

Map[Prime, {1, 2, 3, 4, 5}]

Select[{1, 2, 3, 4, 5}, EvenQ]

Count[{1, 2, 3, 4, 5}, _?OddQ]

{a, b, c} . {x, y, z}

Plus[1, 2, 3, 4, 5]

Common theme: iteration over a list (or other long expression).

Efficient parallelization: do not create one process for each element of the list, but work in chunks.

The New ParallelEvaluate

The General Case

The general mechanism to parallelize all such cases:

Here is ParallelEvaluate:

?ParallelEvaluate

ParallelEvaluate[h[1 + 2, 2 + 3, 3 + 4, 4 + 5, 5 + 6], f, 
      comb]

Special Case 1: Join

For structure-preserving functions f the natural combination function is Join.

ParallelEvaluate[{1, 2, 3, 4, 5}, Prime]

ParallelEvaluate[{1, 2, 3, 4, 5}, 
      Function[list, Map[f, list]]]

Special Case 2: Identity

The default function to perform is the identity (do nothing).

ParallelEvaluate[{1 + 2, 2 + 3, 3 + 4, 4 + 5, 5 + 6}]

Special Case 3: Associative Functions

For associative h, the natural parallelization of

h[a, b, c, d, e]

is

h[h[a, b, c], h[d, e]]

That is, the function is the Identity, but the combiner is h itself.

ParallelEvaluate[1 + 2 + 3 + 4 + 5 + 6 + 7 + 8]

How to Distribute a Table?

The structure over which to parallelize Table[expr,iterator] is the iterator!

Table[i, {i, 1, 7}]

Of course, we should not generate it explicitly, but figure out suitable subiterators.

Join[Table[i, {i, 1, 4}], Table[i, {i, 5, 7}]]

Now, parallelization is easy.

ParallelDispatch[
      Join[Table[i, {i, 1, 4}], Table[i, {i, 5, 7}]]]

Load Balancing

If we know the relative performance of the remote processors, we can take this information into account when figuring out the chunks. Version 2 will allow you to measure and specify this information. This example uses remote kernels of different CPU speeds.

TableForm[{ProcessorID[#], #, ProcessorSpeed[#]} & /@ $Slaves, 
      TableHeadings -> {None, {"ID", "kernel", "speed \
setting"}}]

timings = 
      First /@ RemoteEvaluate[Timing[N[Cos[1], 10000];]] /. 
        Second -> 1

speeds = 1/timings/Min[1/timings]

Round[200  %]

MapIndexed[(ProcessorSpeed[Extract[$Slaves, #2]] = #1) &, 
      speeds]

Parallel Commands by the Dozen

General Idea

Examine each Mathematica command that operates on a list or other long expression.

f[list,arg2,arg3] ⟶ ParallelEvaluate[list,Function[li,f[li,arg2,arg3]]]

Examples

ParallelEvaluate[{1 + 1, 2 + 1, 3 + 1, 4 + 1, 5 + 1},Function[li,Select[li, EvenQ]]]

ParallelEvaluate[{1 + 1, 2 + 1, 3 + 1, 4 + 1, 5 + 1},Function[li,Cases[li, _?OddQ]]]

With[{n = 5}, 
mata = 
        Table[Random[Real, {-1, 1}], {n}, {n}]; 
matb = 
        Table[Random[
            Real, {-1, 1}], {n}, {n}];
]

With[{mata = mata, matb = matb},
par=ParallelEvaluate[mata, Dot[#, matb]&]
];
MatrixForm[par]

Less Obvious

For functions that do not preserve the structure, a custom combiner needs to be chosen.

ParallelEvaluate[{1 + 1, 2 + 1, 3 + 1, 4 + 1, 5 + 1},Count[#, _?OddQ]&,Plus]

ParallelEvaluate[{1, 2, 3, 4, 5, 6, 7},MemberQ[#, 5]&,Or]

ParallelEvaluate[{1, 2, 3, 4, 5, 6, 7}, MemberQ[#, 9] &, Or]

ParallelEvaluate[{1, 2, 3, 4, 5, 6, 7},FreeQ[#, 5]&,And]

ParallelEvaluate[{1, 2, 3, 4, 5, 6, 7}, FreeQ[#, 9] &, And]

Associative Functions

ParallelEvaluate[1 + 2 + 3 + 4 + 5 + 6 + 7 + 8]

ParallelEvaluate[Max[3, 5, 7, 3]]

ParallelEvaluate[GCD[2, 4, 6, 8]]

\(ParallelEvaluate[
      StringJoin["parallel ", "computation ", "is ", "fun."]]\)

It's Always the Same!

As we saw, to parallelize

Count[{1, 2, 3, 4, 5, 6, 7}, _?PrimeQ]

you have to say

ParallelEvaluate[{1, 2, 3, 4, 5, 6, 7},Count[#, _?PrimeQ] &,Plus]

The idea is always the same. Wouldn't this be nice:

ParallelEvaluate[Count[{1, 2, 3, 4, 5, 6, 7}, _?PrimeQ]]

It is quite simple:

Count/:ParallelEvaluate[
        Count[list_, params___]]:=
ParallelEvaluate[list, Count[#, params] &, 
        Plus]

All of these have been overloaded.

$ParallelCommands

The Ultimate Goal

It is still a long way from

ParallelEvaluate[your program goes here ... ]

Waiting for a Result

Idea: try out different methods to solve a problem. The first one to succeed wins, the others are then aborted. We can also allow for one method to return $Failed to indicate that it did not succeed. Waiting then continues until one result is found that is other than $Failed.

It is already supported in V1. Queue each evaluation as a process, then use WaitOne[]. This returns with one of the results. You can then abort the others (this part does not work well in V1).

ids = {Queue[1 + 1], Queue[2 + 2]}

{res, id, ids} = WaitOne[ids]

ResetSlaves[];

res

Parallel Try

Here is a function that combines these steps.

The function ParallelTry[{exprs...}] takes a list of expressions as argument and submits them to remote kernels. The first result other than $Failed is returned, and any remaining computations are aborted.

ParallelTry[exprs_]:=
Module[{res = $Failed, id, ids},
ids=List@@Queue/@Unevaluated[exprs];
While[res === $Failed  &&  Length[ids] > 0,  
{res, id, ids}=WaitOne[ids]
];
ResetSlaves[];
res
]

Finding Irreducible Polynomials

We want to find an irreducible polynomial modulo n.

This generates a random polynomial of a given degree mod p.

p=2

randPoly[n_Integer?Positive, x_, p_] := 
      x^n + Sum[Random[Integer, {0, p - 1}]*x^i, {i, 1, n - 1}] + 
        Random[Integer, {1, p - 1}]

It is irreducible if it does not have any factors other than a numerical content.

irreducibleQ[poly_, p_] := 
      With[{fl = FactorList[poly, Modulus -> p]}, 
        Length[fl] < 3 && fl[[-1, 2]] == 1]

poly = randPoly[10, x, p]

irreducibleQ[poly, p]

The Loop

This loops keeps trying until it finds an irreducible one.

While[! irreducibleQ[poly = randPoly[100, x, p], p], Null]; 
    poly

Here is the function that combines these steps.

findOne[n_, 
        p_] := 
Module[{poly}, 
While[! irreducibleQ[poly = randPoly[n, x, p], p], 
          Null]; 
poly]

findOne[100, p]

In Parallel

ExportEnvironment[randPoly, irreducibleQ, findOne]

Generate list of identical commands, one for each remote kernel.

With[{n = 512, p = 2}, 
Join @@ 
        Table[Hold[
            findOne[n, p]], {Length[$Slaves]}]
]

Let them loose.

ParallelTry[%]

(In this example, 853 polynomials were examined.)

Comparing Algorithms

This example from the new collection of sample applications shows how to compare the different methods for numerical minimization available in Mathematica V5.

f[x_, y_] := (E^Sin)[50 x] + 
        Sin[60 (E^y)] + Sin[70 Sin[x]] + Sin[Sin[80 y]] - 
        Sin[10 (x + y)] + (1/4) ((x^2) + (y^2))

We can run all methods in parallel to compare the results.

TableForm[ParallelMap[tryMethod,methods],TableDepth -> 2,TableHeadings -> {None,  {Method, Timing, Minimum, 
              Position}}]

Here we use ParallelTry to find the fastest method and abort all other calculations. There is no point in including the Automatic method setting, because it implies on of the other methods.

trials = 
        tryMethod /@ Hold @@ 
            Complement[methods, {Automatic}];

ParallelTry @@ trials

Shared Memory

Communication

How can processes communicate with each other?

Mathematica kernels do not share memory, even if they run on the same machine. Our computation model is always a distributed memory one.

Virtual Shared Memory: implementing shared memory on distributed memory machines with a software layer that handles all variable accesses, using message passing.

When a process needs the value of a global (shared) variable or wants to set its value, it sends a request back to the scheduler. The scheduler maintains the variables, updates their values, and sends back the requested values.

Examples

We will work with two slave kernels.

s1 = $Slaves[[1]]; s2 = $Slaves[[2]];

Declare shared variables.

SharedVariables[{x}]

Basic tests. One slave kernel reads the variable, another one increments its value.

x = 0;

RemoteEvaluate[x = x + 1, s1]

RemoteEvaluate[x, s2]

The new value is stored globally.

x

Increment the value in one indivisible step. This operation is important for synchronization.

RemoteEvaluate[x++, s2]

RemoteEvaluate[x, s1]

No special syntax is required to use shared variables. Just do things the normal way. If a variable is shared, all accesses are transparently sent to the master process.

ClearShared[x];

Synchronization

Exclusive access to a variable or other resource can be controlled by a lock. An update of a shared variable implements a lock and allows semaphores and critical sections.

Here is the typical code that implements a critical section with a lock. Each of the five parallel processes accesses the same shared variable y, first reading it, then writing a new value. Before doing this, it acquires a lock, which is released after the critical section. In this way no conflicting updates happen, as can be seen by the output, in which all values from 1 to 5 occur.

SharedVariables[{y, lock}]

y = 0;
ParallelMap[Module[{a},
		Pause[0.2  Random[]];
		While[TestAndSet[lock, #] =!= #, 
                    Pause[0.1  Random[]]];
		a = y;
		Pause[Random[]];
		y = a + 1;
		lock = Null;
		a + 1]&,
	Range[5]]

ClearSlaves[];

Shared Downvalues

Delayed definitions (downvalues) of shared variables can be evaluated on the master kernel or the remote kernel that asked for it. Shared downvalues can be used to access shared resources such as a global queue.

Debugging News

Tracing Events

Options[$DebugObject]

OptionValues[Trace]

SetOptions[$DebugObject,Trace->{SendReceive,Queueing,SharedMemory}]

SetOptions[$DebugObject,TraceHandler->Save]

Running A Computation with Tracing

SharedVariables[x]; x = 5;

Wait[{Queue[x = x + 1], Queue[x = x + 1]}]

Analyzing the Trace

TableForm[TraceList[], TableDepth -> 2, 
      TableHeadings -> {Automatic, {"Trigger", "Event"}}]

newTraceList[]

SetOptions[$DebugObject, TraceHandler -> Print, 
      Trace -> {}]

ClearSlaves[];

The New Examples Directory

From Application Examples, an annotated list of all new examples in the Examples subdirectory: