% Section on Solutions to RLOADs, JOINs, ILOADs in Isca paper

\section{Solutions}

\label{sec-solutions}

\subsection{Caches}

\label{sec-global-caches}

Classically, caches are used to mask memory latency, and we could add
a cache to each node to hold copies of remote locations.  However,
{\em cache coherence} is a then significant problem: multiple copies
of a location residing in different nodes' caches must be kept
consistent.  Further, a processor may still idle when there is a cache
miss.  A possible solution for the distributed cache coherence problem
is to use {\em directories\/}; and for the cache miss problem, to
multiplex between a small number of contexts to cover cache loading.
Implementing this appears non-trivial (see MIT's MASA
\cite{Halstead88a} and Alewife \cite{Agarwal90a}, and
Stanford's DASH~\cite{Lenoski90a}, for example).  In any case, {\em
the proposals in this paper may be seen as orthogonal to cacheing
solutions.} Further, while distributed cacheing helps in the remote
load situation, it does not offer anything for the synchronizing load
problem.

% ----------------------------------------------------------------

\subsection{Multithreading}

Another solution is to multiplex amongst many threads: when one thread
issues a remote load request, the processor begins work on another
thread, and so on.  Clearly, the cost of thread-switching should be
much smaller than the latency of the remote load, or else the
processor might as well wait for the remote load's response.

{\em Continuations on messages\/}: A consequence of multithreading is
that {\em messages should carry continuations\/}.  Suppose, after
issuing a remote load from thread T1, we switch to thread T2 which
also issues a remote load.  The responses may not arrive in the same
order--- for example, the requests may travel different distances,
through varying degrees of congestion, to destination nodes whose
loads differ greatly, \etc\ Thus, each remote load and response should
carry an identifier for the appropriate thread so that it can be
re-enabled on the arrival of a response.  We refer to these thread
identifiers as {\em continuations}.

{\em Adequate continuation namespace\/}:
The longer the latency of remote loads, the more threads we need to
avoid idling the processor.  Thus, it is important that we have a {\em
continuation namespace\/} that is large enough to name an adequate
number of threads waiting for remote responses.

The size of this namespace can greatly affect the programming model.
If the latency of a single remote load is $l$, we would expect that at
any given time, we need no more than about $l$ active threads in the
processor to cover remote loads.  However, if the programmer (or
compiler) cannot predict the scheduling of these threads precisely, he
may need {\em many more than $l$\/} threads in the hope that,
dynamically, $l$ of them are ready to run at any given time.  Thus, if
the continuation namespace is too small, it requires more precise
scheduling of threads, which in turn limits the programming model.

If we generalize our remote transaction model to include not just
remote loads, but (a) synchronizing remote loads and (b) remote
parallel procedure calls, then it becomes extremely difficult for the
programmer/compiler to schedule threads precisely.  Therefore, it is
essential to have a large continuation namespace.

The size of the hardware-supported continuation namespace varies
greatly in existing designs: from 1 in DASH~\cite{Lenoski90a}, 4 in
Alewife~\cite{Agarwal90a}, 64 in the HEP~\cite{Smith78}, and 1024 in
the Tera~\cite{Alverson90a}, to the local memory address space in
Monsoon~\cite{Papadopoulos88a}, Hybrid Dataflow/von
Neumann~\cite{Iannucci88a}, MDP~\cite{Dally87a} and {\archname}.  Of
course, if the hardware-supported namespace is small, one can always
virtualize it by multiplexing in software, but this has an associated
overhead.

\subsection{Fine Grain Multithreading: forks and joins}

So far, we have only talked about efficient multiplexing amongst
existing threads.  However, if thread creation and synchronization are
also sufficiently cheap, the multithreading idea can be advantageously
taken further.  Instead of the {\cf rload-rload}-subtract sequence
earlier,  we could fork separate threads for the two {\cf
rload}s and synchronize them when both have completed.  Then, instead
of taking four message-transit times, we could perhaps do it in two
(since they occur in parallel):
\beginid
    fork M1
