%\chapter{Object Serialization for Marshalling Data in a Java Interface to MPI}
\chapter{Object Serialization for Marshalling Data in mpiJava}
\label{chap:objseri}

\pagestyle{myheadings}
\markright{CHAPTER~\ref{chap:objseri}.  MARSHALLING DATA IN A
JAVA INTERFACE TO MPI} 
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
%In mpiJava, message buffers are restricted to arrays
%with elements of primitive type.  
In this chapter we discuss adoption of the Java object serialization
model for marshalling general communication data in MPI-like APIs.
This approach is compared with a Java transcription of the standard
MPI derived datatype mechanism.  We describe an implementation of the
{\em mpiJava} interface to MPI that incorporates automatic object
serialization.  Benchmark results confirm that current JDK
implementations of serialization are not fast enough for high
performance messaging applications.  Means of solving this problem are
discussed, and benchmarks for greatly improved schemes are presented.

Section \ref{implementation} describes an implementation of
automatic object serialization in mpiJava.  In section \ref{benchmarks}
we discuss benchmarks for this initial implementation.  The
results confirm that naive use of existing Java serialization technology
does not provide the performance needed for high performance message
passing environments.  Section \ref{sec:optimize} illustrates how
various overheads of serialization can be eliminated by customizing the
object serialization stream classes.
The final section draws some conclusion.

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
\section{Motivation}
%Over the past two years
%several groups have independently developed Java bindings to MPI and
%Java implementations of MPI subsets.  With support of several groups working
%in the area, the Java Grande Forum drafted an initial proposal for a
%common MPI-like API for Java \cite{MPIPOSITION}.

A characteristic feature of MPI is its flexible method for describing
message buffers containing mixed primitive fields scattered,
possibly non-contiguously, over the local memory of a processor.
These buffers are described through special objects called {\em derived
datatypes}---run-time analogues of the user-defined types supported by
modern procedural languages.  The standard MPI approach does not map very
naturally into Java.  In the previous section ~\ref{DATATYPE}
we suggested a Java-compatible restriction of the general MPI derived
datatype mechanism, in which all primitive elements of a message
buffer have the same type, and they are selected from the elements of a
one-dimensional Java array passed as the buffer argument.  This approach
preserves some of the functionality of the original MPI mechanism---for
example the ability to describe strided sections of a one dimensional
buffer argument, and to represent a subset of elements selected from
the buffer argument by an indirection vector.  But it does not allow
description of buffers containing elements of mixed primitive types.

This version of the MPI derived datatype mechanism was retained in the
initial draft of \cite{MPIPOSITION}, but its value is not yet
certain.  A more promising approach may be the addition a new basic
datatype to MPI representing a serializable object.  The buffer array
passed to communication functions is still a one-dimensional array, but
as well as allowing arrays with elements of primitive type, the element
type is allowed to be {\tt Object}.  The serialization paradigm of Java
can be adopted to transparently serialize buffer elements at source
and unserialize them at destination.  An immediate application is to
multidimensional arrays.  A Java multidimensional array is an array of
arrays, and an array is an object.  Therefore a multidimensional array
is a one-dimensional array of objects and it can be passed directly as
a buffer array.  The options for representing sections of such an array
are limited, but at least one can communicate whole multidimensional
arrays without explicitly copying them (though there may be
copying inside the implementation).


%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
\section{Datatypes in an MPI-like API for Java\label{review}}

The MPI standard is explicitly object-based.  The C++ binding
specified in the MPI 2 standard collects these objects into suitable
class hierarchies and defines most of the library functions as class
member functions.  The Java API proposed in \cite{MPIPOSITION}
follows this model, and lifts its class hierarchy directly from the C++
binding of MPI.

In our Java version a class {\tt MPI} with only static members acts
as a module containing global services, such as initialization of the
message-passing layer, and many global constants including a default
communicator {\tt COMM\_WORLD}
(It has been pointed out that if multiple MPI threads are allowed in the 
same Java VM, the default communicator cannot be obtained from a static 
variable.  The final version of the API may change this convention.)
The communicator class {\tt Comm} is the single most important class in MPI.
All communication functions are members of {\tt Comm} or its
subclasses.  Another class that is relevant for the discussion below
is the {\tt Datatype} class.  This describes the type of the elements
in the message buffers passed to send, receive, and other
communication functions.  Various basic datatypes are predefined in
the package.  These mainly correspond to the primitive types of Java,
shown in figure \ref{basicdatatype} in section \ref{sec:introAPI}.


The methods corresponding to standard send and receive operations of
MPI are members of {\tt Comm} with interfaces
\small
\begin{verbatim}
        void send(Object buf, int offset, int count,
                  Datatype datatype, int dst, int tag)

        Status recv(Object buf, int offset, int count,
                    Datatype datatype, int src, int tag)
\end{verbatim}
\normalsize
In both cases the {\em actual} argument corresponding to {\tt buf}
must be a Java array with element type compatible with the {\tt datatype}
argument.  If the specified type corresponds to a primitive type,
the buffer must be a one-dimensional array.  Multidimensional arrays
can be communicated directly if an object type is specified, because
an individual array can be treated as an object.  Communication of
object types implies some form of serialization and unserialization.
This could be the built-in serialization provided in current Java
environments, or (as we discuss at length in section \ref{sec:optimize})
it could be some specialized serialization tuned for message-passing.

