Parallel Efficiency:

The Dataflow Advantage

John Morris
Centre for Intelligent Information Processing Systems,
Department of Electrical and Electronic Engineering,
The University of Western Australia,
Nedlands WA 6907, Australia
morris@[Link]

Abstract duced, but this benefit may be negated by techniques such

as placing one shared variable per page to avoid false shar-
Dataflow computation models enable simpler and more ef- ing problems.
ficient management of the memory hierarchy - a key barrier Whatever the coherence unit, a shared memory system will
to the performance of many parallel programs. suffer performance degradation - a fetch of a datum mod-
This paper enumerates some advantages of using the ified by a remote processor may stall a memory transac-
dataflow model; it argues that the programming model is tion for hundreds of clock cycles while the datum is located
simple and easily managed by a programmer and demon- and directories updated. Writes may also take hundreds of
strates some of the efficiencies that the dataflow model al- cycles[1].
lows an underlying run-time system to achieve. Thus convenience for a programmer may have a cost that
improvements in compiler technology may not be able to
surmount. This paper argues that, by substituting an equally
1. Introduction simple and easily comprehended computation model - the
dataflow model - better performance can be obtained with-
out affecting a programmer’s ability to write efficient, read-
Shared memory models are popular: the absence of explicit able programs. Following Arvind and Brock[3], a variety of
message passing directives is widely regarded as making side effect-free systems - applicative, functional, dataflow,
them simpler for the programmer[10, 5]. They do require etc. - are grouped together here under the term dataflow: the
a programmer to add explicit synchronisation directives, arguments apply equally to all side-effect free systems.
e.g. barrier calls, but requirements for some of these can Section 2 sets out some examples of the costs which may
be detected automatically and parallel algorithm descrip- be incurred with some basic examples of shared mem-
tions generally make the need for the remainder manifest. ory programs - chosen to illustrate various aspects of the
To keep a set of fast processors supplied with data from dataflow advantage: these examples are visited again in sec-
a shared memory, a system architect must design a com- tion 4 to show the performance enhancements possible. The
plex memory hierarchy with multiple levels of cache - all dataflow model and its use in a familiar, practical program-
of which must be kept consistent in order to present a con- ming system are described in Section 3.
sistent view of the memory to a programmer. The designer
must also choose a size for the coherence unit (also termed
granularity unit[16]) - the smallest block of memory that 2. Background
will be managed as a separate entity. A small coherence
unit (e.g. a cache line) allows finer grain parallelism and re- To illustrate the problem addressed here, three simple pro-
duces false sharing, but requires complex hardware, such grams are analysed in a shared memory system. The model
as that designed for FLASH[11] and consumes memory for consists of a distributed shared memory machine with hard-
directories. A larger coherence unit (e.g. a page) can be ware or software to maintain memory consistency. Each
managed by software, using page fault hardware, but can program thread sees a single address space. For simplic-
only achieve coarse-grain parallelism due to the load placed ity, a memory hierarchy consisting of one cache level, local
on the communication network. Directory overhead is re- memory and remote memory is assumed. Memory access
cost is assumed to increase by a factor of 10 for each level: PE left PE right
if a cache hit costs 2 machine cycles, a local memory access
costs 20 and a remote memory access 200 cycles - figures
reasonable for modern processors, but the arguments are not w c-w
affected by minor changes to them. Main memory access
01
time is denoted by tacc , so a cache hit costs : tacc and a c
remote memory access tacc. 10
The coherence unit is the smallest block of memory which Figure 1. Division of array to be sorted split-
the coherence hardware or software manages. An update to ting a coherence unit
one word of a coherence unit cannot be distinguished from
updates to neighbouring words in the same coherence unit.
The ideal coherence unit would be a single byte, but because
of efficiency and resource constraints, the smallest unit seen by PEright and the c words of the coherence unit will ‘ping-
in practical machines is a cache line. Some systems use a pong’ between two PEs. Since all PEs will start working on
coherence unit as large as a page. The false sharing phe- their portions of the aray at the same time, this worst-case
nomenon results from separate processing elements (PEs) is very likely. Assuming that an invalidate takes twice as
updating different words in the same coherence unit. The long as a normal remote memory access, an additional cost
size of the coherence unit is denoted c. A small coherence is added to t1 :
= 32
= (20 2) acc = 10
unit, csmall words (as FLASH[11]) and a large one,
clarge =2 10 10 3 words (cf. the page in Treadmarks[2]) t2 w= t wtacc cycles

