A Parallel Heap Object

Mathematica Message Passing Implementations

Anton Antonov

Wolfram Research, Inc.

Introduction

The data structure called heap is used implement priority queues. In a priority queue, the element with highest (or lowest) priority is deleted from the queue, while elements with arbitrary priority are inserted.

Heaps/priority queues are used in certain types of scientific algorithms for numerical integration, integer programming optimization, and searching. Sub-problems of the original problem are prioritized in order to enable the algorithms to reach a result or solution faster.

Parallel implementations of these algorithms that try to achieve good load balancing and optimal resources utilization would in general  require parallel heaps.

In this talk we present:

■ Implementation of the heap data structure

■ Heap visualization

■ Two parallel heap object implementations

■ Examples of using of the parallel heap objects

The code of this presentation is self-contained.

It should be clear how that parallel heap object can be in the contexts of the algorithms pointed above.

Heaps

The implementation below uses leftist trees.

Clear[HeapEmptyQ,RankHeap,HeapTopElement,MakeHeap];
HeapEmptyQ[HP[]]=True;
HeapEmptyQ[h_]:=False;
RankHeap[HP[]]=0;
RankHeap[HP[_,r_,___]]:=r;
HeapTopElement[HP[]]:=None;
HeapTopElement[HP[a_,_,___]]:=a;
MakeHeap[x_,a_HP,b_HP]:=
If[RankHeap[a]≥RankHeap[b],
HP[x,RankHeap[b]+1,a,b],
HP[x,RankHeap[a]+1,b,a]
];

Clear[HeapMerge];
HeapMerge[h_HP,HP[],___]:=h;
HeapMerge[HP[],h_HP,___]:=h;
HeapMerge[HP[x_,r1_,a1_,b1_],HP[y_,r2_,a2_,b2_],OrderFunc_]:=
If[OrderFunc[x,y],
MakeHeap[x,a1,HeapMerge[b1,HP[y,r2,a2,b2],OrderFunc]],MakeHeap[y,a2,HeapMerge[HP[x,r1,a1,b1],b2,OrderFunc]]
];

Clear[HeapInsert,HeapDelete];
HeapInsert[x_,h_HP,OrderFunc_]:=HeapMerge[HP[x,1,HP[],HP[]],h,OrderFunc];
HeapDelete[HP[x_,_,a_,b_],OrderFunc_]:=HeapMerge[a,b,OrderFunc];
HeapDelete[HP[],OrderFunc_]:=HP[];

Heap visualization definition

Clear[HeapNodeRule];
HeapNodeRule[graph_,HP[n_,r_,a_,b_]]:=
Module[{graphNew=graph},
If[a=!=HP[],
graphNew=Append[graphNew,{n,r}->(List@@Take[a,2])];
graphNew=HeapNodeRule[graphNew,a]];
If[b=!=HP[],
graphNew=Append[graphNew,{n,r}->(List@@Take[b,2])];
graphNew=HeapNodeRule[graphNew,b]];
graphNew
];

Clear[GraphHeap];
GraphHeap[h_HP]:=LayeredGraphPlot[HeapNodeRule[{},h],VertexLabeling->True,DirectedEdges->True];

Random heap and distribution functions