Besides object types the draft Java binding proposal retains a model of
MPI derived datatypes.  In C or Fortran bindings of MPI, derived
datatypes have two roles.  One is to allow messages to contain mixed
types.  The other is to allow noncontiguous data to be transmitted.
The first role involves using the {\tt MPI\_TYPE\_\-STRUCT} derived
data constructor, which allows one to describe the physical layout of, say,
a C {\em struct} containing mixed types.  This will not work
in Java, because Java does not expose the low-level layout of its
objects.  In C or Fortran {\tt MPI\_TYPE\_\-STRUCT} also allows one
to incorporate displacements computed as differences between absolute
addresses, so that parts of a single message can come from separately declared
arrays and other variables.  Again there is no very natural way to do
this in Java.  (But effects similar to these uses of {\tt
MPI\_TYPE\_\-STR\-UCT} can be achieved by using {\tt
MPJ.OBJECT} as the buffer type, and relying on object serialization.)

We conclude that in the Java binding the first role of derived dataypes should
probably be abandoned---derived types can only include elements of a
single basic type.  This leaves description of noncontiguous buffers as
the remaining role for derived data types.
Every derived data type constructable in the Java binding has a
uniquely defined {\em base type}.  This is one of the 9 basic types
enumerated above.  A {\em derived datatype} is an object that specifies
two things:  a base type and a sequence of integer displacements. In
contrast to the C and Fortran bindings the displacements can be
interpreted in terms of subscripts in the buffer array argument, rather
than as byte displacements.

For example the type constructor {\tt indexed} is a member of
{\tt Datatype} with interface
\small
\begin{verbatim}
  Datatype indexed(int [] arrayOfBlocklengths,
                   int [] arrayOfDisplacements)
\end{verbatim}
\normalsize
This is a binding of the standard MPI operation {\tt
MPI\_TYPE\_INDEXED}.
It constructs a new datatype representing replication of the original
datatype (to which the method is applied) into a sequence of blocks.
Each block can contain a different number of copies and have a
different displacement. 
The base type of the new datatype will be the same as the base type of
the original type.  If the displacement sequence of the original type was
\begin{displaymath}
\{ \mbox{\em disp}_0, \ldots, \mbox{\em disp}_{n - 1} \}
\end{displaymath}
with extent\footnote{The {\em extent} of a dataype
is a measure of the distance between its smallest and largest
displacement.} {\em ex}, and {\tt B} is {\tt arrayOfBlocklengths}
argument and {\tt D} is {\tt arrayOfDisplace\-ments} argument,
the new datatype will have displacement sequence
\begin{displaymath}
\{
\begin{array}[t]{l}
\mbox{\em disp}_0 + \mbox{\tt D}[0] \cdot \mbox{\em ex},
\ldots,
\mbox{\em disp}_{n - 1} + \mbox{\tt D}[0] \cdot \mbox{\em ex}, \\
\mbox{\em disp}_0 + (\mbox{\tt D}[0] + 1) \cdot \mbox{\em ex},
\ldots,
\mbox{\em disp}_{n - 1} + (\mbox{\tt D}[0] + 1) \cdot \mbox{\em ex}, \\
\ldots , \\
\mbox{\em disp}_0 +
   (\mbox{\tt D}[0] + \mbox{\tt B} [0] - 1) \cdot \mbox{\em ex},
\ldots, \\
\quad \quad \quad \quad \quad \mbox{\em disp}_{n - 1} +
   (\mbox{\tt D}[0] + \mbox{\tt B} [0] - 1) \cdot \mbox{\em ex}, \\
\\
\ldots , \\
\\
\mbox{\em disp}_0 + \mbox{\tt D}[c - 1] \cdot \mbox{\em ex},
\ldots,
\mbox{\em disp}_{n - 1} + \mbox{\tt D}[c - 1] \cdot \mbox{\em ex}, \\
\mbox{\em disp}_0 + (\mbox{\tt D}[c - 1] + 1) \cdot \mbox{\em ex},
\ldots, \\
\quad \quad \quad \quad \quad \mbox{\em disp}_{n - 1} +
   (\mbox{\tt D}[c - 1] + 1) \cdot \mbox{\em ex}, \\
\ldots , \\
\mbox{\em disp}_0 +
   (\mbox{\tt D}[c - 1] + \mbox{\tt B} [c - 1] - 1) \cdot \mbox{\em ex},
\ldots, \\
\quad \quad \quad \quad \quad \mbox{\em disp}_{n - 1} +
   (\mbox{\tt D}[c - 1] + \mbox{\tt B} [c - 1] - 1) \cdot \mbox{\em ex}
\,\, \}
\end{array}
\end{displaymath}
Here, {\em c} is the number of blocks.

In Java the derived dataype constructed by {\tt indexed} has a
potentially useful role.  It allows to send (or receive) messages
containing values scattered randomly in some one-dimensional array.  
The draft proposal incorporates versions of other type constructors from
MPI including {\tt MPI\_TYPE\_VECTOR} for strided sections.
We note, though, that the value of providing strided sections is reduced
because Java has no natural mapping between elements of its
multidimensional arrays and elements of equivalent one-dimensional
arrays.  This thwarts one common use of strided sections, for representing
portions of multidimensional arrays.


%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
\section{Adding Serialization to the API\label{implementation}}

In this section we will discuss the other
option for representing complex data buffers in the Java API of
\cite{MPIPOSITION}---introduction of an {\tt MPJ.OBJECT} datatype.