have been chosen to illustrate the range of possibilities.

To estimate the effect of this on a real program, let us find
2.1. Sorting the point at which t1 =t2 , i.e. the point at which the co-

herence overhead doubles the memory access cost of the

partition phase of the sort. For clarge ,
Assume a parallel sort algorithm in which an array of point-
ers to be sorted is divided among p PEs. Each PE sorts its
t1 = ( )(17 80) acc
n=p = t

2 = 10 acc
section using quick sort and signals a master processor to 4
t t
merge the sorted sections. Here the parallel part of the al-
gorithm - sorting of the sections by individual PEs - alone is
considered. It is assumed that a programmer wants to write if 1 = 2 , then (
t t n=p) = 4 7  104 so each processor needs
:

portable code which will function correctly on a number of to work on  10 elements before the coherence overhead
6

processors and is not prepared to tune a program to one spe- becomes negligible (  5%). For small , (
< c ) = 37 and
n=p

cific processor’s architectural parameters. each processor’s portion must be  103 elements before
A division of the array will generally split a coherence unit. we could ignore coherence overhead. Whilst this would ap-
For two PEs, PEleft and PEright , w words are manipulated pear to be a reasonable problem size, the extraordinary cost
by PEleft and c w words by PEright , cf. figure 1. In a of achieving this small coherence unit needs to be kept in
common variant of quicksort’s partition algorithm, the pivot mind[11]!
is compared from the ends inwards: if an item on the right is This example assumes a sequential consistency model for
less than the pivot it is swapped with one from the left which coherence: if a weaker model, e.g. release consistency com-
is greater. Assuming that data is randomly distributed about bined with write-buffering[5], then much of the communi-
2
the pivot, 50% or w= words in the coherence unit will be cation cost in the above examples can be overlapped with
( )2
updated by PEleft and PEright will update c w = words. computation if the computation is coarse-grained enough.
Now to access every word in a block of n=p words (the This reduces the cost of maintaining consistency at the ex-
portion of the array allocated to PEleft ), PEleft will need pense of a complex system which keeps ‘diffs’ of updated
pages and reconciles them: the overhead for this operation
t1 = ( )(0 1(
n=p : m 1) + 1 ) acc
=m =m t cycles ensures that only coarse grain parallelism can be exploited.
This is clear in the results of Carter et al[5], in which com-
since one in every m words will incur a cache miss (starting munication costs were kept to a few percent of the compu-
with an empty cache - at least as far as the partition opera- tation costs.
tion is concerned) and the remaining m ( 1)
=m reads will However, as will be shown later, a trivial and natural mod-
result in cache hits. ification of the algorithm - using a dataflow computation
But w of these words are shared between two PEs. In the model - eliminates the ‘diffs’ and ensures that the parallel
worst case, each update by PEleft will be followed by one overhead is kept to that for essential data moves.
2.2. Game Playing x a c
b
A game playing program will (a) determine the number of
possible moves, n, from the starting position, (b) allocate
an array to receive ‘scores’ from those moves, (c) allocate
x
moves to PEs to score and (d) when the scores have been
returned, choose the best one.
The array of scores will fit in a small number of coherence
+
units and will be subject to updates from all PEs. In gen-
eral, each coherence unit will change ownership c times as
PEs complete evaluation of moves and update it - and, un-
x
less our underlying architecture and game happen to con-
veniently match (i.e. n=p = ac or cbn=pc =
a, for some
+
integer a), we will not see the convenient migration of the
scores array from one PE to the next as the finite differ-
ence problem allows, but random moves from PE to PE as
the PEs finish evaluations in some random order. Marking
the scores array ‘write-shared’ can clearly reduce the num- Figure 2. A dataflow program
ber of migrations, but again at the expense of considerable
complexity in the RTS itself and overhead when the various
updates are reconciled. of data being stored at intermediate stages in a program,
then you can view the arcs as being the data stores. How-
3. Dataflow programs ever, note that it is not necessary to name these intermedi-
ate stores so that data can be accessed by some subsequent
Functional programming languages have failed to attract computation: the program is explicit about where the data
any significant following in the general computing commu- will be consumed and their is no need for names to remove
nity. ambiguities. A variety of languages - both textual[4, 15]
Although many have been designed and implemented - with and graphical[8] - have been proposed which are partly or
widely varying performance levels - history would suggest wholly based on this dataflow model. Single assignment -
that biases against them are well entrenched. This paper cf. SISAL[4] - is another expression of the dataflow idea: a
attempts to redress some of these problems by demonstrat- value is assigned to each variable once only. The dataflow
ing the performance benefits that can be obtained with a model allows a programmer to expose - with minimal effort
dataflow approach: fortunately, there is at least one lan- - all of the parallelism present in a problem. The communi-
guage with a familiar base in which the benefits of this ap- cation:computation cost ratio in modern systems means that
proach can be easily realised. fine-grain parallelism can rarely be efficiently exploited and
recent work has focussed on efficient medium- or coarse-
grain parallelism. Of particular interest in the context of
3.1. The Dataflow Model
this paper is Cilk - a threaded extension to C based on the
dataflow model[7]. Cilk’s C parentage makes it easy to port
The dataflow computation model relies upon the firing rule: and also to incorporate existing C functions.
A computation is ready to execute when all of its
data has arrived. 3.2. Writing Dataflow Programs

