\input epsf
\input{psadobe}
\makeatletter
\long\def\unmarkedfootnote#1{{\long\def\@makefntext##1{##1}\footnotetext{#1}}}\makeatother
\documentstyle[times]{cmu-art}
\include{psfig}

\begin{document}
\bibliographystyle{acm}

\title{
\begin{flushleft}
\vspace{-0.9in}
\scriptsize {\em 
IEEE Parallel \& Distributed Technology, Volume 2, Number 3, Fall, 1994, pp. 16-26
}
\end{flushleft}
\vspace{0.4in}
\Large\bf Task Parallelism in a High Performance Fortran Framework
}
\author{T. Gross, D. O'Hallaron, and J. Subhlok \\
School of Computer Science\\
Carnegie Mellon University\\
Pittsburgh, PA 15213\\
}
\date{}
\thispagestyle{empty}
\maketitle

\unmarkedfootnote{
This research was sponsored in part by the Advanced Research Projects
Agency/ CSTO monitored by SPAWAR under contract N00039-93-C-0152, and
in part by the Air Force Office of Scientific Research under Contract
F49620-92-J-0131.
}

\begin{abstract}
High Performance Fortran (HPF)  has emerged 
as a standard dialect of Fortran for data parallel computing.
However, for a wide variety of applications, both task and data
parallelism must be exploited to achieve the best possible performance
on a multicomputer.  We present the design and implementation of a
Fortran compiler that integrates task and data parallelism in an HPF
framework.  A small set of simple directives allow users to express
task parallel programs in a variety of domains.  The user identifies
opportunities for task parallelism, and the compiler handles task creation and
management, as well as communication between tasks.  Since a unified
compiler handles both task parallelism and data parallelism, existing
data parallel programs and libraries can serve as the building blocks
for constructing larger task parallel programs.  This paper concludes
with a description of several parallel application kernels that were
developed with the compiler.  The examples demonstrate that exploiting
data and task parallelism in a single framework is the key to achieving
good performance for a variety of applications.
\end{abstract}

\section{Introduction}

Compilation of programs for parallel computers has received
considerable attention for many years.  Several parallelizing
compilers have been developed for data parallel programs, including
Fortran~D~\cite{TsHK93} and Vienna Fortran~\cite{chapman92a}.  High
Performance Fortran~\cite{hpf93} (HPF) has emerged as a standard
dialect of Fortran for data parallel computing.  The core of HPF
contains a set of extensions to describe data mappings and parallel
loops. These allow programmers to write and compile data parallel
programs for a variety of architectures.  However, in its current
form, HPF does not address {\em task parallelism} or {\em
heterogeneous} computing adequately.  Applications that require
different processor nodes to execute different programs, possibly on
different data sets, cannot be programmed effectively in HPF.
There is growing interest in the idea of exploiting both task and data
parallelism
~\cite{agrawal93,chandy94,chapman94,cheung93,choudhary94,crovella93,foster92,subhlok93}.
There are a number of practical reasons for this interest:

{\em Limited scalability:} Many applications, especially in the
domains of image and signal processing, do not scale well when using
data parallelism, because data set sizes are limited by physical
constraints, or because they have a high communication overhead.  For
example, in multibaseline stereo~\cite{webb93aa}, the main data set is
an image whose size is determined by the camera interface. Task
parallelism makes it possible to execute individual computations on a
subset of nodes and thus improves performance, despite limited
scalability.  

{\em Real-time requirements:} Many real-time applications (e.g. in
robot control) have strict latency and throughput requirements. Task
parallelism allows the programmer to partition resources (including
processor nodes) explicitly among the application modules to meet such
requirements. By supporting {\em both } task and data parallelism in a
single framework, the user can tailor the mapping of an application
to a particular performance goal.


{\em Multidisciplinary applications:} Task parallelism can be used to
effectively manage heterogeneity in applications and execution
environments.  There is an increased interest in parallel
multidisciplinary applications where different modules represent
different scientific disciplines and may be implemented for different
parallel machines.  For example, the airshed
model~\cite{mcrae82,mcrae92} represents a ``grand challenge''
application that characterizes the formation of air pollution as the
interaction between wind and reactions among various chemical species.
It is natural to model such interactions using task parallelism; e.g.
one module (or task) models the effect of the wind, and a different
module models the chemical reactions.  Further, the use of task
parallelism is necessary if different modules are designed to execute
on different types of parallel or sequential machines.

%HOW TO SUPPORT IT

There are many ways in which task and data parallelism can be
supported together in a programming environment.  A fundamental design
decision is whether the programmer has to write programs with explicit
communication, or if the responsibility of communication
generation is delegated to the compiler.  One of the benefits provided
by data parallel languages like HPF is that they liberate the
programmer from dealing with the details of communication, which is a
cumbersome and error prone task.  If task parallelism is to find
acceptance, writing task parallel programs must be no harder than
writing data parallel programs, and therefore, in our design, all
communication operations are generated by the compiler. The user
writes programs for a common data space, and the compiler maps the
data objects to the (possibly disjoint) address spaces of the nodes of
the parallel system.  This division of responsibility also allows
communication optimizations by the compiler. Other fundamental design
decisions include whether task management should be static or dynamic,
and strategies for processor allocation and load balancing.

%OUR IMPLEMENTATION
We have designed and implemented task parallelism as directives in a
data parallel language based on HPF. This prototype compiler is called
Fx~\cite{subhlok93,stichnoth94}.\footnote{There are two
explanations for this name. On one hand, the "x" emphasizes that the
language and directives may still undergo further development, and the
"F" emphasizes how irrelevant details of the base language (Fortran)
are.  On the other hand, efficient translation of programs for
parallel machines often brings to mind the use of special effects.}
Our objectives are to develop a system that produces efficient code,
and to use this system to develop applications that need task and data
parallelism.  The current targets for this compiler are an iWarp
parallel machine, networks of workstations running PVM, and the Cray
T3D.  The compiler has been used to develop a variety of task and data
parallel applications, including synthetic aperture radar, narrowband
tracking radar, and multibaseline stereo~\cite{dinda94,subhlok94a}.