It is natural to assume that the elements of buffers
passed to {\tt send} and other output operations are objects
whose classes implement the {\tt Serializable} interface.
There are at least two ways one may consider communicating object types
in the MPI interface
\begin{enumerate} 
\item
Use the standard {\tt ObjectOutputStream} 
to convert the object buffers to byte vectors, and communicate
these byte vectors using the same method as for primitive
byte buffers (for example, this might involve a native method
call to C MPI functions).  At the destination, use the
standard {\tt ObjectInputStream} to rebuild the objects.
\item
Replace naive use of serialization streams with more
specialized code that uses platform-specific knowledge to communicate
data fields efficiently.  For example, one might modify the standard
{\tt writeObject} in such a way that a native method creates an MPI
derived datatype structure describing the layout of data in the object,
and this buffer descriptor could be passed to a native {\tt MPI\_Send} function.
\end{enumerate} 
In the second case our implementation is responsible for prepending
a suitable type descriptor to the message, so that objects can be
reconstructed at the receiving end before data is copied to them.

The first implementation scheme is more strai\-ghtforward, and this
approach will be considered in the remainder of this section.  We
discuss an implementation based on the mpiJava wrappers, combining
standard JDK object serialization methods with a JNI interface to
native MPI.  Benchmark results presented in the next section suggest
that something like the second approach (or some suitable combination
of the two) deserves serious consideration, hence section
\ref{sec:optimize} describes one realization of this scheme.


The original version of mpiJava was a direct Java wrapper for standard
MPI.  Apart from adopting an object-orien\-ted framework, it added a
modest amount of code to the underlying C implementation of MPI.
Derived datatype constructors, for example, simply called the datatype
constructors of the underlying implementation and returned a Java
object containing a representation of the C handle.  A {\tt send} operation or
a {\tt wait} operation, say, dispatched a single C MPI
call.  Even exploiting standard JDK object serialization and a native
MPI package, uniform support for the {\tt MPJ.OBECT} basic type
complicates the wrapper code significantly.

In the new version of the wrapper, every send, receive, or collective
communication operation tests if the base type of the datatype argument
describing a buffer is {\tt OBJECT}.  If not---if the buffer element
type is a primitive type---the native MPI operation is called directly,
as in the old version.  If the buffer is an array of objects, special
actions must be taken in the wrapper.  If the buffer is a send buffer,
the objects must be serialized.  
We {\em also} support MPI-like derived datatypes as described in the
previous section.  On grounds of uniformity, these should be definable
with base type {\tt OBJECT}, just as for primitive elements.  
The message is then some subset of the array of objects passed in the
buffer argument, selected according to the displacement sequence of
the derived datatype.   
% org
%This case must be dealt with in the the Java wrapper, because a native
%{\tt MPI\_Datatype} entity cannot be constructed to directly represent
%Jave objects.  Thus when the base type is {\tt OBJECT} the Java-side
%{\tt Datatype} class requires additional fields; it explicitly
%maintains the displacement sequence as an array of integers.
% 2-col
In the implementation, a method
\small
\begin{verbatim}
  byte [] Object_Serialize(Object   buf,
                           int      offset,
                           int      count,
                           Datatype type)
\end{verbatim}
\normalsize
takes the send buffer and descriptor, and returns a byte vector
containing the serialized data.  At the receiving end a corresponding
{\tt Object\_deserialize} method is called.
Making the Java wrapper responsible for handling derived data types
when the base type is {\tt OBJECT} requires additional fields in the
Java-side {\tt Datatype} class.  In particular the Java object
explicitly maintains the displacement sequence as an array of
integers.
% end 2-col

A further set of changes to the implementation arises because the size
of the serialized data is not known in advance, and cannot be computed
at the receiving end from type information available there.  Before
the serialized data is sent, the size of the data must be communicated
to the receiver, so that a byte receive buffer can be allocated.
We send two physical messages---a header containing size information,
followed by the data%
\footnote{A better protocol would be to eagerly send data for short
messages in the header, assuming some fixed-size buffer
is preallocated at the receiving end.  The two-message protocol
would be reserved for long messages.   This marginally complicates the
implementation but does not essentially change the rest of the discussion,
or the benchmark results presented below, since the latter concentrate on the
asymptotic case.}%
.
This, in turn, complicates the implementation of the various {\tt wait}
and {\tt test} methods on communication request objects, and the {\tt
start} methods on persistent communication requests, and ends up
requiring extra fields in the Java {\tt Request} class.
Comparable changes are needed in the collective communication
wrappers.  A {\tt gather} operation, for example, involving object
types is implemented as an {\tt MPI\_GATHER} operation to collect all
message lengths, followed by an {\tt MPI\_GATHERV} to collect possibly
different-sized data vectors.


%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
\section{Benchmark Results for Multidimensional Arrays\label{benchmarks}}

For the sake of concrete discussion we will
make an assumption that, in the kind of {\em Grande} applications where MPI is
likely to be used, some of the most pressing performance issues
concern arrays
and multidimensional arrays of small objects---especially arrays
of primitive elements such as {\tt int}s and {\tt float}s.  For
benchmarks we therefore concentrated on the overheads
introduced by object serialization when the objects contain many arrays
of primitive elements.  Specifically we concentrated on communication
of two-dimensional arrays with primitive elements.
We note that there some debate about whether the
Java model of multidimensional arrays is the most appropriate one
for high performance computing.  There are various proposals for
optimized HPC array class libraries \cite{MIDKIFF}.  See Section
\ref{sec:DISCUSSION} for some further discussion.