L1: vA = rload pA  ; (A)
    jump N
\null
M1: vB = rload pB  ; (B)
    jump N
\null
N:  join 2 J       ; synchronization of responses
    C = vA - vB
\endid
This kind of parallelism is available in all dataflow
architectures~\cite{Arvind86a}.  {\cf Fork} initiates a new thread at
label {\cf M1}.  The parent thread continues at {\cf L1}, issuing the
remote load (A), which suspends to await the response.  This allows
the just-forked thread at {\cf M1} to run, which issues the remote
load (B), which also suspends to await the response.  When the first
response arrives (say, (A)), the first thread resumes, completing the
instruction at {\cf L1} by storing the arriving value into {\cf vA},
and jumps to {\cf N}.  Similarly, when the second response arrives
(B), the second thread resumes, completing the instruction at {\cf M1}
by storing the arriving value in {\cf vB} and also jumping to {\cf N}.

The {\cf join} instruction at {\cf N} is executed twice, once by each
thread.  Each time, it increments {\cf J} (initialized to 0, we assume)
and compares it to the terminal count (2).  If {\cf J < 2}, the thread
dies, otherwise it continues.  In general, if the terminal count is
$c$, then $c-1$ threads arriving at {\cf N} will die and the only the
last one continues to perform the sequel.

Again, the cost of {\cf fork} and {\cf join} should be small compared
to the latencies of the remote loads that we are trying to cover.
Note that modern RISC techniques such as delayed loads, cache
prefetching, \etc\ are also modest, ``peephole'' examples of the use
of fork-join parallelism to overlap long-latency loads with useful
computation, but they do not generalize well to remote loads in MPA's.

The behavior at {\cf N} is called a {\em join synchronization\/}, and
is very important.  A significant number of messages entering a node
may be destined for join synchronizations.  In an inner-product, at
each index we remote load $A[j]$ and $B[j]$, join the responses, and
multiply.  Note that for all but the last thread entering a join, the
amount of work done is very small--- the thread increments a counter,
tests it and dies.  For a two-way join, this may be as small as
testing and setting one bit.  It is important that the processor not
take a large hiccup to do this.

Exactly the same kind of join synchronization is also needed for {\em
software\/} reasons.  For example, a function that sums the nodes of a
binary tree may have exactly the same structure as the above program
with the two {\cf rload}s simply replaced by remote function calls to
sum the left and right subtrees, respectively, and the subtraction in
the final line replaced by addition.  Now, the latencies include both
communication and the time to perform the subcomputations.

This leads to another observation: for messages that arrive at highly
unpredictable times (as in the tree sum) or at widely separate times
(even if predictable) the rendezvous point {\cf J} for join
synchronization must typically be {\em in memory, not registers.\/}
The reason: processor utilization will be low if registers remain
allocated to a process for unpredictable and possibly very long
durations.

\subsection{Messages and Message-Handling}

Our first approximation to a message format is:
\beginid
    NA: < msg\_type, arg1, arg2, ...>
\endid
Node address NA is used for routing, and may be consumed in transit.
{\cf Msg\_type} specifies how the message should be handled.  The
remaining arguments depend on the message type.  Figure
\ref{fig-rload-messages}
shows the messages for an {\cf rload} where {\cf \&A} and {\cf *A}
mean the address and contents of {\cf A}, respectively.\footnote{Figure
\ref{fig-rload-messages}
depicts the logical information in messages; clever encoding can
compress it significantly.  }
\begin{figure}[htbp]
  \hdivider \\
  \illustration{fig-rload-messages}{1.0 true in}
  \caption{%
     \label{fig-rload-messages}
     Messages to perform a remote load.
          }
  \hdivider
\end{figure}
The values {\cf N1}, {\cf L1} and {\cf CTXT} together constitute a
{\em continuation\/}, \ie\ a complete specification of the thread
awaiting the response.

For fast vectoring to a message handler, it is convenient for the
``message type'' field to {\em be\/} the instruction pointer of the
handler:\footnote{Dally's Message Driven Processor~\cite{Dally87a}
also uses an IP in the first field of a message.}
\beginid
    NA: < IP, arg1, arg2, ...>