There are obvious practical advantages of extending HPF for task
parallelism instead of inventing a new language.  Existing sequential
and data parallel libraries can be used, it is easier to convert
existing programs to task and data parallel programs, and it is easier
to gain user acceptance. Finally, it is important to be able to
compile task and data parallel programs efficiently using existing
compiler technology.
% If the compiler-generated programs are not
%competitive with programs written by hand, it is very difficult to
%gain user acceptance.
In particular, we allow several directives to
help the compiler in generating efficient code, even though some
directives may become obsolete as more sophisticated compilers become
available.

The paper is organized as follows. Section~\ref{sec:motivation}
provides the fundamental motivation for exploiting task and data
parallelism and introduces an example program, which is re-used
throughout the paper to illustrate various concepts.
Section~\ref{sec:usermodel} describes the user model
that Fx presents for writing task and data parallel programs.
%This section discusses
%the tradeoffs necessary to obtain a system that can be implemented
%efficiently together with support for data parallelism.
Section~\ref{sec:compiling} discusses the main issues involved in
compiling such programs and describes how task
parallelism and data parallelism coexist in one 
framework.  Section~\ref{sec:applications} reports  experience
with applications written for the Fx compiler; this section
demonstrates the performance benefits that can be obtained by
finding the right balance between task and data parallelism.
%(compared to an environment that solely supports data parallelism).

\section{Requirements for efficient parallelization}
\label{sec:motivation}

Many applications must exploit both task and data parallelism for
efficient execution on massively parallel programs.  Consider the
following example application kernel (called FFT-Hist) from signal and
image processing.  Input is a sequence of $m$ $512 \times 512$ complex
arrays from a sensor (e.g., a camera).  For each of the $m$ input
arrays, we perform a 2D fast Fourier transform (FFT), followed by some
global statistical analysis of the result, including constructing a
histogram.  The 2D~FFT consists of a 1D FFT on each column of the
array, followed by a 1D~FFT on each row.  The main loop nest in
FFT-Hist is shown in Figure~\ref{fig:example}.
\begin{figure}[htb]
\begin{minipage}[b]{3.5in}
\tt
\begin{tabbing}
tttttttt\=ttt\=ttt\= ttt\=ttt\= \kill 
\>     do i = 1,m \\
\>\>     call colffts(A)\\
\>\>     call rowffts(A)\\
\>\>     call hist(A)\\
\>     enddo\\
\end{tabbing}
\rm
\end{minipage} 
\begin{minipage}[b]{4.0in}
\centerline{\epsfbox{figs/example.eps}}
\end{minipage}
\caption{FFT-Hist example program and task graph}
\label{fig:example}
\end{figure}

For each iteration of the loop, the {\tt colffts} function inputs the
array {\tt A} and performs 1D~FFTs on the columns, the {\tt rowffts}
function performs 1D~FFTs on the rows, and the {\tt hist} function
analyzes and outputs the result.  This is an interesting program
because it represents the structure of a large class of applications
in image and signal processing, and because it illustrates some
important tradeoffs between different styles of mapping programs onto
parallel systems. We use this simple program as a running example
throughout the rest of the paper.

Suppose we have at our disposal parallel versions of the three
functions in Figure~\ref{fig:example} so that each function can run on
1 or more nodes.  A compiler or user has to make a decision on how
many nodes to assign to each function.  Figure~\ref{fig:speedup}
depicts the speedup obtainable for these three functions, as a
function of the number of nodes.  The {\tt colffts} function performs
an independent 1D~FFT on each column of {\tt A}. So if we assign blocks of
columns to nodes, we can run all of the nodes independently, and the
function scales almost linearly up to 512 nodes. The {\tt rowffts}
function behaves in the same way if the array is distributed row-wise
among the nodes. The key point is that neither {\tt colffts} nor {\tt
rowffts} generates any communication, and thus each scales well. On
the other hand, the {\tt hist} function contains significant
communication and thus does not scale well.
\begin{figure}[htb]
%\epsfxsize=3in
\centerline{\epsfbox{figs/speedup.eps}}
\caption{Speedup curves for the functions in FFT-Hist.}
\label{fig:speedup}
\end{figure}

Given that only two out of the three functions scale well, how do we
go about parallelizing the loop nest in Figure~\ref{fig:example}?  One
approach is a purely data parallel mapping: use all of the nodes to
execute {\tt colffts}, then use all of the nodes to execute {\tt
rowffts}, then use all of the nodes to execute {\tt hist}, and so on.
As the number of nodes increases, this purely data parallel approach
works well for the {\tt colffts} and {\tt rowffts} functions but makes
inefficient use of the nodes during the {\tt hist} routine because {\tt hist}
does not scale well.

To achieve good efficiency for functions like the {\tt hist} function,
we must allocate a small number of nodes to it.  So for large parallel
systems, how do we use up the remaining nodes?  The answer is to
exploit a mix of task and data parallelism.

\section{User model for task and data parallelism}
\label{sec:usermodel}

The input language for Fx is based on HPF: The array statments of
Fortran 90 augmented with data layout statements and a FORALL-like
parallel loop construct~\cite{yang93}.  These constructs are described
briefly in Section~\ref{sec:dp}.