Traditionally, dataflow programs are represented by Initially programmers find Cilk’s dataflow model somewhat
dataflow diagrams - a dataflow graph to evaluate ax2 bx c + + daunting, but a simple pardigm shift removes this problem.
is shown in fig 2. Data flows along arcs connecting compu- Rather than focus on the sequence of steps which expresses
tational units (circles in fig 2). When any computational unit an algorithm, one starts with the final result and works back-
has data on all its inputs, it is ready to fire. Once it has fired, wards from that - determining how to produce the final re-
it produces results on its outputs which are directed towards sult and the inputs to that computation, determining what
further computational units which may now become ready compuations produced those inputs, [Link] trivial shift in
to fire. True dataflow programs are functional programs - the way one thinks is easily accomplished because one in-
they do not have any concept of memory or other practical variably has a clear idea of the required outputs and it is thus
details of a computer’s architecture. If you need to think straightforward to work back from the output, throught the
 cilk void merge( cont int res,
byvalue int a[], int n ) f
ultimate, penultimate, etc., transformations to the inputs.
With this sequence of transformations in hand, one simply /* Merge data from a and despatch to res */
sets up a thread for the ultimate computation with empty ... g
‘slots’ into which the penultimate threads will deposit data
/* Parallel sort -
and repeats this process back to the inputs. res - pointer to slot in thread
In Cilk, a spawn statement creates a new context (termed which will use the sorted array
a closure), including pointer to thread code for the context, a - data to be sorted; ns - number of items
arguments, join counter and local variables for the thread. /
cilk void par sort( cont int res, byvalue int
Arguments may be supplied to the closure when it is cre- a[], int ns ) f
ated (see arguments for spawn q sort in fig 3) or posted int i, j, k, np, *c; cont int b;
to it later (indicated by the ? preceding b in the spawn r = spawn merge(res,?b[ns],ns)@0; /* Spawn the
merge in fig 3). The join-counter decrements towards 0 as merge thread */
np = ns / n pe;
arguments are supplied. When it reaches 0, the closure is for(k=0;k<n pe;k++) f /* Spawn the sorting
full and is placed on a ready queue - ready to execute by threads */
the firing rule. A diagram showing the n evaluate clo- spawn seq sort(b,:c[np],np) @ k f
sures each with a continuation pointing to the ‘slot’ in the for(j=0;j<np;j++) c[j] = a[i++];
choose best closure into which it will deposit a score is g
b = b + ns; /* Update the pointer to the merge
shown in fig 5. Cilk’s parallelism is managed entirely by the thread’s slot */
RTS - as each closure contains all of a thread’s data it may g
be run on any PE: a property of any side-effect free sys- g
tem. Note that the closures are single-assignment structures
Figure 3. Cilk code for parallel sort
- they are written only once.