\endid
To keep up with high incoming message rates, a hardware {\em network
interface\/} can accept and queue them, to be consumed by the
processor at its convenience (Figure \ref{fig-msg-queue}).
\begin{figure}[htbp]
  \hdivider \\
  \illustration{fig-msg-queue}{3.7 true in}
  \caption{%
     \label{fig-msg-queue}
     A node with network interface.
          }
  \hdivider
\end{figure}
Of course, this may add to the latency of a remote load.  The {\em
message formatter\/} constructs outgoing messages.

The actual handling of an {\cf rload} request at N2 is simple and
short: read location {\cf A} and send the response message to N1.
Handling the response at N1 may be broken into two parts: (a)
completing the unfinished {\cf rload} instruction, and (b) continuing
the execution of the thread.  Part (a) is always brief.  Part (b) is
also brief if it just executes a failing {\cf join}.

Both N1 and N2 will typically be executing some other thread when
these messages arrive.  How can we handle these messages quickly (so
that latency is low) and with minimum disruption, \ie\ how well is
message-handling integrated with normal processing in the node?  {\em
The answer to this question is crucial to the efficient operation of
an MPA.} The overhead can vary from hundreds of instruction-times in
Thinking Machines' CM-5 and Intel's Touchstone, where message-handling
is not well integrated, down to under ten instruction-times in the
MDP~\cite{Dally87a}, and further down to a single instruction-time in
machines like Monsoon~\cite{Papadopoulos88a} and the
EM-4~\cite{Sakai89a} where message-handling is totally integrated into
the processor pipeline.

There is good reason to insist that message-handling {\em should not
compromise excellent single-thread performance\/} in an MPA node.
First, existing SIMD/SPMD applications have long sequential threads.
Second, even in more dynamic applications, certain sequential critical
sections (such as storage allocators) must be executed as quickly as
possible.  Third, users may wish to develop code on a 1-node MPA.
Fourth, uniprocessor libraries should be portable to MPA nodes without
degrading their performance.  We note that competitive single-thread
performance is an important goal in the Alewife~\cite{Agarwal90a} and
Hybrid Dataflow/von Neumann~\cite{Iannucci88a} machines.

{\em Interrupts are inadequate\/}: An arriving message could interrupt
the processor, which responds, and then resumes normal computation.
Modern processors achieve high speeds through deep pipelining and more
processor state (registers, caches), and any change in context incurs
a large penalty draining, saving, flushing, reloading, \etc\ Thus,
frequent short threads, such as {\cf rload} handlers and messages that
go into failing joins are likely to be severely disruptive.  This may
be alleviated by replicating processor state so that the main thread's
state does not have to be saved and restored during interrupts.
Still, the main thread's performance is affected adversely, because
each incoming message takes away cycles and may require processor
pipelines to drain and refill.  Frequent interrupts for message
handling are thus not desirable.

{\em Separate Coprocessors\/}: We can offload the burden of short,
simple handlers into separate, asynchronous coprocessors, as shown in
Figure
\ref{fig-start-node}.
\begin{figure*}[htbp]
     \illustration{fig-start-node}{3.7 true in}
     \caption{%
        \label{fig-start-node}
        A {\archname} node: separate coprocessors for short threads.
             }
\end{figure*}
This is only a functional decomposition, or logical organization of
the node; we will discuss physical organizations in Section
\ref{sec-implementation}.

We will refer to the original, conventional main processor as the {\em
data processor\/} ({\sc dP}).  It executes threads supplied through
the {\em continuation queue\/} and is optimized for long, sequential
runs.

The {\em remote memory request coprocessor\/} {\sc RMem} handles
incoming {\cf rload} requests.  It accesses local memory, and sends
the response directly, without disturbing {\sc dP}.  Similarly, {\sc
RMem} also handles {\cf rstore} requests.