In the Fx model, a {\em task} corresponds to the execution of a call
to a {\em task-subroutine}. A task-subroutine is a data parallel
subroutine, with well-defined side-effects, contained inside a special
code section in the main program called a {\em parallel section}.
The only allowable side-effect of calling a
task-subroutine is that the values of its actual parameters might
change.  For each lexical call to a task-subroutine, the programmer
provides (1)  hints that indicate  if an actual parameter is read
and/or modified by the task subroutine, and (2) directives that
control the mapping of the task-subroutines onto nodes. These hints and 
directives are described later in Section~\ref{sec:directives}.

The execution model for an Fx program is as follows: The program
begins execution as a single data parallel task running on all nodes.
When the flow of control reaches a parallel section, the tasks
specified by calls to task-subroutines inside the parallel sections
are executed subject to data dependence constraints, i.e., each task
waits for its input, executes, sends its output, and terminates.
Parallelism is obtained by executing different tasks on different sets
of nodes.  When all tasks have terminated, the execution of the
parallel section is over, and the program continues execution as a
single data parallel task.  

Figure~\ref{fig:runtime} depicts some possible executions of FFT-Hist
for $m=4$ iterations on a parallel system.  Details of the
organization of the parallel system do not matter at this time.  For
each node, this figure indicates what function of FFT-Hist is executed
on this node at a given time.  In Figure~\ref{fig:runtime}(a), the
main program starts on all of the nodes.  Once inside the parallel
section, the task-subroutines execute one after the other; each
task-subroutine runs on all of the nodes.  After 4 iterations of the
loop, the main program resumes executing on all the nodes.  Another
possibility is shown in Figure~\ref{fig:runtime}(b), where each
task-subroutine runs on a disjoint set of nodes, and thus the
computation is pipelined.  Notice that the {\tt hist} function takes
about the same time as in the mapping of Figure~\ref{fig:runtime}(a)
-- using more nodes did not shorten execution time in
Figure~\ref{fig:runtime}(b).  Yet another option is depicted in
Figure~\ref{fig:runtime}(c).

\begin{figure}[tp]
%\epsfxsize=4in
\centerline{\epsfbox{figs/runtime.eps}}
\caption{Execution of FFT-Hist on a parallel system}
\label{fig:runtime}
\end{figure}

Since the data relationship of the calling program to the task-subroutines
is well defined, the compiler can map the tasks on different sets of nodes,
and generate communication to maintain data consistency.  For our
application domains, the runtime behavior of tasks can be accurately
predicted before execution, so issues like load balancing and task
migration are currently not of concern to our compiler.  Load balancing can
be influenced by the user's choice of data layout, using the HPF layout
directives.

The basic idea governing the role of directives is that 
the results obtained from parallel execution must be consistent with those
obtained from sequential execution.
The main characteristics of the user model can then be summarized as
follows: (1) There are no new language constructs, only compiler
directives in the form of comments.  (2) There is a common name space
for shared data.  (3) Tasks are represented as calls to data parallel
subroutines with well-defined side-effects.  (4) Communication between
tasks is generated and managed by the compiler.  (5) Sequential
consistency, determinism, and freedom from deadlock are guaranteed by
the compiler.

The rest of this section discusses these ideas in more detail. 
Section~\ref{sec:dp} gives a brief overview of the data parallel 
constructs, and Section~\ref{sec:directives} describes the
directives available to users. 

\subsection{Data parallel constructs}
\label{sec:dp}
Data parallelism is expressed with array
statements (as in HPF) and a FORALL-like parallel loop called the
PDO~\cite{yang93}.  Fx supports BLOCK, CYCLIC, and BLOCK-CYCLIC
distributions in an arbitrary number of array dimensions.  Consider
the following example:
\tt
\begin{tabbing}
tttttttt\=ttt\=ttt\= ttt\=ttt\= \kill 
c\$\>{\bf template} t(n)\\
c\$\>{\bf align} A(i,j) {\bf with} t(i)\\
c\$\>{\bf align} B(i,j) {\bf with} t(j)\\
c\$\>{\bf distribute} t(CYCLIC)\\
\>\\
\>{\bf pdo} i=1,n\\
\>\>A(i,:)\ = A(i,:)\ + B(:,i)\\
\>{\bf enddo}
\end{tabbing}
\rm
This example uses {\tt template}, {\tt align}, and {\tt distribute}
directives to distribute the rows of array {\tt A} and the columns of
array {\tt B} cyclically across the parallel system.  In the example
above, the $ith$ loop iteration uses an array statement to add the
$jth$ column of {\tt B} to the $ith$ row of {\tt A}. Moreover, each
loop iteration is independent and can run in parallel with the other
loop iterations.  

\subsection{Task parallel directives}
\label{sec:directives}

We have not introduced any new language features and rely entirely on
compiler directives for expressing task parallelism.  To simplify the
implementation, the current version of Fx relies on the user to
identify the side effects of the task-subroutines and to specify
them. 
Directives are also used to guide the compiler in making
performance related decisions like program mapping.  In this section,
we describe the directives and hints that are used to express task
parallelism and illustrate their use for the  FFT-Hist example.


\subsubsection*{Parallel sections}
Calls to task-subroutines are permitted only in special code regions
called {\em parallel sections}, denoted by a {\tt begin parallel/end
parallel} pair. For example, the parallel section for the FFT-Hist
example has the following form:
\tt
\begin{tabbing}
tttttttt\=ttt\=ttt\= ttt\=ttt\= \kill 
c\$ \>{\bf begin parallel} \\
\>     do i = 1,m \\
\>\>     call colffts(A)\\
c\$\>\>  {\em input/output and mapping directives}\\ \vspace{3mm}
\>\>     call rowffts(A)\\
c\$\>\>  {\em input/output and mapping directives}\\ \vspace{3mm}
\>\>     call hist(A)\\
c\$\>\>  {\em input/output and mapping directives}\\
\>     enddo\\
c\$\>   {\bf end parallel}
\end{tabbing}
\rm
The code inside a parallel section can only contain loops and
subroutine calls.  These restrictions are necessary to make it
possible to manage shared data and shared resources (including
nodes) efficiently at compile time.

