%\documentstyle[preprint,graphicx,mathptm,epsf,epsfig]{acmconf}
%\documentstyle[graphicx,mathptm,epsf,epsfig]{acmconf}

\documentclass[12pt]{article}
\usepackage{mathptm}
\usepackage{fullpage}
\usepackage[dvips]{graphicx}
\usepackage{epsfig}

% the following line should make it ``16 point spacing''
% \renewcommand{\baselinestretch}{1.6}

% \setlength{\textwidth}{6.55in}
% \setlength{\textheight}{9.05in}
% \setlength{\topmargin}{-0.45in}
% \setlength{\oddsidemargin}{-0.15in}
% \setlength{\evensidemargin}{-0.15in}

\newcommand{\ol}{\par\vspace{1ex}\noindent$\otimes$ }

\begin{document}

\title{Wide-Area Parallel Programming using the \\
       Remote Method Invocation Model}
\author{
Rob van Nieuwpoort 
\hspace{0.3cm} 
Jason Maassen 
\hspace{0.3cm} 
Henri E. Bal \\
\vspace{0.1cm}
Thilo Kielmann
\hspace{0.3cm} 
Ronald Veldema
\vspace{0.3cm}
\\ 
{Department of  Computer Science,
Vrije Universiteit, Amsterdam, The Netherlands}\\
{\small rob@cs.vu.nl \hspace{0.3cm} jason@cs.vu.nl \hspace{0.3cm} bal@cs.vu.nl \hspace{0.3cm} kielmann@cs.vu.nl \hspace{0.3cm} rveldema@cs.vu.nl} \\
\vspace{-0.2cm}
{\small \texttt{http://www.cs.vu.nl/albatross/}}
}
\date{}
\maketitle

\begin{abstract}
Java's support for parallel and distributed processing
makes the language attractive for metacomputing applications, such as parallel
applications that run on geographically distributed (wide-area) systems.
To obtain actual experience with a Java-centric approach to metacomputing,
we have built and used a high-performance wide-area Java system, called Manta.
Manta implements the Java Remote Method Invocation (RMI)
model using different communication protocols
(active messages and TCP/IP) for different networks.
The paper shows how wide-area parallel applications can be expressed
and optimized using Java RMI. Also, it presents performance results
of several applications on a wide-area system consisting
of four Myrinet-based clusters connected by ATM WANs.
We finally discuss alternative programming models, namely 
object replication, Java\-Spaces, and MPI for Java.
% We compare their expressiveness as well as their
% performance impact.
\end{abstract}

\section{Introduction}

Metacomputing is an interesting research area that tries
to integrate geographically distributed computing resources
into a single powerful system.
Many applications can benefit from such an
integration \cite{foster++:98:grid,smarr:1992}.
Metacomputing systems support such applications by addressing
issues like resource allocation, fault tolerance,
security, and heterogeneity.
Most metacomputing systems are language-neutral and
support a variety of programming languages.
Recently, interest has also arisen in metacomputing architectures that are
centered around a single language.
This approach admittedly is restrictive for some applications,
but also has many advantages, such as a simpler design and the
usage of a single type system.
In \cite{Wollrath:1997}, the advantages
of a Java-centric approach to metacomputing are described, including
support for code mobility, distributed polymorphism, distributed garbage
collection, and security.

In this paper, we describe our early experiences in building and using
a high-performance Java-based system for one important class of
metacomputing applications: parallel computing on geographically
distributed resources.
Although our system is not a complete metacomputing environment yet,
it is interesting for several reasons.
The system, called \emph{Manta}, 
focuses on optimizations to achieve high performance with Java.
It uses a native compiler and an efficient, light-weight RMI
(Remote Method Invocation) protocol that achieves a performance close
to that of C-based RPC protocols~\cite{maassen++:99:ppopp_manta}.
We have implemented Manta on a geographically distributed system, called
\emph{DAS}, consisting of four Pentium Pro/Myrinet
cluster computers connected by wide-area ATM links.
The resulting system is an interesting platform for studying
parallel Java applications on geographically distributed systems.

The Java-centric approach achieves a high degree of transparency and hides
many details of the underlying system (e.g., different communication
substrates) from the programmer.
For several high-performance applications, however, the huge
difference in communication speeds between the local and wide-area networks
is a problem.  In our DAS system, for example,
a Java RMI over the Myrinet LAN costs about 40 $\mu$sec, while an RMI over
the ATM WAN costs several milliseconds.
Our Java system therefore exposes the structure of the wide-area system
to the application, so applications can be optimized 
to reduce communication over the wide-area links.

This paper is based on our earlier work as published in
\cite{nieuwpoort++:99:javagrande}.
We show how wide-area parallel applications can be expressed
and optimized using Java RMI and we discuss the performance of several
parallel Java applications on DAS.
We also discuss some shortcomings of the Java RMI model for
wide-area parallel computing
and how this may be overcome by adapting features from alternative
programming models.
The outline of the paper is as follows.
In Section \ref{sec:implementation} we describe the implementation
of Manta on our wide-area system.
In Section \ref{sec:applications} we describe our experiences in
implementing four wide-area parallel applications in Java
and we discuss their performance.
In Section \ref{sec:alternative} we discuss which alternative programming
models may contribute to an RMI-based programming model.
In Section \ref{sec:related} we look at related work and
in Section \ref{sec:conclusions} we give our conclusions.




\section{A wide-area parallel Java system}
\label{sec:implementation}

In this section, we will briefly describe the DAS system and the
original Manta system (as designed for a single parallel machine).
Next, we discuss how we implemented Manta on the wide-area DAS system.
Finally, we compare the performance of Manta and the Sun JDK
on the DAS system.


\subsection{The wide-area DAS system}
\label{sec:das}

We believe that high-performance metacomputing applications will typically run
on collections of parallel machines (clusters or MPPs),
rather than on workstations at random geographic locations.
Hence, metacomputing systems that are used for parallel processing
will be \emph{hierarchically} structured.
The DAS experimentation system we have built reflects this basic assumption,
as shown in Figure~\ref{fig:das}.
It consists of four clusters, located at different universities in
The Netherlands.
The nodes within the same cluster are connected by $1.2$ Gbit/sec
Myrinet~\cite{boden++:95:myrinet}.
The clusters are connected by dedicated 6 Mbit/s wide-area ATM networks.

The nodes in each cluster are 200 MHz/128 MByte Pentium Pros.
One of the clusters has 128 processors, the other clusters have 24 nodes each.
The machines run RedHat Linux version 2.0.36.
The Myrinet network is a 2D torus
and the wide area ATM network is fully connected. 
The system, called DAS, is more fully described in
\cite{plaat++:99:hpca} (and on {\em http://www.cs.vu.nl/das/\/}).


\begin{figure}
\begin{center}
\psfig{figure=das.eps,height=7cm}
\caption{The wide-area DAS system.}
\label{fig:das}
\end{center}
\end{figure}


\subsection{The Manta system}

Manta is a Java system designed for high-performance parallel computing.
Like JavaParty \cite{philippsen:1997}, Manta uses a separate {\sf remote}
keyword to indicate which classes allow their methods to be invoked
remotely. This method is somewhat more flexible and easier to use than
inheriting from {\sf java.rmi.server.UnicastRemoteObject}
(the standard RMI mechanism).
JavaParty requires a preprocessor for implementing this language extension;
for Manta, we have modified our compiler.
Except for this difference, the programming model of Manta is the same
as that of standard RMI.
Manta uses a native compiler and an optimized RMI protocol.
The most important advantage of a native compiler (compared to a JIT)
is that it can do
more aggressive optimizations and therefore generate better code.
The compiler also generates the serialization and deserialization routines,
which greatly reduces the runtime overhead of RMIs.
Manta nodes thus contain the executable code
for the application and (de)serialization routines.
The nodes communicate with each other using Manta's own light-weight RMI
protocol.

The most difficult problem addressed by the Manta system is to
allow interoperability with other JVMs.
One problem is that Manta has its own, light-weight RMI protocol that
is incompatible with Sun's JDK protocol.
We solve this problem by letting a Manta node also communicate
through a JDK-compliant protocol. Two Manta nodes thus communicate
using our fast protocol, while Manta-to-JVM RMIs use the standard RMI protocol.

Another problem concerning interoperability is that Manta uses a native
compiler instead of a byte code interpreter (or JIT).
Since Java RMIs are polymorphic~\cite{waldo:1998}, Manta nodes
must be able to send and receive byte codes to interoperate with JVMs.
For example, if a remote method
expects a parameter of a certain class \emph{C}, the invoker may send it
an object of a subclass of C. This subclass may not yet
be available at the receiving Manta node, so its byte code may have to be
retrieved and integrated into the computation. With Manta, however,
the computation is an executable program, not a JVM.
In the reverse situation,
if Manta does a remote invocation to a node running a JVM,
it must be able to send the byte codes for subclasses that the receiving
JVM does not yet have.
Manta solves this problem as follows. If a remote JVM node sends byte code
to a Manta node, the byte code is compiled dynamically to object code
and this object code is linked into the running application using
the \texttt{dlopen()} dynamic linking interface.
Also, Manta generates byte codes for the classes it compiles
(in addition to executable code). These byte codes are stored
at an http daemon, where remote JVM nodes can retrieve them.
For more details, we refer to~\cite{maassen++:99:ppopp_manta}.

The Manta RMI protocol is designed to minimize serialization and
dispatch  overhead,
such as copying, buffer management, fragmentation, thread switching,
and indirect method calls.
Manta avoids the several stream layers used for serialization by the JDK.
Instead, RMI parameters are serialized directly into a communication buffer.
Moreover, the JDK stream layers are written in Java
and their overhead thus depends on the quality of the interpreter or JIT.
In Manta,  all layers are either implemented as compiled C code or
compiler-generated native code.
Heterogeneity between little-endian and big-endian machines is
achieved by sending data in the native byte order of the sender, and
having the receiver do the conversion, if necessary.%
\footnote{Manta supports the serialization
and deserialization protocols needed
to support heterogeneous systems, but the underlying Panda library does
not yet support heterogeneity, as it does not do byte-conversions on
its headers yet.}

To implement distributed garbage collection,
the Manta RMI protocol also keeps track of object references that
cross machine boundaries.
Manta uses a mark-and-sweep algorithm (executed by a separate thread)
for local garbage collection
and a reference counting mechanism for remote objects.


The serialization of method arguments
is an important source of overhead of existing RMI implementations.
Serialization takes  Java objects and converts (serializes)
them into an array of bytes.
The JDK serialization protocol is written in Java and uses reflection
to determine the type of each object during run time.
With Manta, all serialization code is generated
by the compiler,  avoiding the overhead of dynamic type inspection.
The compiler generates a specialized serialization and deserialization
routine for every class. Pointers to these routines are stored 
in the method table.
% As explained above, Manta sometimes needs to load and compile classes
% dynamically, to implement polymorphic RMIs. In this case, the (de)serialization
% routines are also generated during run time.
The Manta serialization protocol optimizes simple cases.
For example, an array whose elements are of a primitive type
is serialized by doing a direct memory-copy into the message buffer,
which saves traversing the array.
Compiler generation of serialization is one of the major
improvements of Manta over the JDK \cite{maassen++:99:ppopp_manta}.

\begin{figure*}%[p]
\begin{center}
\psfig{figure=panda-atm.eps,width=14cm}
\caption{Wide area communication based on Panda}
\label{fig:manta-atm}
\end{center}
\end{figure*}


\subsection{Manta on the wide area DAS system}


To implement Java on a wide-area system like DAS, the most important
problem is how to deal with the different communication networks
that exist within and between clusters.
As described in Section~\ref{sec:das}, we assume that
wide-area parallel systems are hierarchically structured and consist
of multiple parallel machines (clusters) connected by wide area networks.
The LANs (or MPP interconnects) used within a cluster typically are
very fast, so it is important that the communication protocols used
for intra-cluster communication are as efficient as possible.
Inter-cluster communication (over the WAN) necessarily is slower.

Most Java RMI implementations are built on top of TCP/IP.
Using a standard communication protocol eases the implementation of RMI,
but also has a major performance penalty.
TCP/IP was not designed for parallel processing, and therefore has
a very high overhead on fast LANs such as Myrinet.
For the Manta system, we therefore use different protocols for
intra-cluster and inter-cluster communication.

To obtain a modular and portable system,
Manta is implemented on top of a separate communication library,
called Panda~\cite{bal++:98:tocs}.
Panda provides communication and multithreading primitives that are designed
to be used for implementing runtime systems of various parallel languages.
Panda's communication primitives include point-to-point message passing,
RPC, and broadcast.
The primitives are independent of the operating system or network, which eases
porting of languages implemented on top of Panda.
The implementation of Panda, however, is structured in such a way that
it can exploit any useful functionality provided by the underlying system
(e.g., reliable message passing or broadcast), which makes communication
efficient~\cite{bal++:98:tocs}.

Panda has been implemented on a variety of machines, operating systems,
and networks. 
The implementation of Manta and Panda on the wide-area DAS system
is shown in Figure~\ref{fig:manta-atm}.
For intra-cluster communication over Myrinet,
Panda internally uses the LFC communication system \cite{bhoedjang:1998}.
LFC is a highly efficient, user-space communication substrate for Myrinet,
similar to active messages.

For inter-cluster communication over the wide-area ATM network,
Panda uses one dedicated \emph{gateway} machine per cluster.
The gateways also implement the Panda primitives, but support
communication over both Myrinet and ATM.
A gateway can communicate with the machines in its local cluster, using
LFC over Myrinet. In addition, it can communicate
with gateways of other clusters, using TCP/IP over ATM.
The gateway machines thus forward traffic
to and from remote clusters.
In this way, the existence of multiple clusters is transparent to the
Manta runtime system. Manta's RMI protocol simply invokes Panda's
communication primitives, which internally calls LFC and/or TCP/IP.

The resulting Java system thus is highly transparent, both for the
programmer and the RMI implementor.
The system hides several complicated issues from the programmer.
For example, it uses a combination of active messages and TCP/IP,
but the application programmer
sees only a single communication primitive (RMI).
Likewise, Java hides any differences in processor-types
from the programmer.

As stated before, parallel applications often have to be aware of the structure
of the wide-area system, so they can minimize communication over
the wide-area links.
Manta programs therefore can find out how many clusters there are
and to which cluster a given machine belongs.
In Section~\ref{sec:applications}, we will give several examples
of how this information can be used to optimize programs.


\subsection{Performance measurements on the DAS system}

Table~\ref{tab:lat_bw} shows the latency and throughput
obtained by Manta RMI and Sun JDK RMI over the Myrinet
LAN and the ATM WAN.
The latencies are measured for null-RMIs, which take zero parameters
and do not return a result.
The maximum throughputs are measured for RMIs that take a large
array as parameter.
The Manta measurements were run on the Linux operating system.
With Sun JDK over ATM, we used JDK version 1.1.6 on Linux.
The performance of the JDK on Myrinet was measured on BSD/OS
(using the JDK 1.1.4), because we do not have a Linux port of the JDK
on Myrinet yet.

For intra-cluster communication over Myrinet, Manta is
much faster than the JDK, which uses a slow
serialization and RMI protocol, executed using a byte code interpreter.
Manta uses fast serialization routines generated by the compiler,
a light-weight RMI protocol, and an efficient communication protocol (Panda).
The maximum throughput of Manta % still true ? (achieved with arrays of size 16 KByte)
is 38.0 MByte/sec.
A performance breakdown of Manta RMI and JDK RMI is given in
\cite{maassen++:99:ppopp_manta}.

For inter-cluster communication over ATM, we used the wide area link
between the DAS clusters at VU Amsterdam and TU Delft
(see Figure~\ref{fig:das}),
which has the longest latency (and largest distance) 
of the DAS wide-area links.
The difference in wide-area RMI latency between Manta and the JDK is 1.2 msec.
Both Manta and the JDK achieve a maximum wide-area throughput 
of 0.55 MByte/sec,
which is almost 75\% of the hardware bandwidth (6 Mbit/sec).
%The JDK RMI protocol is able to achieve less than 50\% of the
%wide-area bandwidth.
The differences in wide-area latency %and throughput 
between Manta and the JDK
are due to Manta's more efficient serialization and RMI protocols, since
both systems use the same communication layer (TCP/IP) over ATM.

\begin{table}%[p]
{\footnotesize
\begin{center}
\begin{tabular}{l|rr|rr}
 & \multicolumn{2}{|c}{Myrinet} & \multicolumn{2}{|c}{ATM} \\
 & Latency & Throughput & Latency & Throughput \\
 & ($\mu s$) & (MByte/s) & ($\mu s$) & (MByte/s) \\
\hline
Manta    &   42.3 & 38.0 & 4350 & 0.55 \\
Sun JDK  &   1228 & 4.66 & 5570 & 0.55
\end{tabular}
\caption{Latency and maximum throughput of Manta and Sun JDK}
\label{tab:lat_bw}
\end{center}
}
\end{table}



\section{Application experience}
\label{sec:applications}

We implemented four parallel Java applications that communicate via RMI.
Initially, the applications were designed for homogeneous (local area)
networks. We adapted these single cluster versions to 
exploit the hierarchical structure of the wide-area system
by minimizing the communication overhead over the wide area links, using
optimizations similar to those described in 
\cite{bal++:98:ipps,plaat++:99:hpca}.
Below, we briefly discuss the original (single cluster) applications
as well as the wide-area optimized programs
and we give performance measurements on the DAS system.
%\footnote{All
%applications except ASP are measured on the actual wide-area DAS system.
%ASP was measured on a local
%emulation system described in \cite{plaat++:99:hpca}, which accurately emulates
%the wide-area system on a single 128-node cluster.}
% %%% when used correctly :-(
We only present results for Manta,
as other competitive Java platforms (e.g., the JDK and Kaffe)
are not yet available on the DAS system (using Myrinet and Linux).
% (Sun's JDK 1.1.4 uses only interpreted code,
% the Kaffe JIT does not yet support RMI,
% and we were unable to run Sun's JIT 1.1.6 on Linux.)

For each of the four programs, we will analyze its performance
on the wide-area DAS system, using the following approach.
The goal of wide-area parallel programming is to obtain higher speedups
on multiple clusters than on a single cluster.
Therefore, we have measured the speedups of each program on a single DAS
cluster and on four DAS clusters, the latter with and without wide-area
optimizations.
In addition, we have measured the speedups on a single cluster with
the same total number of nodes, to determine
how much performance is lost by using multiple distributed clusters
instead of one big centralized cluster.
(All speedups are computed relative to the same program on a single machine.)

The results are shown in Figure \ref{fig:performance}.
The figure contains four bars for each application, giving
the speedups on a single cluster of 16 nodes, four clusters of 16 nodes each
(with and without wide-area aware optimizations),
and a single cluster of 64 nodes.
The difference between the first two bars thus indicates
the performance gain by using multiple 16-node clusters 
(at different locations)
instead of a single 16-node cluster, without any change to the application
source.
The performance gain achieved by the wide-area optimizations can be seen
from the difference between the second and the third bar of each application.
Comparing the second and the third bars with the fourth bar shows how much
performance is lost (without and with wide-area optimizations)
due to the slow wide-area network.
(The 64-node cluster uses the fast Myrinet network between all nodes.)



\subsection{Successive Overrelaxation}

Red/black Successive Overrelaxation (SOR) is an iterative method for solving
discretized Laplace equations on a grid. Here, it is used as an example of
nearest neighbor parallelization methods.    
SOR is an iterative algorithm that performs multiple passes over
a rectangular grid, until the grid changes less than a certain value,
or a fixed number of iterations has been reached.
The new value of a grid point is computed using a stencil operation,
which depends only on the previous value of the point
itself and its four neighbors on the grid.

The skeleton code for the single-cluster parallel Java program for SOR
is given in Figure~\ref{sor-code-single}.
The parallel algorithm we use distributes the grid row-wise
among the available processors, so each machine is assigned
several contiguous rows of the grid, denoted by the interval \emph{LB}
to \emph{UB} (for lower bound and upper bound).
Each processor runs a Java thread of class \emph{SOR},
which performs the SOR iterations until the program converges
(see the \emph{run} method).
Each iteration has two phases, for the red and black grid points.
The processes are logically organized in a linear array.
Due to the stencil operations and the row-wise distribution, every process
needs one row of the grid from its left neighbor (row $LB - 1$) and one row
from its right neighbor (row $UB + 1$).
(Exceptions are made for the first and last process, but we have
omitted this from our skeleton code.)

At the beginning of every iteration, each slave process 
exchanges rows with its left and right neighbors and
then updates its part of the grid using this boundary information
from its neighbors.
The row exchange is implemented using a remote object of class \emph{Bin}
on each processor. This object is a buffer that can contain at most
one row. It has synchronized methods to put and get data.

\begin{figure}[!ht]
\scriptsize{
\input{sor_unopt}
}
\caption{Code skeleton for SOR, implementation for single cluster.}
\label{sor-code-single}
\end{figure}


On a local cluster with a fast switch-based interconnect
(like Myrinet), the exchange between neighbors adds little overhead,
so parallel SOR obtains a high efficiency.
On a wide-area system, however, the communication overhead between
neighbors that are located in different clusters will be high,
as such communication uses the WAN.
The Java program allocates neighboring processes
to the same cluster as much as possible, but the first and/or last process
in each cluster will have a neighbor in a remote cluster.
To hide the high latency for such inter-cluster communication,
the wide-area optimized 
program uses split-phase communication for exchanging rows
between clusters, as shown in Figure~\ref{sor-code-wa}.
It first initiates an asynchronous send for its boundary rows and
then computes on the inner rows of the matrix. When this work is finished,
a blocking receive for the boundary data from the neighboring machines is done,
after which the boundary rows are computed.

The optimization is awkward to express in Java, since Java lacks asynchronous
communication. It is implemented by using a
separate thread (of class \emph{SenderThread}) for sending the boundary data.
To send a row to a process on a remote cluster, the row is first
given to a newly created SenderThread; this thread
will then put the row into the Bin object of the destination process
on a remote cluster, using an RMI. During the RMI, the
original SOR process can continue computing, so communication
over the wide-area network is overlapped with computation.
For communication within
a cluster, the overhead of extra thread-switches slightly outweighs
the benefits, so only inter-cluster communication is handled
in this way (see the method \emph{sendRows}).


\begin{figure}[!ht]
\scriptsize{
\input{sor_opt}
}
\caption{Code skeleton for SOR, implementation for wide-area system.}
\label{sor-code-wa}
\end{figure}


The performance of the SOR program is shown in Figure \ref{fig:performance}.
We ran a problem with a grid size of $4096 \times 4096$ and a fixed number
of $64$ iterations.
The program obtains a high efficiency on a single cluster
(a speedup of 61.1 on 64 processors).
Without the optimization, SOR on the wide-area system achieves only a speedup
of $38.1$ on $4 \times 16$ processors. 
With the latency-hiding optimization, 
the speedup increases to $59.7$, which is close to the
speedup on a single 64-node cluster.
Latency hiding thus is very effective for SOR.


\subsection{All-pairs Shortest Paths Problem}

The All-pairs Shortest Paths (ASP) program
finds the shortest path between any pair of nodes in a graph,
using a parallel version of Floyd's algorithm.
The program uses a distance matrix that is divided row-wise among
the available processors.
At the beginning of iteration $k$, all processors need the value
of the $k$th row of the matrix. The most efficient method for
expressing this communication pattern would be to let
the processor containing this row (called the owner) broadcast it to all the
others.
Unfortunately, Java RMI does not support broadcasting, so this
cannot be expressed directly in Java.
Instead, we simulate the broadcast with a spanning tree algorithm
implemented using RMIs and threads.

The skeleton of a single-cluster implementation is shown in
Figure \ref{asp-code-single}.
Here, all processes run a thread of class \emph{Asp}. For broadcasting, they
collectively call their \emph{broadcast} method.
Inside \emph{broadcast}, all threads except the row owner wait until 
they receive the row. The owner initiates the broadcast by invoking
\emph{transfer}, which arranges all processes in a binary tree topology.
Such a tree broadcast is quite efficient inside clusters with fast local
networks. 
\emph{transfer} sends the row to its left and right children in the tree,
using daemon threads of class \emph{Sender}.
A \emph{Sender} calls \emph{transfer} on its destination node which
recursively continues the broadcast.
For high efficiency,
sending inside the binary tree has to be performed asynchronously (via daemon
threads) because otherwise all intermediate nodes would have to
wait until the RMIs of the whole successive forwarding tree completed.
As shown in Figure \ref{fig:performance}, ASP using the binary tree
broadcast achieves almost linear speedup when run on a single cluster.
With a graph of 2000 nodes, it obtains a speedup of 59.3 on a 64-node cluster.

\begin{figure}[!ht]
\scriptsize{
\input{asp_unopt}
}
\caption{Code skeleton for ASP, implementation for single cluster.}
\label{asp-code-single}
\end{figure}

A binary tree broadcast obtains poor performance on the wide-area system,
causing the original ASP program to run much slower on four clusters than on
a single (16-node) cluster (see Figure \ref{fig:performance}).
The reason is that the spanning tree algorithm does not take the topology
of the wide-area system into account, and therefore sends the same row
multiple times over the same wide-area link.
To overcome this problem, we implemented a wide-area optimized broadcast
similar to the one used in our MagPIe collective communication library 
\cite{kielmann++:99:ppopp_magpie}.
With the optimized program, the broadcast data is forwarded to all other
clusters in parallel, over different wide-area links.
We implement this scheme by designating one of the \emph{Asp} processes
in each cluster as a \emph{coordinator} for that cluster.
The broadcast owner asynchronously sends the rows to each \emph{coordinator}
in parallel. This is achieved by one dedicated thread of class
\emph{Sender} per cluster. Using this approach, each row is only sent once
to each cluster. Due to the asynchronous send, all wide-area connections can
be utilized simultaneously.
Inside each cluster, a binary tree topology is used, as in the single-cluster
program.
The code skeleton of this implementation is
shown in Figure \ref{asp-code-wa}.
As shown in Figure \ref{fig:performance}, this optimization significantly
improves ASP's application performance and makes the program run faster
on four 16-node clusters than on a single 16-node cluster.
Nevertheless, the speedup on four 16-node clusters lags far behind the
speedup on a single cluster of $64$ nodes.
This is because each processor performs several broadcasts, for different
iterations of the $k$-loop (see the \emph{run} method).
Subsequent broadcasts from different iterations wait for each other.
A further optimization therefore would be to pipeline the broadcasts
by dynamically creating new \emph{Sender}
threads (one per cluster per broadcast), instead of using dedicated daemon
threads.
However, this would require a large number
of dynamically created threads, even increasing with the problem size.
% Unfortunately, the current Manta runtime system can not handle the
% number of threads we would have needed to implement this additional
% optimization.
If a truly asynchronous RMI would be available, the excessive
use of additional threads could be completely avoided and
the overhead related to thread creation and thread switching would
also disappear.

\begin{figure}[!ht]
\scriptsize{
\input{asp_opt}
}
\caption{Code skeleton for ASP, implementation for wide-area system.}
\label{asp-code-wa}
\end{figure}




\subsection{The Traveling Salesperson Problem}

The Traveling Salesperson Problem (TSP) computes
the shortest path for a salesperson to visit all cities in a given
set exactly once, starting in one specific city.
We use a branch-and-bound algorithm, which prunes a large part of the
search space by ignoring partial routes that are already longer
than the current best solution.
The program is parallelized by distributing the search space over
the different processors.
Because the algorithm performs pruning, however, the amount of computation
needed for each sub-space is not known in advance and varies between
different parts of the search space. Therefore, load balancing
becomes an issue.
In single-cluster systems,
 load imbalance can easily be minimized using a centralized
job queue.
In a wide-area system, 
this would also generate much wide-area communication.
As wide-area optimization we implemented a static work distribution
over the different
clusters. Within each cluster, dynamic load balancing is implemented by
one job queue per cluster.
The job queues are \emph{remote} objects,
so they can be accessed over the network using RMI.
Each job contains an initial path of a fixed number of cities;
a processor that executes the job computes the lengths of all possible
continuations, pruning paths that are longer than the current
best solution.
Each processor runs one \emph{worker} thread that
repeatedly fetches jobs from the job queue of its cluster
(using RMI)
and executes the job, until all work is finished.

The TSP program keeps track of the current best solution found so far,
which is used to prune part of the search space.
Each worker contains a copy of this value.
If a worker finds a better complete route, the program does an RMI to all other
workers to update their copies. (To allow these RMIs, the workers
are declared as remote objects.)

The performance for the TSP program on the wide-area DAS system
is shown in Figure~\ref{fig:performance}, using a 17-city problem.  
As can be seen, the speedup of TSP on the wide-area system is
only slightly inferior than on a single $64$-node cluster. Our wide-area
optimized version almost completely closes the gap and
takes only $2.5\%$ longer than the single, large cluster.

\begin{figure}[t]
\begin{center}
\psfig{figure=apps.eps,height=7cm}
\caption{Speedups of four Java applications on a single cluster of 16 nodes,
4 WAN-connected clusters of 16 nodes (original and optimized program),
and a single cluster of 64 nodes.}
\label{fig:performance}
\end{center}
\end{figure}

\subsection{Iterative Deepening A*}

Iterative Deepening A* is another combinatorial
search algorithm, based on repeated depth-first searches.
IDA* tries to find a solution to a given problem by doing
a depth-first search up to a certain maximum depth. If the
search fails, it is repeated with a higher search depth,
until a solution is found.
The search depth is initialized to a lower bound of the solution.
The algorithm thus performs repeated depth-first searches.
Like branch-and-bound, IDA* uses pruning to avoid searching useless branches.

We have written a parallel IDA* program in Java for solving the 15-puzzle
(the sliding tile puzzle).
IDA* is parallelized by searching different parts of the search tree
concurrently.
The program uses a more advanced load balancing mechanism than TSP,
based on work stealing.
Each machine maintains its own job queue, but machines can get work from
other machines when they run out of jobs.
Each job represents a node in the search space.
When a machine has obtained a job, it first checks whether it can prune
the node. If not, it expands the node by computing the successor
states (children) and stores these in its local job queue.
To obtain a job, each machine first looks in its own job queue; if it
is empty it tries the job queues of some other machines.
We implemented one wide-area optimization:
to avoid wide-area communication for work stealing,
each machine first tries to steal jobs from machines in its own cluster.
Only if that fails, the work queues of
remote clusters are accessed.  In each case, the same mechanism (RMI) is
used to fetch work, so this heuristic is easy to express in Java.

Figure \ref{fig:performance} shows the speedups for the IDA* program.
The program takes about 5\% longer on the wide-area DAS system than
on a single cluster with 64 nodes. The communication overhead is due to 
work-stealing between clusters and to the distributed termination detection
algorithm used by the program.
The gain of the wide-area optimization is small in this case.

% This is due to the synchronous nature of
% work stealing that always needs to wait for RMI results before the computation
% can continue. 
% Unfortunately, the reduced wide-area communication of the optimized
% IDA* is compensated by the disadvantage of slower work distribution between
% clusters. As a consequence,  our wide-area optimization performs only as fast
% as the unoptimized version which selects nodes randomly from all clusters.



\section{Alternative programming models}
\label{sec:alternative}

We have discussed the implementation and wide-area optimization of
four parallel applications using the RMI model.
RMI supports transparent invocation of methods on remote objects and thus
is a natural extension of Java's object model to distributed memory systems.
In this section, we discuss alternative programming models and we compare
them to RMI in terms of expressiveness (ease of programming) and
implementation efficiency.
We will discuss replicated
objects \cite{bal++:98:tocs,lee:98:kan_thesis,macbeth++:98:hyperion,topol++:98:icdcs},
JavaSpaces \cite{freeman++:99:javaspaces}, and
MPI for Java \cite{javagrande:98:mpi_for_java}.
These alternative models have not been implemented in
the Manta system, so we do not provide application measurements.
Our experiences in wide-area parallel programming using replicated objects
and MPI in combination with other languages (Orca and C) are described
elsewhere\cite{bal++:98:ipps,kielmann++:99:ppopp_magpie,plaat++:99:hpca}.
We first briefly describe the four programming models.

\subsection{The programming models}

\paragraph{RMI.} 

With RMI, parallel applications strictly follow Java's object-oriented
model in which client objects invoke methods on server objects in
a location-transparent way.
% RMI simply makes remoteness of server objects transparent to the clients.
Each remote object is physically located at one machine.
% which may access this object like any other local object.
Although the RMI model hides object remoteness from the programmer,
the actual object location strongly impacts application performance.


\paragraph{Replication.}

From the client's point of view,
object replication is conceptually equivalent to the RMI model.
The difference is in the implementation: objects may be physically
replicated on multiple processes. The advantage is that
read-only operations can be performed locally,
without any communication. The disadvantage is that write operations
become more complex and have to keep object replicas consistent.

\paragraph{JavaSpaces.}

JavaSpaces adapt the Linda model \cite{gelernter:85:generativeness}
to the Java language. Communication occurs
via shared data {\em spaces\/} into which {\em entries\/}
(typed collections of Java objects) may be written. Inside a space,
entries may not be modified, but they may be read or removed (taken)
from a space.
A reader of an entry provides a {\em template\/}
that matches the desired entry type and also desired object values stored
in the entry.
Wildcards may be used for object values.
Additionally, a space may notify an object whenever an entry
matching a certain template has been written.
{\em Space\/} objects may be seen as objects that are remote to all
communicating processes; read and write operations are
implemented as RMIs to space objects.
JavaSpaces also supports a transaction model,
allowing multiple operations on space objects to be combined in a
transaction that either succeeds or fails as a whole. This feature
is especially useful for fault-tolerant programs.
% systems, an issue
% that may become important in wide-area parallel computing.


\paragraph{MPI.}

With the Message Passing Interface (MPI) language binding to Java,
communication is expressed using message passing rather than
remote method invocations.  Processes send messages
(arrays of objects) to each other. Additionally, MPI defines
collective operations in which all members of a process
group collectively participate; examples are broadcast and related
data redistributions, reduction computations (e.g.,\ computing global sums),
and barrier synchronization.

\subsection{Comparison of the models}

To compare the suitability of the models for wide-area parallel computing,
we study three important aspects.
Table~\ref{tab:models} summarizes this comparison.
In general, we assume that the underlying programming system (like Manta)
exposes the physical distribution of the clustered wide-area system.
This information
% is independent of the programming models we discuss in this section.
may be used either by application programs (as with Manta)
or by programming platforms designed for wide-area systems (e.g.,
the MagPIe library \cite{kielmann++:99:ppopp_magpie}).



\paragraph{Synchronous vs. asynchronous communication.}

Remote method invocation is a typical example of synchronous communication.
Here, the client has to wait until the server object has returned the 
result of the invoked method. This enforces rendezvous-style synchronization
of the client with another, possibly remote process.
With asynchronous communication, the client may immediately continue its
operation after the communication has been initiated. It may later check or
wait for completion of the communication operation.
Asynchronous communication is especially important for wide-area computing,
where it can be used to hide the high message latencies by
overlapping communication and computation.

MPI provides asynchronous sending and receiving.
% (e.g.,\ Comm.isend and Comm.irecv). 
The other three models, however, rely on synchronous method invocation
so applications have to simulate asynchronous communication 
using multithreading.
% by spawning an additional thread for each asynchronous operation. 
For local-area communication,
the corresponding overhead for thread creation and context switching
may exceed the cost of a synchronous RMI. To cope with this problem, the
optimized code for SOR gets rather complicated, as has been shown in the
previous section.
The broadcast implementation in ASP also requires asynchronous
communication, both in the original (single cluster) and wide-area optimized
version.
(With synchronous RMI, a broadcast sender would have to wait for the whole
spanning tree to complete.)
The TSP code also needs to treat local and remote communication differently
when updating the value of the currently best solution.
Fortunately, TSP is less sensitive to this
problem because these updates occur infrequently.
In IDA*, work stealing is always synchronous but fortunately infrequent.
So, IDA* is hardly affected by synchronous RMI.

In conclusion, for wide-area parallel computing on hierarchical systems,
directly supporting asynchronous communication
(as in MPI) is easier to use and more efficient than using synchronous
communication and multithreading.


\paragraph{Explicit vs. implicit receipt.}

A second issue is the way in which messages are received and method
invocations are served. With RMI and with replicated objects, method
invocations cause upcalls on the server side, which is a form
of implicit message receipt.
% freeing application programmers from the burden of explicitly serving them.
MPI provides only explicit message receipt, making the implementations of
TSP and IDA* much harder. Here, incoming messages (for updating the global
bound or for job requests) have to be polled for by the applications. 
This complicates the programs and makes them also less efficient, because
finding the right polling frequency
(and the best places in the applications
to put polling statements at) is difficult \cite{bal++:98:tocs}.
With clustered wide-area systems, polling has to simultaneously satisfy the
needs of LAN and WAN communication, making the problem of finding the right
polling frequency even harder.
With programming models like RMI that support implicit receipt,
application-level polling is not necessary.

JavaSpaces provide explicit operations for reading and removing entries
from a space, resulting in the same problems as with MPI.
Additionally, a space object may notify a potential reader whenever an
entry has been written that matches the reader's interests.
Unfortunately, this notification simply causes an upcall at the receiver
side which in turn has to actually perform the read or take operation.
This causes a synchronous RMI back to the space object. The additional
overhead can easily outweigh the benefit of implicit receipt,
especially when wide-area networks are used.


\paragraph{Point-to-point vs. collective communication.}

Wide-area parallel programming systems can ease the task of the programmer
by offering higher-level primitives that are mapped easily onto the
hierarchical structure of the wide-area system. In this respect,
we found MPI's collective operations (e.g., broadcast and reduction) to be of
great value. The MagPIe library \cite{kielmann++:99:ppopp_magpie} optimizes
MPI's collective operations for clustered wide-area systems by exploiting
knowledge about how groups of processes interact.
For example, a broadcast operation defines
data transfers to all processes of a group.
MagPIe uses this information to implement a broadcast that optimally utilizes
wide-area links; e.g., it takes care that data is sent only once to
each cluster. 
% Object replication allows similar optimizations for write operations on
% replicated objects, coming close to a broadcast operation.
Replicated objects that are implemented with a write-update
protocol~\cite{bal++:98:tocs} can use broadcasting for write operations,
and thus can also benefit from wide-area optimized broadcast.

Programming models without group communication (like RMI and JavaSpaces)
cannot provide such wide-area optimizations inside a runtime system.
Here, the optimizations are left to the application itself, making
it more complex. The broadcast implemented for ASP is an example
of such an application-level optimization that could be avoided by having
group communication in the programming model.
The MagPIe implementation of ASP \cite{kielmann++:99:ppopp_magpie},
for example, is much simpler than the (wide-area optimized) Java version.
Similarly, TSP's global bound
could be updated using a pre-optimized group operation.


\begin{table}%[p]
{\footnotesize
\begin{center}
\begin{tabular}{lllll}
     & RMI & replication & JavaSpaces & MPI \\
\hline
send & synchronous & synchronous & synchronous & synchronous and asynchronous \\
receive & implicit & implicit & explicit and implicit & explicit \\
collective communication & no & broadcast & no & yes
\end{tabular}
\caption{Aspects of programming models}
\label{tab:models}
\end{center}
}
\end{table}

\subsection{Summary}

The remote method invocation model provides a good starting point for
wide-area parallel programming, because it integrates communication into 
the object model. Another benefit is RMI's implicit receipt capability
by which RMIs are served using upcalls. This contributes
to expressiveness of the programming model as well as to program efficiency,
especially with wide-area systems.

To be fully suited for parallel systems, RMI needs two augmentations.
The first one is asynchronous method invocation, especially important with
wide-area systems. The second extension is the implementation of
collective communication operations that transfer the benefits of MPI's
collective operations into Java's object model.

Although promising at first glance, JavaSpaces does not really make
wide-area parallel programming easier,
because operations on space objects are neither asynchronous nor collective.
As fault-tolerance becomes more important in wide-area computing,
the JavaSpaces model along with its transaction feature may receive
more attention.



\section{Related work}
\label{sec:related}

We have discussed a Java-centric approach to writing wide-area parallel
(metacomputing) applications.
Most other metacomputing systems
(e.g., Globus \cite{Foster:1997} and Legion~\cite{Grimshaw:1997})
support a variety of languages.
The SuperWeb \cite{SuperWeb97b}, the Gateway system \cite{Haupt:1999},
Javelin~\cite{Javelin97}, Javelin++ \cite{Neary:1999}, and
Bayanihan~\cite{sarmenta++:99:bayanihan}
are examples of global computing infrastructures that support Java.
A language-centric approach makes it easier to deal with heterogeneous
systems, since the data types that are transferred over the networks
are limited to the ones supported in the language (thus obviating the need
for a separate interface definition language) \cite{Wollrath:1997}.

Most work on metacomputing focuses on how to build the necessary
infrastructure \cite{Arabe:1996,Foster:1997,Grimshaw:1997,Reinefeld:1997}.
In addition, research on parallel algorithms and applications
is required, since the bandwidth and latency differences
in a metacomputer can easily exceed three orders of magnitude
\cite{foster++:98:wide-area-mpi,Foster:1997,Grimshaw:1997,plaat++:99:hpca}.
Coping with such a large non-uniformity in the interconnect
complicates application development.
The ECO system addresses this problem by automatically generating
optimized communication patterns for collective operations
on heterogeneous networks~\cite{Lowekamp:1996}.
The AppLeS project favors the integration of workload scheduling into
the application level \cite{berman++:96:apples}.

In our earlier research, we experimented with optimizing
parallel programs for a hierarchical interconnect,
by changing the communication structure \cite{bal++:98:ipps}.
Also, we studied the sensitivity of such optimized programs to large
differences in latency and bandwidth between the LAN and WAN 
\cite{plaat++:99:hpca}.
Based on this experience, we implemented collective communication operations
as defined by the MPI standard, resulting in improved application
performance on wide area systems
\cite{kielmann++:99:ppopp_magpie}.
Some of the ideas of this earlier work have been applied in
our wide-area Java programs.
% In this work, we investigate application performance on wide area systems
% when Java's RMI mechanism is used for communication.

There are many other research projects for parallel programming in Java.
% (see, for example, the JavaGrande Forum at {\em http://www.javagrande.org/}). 
Titanium \cite{yelick:1998} is a Java-based
language for high-performance parallel scientific computing.
% It extends
% Java with features like immutable classes, fast multidimensional array
% access and
% an explicitly parallel SPMD model of communication. The Titanium
% compiler translates Titanium into C. Titanium is built on
% the Split-C/Active Messages back-end.
The JavaParty system \cite{philippsen:1997} is designed to ease
parallel cluster programming in Java. In particular, its goal is to run
multi-threaded programs with as little change as possible on a
workstation cluster.
Hyperion~\cite{macbeth++:98:hyperion} also uses the standard Java thread model
as a basis for parallel programming.
Unlike JavaParty, Hyperion caches remote objects to improve performance.
% Modifications on cached objects are propagated when a thread enters or
% leaves a monitor. 
% JavaParty is implemented on top of Java RMI, and thus suffers from
% the same performance problem as RMI.
The Do! project tries to ease parallel programming in Java using
parallel and distributed frameworks \cite{launay:1998}.
Ajents \cite{Izatt:1999} is a parallel programming environment that
supports object migration.
Java/DSM \cite{yu:1997} implements a JVM on top of a
distributed shared memory system. 
Breg et al.~\cite{Breg:1998} study RMI performance and
interoperability.  Krishnaswamy et al.~\cite{Krishnaswamy:1998} 
improve RMI performance somewhat with caching and by using UDP instead of TCP. 
Nester et al.~\cite{Nester:1999} present new RMI and serialization packages
(drop-in replacements) designed to improve RMI performance.
The above Java systems are designed for single-level (``flat'')
parallel machines.
The Manta system described in this paper,
on the other hand, is designed for hierarchical systems and uses
different communication protocols for local and wide area networks.
It uses a highly optimized RMI implementation,
which is particularly effective for local communication.


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

We have described our experiences in building and using a
high-performance Java system that runs on a geographically distributed
(wide-area) system.
The goal of our work was to obtain actual experience with a Java-centric
approach to metacomputing.
Java's support for parallel processing and heterogeneity
make it an attractive candidate for metacomputing.
The Java system we have built, for example, is highly transparent:
it provides a single communication primitive (RMI) to the user,
even though the implementation uses several communication networks and
protocols.

Our Manta programming system is designed for hierarchical wide-area systems,
for example clusters or MPPs connected by wide-area networks.
Manta uses a very efficient (active message like) communication protocol
for the local interconnect (Myrinet) and TCP/IP for wide-area communication.
The two communication protocols are provided by the Panda library.
Manta's light-weight RMI protocol is implemented on top of Panda and
is very efficient.

We have implemented several parallel applications on this system, using
Java RMI for communication. In general, the RMI model was easy to use.
To obtain good performance, the programs take the hierarchical structure
of the wide-area system into account and minimize the amount of communication
(RMIs) over the slow wide-area links.
With such optimizations in place, the programs can effectively use
multiple clusters, even though they are connected by slow links.

We compared RMI with other programming models, namely object replication,
JavaSpaces, and MPI for Java.
We identified several shortcomings of the RMI model. In particular,
the lack of asynchronous communication and broadcast complicates programming.
MPI offers both features (and further useful collective operations)
but lacks RMI's clean object-oriented model.
Object replication is closer to pure RMI but offers only broadcast-like
object updating.
A careful integration of these features into an RMI-based system is
our next goal for a Java-centric programming platform
for wide-area parallel programming.


{\small
\section*{Acknowledgements}
This work is supported in part by a SION grant from the Dutch research
council NWO, and by a USF grant from the Vrije Universiteit.  
The wide-area DAS system is an initiative of
the Advanced School for Computing and Imaging (ASCI).
We thank Aske Plaat, Raoul Bhoedjang, Rutger Hofman, Ceriel Jacobs,
and Cees Verstoep for their contributions to this research.
We thank Cees Verstoep and John Romein for keeping the DAS in good shape,
and Cees de Laat (University of Utrecht) for getting the wide area links
of the DAS up and running.
}

\small

\bibliographystyle{abbrv}
\bibliography{cpe99}

\end{document}
