\section{Introduction}

This paper presents the parallel continuation machine (PCM), a dynamic
execution model creating patterns of locality which are necessary to
harvest the computing power of today's message passing architectures.
We will first concentrate on explaining the key ideas underlying the
implementation, and then demonstrate how they give rise to extremely
efficient parallel programs via two real-world examples.

Parallel programs can be classified along several dimensions, such as
grain-size, communication regularity, and whether the execution
depends on runtime data. We believe that existing programming models,
such as data parallel programming and explicit message passing, have
been successful in addressing the needs of programs with simple static
communication patterns. For these programs it is usually possible to
carefully orchestrate communication and computation to statically
optimize the overall performance.

On the other hand, it proves far more difficult to find static
solutions leading to high machine utilizations for parallel
applications whose communication patterns are either highly irregular or
dependent on dynamic information. In this paper, we are mostly
interested in investigating the needs and characteristics of these
classes of programs, which must rely on runtime mechanisms to
enable efficient solutions.

Multi-threaded computation models have typically been proposed as a
general solution to exploit dynamic, unstructured parallelism. In such
a model, dynamically created instances of sequential threads of
execution cooperate in solving the problem at hand. To efficiently
execute such an application, it is necessary to have efficient runtime
thread placement and scheduling techniques. Although this general
thread placement problem is known to be NP-hard~\cite{garey79}, it is
possible to implement schedulers based on simple heuristics, achieving
good machine utilizations at reasonable cost. These heuristics
usually work well for a broad class of applications, making it
possible to implement the scheduling and placement task as a fairly
generic service that resides at the core of the runtime system.

Unfortunately, current parallel architectures present severe obstacles
to achieving acceptable results for multi-threaded applications. These
machines typically consist of a collection of computing nodes that
communicate with each other through a message-based communication
network. The individual RISC-like nodes have been optimized for
sequential computations with a high locality of reference, and the cost of
network accesses are often one or more orders of magnitude more expensive
than local memory accesses.
Multi-threaded programs, however, often require frequent global
communications. In addition, by exposing multiple threads a parallel
program tends to both lose locality and incur extra runtime overhead in
synchronization and scheduling.

Several research machines, such as the HEP~\cite{hep}, the Monsoon dataflow
system~\cite{ets90}, or the forthcoming Tera
machine~\cite{tera}, are designed to overcome these problems in hardware.
They provide  highly integrated, low overhead, message interfaces as well
as hardware support for scheduling and synchronization. 
Disregarding the debate of whether such machines are commercially or
technically viable, the problem of programming most of the current
parallel machines, which have no special hardware support for
multi-threading, still remains.

We are thus left with the only alternative solution, namely software. The
programming challenge, in view of the above difficulties, is to minimize
network communication and to provide longer sequential threads to offset
the runtime scheduling and synchronization overhead.

The static set of sequential threads making up the multithreaded
program can either be generated {\em implicitly} by a sophisticated
compiler, or {\em explicitly} by the programmer. Programming languages
advocating the implicit style, such as Id~\cite{Id91} and
Sisal~\cite{sisal85}, usually take a high level description of the
actual problem, extract the available parallelism from the declaration
and partition it into sequential threads. While implicit programming
languages simplify the programming task, they usually fail to produce
threads long enough to adequately amortize the overhead introduced by
a dynamic execution model.

In terms of reducing network communication, the execution of threads
must exhibit communication locality. This precludes scheduling
policies such as round-robin or random placement, but favors solutions
such as {\em work stealing} where all threads are created locally per
default, but may later migrate to other nodes upon demand. Other
solutions may be built around explicit policies specialized for the
application program at hand.

The PCM model presented in this paper is aimed at solving the
aforementioned problems. The intended target architectures are simple
message passing machines which support the implementation of 
low-overhead communication layers such as {\em Active
Messages}~\cite{am92}. We do not assume any additional hardware
support.