3.3. Controlled Context once. This write-once (or single assignment) rule simplifies
the management of memory. In the case of the quicksort
From a performance point of view, the key benefit of the partition phase, instead of updating the pointer array, they
dataflow programming model is that the underlying soft- are copied into a second one. The RTS knows that this ar-
ware (RTS) or hardware (e.g. that responsible for ensuring ray is ‘write-once’ and can direct the underlying coherence
global cache coherence) can trivially determine coherence hardware or software to treat it appropriately. The ‘ping-
requirements and that they are generally considerably re- ponging’ of coherence units between PEs can be avoided
duced from those of a shared-memory program. In Cilk, altogether.
all memory allocated for closures is single-assignment and
In Cilk’s dataflow model, this is the only way to write the
does not need to be kept coherent: the RTS sends messages
program: the data in a partition is copied into a new con-
to the PE on which the closure resides to add data to the
text on which the next thread works. A key point here is
closure. A Cilk program runs like a message-passing one - that this modification is a simple and natural one to make
and can achieve the same level of efficiency. (it could even be applied to a conventional expression of
But, importantly, no burden is placed on the programmer
the problem) because it reflects an easy way of looking at
other than reversing the order in which a large computa-
the problem. The Cilk parallel sort in fig 3 splits the data
tion is ‘conceived’. Cilk has a send argument directive
into n pe sections and calls the library qsort function in
to post data to empty slots in closures, but the receive is the seq sort thread: the library qsort is assumed to be
implicit. (The Cilk version used here requires a sync di- heavily optimised. However to illustrate the simplicity of
rective also, but the need for this can be inferred from code the dataflow approach, a threaded version of quick sort in
analysis - as in our Dataflow Java[13]).
Cilk is shown in fig 4: the elegance of the ‘dataflow’ par-
tition operation here should be compared with any conven-
4. Examples tional quick sort partition function!
Here, coherence problems are eliminated by copying data:
Here we will re-examine the examples presented earlier. shared data (the data before the partition) is always read-
only, the compiler and RTS understand this and can sim-
4.1. Quicksort ply replicate data as required. Since this must happen in
any distributed memory system (shared or not) to trans-
In a dataflow approach, structures are never updated but fer the data to remote PEs, the dataflow approach does no
transformed into new structures. A value may be used by unneeded work and introduces no overhead. There is no
many subsequent computations but it is only ever created ‘memory’ in the dataflow model: thus the dataflow program
cilk void q sort( cont int res,
evaluate
byvalue int offset[1],
byvalue int a[], byvalue int n[1]) f choose_best
int k, n1, n2, off, nt = n[0], pivot = a[0]; evaluate
int *offl, *nl, *lower, *offr, *upper, *nu;
jc = 0
off = offset[0]; evaluate
if ( nt <= 1 ) f if ( nt == 1 ) f best
move jc = 0
res = res + off; evaluate
send argument( res, pivot );
g jc = 0
g
else f /* Spawn lower / upper sorts */ jc =evaluate
0
spawn q sort(res,:offl[],:lower[nt],:nl[]) f
spawn q sort(res,:offr[],:upper[nt],:nu[]) f
n1 = n2 = 0; offl[0] = off;
/* Distribute data to lower or upper */
for(k=1;k<nt;k++) f
jc = n jc = 0

if( a[k] < pivot ) lower[n1++] = a[k];