The ``ping-pong'' method was used to time
point-to-point communication of an $N$ by $N$ array of primitive
elements treated as a one dimensional array of objects, and compare it
with communication of an $N^2$ array without using serialization.  As
an intermediate case we also timed communication of a $1$ by $N^2$
array treated as a one-dimensional (size 1) array of objects.
This allows us to extract an estimate of the overhead to ``serialize''
an individual primitive element.
The code for sending and receiving the various array shapes
is given schematically in Figure \ref{fig:shapes}.

\begin{figure*}[t]
\Line \\ \\
\centerline{\bf $N^2$ float vector}
\centerline{
\begin{minipage}[t]{0in}
\footnotesize
\tt
\begin{tabbing}
\hspace{3.1in} \\
float [] buf = new float [N * N] ; \\
MPJ.COMM\_WORLD.send(\=buf, 0, N * N, \\
                     \>MPJ.FLOAT, \\
                     \>dst, tag) ;
\end{tabbing}
\end{minipage}
\begin{minipage}[t]{0in}
\footnotesize
\tt
\begin{tabbing}
\hspace{0in} \\
float [] buf = new float [N * N] ; \\
MPJ.COMM\_WORLD.recv(\=buf, 0, N * N, \\
                     \>MPJ.FLOAT, \\
                     \>src, tag) ;
\end{tabbing}
\end{minipage}
}

\vspace{0.5in}
\centerline{\bf $N \times N$ float array}
\centerline{
\begin{minipage}[t]{0in}
\footnotesize
\tt
\begin{tabbing}
\hspace{3in} \\
float [] [] buf = new float [N] [N] ; \\
MPJ.COMM\_WORLD.send(\=buf, 0, N, \\
                    \>MPJ.OBJECT, \\
                    \>dst, tag) ;
\end{tabbing}
\end{minipage}
\begin{minipage}[t]{0in}
\footnotesize
\tt
\begin{tabbing}
\hspace{0in} \\
float [] [] buf = new float [N] [] ; \\
MPJ.COMM\_WORLD.recv(\=buf, 0, N, \\
                    \>MPJ.OBJECT, \\
                    \>src, tag) ;
\end{tabbing}
\end{minipage}
}

\vspace{0.5in}
\centerline{\bf $1 \times N^2$ float array}
\centerline{
\begin{minipage}[t]{0in}
\footnotesize
\tt
\begin{tabbing}
\hspace{3in} \\
float [] [] buf = new float [1] [N * N] ; \\
MPJ.COMM\_WORLD.send(\=buf, 0, 1, \\
                    \>MPJ.OBJECT, \\
                    \>dst, tag) ; \\
\end{tabbing}
\end{minipage}
\begin{minipage}[t]{0in}
\footnotesize
\tt
\begin{tabbing}
\hspace{0in} \\
float [] [] buf = new float [1] [] ; \\
MPJ.COMM\_WORLD.recv(\=buf, 0, 1, \\
                    \>MPJ.OBJECT, \\
                    \>src, tag) ;
\end{tabbing}
\end{minipage}
}
\Line 
\caption{Send and receive operations for various array shapes.
\label{fig:shapes}}
\end{figure*}

As a crude timing model for these benchmarks, one can assume that there
is a cost $t^{\mbox{\scriptsize T}}_{\mbox{\scriptsize ser}}$ to
serialize each primitive element of type {\tt T}, an additional cost
$t^{\mbox{\scriptsize vec}}_{\mbox{\scriptsize ser}}$ to serialize each
subarray, similar constants
$t^{\mbox{\scriptsize T}}_{\mbox{\scriptsize unser}}$ and
$t^{\mbox{\scriptsize vec}}_{\mbox{\scriptsize unser}}$ for
unserialization, and a cost
$t^{\mbox{\scriptsize T}}_{\mbox{\scriptsize com}}$ to 
physically tranfser each element of data.
Then the total time for benchmarked communications should be
\begin{eqnarray}
t^{\mbox{\small \tt T[$N^2$]}}     & = & c +
    t^{\mbox{\scriptsize T}}_{\mbox{\scriptsize com}} N^2\label{eqn:mod1} \\
t^{\mbox{\small \tt T[1][$N^2$]}} & = & c' +
   (t^{\mbox{\scriptsize T}}_{\mbox{\scriptsize ser}} +
    t^{\mbox{\scriptsize T}}_{\mbox{\scriptsize com}} +
    t^{\mbox{\scriptsize T}}_{\mbox{\scriptsize unser}}) N^2
                                                      \label{eqn:mod2} \\
t^{\mbox{\small \tt T[$N$][$N$]}} & = & c'' +
   (t^{\mbox{\scriptsize vec}}_{\mbox{\scriptsize ser}} +
    t^{\mbox{\scriptsize vec}}_{\mbox{\scriptsize unser}}) N + \nonumber \\
&& (t^{\mbox{\scriptsize T}}_{\mbox{\scriptsize ser}} +
    t^{\mbox{\scriptsize T}}_{\mbox{\scriptsize com}} +
    t^{\mbox{\scriptsize T}}_{\mbox{\scriptsize unser}}) N^2
                                                      \label{eqn:mod3}