A parallel section corresponds to a mapping of task-subroutines to
nodes. The corresponding mapping outside the parallel section
is a simple data parallel mapping, where every routine is mapped
to all nodes. The current implementation does not allow
nesting of parallel sections.

\subsubsection*{Input/output directives}

The user includes {\tt input} and {\tt output} hints to 
define the side-effects of a task-subroutine, i.e., the data space
that the subroutine accesses and modifies.  Every variable whose value
at the call site may potentially be used by the called subroutine must
be added to the input parameter list of the task-subroutine.
Similarly, every variable whose value may be modified by the called
subroutine must be included in the output parameter list.  A variable
in the input or output parameter list can be a scalar, an array, or an
array section.  An array section must be a legal Fortran~90 array
section, with the additional restriction that all the bounds and step
sizes must be constant.

For example, the input and output directives for the call to {\tt
rowffts} have the form:
\tt
\begin{tabbing}
tttttttt\=ttt\=ttt\= ttt\=ttt\= \kill 
\>\>     call rowffts(A)\\
c\$\>\>  {\bf input} (A),  {\bf output} (A)\\
c\$\>\>  {\em mapping directives}\\
\end{tabbing}
\rm
This tells the compiler the subroutine {\tt rowffts}
can potentially use values of, and write to the parameter array {\tt A}.
As another example, the input and output directives
for the the call to {\tt colffts} has the form:
\tt
\begin{tabbing}
tttttttt\=ttt\=ttt\= ttt\=ttt\= \kill 
\>\>     call colffts(A)\\
c\$\>\>  {\bf output} (A)\\
c\$\>\>  {\em mapping directives}\\
\end{tabbing}
\rm

This tells the compiler that subroutine {\tt colffts} does not use the value
of any parameter that is passed but  can potentially write to array {\tt A}
(which is set to values read from a sensor by {\tt colffts}).

\subsubsection*{Mapping directives}

Exploiting task and data parallelism together opens a variety of ways
to map a computation onto a parallel machine. In the Fx model, we
characterize mappings in terms of three attributes: {\em clustering},
{\em degree of replication}, and {\em node allocation}.

A {\em clustering} is an assignment of task-subroutines to {\em
modules}.  At run time, each task-subroutine in a module runs on the
same set of nodes, and each module runs on a unique set of nodes.  For
example, Figure~\ref{fig:newforms}(a)--(c) shows three possible
clusterings of FFT-Hist.  Figure~\ref{fig:newforms}(a) shows the
familiar {\em data parallel mapping} where all task-subroutines are
assigned to the same module; this corresponds to the schedule in
Figure~\ref{fig:runtime}(a).  Figure~\ref{fig:newforms}(b) shows a
{\em purely task parallel mapping} where each task-subroutine is
assigned to a different module; this corresponds to the schedule in
Figure~\ref{fig:runtime}(b).  Figure~\ref{fig:newforms}(c) shows a
mapping that is a mix of both.
\begin{figure}[ht]
\centerline{\epsfbox{figs/newforms.eps}}
\caption{Combinations of task and data parallel mappings.}
\label{fig:newforms}
\end{figure}

If the data sets in the input sequence of a module are independent,
and the module carries no internal state, then that module can be {\em
replicated}. Each copy of the module is called a {\em module
instance}.  If the module is replicated into $k$ instances, then we
say that the mapping uses {\em k-way} replication, or equivalently,
that the {\em degree of replication} for that module is $k$.
Module instances execute the calls to the corresponding subroutines in
a round robin order such that each instance executes only {\em 1/k}th
of the total number of calls (except for boundary conditions). For
example, Figures~\ref{fig:newforms}(d)--(e) show mappings with 2-way
replication for all modules; one replicated instance executes the
even-numbered iterations, and the other replicated instance executes
the odd-numbered iterations .  In Figure~\ref{fig:newforms}(f), the
first module is not replicated (i.e., there is only one instance), and
the second module is replicated into 4 instances; this corresponds to
the schedule in Figure~\ref{fig:runtime}(c).

Finally, there is an assignment of  nodes to module
instances.  This attribute is approximated graphically in
Figure~\ref{fig:newforms} by the relative sizes of the rectangles.
For example, in Figure~\ref{fig:newforms}(c), each module instance is
assigned half of the nodes.  In Figure~\ref{fig:newforms}(f), the
single instance of the first module is assigned 
24 of the available 64 nodes, and
each instance of the second module is assigned 10 nodes each.

Often the programmer has a good idea of how a computation should be
mapped but does not want to deal with the low level details of the
mapping.  To allow a programmer to pass this information to the
compiler, we include mapping directives.  By their very nature, the
effect of such mapping directives is machine specific (the directives
are not).  For example, a user may want to indicate that some sets of
tasks be mapped to physically adjacent nodes.  The number of nodes
that are adjacent depends on the architecture of the target machine (4
for a 2D-torus, 6 for a 3D-torus, etc.), but in our experience, such
machine-specific hints can improve the performance dramatically.
Nevertheless, these directives have no semantic meaning; if ignored by
the compiler, performance may suffer but correctness is maintained.