else upper[n2++] = a[k]; Figure 5. Structure of the Cilk program for
g
off = off + n1; offr[0] = off+1; nu[0] = n2; game playing
g
nl[0] = n1; Each evaluate closure has a continuation to an empty slot
g in the choose best closure. Initially the join counter for
choose best is n. The evaluate closures, however, are sup-
res = res + off; /* Calculate pivot slot address
*/
send argument( res, pivot ); /* Send the pivot */ plied with all their data when they’re spawned, so have join coun-
g ters of 0.
g
Figure 4. Cilk code for threaded quick sort known and have been used in these optimisations. As one
example, Skeppstedt and Stenstrom[17] use them to reduce
the cost of invalidations when the compiler can detect a
in fig 3 doesn’t use nor require shared memory and appears load followed by a store to the same location and issue
more like a message passing version - without explicit re- a hint to the cache that an exclusive copy of the datum
ceives or synchronisations. should be loaded. However, this paper, like many others,
for example[9], focusses on optimisation of existing pro-
4.2. Games grams. There seems to be a general reluctance to re-write
programs to fit into simpler, more efficient models!
In a game-playing program, the random migration of the fi- Other techniques introduce additional hardware or soft-
nal scores array as individual PEs update it (or alternatively ware primarily to reduce message traffic - the major bot-
the management of ‘diff’s when the array has been marked tleneck to speed-up and scalability. For example, inval-
write-shared) is completely avoided. idation techniques have been described by a number of
A Cilk program creates a choose best thread with empty researchers[12],.. They rely on the transparency of cache -
slots for the scores, cf. fig 5. It then spawns multiple for example, if a block is invalidated in any cache too early,
evaluate threads to evaluate individual moves. As each then performance will suffer as the block is fetched again,
evaluate thread completes its evaluation, it posts a re- but program correctness is not affected. They work with
sult back to the choose best thread. Again the need whatever a programmer or compiler throws at them and
for shared memory has disappeared: data is transferred be- generally rely on observation of reference patterns rather
tween PEs with send directives which post data into empty that attempting make use of any model of execution to guide
slots in a waiting closure. Receive and synchronisation are their actions.
implicit: the choose best thread ‘fires’ when the last
evaluate thread has posted its result. Thus no overhead
6. Conclusion
messages are sent.
The simple examples presented here show the potential to
5. Related work use the dataflow computation model in a familiar environ-
ment - the C-based Cilk system - and avoid significant over-
There is also a considerable body of work on static and heads without requiring the programmer to bother with di-
dynamic detection of data access patterns to reduce cache rectives relating to coherence requirements. In a dataflow-
coherence overheads, Dataflow techniques are also well- based model, the RTS is able to use its knowledge of the
model semantics to infer the ‘write-once’ characteristics of [4] D. C. Cann, J. T. Feo, A. D. W. Bohoem, and R. R. Old-
many structures and treat them appropriately. In Cilk, this ehoeft. SISAL Reference Manual: Language Version 2.0,
is achieved by generating messages to update parts of the 1992.
structure in some fixed part of the distributed memory. Cilk [5] J. B. Carter, J. K. Bennet, and W. Zwaenepoel. Techniques
for Reducing Consistency-Related Communication in Dis-
programs run without dependence on a memory model: true
tributed Shared-Memory Systems. ACM Transactions on
dataflow programs have no addressable memory in the nor-
Computer Systems, 13(3):205 – 243, 1995.
mal sense. Thus they gain the efficiency of message passing [6] A. L. Cox and R. J. Fowler. Adaptive Cache Coherency for
systems because they do not require overhead transactions Detecting Migratory Shared Data. In Proceedings of The
for memory coherence. Although they occasionally gener- 20th Annual International Symposium on Computer Archi-
ate more local copying than a shared memory system might tecture, San Diego, U.S.A., pages 98–108, May 1993.
- and thus make less effective use of any cache present - [7] M. Halbherr, Y. Zhou, and C. Joerg. MIMD-Style Paral-
they compensate for this by only copying data between re- lel Programming based on Continuation-Passing Threads.
mote memories when the copies are necessary. Other mech- Technical Report 387, MIT Laboratory for Computer Sci-
anisms developed primarily for shared memory systems, ence, 1994.
[8] N. Harvey and J. Morris. Nl: A parallel programming visual
such as combining updates into single messages, are also
language. Australian Computer Journal, 28:2–12, 1996.
possible1 [9] G. Jin, Z. Li, and F. Chen. An efficient solution to the cache
Although shared-memory systems have been shown to thrashing problem caused by true data sharing. IEEE Trans
be capable of approaching the efficiency and speed of Computers, 47(5):527–543, 1998.
message-passing ones, the best results to date have still [10] A. C. Klaiber and H. M. Levy. A comparison of message
shown them lagging - sometimes by small amounts, but passing and shared memory architectures for data parallel
nevertheless always lagging. Cilk programs function more programs. In Proceedings of the 21st International Sympo-
like message-passing ones; they send messages to post data sium on Computer Architecture, pages 94–105, 1994.
into closures. No coherence problems arise due to the [11] J. Kuskin, D. Ofelt, M. Heinrich, J. Heinlein, R. Si-
moni, K. Gharachorloo, J. Chapin, D. Nakahira, J. Baxter,
model semantics. Thus Cilk programs will generally run
M. Horowitz, A. Gupta, M. Rosenblum, and J. Hennessy.
at rates similar to message-passing ones - without requiring The Stanford FLASH Multiprocessor. In Proceedings of The
the programmer to add the messaging code and synchroni- 21th Annual International Symposium on Computer Archi-
sations! This paper has focussed on Cilk partly because its tecture, Chicago, Il, pages 302–313, 1994.
C heritage makes it a practical - it is able to take advantage [12] A. R. Lebeck and D. A. Wood. Dynamic self-invalidation:
of C compiler technology and it is possible to easily control Reducing coherence overhead in shared-memory multipro-
granularity - and acceptable - it is able to use much existing cessors. In Proc. of the 22nd Annual Int’l Symp. on Com-
code: however the general thesis of this paper applies to all puter Architecture (ISCA’95), pages 48–59, Jun 1995.
other side-effect free systems. [13] G. Lee and J. Morris. Dataflow java: Implicitly parallel java.
Technical Report TR99-1, Centre for Intelligent Information
The sorting and finite differencing examples shows that the
Processing Systems, University of Western Australia, 1999.
dataflow model can sometimes be applied to shared mem- [14] Motorola. PowerPC 601: RISC Microprocessor User’s
ory environments also: emphasising that the dataflow model Manual. Motorola, Inc, 1992.
should be part of a parallel programmer’s tools - whatever [15] R. S. Nikhil. Id Reference Manual, Version 90.1. CSG
the programming environment. Memo 284-2, Laboratory for Computer Science, MIT, Cam-
bridge MA, Sep 1990.
[16] J. Protic, M. Tomasevic, and V. Milutinovic. Distributed
References shared memory: Concepts and systems. IEEE Parallel and
Distributed Technology, 4(2):63–79, 1996.
[1] G. A. Abandah and E. S. Davidson. Characterizing dis- [17] J. Skeppstedt and P. Stenstrom. Using dataflow analysis
tributed shared memory performance: A case study of the techniques to reduce ownership overhead in cache coherence
convex SPP1000. IEEE Transactions on Parallel and Dis- protocols. ACM Trans Prog Lang Systems, 18(6):659–682,
tributed Systems, 9(2):206–216, Feb 1998. 1996.
[2] C. Amza, A. L. Cox, S. Dwarkadas, P. Keleher, H. Lu, R. Ra-
jamony, W. Yu, and W. Zwaenepoel. TreadMarks: Shared
Memory Computing on Networks of Workstations. IEEE
Computer, 29(2):18–28, 1996.
[3] Arvind and D. J. Brock. Resource managers in functional
programming. Journal of Parallel and Distributed Comput-
ing, 1(1):5–21, Aug 1984.
1 Simple modifications to the Cilk RTS would enable it to combine mes-

sages destined for the same PE.