The {\em synchronization coprocessor\/} {\sc sP} handles {\cf rload}
responses returning to this node.  The contents of the message are
loaded directly into its registers, with the first message field (IP)
loaded into its PC, so that it is vectored directly to the appropriate
handler.  The basic action of {\sc sP} is thus to repeatedly load a
message from its message queue and run it to completion; the run is
always non-suspensive and is expected to be very short.  {\sc sP}
completes the unfinished {\cf rload} by storing the value from the
message (now in {\cf sV1}) into the destination location.  After this,
it places the {\cf <L1+1,CTXT>} continuation into the {\em
continuation queue}, to be picked up later by the main processor {\sc
dP}.  Even better, if the next instruction is a {\cf join}, {\sc sP}
executes it and enqueues the continuation only if it succeeds.  Thus,
the main processor {\cf dP} does not have to execute disruptive {\cf
join} instructions at all.

\subsection{Handling Synchronized Loads}

We observed the need for synchronization if the remote loads for {\cf
A} and {\cf B} must wait for corresponding producer computations.  We
follow the well-known idea of appending {\em presence bits\/} to each
memory location indicating whether it is {\sc full} or {\sc
empty}.\footnote{Our development does not depend on it--- presence
bits could be interpreted in ordinary memory.} The variables {\cf A}
and {\cf B} are initially marked {\sc empty}.  When values are written
there, they are marked {\sc full}, \eg\ by the {\sc RMem}s when they
perform the stores.  Thus, the remote loads should succeed only after
the locations become {\sc full}.

{\em Busy waiting\/}: When N1's {\cf rload} request for {\cf A}
arrives, N2 could simply return the entire location, presence bits and
all.  N1 would then test the response, and may choose to try again,
perhaps after some delay.  Unfortunately, this can waste network
bandwidth, and may also waste cycles on N1 in order to test and
perform the retries.  This is essentially the solution in the
Tera~\cite{Alverson90a} and Alewife~\cite{Agarwal90a}, using traps
instead of explicit tests.

{\em A data driven solution\/}: We extend the presence bits to have an
additional state, {\sc pending}.  For a remote synchronizing load, node N1
executes an {\cf iload} instruction instead of {\cf rload}:
\beginid
    L1: vA = iload pA
\endid
The only difference in the request message is a different IP:
\beginid
    N2: <iload\_request, \&A, N1,L1,CTXT>
\endid
At N2, the {\cf iload\_request} handler tests the presence bits of
{\cf A}.  If {\sc full}, it responds just like the {\cf
rload\_request} handler.  If {\sc empty}, it attaches the arguments
({\cf N1,L1,CTXT}) to location {\cf A}, marks it {\sc pending}, and
{\em produces no response}.  When the producer later stores the value
at {\cf A}, N2 again checks the presence bits.  If {\sc empty}, it
simply stores the value there and marks it {\sc full}.  If {\sc
pending}, it not only stores the value there, but also constructs and
sends the response to the waiting load.  In all cases, N2's response
message for an {\cf iload} looks identical to an {\cf rload} response.

In general, many loads may be pending at {\cf A}, so N2 needs to
remember a set of pending requests at location {\cf A}.  Thus, one of
the costs is that N2 must perform some storage management to construct
this set.  However, a simple free-list strategy may be used, since set
components are of fixed size and are not shared.  Thus, it does not
matter if the remote load request arrives before the value is ready---
it simply waits there until it is ready.  There is no additional
message traffic for the synchronizing remote load, and N1 does no busy
waiting at all.  As far as N1 is concerned, an {\cf iload} is just an
{\cf rload} that may take a little longer--- the interface is
identical.

This is essentially the dataflow notion of {\em
I-structures}~\cite{Arvind90a}, but we note that it is more general---
a variety of other synchronization primitives (such as
Fetch-and-$\phi$~\cite{Almasi89}) could be implemented using the same
methodology.  Finally, note that these message handlers are still
simple enough to be handled entirely by the {\sc RMem} coprocessor,
without disturbing {\sc dP}.

% ----------------------------------------------------------------