\end{eqnarray}
These formulae do not attempt to explain the constant initial overhead,
don't take into account the extra bytes for type description that
serialization introduces into the stream, and ignore possible
non-linear costs associated with analysing object graphs, etc.
Empirically these effects are small for the range of $N$ we consider.

All measurements were performed on a cluster of 2-proc\-essor, 200 Mhz
UltraSparc nodes connected through a Sun\-ATM-155/MMF network.  The
underlying MPI implementation was Sun MPI 3.0 (part of the Sun HPC
package).  The JDK was jdk1.2beta4.  Shared memory results quoted are
obtained by running two processes on the processors of a single node.
Non-shared-memory results are obtained by running peer processes in
different nodes.

In a series of measurements, element serialization and unserialization
timing parameters were estimated by independent benchmarks of the
serialization code.  The parameters $t^{\mbox{\scriptsize
vec}}_{\mbox{\scriptsize ser}}$ and $t^{\mbox{\scriptsize
vec}}_{\mbox{\scriptsize unser}}$ were estimated by plotting the
difference between serialization and unserialization times for {\tt
T[1][$N^2$]} and {\tt T[$N$][$N$]}
%\footnote{
Our timing
model assumed the values of these parameters is independent of the
element type.  This is only approximately true, and the values quoted in
the table and used in the plotted curves are averages.
Separately measured values for {\tt byte} arrays were smaller than
these averages, and for {\tt int} and {\tt float} arrays they were
larger.
%}.
The raw communication speed was
estimated from ping-pong results for $t^{\mbox{\small \tt T[$N^2$]}}$.
Table \ref{tab:params}
contains the resulting estimates of the various parameters for {\tt byte}
and {\tt float} elements.

\begin{table}[t]
\Line \\ \\
\centerline{
\begin{tabular}{ l @{\hspace{0.1in}=\hspace{0.1in}} l @{\hspace{0.4in}}
                 l @{\hspace{0.1in}=\hspace{0.1in}} l @{\hspace{0.4in}} 
                 l @{\hspace{0.1in}=\hspace{0.1in}} l }
$t^{\mbox{\scriptsize byte}}_{\mbox{\scriptsize ser}}$    & 0.043$\mu$s &
$t^{\mbox{\scriptsize float}}_{\mbox{\scriptsize ser}}$   & 2.1$\mu$s &
$t^{\mbox{\scriptsize vec}}_{\mbox{\scriptsize ser}}$     & 100$\mu$s \\[0.2in]
$t^{\mbox{\scriptsize byte}}_{\mbox{\scriptsize unser}}$  & 0.027$\mu$s &
$t^{\mbox{\scriptsize float}}_{\mbox{\scriptsize unser}}$ & 1.4$\mu$s &
$t^{\mbox{\scriptsize vec}}_{\mbox{\scriptsize unser}}$   & 53$\mu$s \\[0.2in]
$t^{\mbox{\scriptsize byte}}_{\mbox{\scriptsize com}}$    & $0.062\mu\mbox{s}^\dag$ &
$t^{\mbox{\scriptsize float}}_{\mbox{\scriptsize com}}$   & $0.25\mu\mbox{s}^\dag$ \\[0.2in]
$t^{\mbox{\scriptsize byte}}_{\mbox{\scriptsize com}}$    & $0.008\mu\mbox{s}^\S$ &
$t^{\mbox{\scriptsize float}}_{\mbox{\scriptsize com}}$   &
                 $0.038\mu\mbox{s}^\S$ \\ [0.07in]
\end{tabular}
}
\\ \\ \Line 
\caption{Estimated parameters in serialization and communication timing
model.  The $t^{\mbox{\scriptsize T}}_{\mbox{\scriptsize com}}$ values
are respectively for non-shared memory ($\dag$) and shared memory ($\S$)
implementations of the underlying communication.
\label{tab:params}}
\end{table}

Figures \ref{fig:pingpong1} and \ref{fig:pingpong2} on Page \pageref
{fig:pingpong1} and \pageref{fig:pingpong2} plot actual measured
times from ping-pong benchmarks for the mpiJava sends and receives of
arrays with {\tt byte} and {\tt float} elements.  In the
plots the array extent, $N$, ranges between 128 and 1024.  The measured
times for $t^{\mbox{\small \tt T[$N^2$]}}$, $t^{\mbox{\small \tt
T[1][$N^2$]}}$ and $t^{\mbox{\small \tt T[$N$][$N$]}}$ are compared
with the formulae given above (setting the $c$ constants to zero).  The
agreement is good, so our parametrization is assumed to be realistic in
the regime considered. 


\afterpage{\clearpage}
\begin{figure}[ht]
\centerline{\psfig{figure=Figs/nonsh_byte.eps,width=3.5in}}
\centerline{\psfig{figure=Figs/nonsh_float.eps,width=3.5in}}
\caption{Communication times from ping-pong benchmark in non-shared-memory
case.  The lines represent the model
defined by Equations \ref{eqn:mod1} to \ref{eqn:mod3} in the text, with
parameters from Table \ref{tab:params}.}
\label{fig:pingpong1}
\end{figure}


\afterpage{\clearpage}
\begin{figure}[ht]
\centerline{\psfig{figure=Figs/sh_byte.eps,width=3.5in}}
\centerline{\psfig{figure=Figs/sh_float.eps,width=3.5in}}
\caption{Communication times from ping-pong benchmark in shared-memory
case.  The lines represent the model
defined by Equations \ref{eqn:mod1} to \ref{eqn:mod3} in the text, with
parameters from Table \ref{tab:params}.}
\label{fig:pingpong2}
\end{figure}