\begin{figure}[htbp]
\centerline{\psfigure{commcost}{4.0in}}
\caption{\label{commcost}maximizing communication locality}
\end{figure}

We have provided C language extensions where threads can be specified
along with conventional sequential C code. A program consists simply
of a collection of {\em threads}, which are pieces of sequential code
which are guaranteed to terminate once they have been scheduled.
Threads represent the basic scheduling units that can be executed on
any processor. By exposing threads in this way, we can experiment with
various static and dynamic scheduling policies to optimize the overall
machine utilization.  In particular, we have found two strategies
which can have a great effect on the performance of a program.
\begin{itemize}
\item A thread can be made to directly transfer control to another
thread, similar to the control transfer mechanism used to implement
tail-recursions. This mechanism bypasses dynamic scheduling
entirely, thus avoiding all of its associated costs.
\item The scheduler uses work stealing as its default policy. This creates
patterns of locality in which threads can pass arguments through local
memory rather than across the network most of the time, thus greatly
reducing the communication frequency.
\end{itemize}

We show the effect of these optimizations in Figure \ref{commcost}. As
this diagram suggests, there are three communication levels, namely
register communication, memory communication and network
communication. Preferably, we would like to transfer data through
registers as much as possible, but the very nature of a dynamic
execution model will force us to resort to memory communication or,
even worse, to network communication. Note that an application
increases the working set whenever it exposes additional parallelism,
making it impossible to keep the entire working set in registers. The
implementation goal will be to provide an optimal compromise between
increasing the sequentiality of the application to increase its
locality and exposing enough parallelism to enable dynamic load
balancing. The second optimization, work stealing, will then attempt
to minimize the work migration frequency, thereby minimizing network
communication.

\subsection{Related Work}

The idea of continuation passing has been with us for some time now,
since Steele and Sussman \cite{lambda76} forcefully argued its
advantage in optimizing function calls. Several compilers for
sequential languages convert programs into continuation passing style
\cite{appel92} before optimization and generating machine code. It is
also used in parallel programming, for example, in \cite{gm90}. We
note that dataflow diagrams are essentially CPS in disguise
\cite{ttda87}, with additional structures to implement function calls.

Active Messages were popularized by von Eicken et~al \cite{am92},
although our current CM-5 implementation is based on a faster version
in the Strata package \cite{strata}. There are numerous message
passing libraries for conventional sequential languages, which
implement the functionality of the underlying network hardware, but
none of them matches Active Messages in terms of raw performance.

There also exist many parallel coordination languages, for example PVM
\cite{pvm} and Linda \cite{linda}, which provide additional support
for scheduling and synchronization.  These packages typically arise
from the needs of distributed computing, therefore are suitable mostly
for coarser grained parallelism.

Parallel programming languages are more often modeled after
communicating sequential processes, which gives rise to preemptive
threads. For example, Mul-T \cite{alewife} has blocking tasks which
are context switched when an embeded future construct has not yet been
computed to provide a value. An important feature of the Mul-T runtime
system worth mentioning is that it uses lazy task creation, supported
by work stealing, to increase the granularity of tasks \cite{kh91}.

The evolution of parallel computing based on the dataflow model has
advocated the use of non-preemptive threads, also called
micro-threads. Models based on this approach \cite{ets90,tam91} also
superimposes upon threads a runtime environment consisting of a tree
of frames (the parallel equivalent of the call stack in a sequential
program) to support the function call abstraction. PCM is in fact the
result of optimizing away this super-structure, replacing it with
structures more resource-efficient and flexible in scheduling.\\

The rest of this paper is structured as follows: Section~\ref{language-sec}
introduces the PCM thread specification language and present in some detail
all of its components. Section~\ref{scheduler-sec} introduces a cost model,
intended to clarify the costs
involved in dynamic execution.  Section~\ref{case-studies} contains in depth
explanations of two actual problems, implemented with the PCM package.
Section~\ref{future-work-sec} states some possible future work and
Section~\ref{conclusion-sec} concludes the paper. 