Fx includes the  {\tt processor} and {\tt origin} directives to describe the
clustering of task-subroutines into modules, the allocation of nodes to
modules, and the replication of modules.
The {\tt processor} directive
states how many nodes should be assigned to a task-subroutine.  The {\tt
origin} directive states the location(s) of the task-subroutine in the
parallel system.  In the current implementation, only rectangular subarrays
can be assigned to task-subroutines, and the parallel system is assumed to
be organized as a two dimensional space, with node (0,0) at the top left of
the system. Hence {\tt processor} and {\tt origin} directives contain
pairs of numbers referring to the size and location of a rectangular
subarray of nodes, respectively.
For example, to map FFT-Hist as shown in Figure~\ref{fig:newforms}(f)
onto an $8 \times 8$ array of nodes, we use the following mapping
directives:
\tt
\begin{tabbing}
tttttttt\=ttt\=ttt\= ttt\=ttt\= \kill 
c\$\>{\bf begin parallel} \\
\>     do i = 1,m \\
\>\>     call colffts(A)\\
c\$\>\>  {\bf output} (A)\\
c\$\>\>  {\bf processor} (8,3)\\
c\$\>\>  {\bf origin} (0,0)\\

\>\>     call rowffts(A)\\
c\$\>\>  {\bf input} (A),  {\bf output} (A)\\
c\$\>\>  {\bf processor} (2,5)\\
c\$\>\>  {\bf origin} (0,3), (2,3), (4,3), (6,3)\\

\>\>     call hist(A)\\
c\$\>\>  {\bf input} (A)\\
c\$\>\>  {\bf processor} (2,5)\\
c\$\>\>  {\bf origin} (0,3), (2,3), (4,3), (6,3)\\
\>     enddo\\
c\$\>   {\bf end parallel}
\end{tabbing}
\rm
These directives instruct the compiler that {\tt colffts} should be
allocated an $8 \times 3$ module of nodes, with the top-left corner of
the module at node (0,0). Task-subroutines {\tt rowffts} and {\tt
hist} are to be placed on the same $2 \times 5$ module, replicated 4
ways, with the top-left corner of the 4 module instances starting at
nodes (0,3), (2,3), (4,3), and (6,3), respectively. The replicated
instances of the {\tt rowffts-hist} module are called in round-robin
order. So, instance 0 gets the first data set, instance 1 gets the
second data set, and so on.

The current implementation of Fx only supports homogeneous parallel
system, for which the size and location of a subarray is sufficient
information to map a task-subroutine.  In a heterogeneous environment
with different machines, additional information is needed.

\section{Compiling task parallel programs}
\label{sec:compiling}

The compiler must perform a set of steps to support task parallelism:
(1) Identify the task structure of the program and determine the
placement of task-subroutines.  This step determines the mapping of the
application on the parallel system. (2) Determine the communication links
between the task-subroutines and identify the data to be transferred.
(3) Generate and schedule inter-task communication.  (4) Generate a
final program along with variable declarations to manage the shared
address space.

The different tasks in the program are obtained by examining the statements
in a parallel section.  The placement of the tasks in the parallel
system is  obtained  from the mapping directives, i.e. the
{\tt processor} and {\tt origin} directives.
The dependences between tasks are identified by data flow
analysis over array sections using the information in the input and output
directives supplied by the user.  The task dependence edges are also the
communication edges, and the actual communication is generated using the
task-communication graph and the data distribution information that is
present in the form of alignment and distribution directives inside
task-subroutines.  Declaration and distribution of array variables, and node
assignments determine the amount of memory allocated for array
variables on individual nodes.

\subsection{Mapping criterion}
The mapping of the tasks of a parallel program to the processor nodes
is an important determinant of performance. The directives 
that control mapping may be provided by the user, or generated by 
an automatic mapping tool. The situation is analogous to the data 
layout directives in HPF. The mapping process is discussed in more
detail in~\cite{subhlok94a}, and an automatic mapping tool is discussed
in~\cite{subhlok93b}. Here we briefly discuss the basic mapping criterion,
which is the same whether the mapping is done by hand or by an automatic
tool.

In our experience, the following three dimensions have the biggest
impact on the quality of a mapping:

\begin{description}
\item[Scalability:] When a computation or a subroutine is not scalable, 
better node efficiency is achieved by using a smaller number of
nodes for each computation instance.  

\item[Memory requirements:] The minimum number of nodes needed 
for a computation is bounded by memory requirements. This is an 
important consideration that is often overlooked in the mapping literature.

\item[\bf Inter-task communication:] The nature and cost of inter-task
communication depends on the mapping. If two tasks are placed 
in the same module, the cost of the inter-task communication is different
than if they are placed in different modules.
\end{description}

The major steps in generating a mapping are:(1) Cluster
task-subroutines into modules.  (2) Allocate nodes to modules.
(3) Replicate modules into module instances.  

\subsection{Example}
We qualitatively discuss the mapping of the FFT-Hist example program.
Task-subroutines {\tt
rowffts} and {\tt hist} are clustered into the same module to save
the cost of data
transfer between them. The data transfer cost is zero if these two
task-subroutines are included in the same module and hence execute on the
same nodes. The cost of communication between {\tt colffts} and {\tt
rowffts} does not decrease if they are mapped to the same module, since a
matrix transpose is required even if they are mapped to the same set of
nodes. So these two task-subroutines are kept in separate modules to reduce
the memory requirements. 

Nodes are then allocated to the two modules in proportion to the computation
load.
The {\tt rowfft-hist} module is allocated 40 processors and then
replicated to 4 instances (at least 10 processors are needed 
for each instance due to memory requirements) .
Each instance runs on 10 nodes, rather than having a
single instance running on 40 nodes. This is an important step since the
{\tt hist} routine does not scale well, and its performance improves only
slightly from 10 to 40 nodes.  Replication is not applied to the {\tt
colffts} module, since it scales nearly linearly.  Figure~\ref{fig:final}
shows the steps and the resulting mapping for a 64-node parallel system.
The quantitative measurements used by our automatic tool to arrive at
this mapping is discussed in~\cite{subhlok93b}.