%\begin{figure*}[t]
%\centerline{\psfig{figure=Figs/nonsh_byte.eps,width=5.0in}
%            \psfig{figure=Figs/sh_byte.eps,width=2.4in}}
%\centerline{\psfig{figure=Figs/nonsh_float.eps,width=2.4in}
%            \psfig{figure=Figs/sh_float.eps,width=2.4in}}
%\caption{Communication times from ping-pong benchmark in non-shared-memory
%and shared-memory cases.  The lines represent the model
%defined by Equations \ref{eqn:mod1} to \ref{eqn:mod3} in the text, with
%parameters from Table \ref{tab:params}.\label{fig:pingpong2}}
%\end{figure*}
%

According to table \ref{tab:params} the overhead of Java serialization
nearly always dominates other communication costs.  In the worst
case---floating point numbers---it takes around 2 microseconds to
serialize each number and a smaller but comparable time to
unserialize.  But it only takes a few hundredths of a microsecond to
communicate the word through shared memory.  Serialization slows
communication by nearly two orders of magnitude.  When the underlying
communication is over a fast network rather than through shared memory
the raw communication time is still only a fraction of a microsecond,
and serialization still dominates that time by about one order of
magnitude.  For {\tt byte} elements serialization costs are smaller,
but still larger than the communication costs in the fast network
and still much larger than the communication cost through shared
memory.  Serialization costs for {\tt int} elements are intermediate.

The constant overheads for serializing each subarray, cha\-racterized by
the parameters $t^{\mbox{\scriptsize vec}}_{\mbox{\scriptsize ser}}$
and $t^{\mbox{\scriptsize vec}}_{\mbox{\scriptsize unser}}$ are also
quite large, although, for the array sizes considered here they only
make a dominant contribution for the {\tt byte} arrays, where
individual element serialization is relatively fast.


%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
\section{Reducing Serialization Overheads for Arrays\label{sec:optimize}}

The work of \cite{PHILIPPSEN99} and others mentioned in section
\ref{EFFRMIOBJSERI}, has established that there 
is considerable scope to optimize the JDK serialization software.
Here we pursue an alternative that is interesting from the point of
view of ultimate efficiency in messaging APIs, namely to replace calls
to the {\tt writeObject}, {\tt readObject} methods with specialized,
MPI-specific, functions.  A call to standard {\tt write\-Object}, for
example, might be replaced with a native method that creates a native MPI
derived datatype structure describing the layout of data in the object.
This would provide the conceptually straightforward object serialization
model at the user level, while retaining the option of fast (``zero-copy'')
communication strategies inside the implementation.

Implementing this general scheme for every kind of Java object is
difficult or impractical because the JVM hides the internal representation
of most objects.  Less ambitiously, we can attempt to eliminate the
serialization and copy overheads for arrays of primitive
elements embedded in the serialization stream.  The general idea is
to produce specialized versions of {\tt ObjectOutputStream} and {\tt
ObjectInputStream} that yield byte streams identical to the standard
version {\em except} that array data is omitted from those streams.
The ``data-less'' byte stream is sent as a header.  This allows the
objects to be reconstructed at the receiving end.  The array data is then
sent separately using, say, suitable native {\tt MPI\_TYPE\_STRUCT} types
to send all the array data in one logical communication.  In this way
the serialization overhead parameters measured in the benchmarks of the
previous section can be drastically reduced or eliminated. An implementation
of this protocol is illustrated in Figure \ref{fig:arrayProtocol}.

\begin{figure}[tp]
\centerline{\psfig{figure=Figs/arrayProtocol.eps,width=5.3in}}
\caption{Improved protocol for handling arrays of primitive elements.
\label{fig:arrayProtocol}}
\end{figure}

A customized version of {\tt ObjectOutputStream} called {\tt
ArrayOutputStream} behaves in exactly the same way as the original stream
except when it encounters an array.  When an array is encountered
a small object of type {\tt ArrayProxy} is placed in the stream.
This encodes the type and size of the array.  The array reference
itself is placed in a separate container called the ``data vector''.
When serialization is complete, the data-less byte stream is sent to
the receiver.  A piece of native code unravels the data vector and sets
up a native derived type, then the array data is sent.  At the receiving
end a customized {\tt ArrayInputStream} behaves exactly like an {\tt
ObjectInputStream}, except that when it encounters an {\tt ArrayProxy}
it allocates an array of the appropriate type and length and places a
handle to this array in the reconstructed object graph {\em and} in a
data vector container.  When this phase is completed we have an object
graph containing uninitialized array elements and a data vector, created
as a side effect of unserialization.  A native derived data type is
constructed from the data vector in the same way as at the sending
end, and the data is received into the reconstructed object in a single
MPI operation.

Our implementation of {\tt ArrayOutputStream} and {\tt ArrayInputStream}
is straightforward.  The standard {\tt ObjectOutputStream}
provides a method, {\tt replace\-Object}, which can be overridden in
subclasses.  {\tt ObjectInputStream} provides a corresponding
{\tt resolveObject} method.  Implementation of the customized streams
is sketched in Figure \ref{fig:arrayStreams}.

