Parallel Computing in the Wolfram Language: 
Local Optimization and Cluster Computing

Kevin Daily
Wolfram Research
training@wolfram.com

© Wolfram Research, Inc.

Outline

Local Optimization

Table, Map, element-wise operations

Compile

Wolfram kernel locations and licensing

Local

Remote

gridMathematica / Lightweight Grid

Mathematica on a cluster

ClusterIntegration` Package (almost obsolete)

Guidelines for using remote kernels via modern clusters (Torque, PBS, and SLURM)

Step-by-step using SLURM

Optimizing a single thread

The controlling kernel has optimized many functions to work quickly on a single core. Almost all math functions are already Listable. For real machine size numbers the Wolfram Language uses the highly optimized Intel MKL library, where possible.

elements=1.*Range[-1000,1000];

Tables, Maps, and element-wise arithmetic

RepeatedTiming[Do[Table[elements[[i]]+elements[[i]],{i,Length[elements]}],{(10^3)}]]

RepeatedTiming[Do[Table[i+i,{i,elements}],{(10^3)}]]

RepeatedTiming[Do[Map[#+#&,elements],{(10^3)}]]

RepeatedTiming[Do[elements+elements,{(10^3)}]]

Mapping vs Listable

RepeatedTiming[Do[Table[Sin[i],{i,elements}],{(10^3)}]]

RepeatedTiming[Do[Map[Sin,elements],{(10^3)}]]

RepeatedTiming[Do[Sin[elements],{(10^3)}]]

Compile

Useful if functions are all numeric or boolean.

RepeatedTiming[Do[FoldList[#1+1/(#2+0.1)&,elements],{(10^3)}]]

test=Compile[{{x,_Real,1}},FoldList[#1+1/(#2+0.1)&,x]]

RepeatedTiming[Do[test[elements],{(10^3)}]]

You can examine your compiled code. If “MainEvaluate” is present, then the code was not fully converted to low-level evaluations.

<<CompiledFunctionTools`

CompilePrint[test]

Compile with multiple threads