\begin{figure}[htb]
\begin{minipage}[b]{2.5in}
\centerline{\epsfbox{figs/steps.eps}}
\end{minipage}
\begin{minipage}[b]{3.0in}
\hspace*{.8in}
\vspace*{-.3in}
{\sf Final mapping onto 64 nodes}
\centerline{\epsfbox{figs/fft_mod.eps}}
\end{minipage}
\caption{Mapping steps and the final mapping of the example program}
\label{fig:final}
\end{figure}


In summary, a combined task and data parallel mapping is often
needed to achieve the best performance, and the choice is based on
measurable program properties.  For example, Figure~\ref{fig:perf}
shows the performance of the mapping in Figure~\ref{fig:final} on a
64-node iWarp system, relative to a direct data parallel mapping.
The mapping in Figure~\ref{fig:final}, which consists of a mix of
task and data parallelism, outperforms the straightforward data
parallel mapping by a factor of two.
\begin{figure}[htb]
\begin{center}
\begin{tabular}{|c|c|}   \hline
Program Mapping & Speedup over data parallel mapping\\\hline
Data Parallel (Fig. \protect \ref{fig:newforms}(a)) & 1 \\ \hline
Task Parallel (Fig. \protect \ref{fig:newforms}(b)) & 1.43 \\ \hline
Mixed Mapping (Fig. \protect \ref{fig:final}) & 1.95 \\
\hline
\end{tabular}
\end{center}
\caption{Speedup for different mappings of the FFT-Hist example}
\label{fig:perf}
\end{figure}


\section{Experience with Fx applications}
\label{sec:applications}

We used Fx for problems in a variety of domains: medical image processing,
synthetic aperture radar, narrowband tracking radar, computer vision, and
airshed modelling~\cite{dinda94}.  This section describes our experience
with a subset of these applications and kernels: 2D fast Fourier transform,
narrowband tracking radar, and multibaseline stereo
imaging~\cite{dinda94,subhlok94a}.  The programs are compiled for a 64-node
iWarp system~\cite{iwarp,iwarpcomm}. Since the details of the target machine
are not relevant in this context, we present the results as speedup over a
purely data parallel implementation.  In all the examples, significant
performance benefits are realized by compiling the programs with a mix of
task and data parallelism.

\subsection{Fast Fourier transform}
The FFT-Hist example from the previous sections consists of a 2D~FFT
(task-subroutines {\tt colffts} and {\tt rowffts}), followed by a
histogram. The 2D~FFT is an interesting application in its own right;
even though it shares much of the same code with the FFT-Hist example,
it scales differently, and thus its best mapping is quite different.

Figure~\ref{fig:fftnumbers1} shows the speedups for different mappings
of the FFT program relative to a simple data parallel mapping, for
different problem sizes.  
\begin{figure}[htp]
\begin{center}
\begin{tabular}{|l|c|c|c|c|c|c|}
\hline
&\multicolumn{3}{c|}{one module} &  \multicolumn{3}{c|}{two modules} \\ 
\cline{2-7} 
size     & pure data   & 2-way  & 4-way & no     & 2-way       & 4-way \\
  & parallel    & replication & replication  & replication & replication & replication\\
\hline
$64 \times 64$   & 1 & 1.88 & 2.61 & 2.08 & 2.65 & 3.04 \\
$128 \times 128$ & 1 & 1.15 & 1.28 & .91 & 1.12 & 1.29\\
$256 \times 256$ & 1 & 1.09 & 1.16 & .79 & .94 & 1.08\\ 
\hline
\end{tabular}
\end{center}
\caption{
Speedup of 2D~FFT relative to a data parallel mapping.
%(The mapping of Figure~\protect{\ref{fig:ffttask}} is underlined.)
}
\label{fig:fftnumbers1}
\end{figure}
The numbers are relative only to numbers in
the same row and are not comparable across rows.  Notice that the
optimal clustering depends on the problem size, but a higher degree of
replication always improves performance.  For example, for the $128
\times 128$ 2D~FFT (a size frequently encountered) a 4-way replication
of two modules, as shown in Figure~\ref{fig:ffttask}, minimizes
execution time.  This mapping differs from the mapping of
FFT-Hist in Figure~\ref{fig:final}.  

As the problem
size increases, the pure data parallel mapping begins to perform
better relative to the mixed mappings.  The reason is due to
differences in the scalability of inter-task communication; in our
implementation, communication between task-subroutines in the same
module scales better than communication between task-subroutines in
different modules.  The crucial point here is that the best mapping
depends on the input size; no single approach works best in all cases.
\begin{figure}[tp]
%\epsfxsize=4in
\centerline{\epsfbox{figs/ffttask.eps}}
\caption{FFT task graph and mapping}
\label{fig:ffttask}
\end{figure}


\subsection{Narrowband tracking radar}

The narrowband tracking radar benchmark was developed by researchers
at MIT Lincoln Labs to measure the effectiveness of various
multicomputers for their radar applications~\cite{shaw92}. It is a
particularly interesting benchmark for studying task parallelism
because of its hard real-time requirements, and because the size of
the input data set is limited by physical properties of the radar
sensor.  The amount of available low-level data parallelism is
limited, so additional parallelism must come from higher-level task
parallelism.

The radar program inputs data from a single sensor along $c=4$
independent {\em channels}.  Every 5 milliseconds, for each channel,
the program receives $d=512$ complex vectors of length $r=10$, one
after the other in the form of an $r \times d$ complex matrix {\tt A}
(assuming the column major ordering of Fortran).  At a high-level,
each input matrix {\tt A} is processed in the following way: (1) {\em
Corner turn} the $r \times d$ input matrix to form a $d \times r$
matrix.  (2) Perform $r$ independent $d$-point FFTs.  (3) Convert the
resulting complex $d \times r$ matrix to a real $w \times r$
submatrix, $w=40$, by replacing each element $a + ib$ in the $w \times
r$ submatrix with its scaled magnitude $\sqrt{a^2 + b^2}/d$.  (4)
Threshold each element $a_{jk}$ of the submatrix using a cutoff that
is a function of $a_{jk}$ and the sum of the submatrix elements.  The
Fx version of the radar program operating on a stream of $m$ input
data sets has the following form:

\tt
\begin{tabbing}
tttttttt\=ttt\=ttt\= ttt\=ttt\= \kill
c\$ \>{\bf begin parallel} \\
\>     do i = 1,m \\
\>\>       call get(A)\\
c\$\>\>     {\bf output:} A\\
\>\>       call compute(A,B)\\
c\$\>\>     {\bf input:} A\\
c\$\>\>     {\bf output:} B\\
\>     enddo\\
c\$\>   {\bf end parallel}
\end{tabbing}
\rm
\normalsize
The program consists of a parallel section with calls to two
task-subroutines inside a loop that iterates $m$ times.  
Figure~\ref{fig:radartask} shows the task graph. Task-subroutine {\tt
get} acquires the data from all 4 channels and sends it to
task-subroutine {\tt compute}, a data parallel routine that performs
steps (1)--(4) above.  We will assume for purposes of discussion that
the {\tt get} task-subroutine must run on one node, and that it
must be assigned to the node that is connected to the radar sensor.
The data parallelism in the {\tt compute} task-subroutine is in the
form of a parallel loop where each loop iteration operates on a single
column of the corner-turned data set. Since there are only $r=10$ of
these columns for each of the 4 channels, the amount of loop-level
parallelism is quite small.
\begin{figure}[tp]
%\epsfxsize=4in
\centerline{\epsfbox{figs/radartask.eps}}
\caption{Radar task graph and mapping}
\label{fig:radartask}
\end{figure}

Since the {\tt get} task-subroutine must run on exactly one node, we can
only replicate the {\tt compute} task-subroutine if the two task-subroutines
are clustered into different modules.  The {\tt compute} task-subroutine can
use at most 10 nodes efficiently, so we want to use up nodes by using
replication.  A mapping of the program that uses 4-way replication of the
{\tt compute} task-subroutine is shown in Figure~\ref{fig:radartask}.

Figure~\ref{fig:radarnumbers} gives the measured performance of the Fx
radar program when compiled with different degrees of replication.
The linear speedups illustrate the value of replication for programs
like the radar program that operate on small data sets.
\begin{figure}
\begin{center}
\begin{tabular}{|c|c|c|c|}
\hline
replication & 1-way & 2-way & 4-way\\
\hline
\hline
speedup & 1 & 2.000 & 3.996 \\
\hline
\end{tabular}
\end{center}
\caption{Speedup of radar for different degrees of replication.}
\label{fig:radarnumbers}
\end{figure}

\subsection{Multibaseline stereo}
The multibaseline stereo example is based on an algorithm developed at Carnegie
Mellon for depth perception by using more
than two cameras~\cite{okutomi93}. It is an interesting program for
studying task parallelism because it contains significant amounts of
both inter-task and intra-task communication~\cite{webb93bb}, and the
size of data sets is fixed.  Our implementation is adapted from a
previous data parallel implementation written in a specialized image
processing language~\cite{webb93aa}.

Input consists of three $m \times n$ images acquired from three
horizontally aligned, equally spaced cameras. One image is the {\em
reference image}, the other two are {\em match images}. For each of 16
disparities, $d = 0,\ldots,15$, the first match image is shifted by
$d$ pixels, the second image is shifted by $2d$ pixels. A {\em
difference image} is formed by computing the sum of squared
differences between the corresponding pixels of the reference image
and the shifted match images.  Next, an {\em error image} is formed by
replacing each pixel in the difference image with the sum of the
pixels in a surrounding $13 \times 13$ window. A {\em disparity image}
is then formed by finding, for each pixel, the disparity that
minimizes error.  Finally, the depth of each pixel is displayed as a
simple function of its disparity.  The Fx version of the stereo
program operating on a stream of $s$ input data sets has the following
form:
\tt
\begin{tabbing}
tttttttt\=ttt\=ttt\= ttt\=ttt\= \kill
c\$ \>{\bf begin parallel} \\
\>     do i = 1,s \\
\>\>       call dgen(R,M1,M2)\\
c\$\>\>    {\bf output:} R,M1,M2\\
\>\>       do d = 0,15\\
\>\>\>		call diff(R,M1,M2,DIFF,d)\\
c\$\>\>\>       {\bf input:} R,M1,M2\\
c\$\>\>\>       {\bf output:} DIFF\\
\>\>\>		call error(DIFF,ERR(:,:,d),d)\\
c\$\>\>\>       {\bf input:} DIFF\\
c\$\>\>\>       {\bf output:} ERR(:,:,d)\\
\>\>       enddo\\
\>\>       call min(ERR,DISP)\\
c\$\>\>    {\bf input:} ERR\\
c\$\>\>    {\bf output:} DISP\\
\>     enddo\\
c\$\>   {\bf end parallel}
\end{tabbing}
\rm
\normalsize
Figure \ref{fig:stereotask} shows the task graph.  Task-subroutine
{\tt dgen} acquires three $256 \times 240$ images from the cameras.
Each of the 16 instances of the {\tt diff} task-subroutine is a
perfectly data parallel routine that converts the three input images
to a difference image. Each instance of the {\tt error}
task-subroutine is a data parallel routine that sums over a window of
pixels in the difference image to produce an error measure for each
pixel. Each image is distributed by rows within each task, so a
node needs to exchange rows with its neighbors before the error
image can be produced. The outputs from the {\tt error} tasks are
passed to {\tt min}, which applies a {\em min}-reduction to produce
the disparity image, and then displays the corresponding depth image.