\begin{figure}[t]
\Line \\ \\
\centerline{
\begin{minipage}[t]{0in}
%\footnotesize
\small
\tt
\begin{tabbing}
XXX \= XXX \= XXX \=  \kill
class ArrayOutputStream extends ObjectOutputStream \{ \\
  \> Vector dataVector ;\\
\\
  \> public Object replaceObject(Object obj) \{ \\
  \>  \> if(obj instanceof int []) \{ \\
  \>  \>  \> dataVector.addElement(obj) \\
  \>  \>  \> return new ArrayIntProxy(((int []) obj).length) ; \\
  \>  \> \} \\
  \>  \> {\em \ldots \ deal with other primitive array types \ldots} \\
  \>  \> else \\
  \>  \>  \> return obj \\
  \> \} \\
\} \\
\\ \\
class ArrayInputStream extends ObjectInputStream \{ \\
  \> Vector dataVector ;\\
\\
  \> public Object resolveObject(Object obj) \{ \\
  \>  \> if(obj instanceof ArrayIntProxy) \{ \\
  \>  \>  \> int dat = new int [((ArrayIntProxy) obj).length] ; \\
  \>  \>  \> dataVector.addElement(dat) \\
  \>  \>  \> return dat ; \\
  \>  \> \} \\
  \>  \> {\em \ldots \ deal with other array proxy types \ldots} \\
  \>  \> else \\
  \>  \>  \> return obj \\
  \> \} \\
\}
\end{tabbing}
\end{minipage}
} \\ \\
\Line 
\caption{Pseudocode for {\tt ArrayOutputStream} and
{\tt ArrayInputStream}\label{fig:arrayStreams}}
\end{figure}

Figures \ref{fig:pingpong3} and \ref{fig:pingpong4} on Page
\pageref{fig:pingpong3} and \pageref{fig:pingpong4} show the effect
this change of protocol has 
on the original timings.  As expected, eliminating the overheads of
element serialization dramatically speeds communication of float arrays
(for example) treated as objects, bringing bandwidth close to the raw
performance available with {\tt MPJ.FLOAT}.

\afterpage{\clearpage}
\begin{figure}[H]
\centerline{\psfig{figure=Figs/nonsh_byte_opt1.eps,width=3.5in}}
\centerline{\psfig{figure=Figs/nonsh_float_opt1.eps,width=3.5in}}
\caption{Ping-pong timings with primitive array data sent separately (solid
points) in distributed  memory mode, compared with the unoptimized results from Figure \ref{fig:pingpong1}
(open points).  Recall that the goal is to bring times for ``object-oriented''
sends of arrays down to the ``native'' send times,
most closely approximated by the triangular points.}
\label{fig:pingpong3}
\end{figure}


\afterpage{\clearpage}
\begin{figure}[H]
\centerline{\psfig{figure=Figs/sh_byte_opt1.eps,width=3.5in}}
\centerline{\psfig{figure=Figs/sh_float_opt1.eps,width=3.5in}}
\caption{Ping-pong timings with primitive array data sent separately (solid
points) in shared memory mode, compared with the unoptimized results from Figure \ref{fig:pingpong2}
(open points).  Recall that the goal is to bring times for ``object-oriented''
sends of arrays down to the ``native'' send times,
most closely approximated by the triangular points.}
\label{fig:pingpong4}
\end{figure}


Each one-dimensional array in the stream needs some separate processing
here (associated with calls to {\tt replace\-Object}, {\tt resolveObject},
and setting up the native {\tt MPI\_TYPE\_STRUCT}).  Our fairly
simple-minded prototype happened to increase the constant overhead
of communicating each subarray (parametrized by $t^{\mbox{\scriptsize
vec}}_{\mbox{\scriptsize ser}}$ and $t^{\mbox{\scriptsize
vec}}_{\mbox{\scriptsize unser}}$ in the previous section).  
As mentioned at the end of section \ref{benchmarks},
this overhead typically dominates the time for communicating
two-dimensional byte arrays 
(where the element serialization cost is less extreme), so performance
there actually ends up being worse.  A more highly tuned implementation
could probably reduce this problem.  Alternatively we can go a
step further with our protocol, and have the serialization stream
object directly replace {\em two-dimensional} arrays of primitive
elements\footnote{Defined to be arrays of objects, each element being
an array of primitive type of the same type and length.}.  The benefits
of this approach are shown in Figures \ref{fig:pingpong5} and
\ref{fig:pingpong6} on Page \pageref{fig:pingpong5} and
\pageref{fig:pingpong6} .

\afterpage{\clearpage}
\begin{figure}[H]
\centerline{\psfig{figure=Figs/nonsh_byte_opt2.eps,width=3.5in}}
\centerline{\psfig{figure=Figs/nonsh_float_opt2.eps,width=3.5in}}
\caption{Timings allowing {\em two-dimensional} array proxies in the
object stream (solid points) in distributed  memory mode,
compared with the unoptimized results from Figure \ref{fig:pingpong1}
(open points).
Sends of two-dimensional Java arrays (solid circles) are now much closer to
the native bandwidth (of which the triangular points are representative).
\label{fig:pingpong5}}
\end{figure}


This process could continue almost indefinitely---adding special cases for
arrays and other structures considered critical to Grande applications.
Currently we do not envisage pushing this approach any further than
two-dimensional array proxies.  Of course three-dimensional
arrays and higher will automaticall benefit from the optimization of
their lower-dimensional component arrays.  Recognizing a rectangular
two-dimensional arrays already adds some unwanted complexity to the
serialization process.
It can also introduce some unexpected
behaviour.  Our version subtly alters the semantics of serialization,
because it does not detect aliasing of rows (either with other rows
of the same two-dimensional array, or with one-dimensional primitive
arrays elsewhere in the stream).  Hence the reconstructed object graph
at the receiving end will not reproduce such aliasing.  Whether this is
a serious problem is unclear.