RepeatedTiming[Do[Map[If[#>0,#^2,#^4]&,elements],{(10^3)}]]

RepeatedTiming[Do[Table[If[x>0,x^2,x^4],{x,elements}],{(10^3)}]]

cNonListable=Compile[{{x,_Real,1}},Table[If[i>0,i^2,i^4],{i,x}]];
RepeatedTiming[Do[cNonListable[elements],{(10^3)}]]

cListable=Compile[{{x,_Real}},If[x>0,x^2,x^4],RuntimeAttributes->{Listable}];
RepeatedTiming[Do[cListable[elements],{(10^3)}]]

cListableP=Compile[{{x,_Real}},If[x>0,x^2,x^4],RuntimeAttributes->{Listable},Parallelization->True];
RepeatedTiming[Do[cListableP[elements],{(10^3)}]]

Parallel Computation Architecture in the Wolfram Language

The Wolfram Language provides a high-level, parallel language interface that allows you to connect multiple Wolfram kernels using the Wolfram Symbolic Transfer Protocol (WSTP) and provides a framework to distribute calculations and collect results.

Colloquially this is the “hub-and-spoke” model. The controlling kernel splits up and sends out computations to subkernels. The subkernels work on their own computation and send the result back to the controlling kernel. No communication occurs between the subkernels.

Types of processes

There are two types of processes available in Mathematica: Controlling processes and subordinate (sub) processes. Subprocesses are also known as computation processes.

controlling front end: This is the notebook interface that the user directly interacts with. This presentation is an example of the front end.

controlling kernel: Every computation that you run requires a controlling kernel.

subkernel: These kernels are launched as a child process of a controlling kernel with communcation via the Wolfram Symbolic Transfer Protocol (WSTP).

sub front end:  Certain types of Wolfram Language expressions require a front end to execute e.g., Import/Export or various plotting commands. In cases when you want to run these expressions in parallel, a sub-frontend launches automatically.

Wolfram subkernels

Locations

Local kernels

Mathematica detects the number of physical CPU cores

launched automatically, one subkernel per physical CPU core

easily modified to launch more in case of logical CPU cores

running subkernels on Intel Hyper Threading is not recommended

Remote kernels

require some manual configuration

easiest to use between Linux computers; have public/private key

can use e.g. PuTTY/plink to communicate between Windows and Linux

gridMathematica / Lightweight Grid kernels

requires installing gridMathematica on every computer of the cluster

runs an Apache Tomcat server to track kernel usage

does not take the place of a cluster manager (no queueing system)

can restrict usage by IP address

Licensing

Every launched kernel first checks whether it has an available license

Many organizations already have a license that can support kernels launched in a cluster environment

Please communicate your needs with your Wolfram Sales Associate

Optimizing across kernels

There are two main considerations when working with multiple kernels:

overhead of communicating between kernels

duplication of memory on each kernel

WSTP works best when transferring a minimal amount of data. Some data transfer is automatic, such as when defining a variable.

LaunchKernels[2]

ParallelEvaluate[Information[k]];

k=2;
ParallelEvaluate[k]

ParallelEvaluate evaluates only on subkernels unless a shared variable is declared. It can also cause subkernel states to become out of sync. This is a typical consideration in parallel computation.

k=3;
ParallelEvaluate[k=4,1];ParallelEvaluate[k=5,2];
{k,ParallelEvaluate[k,1],ParallelEvaluate[k,2]}

Embarrassingly parallelizable problems are the most efficient in the Wolfram Language.

fct[a_]:=NIntegrate[Boole[x≤2y+3z+w+a&&y>x-z+w],{x,-2,2},{y,-2,2},{z,-2,2},{w,-2,2}]

{time,result}=AbsoluteTiming[fct[1]]

AbsoluteTiming[Table[fct[i],{i,1,10}];]

DistributeDefinitions[fct];
AbsoluteTiming[ParallelTable[fct[i],{i,1,10}];]

Using ParallelSubmit and WaitAll creates your own local computation queue. As soon as a kernel becomes available it starts the next computation in the queue.

queue=Table[ParallelSubmit[{i},fct[i]],{i,1,10}]

AbsoluteTiming[WaitAll[queue]]

Using Remote Kernels

Launching remote kernel on Linux from Windows using PuTTY

Must have plink (PuTTY link) on PATH

can use “-pw” option to directly put in remote login password, but this makes your password visible

can still setup public/private keys; use Pageant (PuTTY agent) to organize your keys

First try the connection via RunProcess.

SetOptions[$Parallel,Tracers->{MathLink}];

Needs["SubKernels`RemoteKernels`"]

$RemoteCommand="plink wolfram@wtgclustermaster -C \"wolfram -wstp -linkmode Connect `4` -linkname `2` -subkernel -noinit -nopaclet >& /dev/null &\"";

LaunchKernels[RemoteMachine["wtgclustermaster"]]

Managed clusters: ClusterIntegration` package

CellPrint[ExpressionCell[
Pane[
Grid[{
{Style["Pros",Bold,18,Darker[Green]],Style["Cons",Bold,18,Darker[Red]]},
{"• Allows for the familiar notebook interface with many subkernels","• Every subkernel is submitted to the queue individually"},
{"","• Each kernel can wait in the cluster's queue for a long time (depends on cluster usage)"},
{"","• Most of the listed cluster managers no longer exist"}
},Alignment->Left,ItemSize->35,BaseStyle->Directive[FontFamily->"Arial",LineIndent->0.4],Frame->True],
Full,Alignment->Center
],"Output",GeneratedCell->False,TextAlignment->Left]]

Additional considerations:

The package only makes a select few clusters available. Almost all of which are no longer actively used.

CCS — Microsoft Windows Compute Cluster Server 2003

HPC — Microsoft Windows High Performance Computing Server 2008

LSF — Platform Load Sharing Facility

PBS — Altair Portable Batch System Pro (not modern Torque)

SGE — Sun Grid Engine

Managed clusters: interactive sessions

CellPrint[ExpressionCell[
Pane[Grid[{
{Style["Pros",Bold,18,Darker[Green]],Style["Cons",Bold,18,Darker[Red]]},
{"• Can provide the front end Mathematica environment","• Front end is running on the cluster"},
{"• Can use the same Wolfram Language expressions as seen earlier to connect to remote kernels","• Slower than using the front end on your own computer due to lag over the network"},
{"• Often used to debug code before making a production run","• Interactive jobs waste enormous amount of CPU time"}
},Alignment->Left,ItemSize->35,BaseStyle->Directive[FontFamily->"Arial",LineIndent->0.4],Frame->True],
Full,Alignment->Center
],"Output",GeneratedCell->False,TextAlignment->Left]]

Managed clusters: SLURM

See also Wolfram Support Article: How do I use Mathematica in a managed high-performance cluster?

Assumptions

Linux cluster with password-less SSH between each node

Every node has the same file system (path to Wolfram kernel is the same)

Example job script

The job script launches a Mathematica controlling kernel.

‘ulimit’ does have to be set to permissive; this example forces it in the script

#!/bin/bash
#SBATCH -J math
#SBATCH -N 2
#SBATCH -t 6-00:00:00

module load mathematica

ulimit -u 1000
ulimit -v unlimited
SCRIPT="hpc.wl"

echo "math -script $SCRIPT" | /bin/bash

Example Mathematica script with debug statements (hpc_SLURM.wl)

(*DEBUG*) (*turn on WSTP link debugging*)
(*DEBUG*) Needs["Parallel`Debug`"];
(*DEBUG*) SetOptions[$Parallel,Tracers->{MathLink}];

(* get list of provided resources i.e. list of node names *)
(* can also use SLURM_NODELIST for backwards compatibilty *)
nodes = ReadList["!scontrol show hostname $SLURM_JOB_NODELIST",String];  

(* get number of tasks per node; output is e.g. "2(x2),1(x2)" *) 
tasks = Environment["SLURM_TASKS_PER_NODE"];

(* parse tasks to e.g. {2,2,1,1} expression *) 
jobsPerNode = Flatten@ ToExpression[
	StringReplace[StringSplit[tasks,","], a:DigitCharacter ~~ "(x" ~~ b:DigitCharacter ~~")" :> ToString@ Table[a,ToExpression[b]]]	
];

(* setup remote command to launch kernels *)
kernelLocation = "wolfram -wstp -linkmode connect `4` -linkname `2` -subkernel -noinit >& /dev/null &";

(*DEBUG*) (* print job resources and Wolfram Kernel path *)
(*DEBUG*) Print["nodes=", nodes];
(*DEBUG*) Print["Hosts counts = ", jobsPerNode];
(*DEBUG*) Print["Remote command ", kernelLocation];

(*launch remote kernels from each node, equal to the number of resources on the node*)
Needs["SubKernels`RemoteKernels`"];
MapThread[LaunchKernels[RemoteMachine[#1, "ssh -x -f " <> #1 <> " \"" <> kernelLocation <> "\"", #2]]&, {nodes, jobsPerNode}]

(*DEBUG*) (*check that kernels were launched correctly*)
(*DEBUG*) Print["Kernels = ", Kernels[]];
(*DEBUG*) Print["Kernel count = ", $KernelCount];

(* ========== Begin user code ========== *)

(* For example: have every kernel say where it is located *)
result = ParallelEvaluate[{$MachineName,$KernelID}];
Save["result.txt", result];

(* ========== End user code ========== *)

(* clean up use of subkernels *)
CloseKernels[];
Quit[];

Managed clusters: Torque / PBS

See also Wolfram Support Article: How do I use Mathematica in a managed high-performance cluster?

Assumptions

Linux cluster with password-less SSH between each node

Every node has the same file system (path to Wolfram kernel is the same)

Example job script

The job script launches a Mathematica controlling kernel (submitted via cluster’s qsub command):

#!/bin/bash
#PBS -N Job_name
#PBS -l walltime=10:30
#PBS -l nodes=4
#PBS -l ppn=6
#PBS -m be

module load mathematica

math -script hpc.wl

Example Mathematica script without debug comments (hpc_TORQUE.wl)

(* get list of provided resources, i.e. node names;  output is e.g. {NODE1, NODE1, NODE2, NODE3} *) 
h = ReadList[Environment["PBS_NODEFILE"], "String"];

(* for convenience, convert to association of node counts;  output is e.g. <|NODE1 -> 2, NODE2 -> 1, NODE3 -> 1|> *)
hosts = Counts[h];

(* launch remote kernels from each node, equal to the number of resources on the node *)
Needs["SubKernels`RemoteKernels`"];
Map[ LaunchKernels[RemoteMachine[#, hosts[#]]]&, Keys[hosts] ];

(* ========== Begin user code ========== *)

(* For example: tabulate the first 1000 prime numbers *)
result = ParallelTable[Prime[i],{i,1000}];
Save["result.txt", result];

(* ========== End user code ========== *)

(* clean up use of subkernels *)
CloseKernels[];
Quit[];

Example Mathematica script with debug comments

(*DEBUG*) (*turn on WSTP link debugging*)
(*DEBUG*) Needs["Parallel`Debug`"];
(*DEBUG*) SetOptions[$Parallel,Tracers->{MathLink}];

(* get list of provided resources, i.e. node names;  output is e.g. {NODE1, NODE1, NODE2, NODE3} *) 
h = ReadList[Environment["PBS_NODEFILE"], "String"];

(* for convenience, convert to association of node counts;  output is e.g. <|NODE1 -> 2, NODE2 -> 1, NODE3 -> 1|> *)
hosts = Counts[h];

(* === Start optional code to make room for controlling kernel === *)

local = StringSplit[Environment["HOSTNAME"],"."][[1]];
hosts[local]--;

(* === End optional code to make room for controlling kernel === *)

(*DEBUG*) (* for debugging purposes: print job resources, and Wolfram Kernel path, to screen *)
(*DEBUG*) Print["Hosts = ", Keys[hosts]];
(*DEBUG*) Print["Hosts counts = ", Values[hosts]];
(*DEBUG*) Print["Local = ", local];
(*DEBUG*) Print["$RemoteCommand = ",$RemoteCommand];

(* launch remote kernels from each node, equal to the number of resources on the node *)
Needs["SubKernels`RemoteKernels`"];
Map[ If[hosts[#] > 0, LaunchKernels[RemoteMachine[#, hosts[#]]]]&, Keys[hosts]];

(*DEBUG*) (*check that kernels were launched correctly*)
(*DEBUG*) Print["Kernels = ", Kernels[]];
(*DEBUG*) Print["Kernel count = ", $KernelCount];

(* ========== Begin user code ========== *)

(* For example: tabulate the first 1000 prime numbers *)
result = ParallelTable[Prime[i],{i,1000}];
Save["result.txt", result];

(* ========== End user code ========== *)

(* clean up use of subkernels *)
CloseKernels[];
Quit[];