\begin{figure}
%\epsfxsize=4in
\centerline{\epsfbox{figs/stereotask.eps}}
\caption{Stereo task graph and mapping}
\label{fig:stereotask}
\end{figure}

A mapping of the stereo program that uses 4-way replication is shown
in Figure~\ref{fig:stereotask}.  Figure~\ref{fig:stereonumbers1} shows
the measured performance of the Fx stereo program compiled as 1 (i.e.
purely data parallel), 2, and 4 replicated modules. The higher
throughput of the 4-way replicated case validates the decision to use
replication.  However, while a 4-way replication roughly doubles the
throughput, it roughly doubles the latency too. Depending on the
requirements of a particular application of the stereo program, this
may or may not be a reasonable tradeoff. A system striving to minimize
latency would potentially arrive at a different mapping.

\begin{figure}[tp]
\begin{center}
\begin{tabular}{|c|c|}
\hline
degree of   & \\
replication & speedup\\
\hline
\hline
1 & 1 \\
2 & 1.45\\
4 & 1.93\\
\hline
\end{tabular}
\end{center}
\caption{Speedup of stereo for different degrees of replication}
\label{fig:stereonumbers1}
\end{figure}


\section{Comparison with related work}\label{sec:related}

The approach that we have taken towards task parallelism 
can be summarized by the following key features:
\begin{itemize}
\item Task parallelism is integrated with a data parallel compiler,
      and data parallel subroutines are units for task parallelism.

\item Task parallelism is expressed with high level directives,
      and communication and task management is done by the compiler.
\end{itemize}

Task parallelism that can be expressed in our system is constrained in two
significant ways. First, communication between task-subroutines is permitted
only at procedure boundaries (through procedure arguments). Since we are
using data parallelism with its own compiler-generated communication inside
subroutines, there is some justification that explicit communication between
task-subroutines is less important. This constraint considerably simplifies
the programming model and the compiler.
Second, the mapping of tasks to nodes is
fixed at compile time. This makes it easier to generate efficient parallel
programs with low execution overheads, but makes the method not suitable for
dynamic computations.

Coordination languages like Linda~\cite{CaGe88,linda-dp} and
Fortran~M~\cite{foster92} provide a communication interface to build
task parallel programs, with facilities for more general inter-task communication. In
contrast, Fx task parallelism is more restricted but is closely
integrated with a data parallel compiler, and communication is
exclusively generated by the compiler.

Jade~\cite{JADE-LANGUAGE} and PYRROS~\cite{YaGe92} capture all parallelism as
fine grain task parallelism; these systems create and schedule tasks
dynamically. A new language is developed in Jade, and fine grain
directives are required in PYRROS. While these systems can support a
richer variety of parallelism, particularly dynamic programs, writing
programs is more cumbersome because a fine grain control of
parallelism by the programmer is required, or because they do not use
a standard data parallel layer like HPF, which we found invaluable for
ease of development and user acceptance.

HPF~\cite{hpf93} can be used to develop task parallel MIMD programs using
{\em INDEPENDENT} parallel loops, but no support is available for expressing
data transfers between tasks.  Chapman et. al.~\cite{chapman94} propose a
similiar, but
more general and dynamic approach to task parallelism than ours.  Fx
emphasizes simplicity to obtain efficient code through compilation; it will
be interesting to compare the performance once results from an
implementation are reported.

The node mapping problem introduced in the paper has been addressed in more
detail in related publications~\cite{subhlok93b,subhlok94a}. This problem is
quite different from the many partitioning problems addressed in the
literature (e.g.,~\cite{Sark89,bokhari88,choudhary94}) due to one or more of
the following reasons: (1) Task-subroutines are to be mapped to  groups of
nodes, not individual nodes. (2) The computation and communication costs are
functions of the number of nodes, not constants. (3) The objective is to
maximize throughput for a sequence of inputs, not to minimize execution time
of a fixed set of tasks.

\section{Conclusions}\label{sec:conclusions}

Both task and data parallelism are important for practical applications and
are necessary to make the best use of a parallel system.  We demonstrate
that a set of simple directives is sufficient to capture task parallelism
for representative applications in computer vision,  signal processing, and
multidisciplinary scientific computing. Without task parallelism, it may be
impossible to utilize a large number of nodes efficiently.  Applications in
these domains often exhibit only a limited amount of data parallelism due to
the fixed size of their input sets, are subject to real-world latency
constraints, or are structured so that individual components scale differently.


The Fx compiler is a prototype  system that integrates both data and task
parallelism, and our experience demonstrates that task parallelism can be
supported effectively  in an HPF framework.  The current design reflects our
desire to obtain a working system that can serve as a basis for further
experimentation with the limited resources available.  The Fx compiler
represents approximately a 10 person-year effort (this includes dealing with
task as well as data parallelism). The design contains some limitations:
task parallelism is subject to several constraints, and the programmer has
limited control over execution and communication. However the  design and
compiler have proven adequate for interesting applications.

We take the approach that the user provides a high level specification of
task parallelism via directives, and the compiler manages the execution of
tasks as well as communication between them.  This  extends one of the
attractions of HPF in that it frees the user from dealing with the details
of communication in the parallel program. Furthermore, this approach
provides the compiler opportunities for optimizing communication and mapping
of the program.  Our prototype demonstrates the benefits of task
parallelism; programs with both styles of parallelism exhibit improved
performance over  data or task parallelism alone.   Fx presents a simple
approach to task parallelism that considerably enhances the power of a data
parallel language like HPF.


\bibliography{/afs/cs/project/iwarp/member/droh/bib/refs}
\end{document}