\afterpage{\clearpage}
\begin{figure}[H]
\centerline{\psfig{figure=Figs/sh_byte_opt2.eps,width=3.5in}}
\centerline{\psfig{figure=Figs/sh_float_opt2.eps,width=3.5in}}
\caption{Timings allowing {\em two-dimensional} array proxies in the
object stream (solid points) in shared memory mode,
compared with the unoptimized results from Figure \ref{fig:pingpong2}
(open points).
Sends of two-dimensional Java arrays (solid circles) are now much closer to
the native bandwidth (of which the triangular points are representative).
\label{fig:pingpong6}}
\end{figure}


%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
\newpage

\section{Discussion\label{sec:DISCUSSION}}

In Java, the object serialization model for data marshalling has various
advantages over the MPI derived type mechanism.  It provides much
(though not all) of the flexibility of derived types, and is presumably
simpler to use.  Object serialization provides a natural way to deal
with Java multidimensional arrays.  Such arrays are likely to be common
in scientific programming.

Our initial attempt to add automatic object serialization to our MPI-like
API for Java was impaired by poor performance of the serialization code
in the current Java Development Kit.  Buffers were serialized using
standard technology from the JDK.  The benchmark results from section
\ref{benchmarks} showed that this implementation introduces very large
overheads relative to underlying communication speeds on fast networks
and symmetric multiprocessors.  Similar problems were reported in the
context of RMI implementations in \cite{JGFreport}.  In the context
of fast message-passing environments (not surprisingly) the issue is
even more critical.  Overall communication performance can easily be
downgraded by an order of magnitude or more.

In our benchmarks and tests the organization of primitive elements---their
byte-order, in particular---was the same in sender and receiver.  This is
commonly the case in MPI applications, which are often run on homogenous
clusters of computers.  Hence it should be possible to send the bytes
with no format conversion at all.  More generally an MPI-like package
can be assumed to know in advance if sender and receiver have different
layouts, and need only convert to an external representation in the case
that they do.  Presuming we are building on an underlying native MPI in
the first place, then, a reasonable assumption is that the conversions
necessary for, say, communication of float arrays between little-endian
and big-endian machines in a heterogenous cluster are dealt with inside
the native MPI.  This may degrade the effective native bandwidth to a
greater or lesser extent, but should not impact the Java wrapper code.
In any case, to exploit these features in the native library, we
need a way to marshal Java arrays that avoids performing conversions
inefficiently in the Java layer.

The standard Java serialization framework allows the programmer to
provide optimized serialization and unserialization methods for particular
classes, but in scientific programming we are often more concerned with
the speed of operations on arrays, and especially arrays of primitive
types.  The standard Java framework for serialization does not provide
a direct way to handle arrays, but in section \ref{sec:optimize} we
customized the object streams themselves by suitably defining the {\tt
replaceObject}, {\tt resolveObject} methods.  Primitive array data
was removed from the serialization stream and sent separately using
{\em native} derived datatype mechanisms of the underlying MPI, without
explicit conversion or explicit copying.  This dramatically reduced the
overheads of treating Java arrays uniformly as objects at the API level.
Order of magnitude degradations in bandwidth were typically replaced by
fractional overheads.

A somewhat different approach was taken by the authors of \cite{NESTER99}.
Their remote method invocation software, KaRMI,
incorporates an extensive reimplemention of the JDK serialization code, to
better support their optimized RMI.  Their ideas for
optimizing serialization can certainly benefit message-based APIs as well, 
and KaRMI does also reduce copying
compared with standard RMI.  But we believe they do not immediately
support the ``zero-copy'' strategy we strive for here, whereby large
arrays are removed from the serialization stream and dealt with
separately by platform-specific software.

Given that the efficiency of object serialization can be improved
dramatically---although probably it will always introduce a non-zero
overhead---a reasonable question is whether an MPI-like API for Java needs
to retain anything like the old derived datatype mechanism of MPI at all?

The MPI mechanism still allows non-contiguous sections of a buffer
array to be sent directly.  Although implementations of MPI derived
types, even in the C domain, have often had disappointing performance
in the past, we note that VIA provides some low-level support for
communicating non-contiguous buffers, and recently there has been interest
in producing Java bindings of VIA \cite{JAVIA,JAGUAR}.  So perhaps in the
future it will become possible to support derived types quite efficiently
in Java.  We have emphasized the use of object serialization as a way of
dealing with communication of Java multidimensional arrays.  Assuming the
Java model of multidimensional arrays (as arrays of arrays), we suspect
serialization is the most natural way of communicating them.  On the
other hand there is an active discussion (especially in Numerics Working
Group of the Java Grande Forum) about how Fortran-like multidimensional
rectangular arrays could best be supported into Java.  A reasonable
guess is that multidimensional array sections would be represented as
strided sections of some standard one-dimensional Java array.  In this
case the best choice for communicating array sections may come back to
using MPI-like derived datatypes similar to {\tt MPI\_TYPE\_VECTOR}.

In any case---whether or not a version of MPI derived data types survive
in Java---the need to support object serialization in a message-passing
API seems relatively clear.