Clear[MakeRandomHeap];
MakeRandomHeap[nelems_?IntegerQ,randelemfunc_,orderFunc_]:=
Block[{data},
data=Table[randelemfunc[i],{i,nelems}];
Fold[HeapMerge[#1,MakeHeap[#2,HP[],HP[]],orderFunc]&,MakeHeap[First[data],HP[],HP[]],Rest[data]]
];

Clear[TakeFromHeap];
TakeFromHeap[heapArg_HP,offset_,step_,orderFunc_]:=
Block[{newheap,heap=heapArg},
heap=Nest[HeapDelete[#,orderFunc]&,heap,offset];
newheap=MakeHeap[HeapTopElement[heap],HP[],HP[]];
While[!HeapEmptyQ[heap],
heap=Nest[HeapDelete[#,orderFunc]&,heap,step];
If[!HeapEmptyQ[heap],
newheap=HeapInsert[HeapTopElement[heap],newheap,orderFunc]
];
];
newheap
];

Clear[MakeLargeElement];
MakeLargeElement[___]:={RandomReal[],RandomInteger[{0,100},{3}]}

h1=MakeRandomHeap[3,MakeLargeElement,(#1[[1]]>#2[[1]]&)]

h2=MakeRandomHeap[3,MakeLargeElement,(#1[[1]]>#2[[1]]&)]

HeapMerge[h1,h2,(#1[[1]]>#2[[1]]&)]

h=MakeRandomHeap[12,RandomInteger[{0,100}]&,Greater]

GraphHeap[h]

h1=TakeFromHeap[h,1,2,Greater]

GraphHeap[h1]

Operations for a parallel heap

Operations

■ Distribution of a heap from one processor to the others.

■ Collecting the heaps from all processors.

■ Consequent applications of the operations above, e.g. collect, merge, re-distribute -- this gives room for at least two different implementations (shown below).

Usage

A parallel algorithm that uses heaps goes through these steps:

1. The initial problem (domain, solution set) is partitioned into smaller sub-problems that are put in heap. The number of these sub-problems is, say, the same as the number of processors.

2. The heap is distributed among all processors. I.e. each processor gets at least one sub-problem from the heap.

3. Each processor works on its sub-problem for certain number of steps. Each processor changes its heap during that process.

4. The heaps from all processors are merged into one heap -- go to step 2. and use that heap for re-distribution.

ParallelHeap object definition

Module[{counter=0}, NewID[]:=counter++;ResetID[]:=counter=0;]

CreateParallelHeap[orderFunc_]:=Block[{obj},obj=ParallelHeap[NewID[]];obj["OrderFunction"]=orderFunc;obj["Heap"]=HP[];obj];

Clear[ParallelHeap];
ParallelHeap[d___]["Initialize"[]]:=
If[ListQ[$Slaves],
$Failed,
ParallelHeap[d]["Initialize"[ ConnectKernels[$Slaves]]]
];
ParallelHeap[d___]["Initialize"[env_PMPObject]]:=
Block[{},
ParallelHeap[d]["PMPObject"]=env;
ExportEnvironment[MakeLargeElement,HeapEmptyQ,RankHeap,HeapTopElement,MakeHeap,HeapInsert,HeapDelete,HeapMerge,MakeRandomHeap,TakeFromHeap,ParallelHeap];
];
ParallelHeap[d___]["Distribute"[heapArg_HP]]:=
Block[{counter=0,heap=heapArg},
While[!HeapEmptyQ[heap],
With[{topelem=HeapTopElement[heap],obj=ParallelHeap[d]},
RemoteEvaluate[
ParallelHeap[d]["Heap"]=HeapInsert[topelem,ParallelHeap[d]["Heap"],ParallelHeap[d]["OrderFunction"]]
,$Slaves[[counter+1]]]
];
heap=HeapDelete[heap,ParallelHeap[d]["OrderFunction"]];
counter=Mod[counter+1,Length[$Slaves]];
]
];
ParallelHeap[d___]["CollectAndDistribute"[]]:=
Block[{},
If[!NumberQ[$PMPKernelRank],
res=RemoteEvaluate[
ParallelHeap[d]["Heap"]=PMPFoldBroadcast[HeapMerge[#1,#2,ParallelHeap[d]["OrderFunction"]]&,ParallelHeap[d]["Heap"]],ParallelHeap[d]["PMPObject"]];
With[{nproc=Length[$Slaves]},RemoteEvaluate[ParallelHeap[d]["Heap"]=TakeFromHeap[ParallelHeap[d]["Heap"],$PMPKernelRank-1,nproc,ParallelHeap[d]["OrderFunction"]]]
];
];
ParallelHeap[d]
];

Sub-classing ParallelHeap object -- ParallelHeapTwo definition

Here we use static inheritance to create an object that has the same feel and behavior as the parallel heap object defined above, but uses a different implementation of the operation "CollectAndDistribute".

(*subclass that uses diffrent implementation of the CollectAndDistribute operation *)
SubValues[ParallelHeapTwo]=ReplaceRepeated[SubValues[ParallelHeap],ParallelHeap->ParallelHeapTwo];

(*ParallelHeapTwo creator*)
CreateParallelHeapTwo[orderFunc_]:=Block[{obj},obj=ParallelHeapTwo[NewID[]];obj["OrderFunction"]=orderFunc;obj["Heap"]=HP[];obj];

(*pos is position of the overridden method *)
pos=Position[First/@SubValues[ParallelHeapTwo],"CollectAndDistribute"[___],∞][[1,1]];
SubValues[ParallelHeapTwo]=Drop[SubValues[ParallelHeapTwo],{pos}];
ParallelHeapTwo[d___]["CollectAndDistribute"[]]:=
Block[{resheap},
If[!NumberQ[$PMPKernelRank],
resheap=First@RemoteEvaluate[PMPFold[HeapMerge[#1,#2,ParallelHeapTwo[d]["OrderFunction"]]&,ParallelHeapTwo[d]["Heap"]],ParallelHeapTwo[d]["PMPObject"]];
RemoteEvaluate[ParallelHeapTwo[d]["Heap"]=HP[],ParallelHeapTwo[d]["PMPObject"]];
ParallelHeapTwo[d]@"Distribute"[resheap];
];
ParallelHeapTwo[d]
];

Slaves launch and connection

Launch of the slaves

kernels= Table[LaunchSlave["localhost"],{3}]
env= ConnectKernels[kernels]

$Slaves

ParallelHeap object creation

Creation of the parallel heap object:

ResetID[];
pheap=CreateParallelHeap[Greater]

Initialization -- the parallel heap object has copied on the slaves:

pheap@"Initialize"[env]

Various queries about the parallel heap objects in the slaves...

RemoteEvaluate[ParallelHeap[0]@"OrderFunction",env]

RemoteEvaluate[ParallelHeap[0],env]

RemoteEvaluate[ParallelHeap[0]@"Heap",env]

RemoteEvaluate[Information[ParallelHeap],env]

Heap distribution

Make a heap with a random sequence of integers:

h=MakeRandomHeap[12,RandomInteger[{0,100}]&,Greater]

Make a heap with a permutation of consequent numbers:

list=RandomSample[Range[1,12]];counter=1;
h=MakeRandomHeap[12,list[[counter++]]&,Greater]

Visual representation:

GraphHeap[h]

Distribute the heap among the slaves:

pheap@"Distribute"[h]

Show the heaps on the slaves:

sheaps=RemoteEvaluate[ParallelHeap[0]@"Heap",env];
GraphHeap/@sheaps

Slave heaps changes

Function that makes changes to a heap:

Clear[HeapManipulate];
HeapManipulate[obj_ParallelHeap,{minn_,maxn_}]:=
Block[{h,n},
n=RandomInteger[{minn,maxn}];
h=MakeRandomHeap[n,RandomInteger[{0,100}]&,Greater];
obj["Heap"]=HeapMerge[obj["Heap"],h,Greater];
];

Export the definition of a function specially made to change the heaps on the slaves in a random way:

ExportEnvironment[HeapManipulate]

Do some random changes of the heaps the slaves:

RemoteEvaluate[HeapManipulate[ParallelHeap[0],{2,5}],env]

Show the heaps on the slaves:

sheaps=RemoteEvaluate[ParallelHeap[0]@"Heap",env];
GraphHeap/@sheaps

Heap (work load) re-distribution

Merger the slave heaps into one and re-distribute them:

pheap@"CollectAndDistribute"[]

Show the heaps on the slaves:

res=RemoteEvaluate[ParallelHeap[0]@"Heap",env];
GraphHeap/@res

Summary

The object implementation allows the use of different implementations of the parallel heap operations. These implementations might be chosen with regard to the parallel computer architecture and/or the size of the heaps and the communications/computations speed ratio.

Both objects used the same code that illustrates the parallel heap operations.

The code of this presentation is self-contained.

For more information about heaps implemented with leftist trees see: 
Algorithms: A Functional Programming Approach, Fethi A. Rabhi, Guy Lapalme,
Handbook of Data Structures and Applications, Dinesh P. Mehta, Sartaj Sahni.

Mathematica has the heap structure operations implemented/explained above in the NIntegrate context.

A notebook describing how to use an object-oriented style of programming that is inherent to Mathematica is available upon request. (The notebook shows exampels dynamic and static inheritance and design patterns code.)