%\documentclass[11pt]{report} 
\documentstyle[10pt,iohdr,psfig,twoside]{article}

%\usepackage{psframe}
%\usepackage{amsmath}
%\usepackage{theorem}
%\usepackage{sections}
%\usepackage{iohdr,encIOA2}
%\usepackage{psfig}

\setlength{\oddsidemargin}{0.775in}
\setlength{\evensidemargin}{0.775in}
\setlength{\topmargin}{-0.0in}
\setlength{\textwidth}{5.00in}
\setlength{\parskip}{\smallskipamount}
\setlength{\columnsep}{0.33in}

\brokenpenalty=10000
\clubpenalty=10000
\widowpenalty=10000

\renewcommand{\topfraction}{0.7}
\renewcommand{\bottomfraction}{0.7}

\bibliographystyle{abbrv}

\newcommand{\comment}[1]{\textsf{[[[Comment: #1]]]}}
\newcommand{\remove}[1]{}
\newcommand{\extra}[1]{\textsf{#1}}

\newcommand{\set}[1]{\{ {#1} \}}
\newcommand{\union}{\cup}
\newcommand{\intersect}{\cap}
\newcommand{\Union}{\bigcup}
\newcommand{\Intersect}{\bigcap}

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
%%%% Theorem and Proof Commands %%%%
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%

%\theorembodyfont{\upshape}

\newtheorem{theorem}{Theorem}[section]

\newtheorem{lemma}[theorem]{Lemma}

\newtheorem{corollary}[theorem]{Corollary}

\newtheorem{proposition}[theorem]{Proposition}  

\newtheorem{remark}[theorem]{Remark}

\newtheorem{claim}[theorem]{Claim}

\newtheorem{invariant}[theorem]{Invariant}

\newtheorem{Definition}{Definition}[section]

%AAS
%\newcommand{\bfseries}{\bf}
%\newcommand{\emph}[1]{{\em #1}}
%\newcommand{\textbf}[1]{{\bf #1}}
%\newcommand{\mathcal}[1]{{\cal #1}}
%\newcommand{\textsl}[1]{{\sl #1}}

\newcommand{\ms}[1]{%
        \relax\ifmmode
                \mathord{\mathcode`\-="702D\it #1\mathcode`\-="2200}%
        \else
%               $\mathord{\mathcode`\-="702D\it #1\mathcode`\-="2200}$%
%                {\itshape #1}%
                {\it #1}%
        \fi
}


\newcommand{\F}{\vspace*{\smallskipamount}}
\newcommand{\FF}{\vspace*{\medskipamount}}
\newcommand{\FFF}{\vspace*{\bigskipamount}}
\newcommand{\B}{\vspace*{-\smallskipamount}}
\newcommand{\BB}{\vspace*{-\medskipamount}}
\newcommand{\BBB}{\vspace*{-\bigskipamount}}
\newcommand{\G}{\hspace*{-0.3em}}
\newcommand{\GG}{\hspace*{-0.5em}}
\newcommand{\Ht}{\hspace*{0.5em}}
\newcommand{\T}{\hspace*{1em}}
\newcommand{\TT}{\hspace*{2em}}
\newcommand{\TTT}{\hspace*{3em}}
\newcommand{\TTTT}{\hspace*{4em}}
\newcommand{\TTTTT}{\hspace*{5em}}

\newcommand{\Section}[1]{\BBB\section{#1}\BB}
\newcommand{\Subsection}[1]{\BBB\subsection{#1}\BB}%\vspace*{1mm}}
\newcommand{\Subsubsection}[1]{\BBB\subsubsection{#1}\BB}
\newcommand{\Subsubsubsection}[1]{\BBB\subsubsection*{#1}\BB}

\newcommand{\Item}{\B\item}

\newcommand{\ensuremath}{}
\newcommand{\text}[1]{{\rm #1}}
\newcommand{\implies}{\Rightarrow}

\newcommand{\pf}{\par\noindent{\bfseries Proof:}~}


\newcommand{\qed}{\hfill{\rule{3mm}{3mm}}\medskip}

\newenvironment{proof}{\pf}{\qed}

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%

\newcommand{\action}[1]{\ms{#1}}
\newcommand{\statevar}[1]{\ms{#1}}

\newcommand{\arrow}[1]{\raisebox{0ex}[2ex]{$-\!\!\!@>{\!\!\!\!#1\,}>>\,$}}
\newcommand{\sarrow}[2]{\raisebox{0ex}[2ex]{$-
\!\!\!@>{\!\!\!\!#1\,}>>\!_{#2}\:$}}

\newcommand{\request}{\action{request}}
\newcommand{\response}{\action{response}}

\newcommand{\send}{\action{send}}
\newcommand{\receive}{\action{receive}}
\newcommand{\doit}{\action{do\_it}}
\newcommand{\solidify}{\action{solidify}}

\newcommand{\enter}{\action{enter}}
\newcommand{\calc}{\action{calculate}}
\newcommand{\stabilize}{\action{stabilize}}
\newcommand{\addcons}{\action{add\_constraints}}



\newcommand{\wait}{\statevar{wait}}
\newcommand{\rept}{\statevar{rept}}


\newcommand{\potrept}{\statevar{potential\_rept}}

\newcommand{\channel}{\statevar{channel}}
\newcommand{\pending}{\statevar{pending}}
\newcommand{\rcvd}{\statevar{rcvd}}
\newcommand{\gossip}{\statevar{gossip}}
\newcommand{\done}{\statevar{done}}
\newcommand{\solid}{\statevar{solid}}

\newcommand{\ops}{\statevar{ops}}
\newcommand{\po}{\statevar{po}}
\newcommand{\newpo}{\statevar{new-po}}
\newcommand{\stabilized}{\statevar{stabilized}}

\newcommand{\id}{\ms{id}}
\newcommand{\op}{\ms{op}}
\newcommand{\prev}{\ms{prev}}
\newcommand{\state}{\ms{state}}
\newcommand{\val}{\ms{val}}

\newcommand{\datatrans}{\ensuremath{f}}
\newcommand{\datatransextend}{\ensuremath{f^+}}
\newcommand{\datastate}{\statevar{data\_state}}
\newcommand{\responses}{\statevar{responses}}
\newcommand{\mkseq}[1]{\ensuremath{\langle #1 \rangle}}

\newcommand{\mkmsg}[1]{\ensuremath{\langle #1 \rangle}}
\newcommand{\reqmsg}[1]{\mkmsg{\text{``request"}, #1}}		% args: id
\newcommand{\respmsg}[1]{\mkmsg{\text{``response"}, #1}}	% args: id, v
\newcommand{\gossipmsg}[1]{\mkmsg{\text{``gossip"}, #1}}	% args: R, D, L, S

\newcommand{\minlbl}{\statevar{minlabel}}

\newcommand{\frontend}{\ms{frontend}}
\newcommand{\client}{\ms{client}}

\newcommand{\minlabel}{\ms{minlabel}}
\newcommand{\strict}{\ms{strict}}
% \newcommand{\val}{\ms{val}}
\newcommand{\outcome}{\ms{outcome}}
\newcommand{\valset}{\ms{valset}}


\newcommand{\setinsert}[2]{\ensuremath{#1 \gets #1 \cup \{ #2 \}}}
\newcommand{\setdelete}[2]{\ensuremath{#1 \gets #1 - \{ #2 \}}}

\newcommand{\CSC}{\mathit{CSC}}
\newcommand{\LC}{\mathit{local\_cons}}
\newcommand{\MC}{\mathit{msg\_cons}}
\newcommand{\SC}{\mathit{system\_cons}}
\newcommand{\horline}{\noindent\rule{\textwidth}{.2mm}}

\newcommand{\opdescs}{\ensuremath{\mathcal{O}}}
\newcommand{\labels}{\ensuremath{\mathcal{L}}}

\newcommand{\requested}{\statevar{requested}}
\newcommand{\responded}{\statevar{responded}}

\newcommand{\stabstate}{\ms{stable-state}}
\newcommand{\stabval}{\ms{stable-value}}
\newcommand{\maxstab}{\ms{max-stable}}
\newcommand{\currentstate}{\ms{current-state}}
\newcommand{\readonly}{\ms{read-only}}
\newcommand{\writeonly}{\ms{write-only}}

\newcommand{\states}{\ensuremath{\Sigma}}
\newcommand{\SpecAut}{\ensuremath{\ms{Spec}}}
\newcommand{\AlgAut}{\ensuremath{\ms{AbsAlg}}}
\newcommand{\UserAut}{\ensuremath{\ms{Users}}}
\newcommand{\SafeUserAut}{\ensuremath{\ms{SafeUsers}}}
\newcommand{\CommuteAut}{\ensuremath{\ms{Commute}}}

\begin{document}

\pagestyle{empty}
\setlength{\textheight}{6.0in}

%\title{\BBB\sf Implementing and Evaluating\\an Eventually-Serializable
%Data Service%
\title{Implementing an Eventually-Serializable
Data Service as a Distributed System Building Block%
\thanks{%
Appeared in M. Mavronicolas, M. Merritt, and N. Shavit, editors,
Networks in Distributed Computing, volume 45 of DIMACS Series in
Discrete Mathematics and Theoretical Computer Science, pages
43--72. American Mathematical Society, 1999.
}}

\author{%
Oleg M. Cheiner%
\thanks{%
This author was supported by an
NSF Graduate Research Fellowship.
}
\and
Alex A. Shvartsman%
\thanks{%
This work was supported by the contracts
ARPA F19628-95-C-0118,
NSF 922124-CCR,
ONR-AFOSR F49620-94-1-01997,
and by the GTE Laboratories.
}}

\date{1999}
%\date{}

\maketitle

\BBB\BB


\begin{abstract}
This work presents an implementation of a distributed system
building block that is formally specified as
the Eventually-Serializable Data Service (ESDS)~\cite{esds}
proposed by Fekete \emph{et al}.
ESDS deals with replicated objects that allow the users of the service to relax 
consistency requirements in return for improved responsiveness, while providing 
guarantees of eventual consistency of the replicated data. 
The ESDS paper ~\cite{esds} includes a formal service specification
and an abstract algorithm implementing the service.
The algorithm is given in terms of I/O automata of Lynch and Tuttle~\cite{Lynch}.
An important consideration in formulating ESDS was
that it could be employed in building real systems.

The work described here makes the following contributions.
We develop an optimized implementation of ESDS and explore its behavior.
We combine the implementation with different data types and
clients, thus demonstrating the utility of the service as a 
{\em building block\/} suitable for serving as a distributed operating
system component.  The implementation has been experimentally evaluated on a network
of workstations. The results confirm that the designed trade-off between 
{\em consistency\/} and {\em performance\/} is present in
the implemented service.
To make the implementation process less error prone, we develop and use a framework for
mapping algorithms formally specified using I/O automata to distributed programs.
The framework includes a set of conversion rules and a core set of
object common to all target implementations.

\end{abstract}

%\renewcommand{\baselinestretch}{1.5}

\thispagestyle{empty}

\setlength{\textheight}{7.875in}

\section{Introduction}

Extant network technology enables the creation of very large distributed
platforms. Developing sophisticated distributed
applications for such environments still presents a challenge, despite the
availability of distributed middleware packages, such as DCE~\cite{OSF/DCE-UUID}.
In many cases, middleware provides functions at
a lower-level of abstraction than those required by complex 
applications. Even when higher-level functions are available,
they often have informal specifications, thus providing only a modest benefit.
In particular, it is often difficult to reason about the properties of
distributed applications that are built using middleware, and about
the compositional semantics of the resulting applications.

Specification of building blocks for sophisticated distributed systems and 
development of supporting algorithms is an area of active research.
However, even when specifications and algorithms are formally stated, 
deriving a distributed implementation from a specification is an error-prone
process. It is often difficult to forecast the behavior patterns of distributed
implementations in realistic deployment scenarios, especially when the performance
bottleneck cannot be readily assessed and when the scalability of
the underlying platforms becomes a limiting factor.

In this paper we present an implementation of a formally specified
building block for a flexible eventually-serializable data service,
or ESDS for short, proposed by Fekete \emph{et al}. in~\cite{esds}.
The implementation is derived with the help
of a framework that is designed to make the derivation process less
error-prone, and to reduce the need to reimplement common functions
when working with I/O automata specifications~\cite{I/O-Automata}.
We also present experimental results of exploring the scalability
of the service and its consistency/performance tradeoff.


\pagestyle{myheadings}
\markboth{\centerline{\small IMPLEMENTING ESDS AS A BUILDING BLOCK}}{\centerline{\small OLEG CHEINER, ALEX SHVARTSMAN}}

\subsection{Background}

Replication is used in distributed systems to improve availability and
to increase throughput. The disadvantage of replication is the additional
effort required to maintain consistency among replicas when serializing
client operations.
Several notions of consistency have been defined.
The strongest notion is \emph{atomicity}, 
requiring that replicas emulate a single centralized object. 
Methods to achieve atomicity include 
write-all/read-one \cite{BernsteinHG87}, 
primary copy \cite{AlsbergD76,Stonebraker79,OkiL88}, 
majority consensus \cite{Thomas79}, 
and quorum consensus \cite{Gifford79}.
Achieving atomicity often incurs a high cost, while
some applications, such as directory services~\cite{ISO500,DNS},
are willing to tolerate some transient inconsistencies. 
This gives rise to different notions of consistency.
\emph{Sequential consistency}~\cite{Lamport79},
guaranteed by systems such as Orca~\cite{BalKT92}, 
allows operations to be reordered as long as they remain 
consistent with the view of isolated clients. 

Improving performance by providing weaker consistency
may lead to more complicated semantics. 
While in practice, replicated systems are often incompletely or 
ambiguously specified, it remains important to
provide formal consistency guarantees.
Ladin, Liskov, Shrira, and Ghemawat~\cite{Ladin92} 
define a replicated data service.
They specify general conditions for such a service,
and present an algorithm based on \emph{lazy replication}, 
in which operations received by each replica are 
\emph{gossiped} in the background. 
Building on the work of Ladin \emph{et al}.,
Fekete \emph{et al}.~\cite{esds} developed a flexible eventually-serializable
data service.

The definition of ESDS includes a specification of the
data service and an abstract algorithm that implements it.
ESDS relaxes consistency guarantees provided by serializable
data services to improve system efficiency and availability. It
also provides provable guarantees of long-term consistency of the data. An
important consideration in the design of ESDS has been to make
it suitable for employment in building real systems. In this work we
implement and explore an optimized version of ESDS. 

\subsection{Our Contributions}%Experimental ESDS Implementation

We implement an optimized version
of ESDS and explore its behavior in a distributed setting.
We combine the implementation of ESDS with different data types and
clients, thus demonstrating the suitability of the service as a general
building block in a distributed computing environment.
The implementation has been experimentally evaluated on a network
of workstations. In this setting, the implementation scaled well with the number of
processors, which reflects a designed trade-off between consistency and
performance.

The ESDS algorithm is specified as a composition of I/O
Automata~\cite{I/O-Automata}. The I/O automata notation is
a declarative description language used to specify
state machines.  To implement ESDS, we need 
to convert the abstract algorithm to a design specification for 
a distributed program. To our knowledge, no general method for converting 
I/O automata specifications to distributed programs has been proposed before. 

To assist us in the implementation process, we formulate and utilize
a framework for mapping algorithms
specified using I/O automata to distributed
programs. The goal of the framework is to streamline the mapping of a specification to
a message-passing implementation, thus reducing the number of errors that
might be accidentally introduced. The framework also implements
several common functions, thus eliminating the need to reimplement such functions
when working with I/O automata specifications.
We believe that the techniques in the framework are general and that they 
can be used to implement many other algorithms specified as I/O automata. 

One of our main design goals was to confirm that the ESDS algorithm is suitable
for implementation as a building block from which concrete
applications can be built with minimal effort.  The algorithm is specified
to be independent of the serial data type of the replicated data object to
facilitate its use as a building block. Our implementation of ESDS uses
object-oriented techniques to ensure that this independence is preserved.
We build four distinct applications on top of ESDS to demonstrate the viability
of our design as a generic building block for real distributed data services.

The ESDS service has been developed and tested on a network of Sun workstations 
running SunOS 4.1.4.  Four clients for sample 
ESDS service applications were developed. One client was developed for Win32 and 
tested under Windows 95 on an Intel Pentium machine. Three other clients ran under 
SunOS 4.1.4.
We use MPI (Message Passing Interface) Standard~\cite{MPI-intro} to implement
communication between distributed components of the ESDS implementation.
In selecting MPI, we took into account its suitability for implementing I/O automata,
the simplicity of communication semantics,
the availability of development tools, and portability.


We have instrumented an optimized implementation of ESDS with tools for monitoring 
interesting parts of the state of the data service and collecting information about 
performance characteristics of the system. Characteristics of interest include response 
time to user requests, system throughput, and deviation from strict consistency in system 
responses.

The empirical tests have provided data on the behavior of the implementation with varying 
number of participating replicas and with varying system load.
In addition, the tests have confirmed 
that ESDS offers a tradeoff between consistency and performance, and that it is 
possible to shift the tradeoff balance in either direction according to the user's needs.

%\noindent
The rest of this paper is organized as follows.
In Section~\ref{chapter:Models-definitions-platforms} we
overview the I/O automata model and introduce names for
variants of the ESDS algorithm and its implementations that we use thereafter.
Section~\ref{chapter:IO-to-program-framework} describes the framework for converting 
I/O automata to distributed implementations. Section~\ref{ESDS-algorithm} describes
the ESDS algorithm.
The ESDS implementation is dealt with in~Section~\ref{chapter:ESDS-prototype-design}.
Section~\ref{chapter:Empirical-testing} discusses the empirical results. Our conclusions 
are presented in Section~\ref{chapter:Conclusions}, while Section~\ref{section:Future-work}
contains suggestions for future work.



\section{Models and Definitions} %, and Platforms}

\label{chapter:Models-definitions-platforms}

The \emph{Input/Output automaton}, or I/O automaton for short~\cite{Lynch,I/O-Automata}
is a general model used for formal descriptions of asynchronous and distributed algorithms. 
The model provides a precise way of describing and reasoning 
about asynchronous interacting components. We provide a consise description of the model
and refer the reader to~\cite{Lynch,I/O-Automata} for more details.

An I/O automaton
is a \emph{state machine} in which the \emph{transitions} are associated with 
named \emph{actions}. The actions are classified as either \emph{input},
\emph{output} or \emph{internal}. Roughly speaking, the 
input and output actions are used for interaction with the automaton's environment. The
internal actions work on the automaton's local state.

I/O automata code is given in a \emph{precondition-effect} style.
For each action, the code specifies the \emph{preconditions} under 
which the action is permitted to occur, as a predicate on the automaton state,
and the \emph{effects} that occur as the result.
The input actions are always enabled (i.e. the precondition clause of an
input action is always \emph{true}).
The code in the effects clause gets executed atomically.

The \emph{composition} operation allows an automaton representing a complex system to
be constructed by composing automata representing individual system
components. The composition identifies actions with the same name in different
component automata. When any component automaton performs a step involving 
an action $\pi$, so do all component automata that include the action $\pi$. 
The states and start states of the composition automaton are vectors of
states and start states, respectively, of the component automata.

When we compose a collection of automata, output actions of the components
become output actions of the composition, internal actions of the components
become internal actions of the composition, and actions that are inputs to
some components but outputs of none become input actions of the composition.

We now name and briefly describe the different variations of the ESDS algorithm 
and its implementations; we also introduce additional terminology
that will be used in describing
interactions of automata in a composition.

The component automata in a composition communicate 
by means of input and output actions with the same 
name. We distinguish between two types of actions in an I/O automata composition.

%\begin{Definition}
%\rm
Let an I/O automaton $A$ be a composition of I/O automa\-ta $A_1$, $A_2$, $\ldots$, 
$A_m$. If there is an output action $X \in A$ that occurs as an output action in some 
$A_i$ and as an input action in some other automaton $A_j$ $(i \neq j)$, we call $X$ an 
\emph{input/output combination}, or \emph{I/O combination} for short. We call $A_i$ 
the \emph{output end} with respect to $X$, and we call $A_j$ the \emph{input end} with 
respect to $X$. Any action $Y \in A$ that appears in one and only one automaton $A_k$ is called a 
\emph{regular} action.

%\end{Definition}

\begin{figure}
%\centerline{\psfig{figure=IOComp.eps,bbllx=13bp,bblly=13bp,bburx=360bp,bbury=275bp,clip=}}
\begin{center}

%TexCad Options
%\grade{\off}
%\emlines{\off}
%\beziermacro{\off}
%\reduce{\on}
%\snapping{\on}
%\quality{1.00}
%\graddiff{0.01}
%\snapasp{1}
%\zoom{1.00}
\unitlength 1.00mm
\linethickness{0.4pt}
\begin{picture}(93.00,41.00)
\put(46.00,41.00){\makebox(0,0)[cc]{{\small\bf Algorithm A}}}
\put(15.00,34.00){\makebox(0,0)[cc]{{\small\bf Automaton $A_1$}}}
\put(77.00,34.00){\makebox(0,0)[cc]{{\small\bf Automaton $A_2$}}}
\put(2.00,4.00){\framebox(29.00,26.00)[cc]
{\parbox{1in}{\small State: $StA_1$\\Transitions\\\T Output W\\\T Output Y\\\T Input Z}}}
\put(64.00,4.00){\framebox(29.00,26.00)[cc]
{\parbox{1in}{\small State: $StA_2$\\Transitions\\\T Internal X\\\T Input Y\\\T Output Z}}}
\put(27.00,13.00){\vector(1,0){40.00}}
\put(67.00,9.00){\vector(-1,0){40.00}}
\end{picture}

\end{center}
\BBB\BBB
\caption{I/O automata composition and input/output combinations}
\label{figure:io_composition}
\end{figure}

Figure~\ref{figure:io_composition} gives an example of an automaton $A$ composed of 
two component automata, $A_1$ and $A_2$. In the composition, $W$ and $X$ are 
regular actions, while $Y$ and $Z$ are I/O combinations.

I/O automata are typically used to specify distributed algorithms involving a 
collection of nodes by encapsulating the behavior 
of each node $I$ as a separate automaton $A_i$. The entire algorithm is then represented by 
the composition $A$ of component automata $A_1$, $A_2$, ..., $A_m$. The internal 
actions of $A_i$ represent local processing at the corresponding node. The I/O combinations
represent communication between 
the nodes. The input and output actions of each automaton that do not participate in an 
I/O combination represent the interaction of the corresponding node with its 
external environment. We assume this representation of distributed algorithms as
I/O automata when discussing conversion of such algorithms to distributed programs in
Section~\ref{chapter:IO-to-program-framework}.

In the rest of the paper, we discuss several variants of the ESDS algorithm ans its
implementations:

\begin{description}

\Item[ESDSAlg:] this refers to the unoptimized abstract algorithm for 
ESDS (described in Section~\ref{ESDS-algorithm}).

\Item[SimpleESDSAlg:] this is a simplified version of \emph{ESDSAlg} that replaces 
channel automata with I/O combinations. This version restricts concurrency, but
leads to a simpler implementation.

\Item[ESDSOptAlg:] this is an optimized version of \emph{ESDSAlg}. The 
optimizations pursued in \emph{ESDS\-OptAlg} and a corresponding I/O automata
description of them are presented in Section~4.2.
%\ref{section:Abstract-ESDS-optimizations}.

\Item[ESDSImpl, SimpleESDSImpl, and ESDSOptImpl] are 
distributed programs that implement \emph{ESDSAlg}, 
\emph{SimpleESDSAlg}, and \emph{ESDSOptAlg} respectively.

\end{description}


\section{Converting I/O Automata to Programs}

\label{chapter:IO-to-program-framework}

I/O automata have been effectively used for describing message-passing distributed 
algorithms and in proving their correctness properties. We develop a framework for
converting commonly occurring algorithms that are specified using 
I/O automata compositions into distributed message-passing implementations using
an imperative language (chosen in this work to be C++).

We call the I/O automata composition being converted the \emph{source 
composition}. We call the algorithm represented by the source composition the 
\emph{source algorithm}. The result of the conversion is the 
\emph{target program}.

A key goal for our framework is to facilitate a faithful implementation 
of the source algorithm. Also, one must be able to directly relate each
part of the target program to the corresponding parts of the source algorithm.
Due to these requirements, the techniques discussed here are rather 
conservative, and they will usually lead to an overspecification of the
algorithm specified using I/O automata. Nevertheless, this would not
preclude one from representing a large and interesting 
subset of behaviors of the source algorithm in the target implementation.

The conversion rules presented in this sections are intended to help
the programmer with the implementation task and to eliminate redundant
software development work.  Formalizing the rules is a task for future work.

\subsection{Overall Approach}

In the target program produced from the source composition $A$, each of $A$'s 
component automata $A_i$ is represented by a sequential process $P_i$\footnote{It is also 
possible to combine several automata to run as a single process if there is a reason to do 
so. For example, the algorithm may be split into automata to make the description modular.
In this case, the component automata may not be meant to run in parallel.}.
Each action of the source composition has a corresponding fragment of code in the 
target program that implements the action. The conversion techniques ensure that these
code fragments are atomic with respect to each other.

When dealing with a composition $A$ of I/O automata, we
assume that $A$ consists of component automata $A_1, \ldots, A_m$.
We further assume that an action named $X$ belonging to 
automaton $A_i$ has preconditions clause \emph{PXA}$_i$ and effects 
clause \emph{EXA}$_i$. A component automaton $A_i$ from composition $A$
will correspond one-to-one to an implementation process $P_i$.
%
The local state of a component automaton $A_i$ is represented by the local state variables 
of the corresponding process $P_i$. We do not make provisions for representing the
global state of $A$\footnote{By global state of $A$, we mean state that is accessible to more
than one component automaton.}. If $A$ utilizes global state, it may not be
easily implementable as a message-passing distributed program. Global state must
be removed from such algorithms if one wishes to apply these techniques to them.


\subsection{Converting Preconditions Clauses to Procedures}

\label{section:Preconditions-to-procedures}

The purpose of the preconditions clause \emph{PXA}$_i$ in an action $X$ is to 
determine whether the state transition \emph{EXA}$_i$ is enabled in the current 
automaton state. The preconditions clause is converted to a predicate procedure 
\emph{Enabled} that checks the current state of the automaton and returns \emph{true} if 
the action is enabled and \emph{false} otherwise.

\subsection{Converting Effects Clauses to Procedures}

\label{section:Effects-to-procedures}

The effects clause \emph{EXA}$_i$ describes the state transition represented by 
action $X$. The effects clause is converted to a procedure named \emph{Transition}. 
\emph{Transition} requires $X$ to be enabled and the desired state transition to be 
chosen among all enabled transitions represented by $X$.
\emph{Transition}'s effect on the 
state of $P_i$ must correspond to the effects of \emph{EXA}$_i$ on the state of $A_i$.

\subsection{Converting Regular Actions to Procedures}

Conversion of a regular action or an input action to code is straightforward. All that needs 
to be developed are the \emph{Enabled} and \emph{Transition} procedures that 
implement the preconditions and effects clauses of the action, respectively (for an input 
action, \emph{Enabled} is the constant \emph{true}). The implementation of \emph{Transition}
needs to be atomic to preserve I/O automata semantics.

\begin{figure}
\BBB
%\centerline{\psfig{figure=IOtoCode.eps,bbllx=13bp,bblly=13bp,bburx=470bp,bbury=310bp,clip=,width=435bp}}
\begin{center}
\parbox{0.05\textwidth}{~}
\parbox[b]{0.45\textwidth}{\small{\bf Automaton} $A_i$\\
\framebox[2.25in][l]{\parbox{1.5in}{{\bf Output} $X$\\
\T Precondition: {\em PXA}$_i$\\
\T Effects: {\em EXA}$_i$}}
}
\parbox[b]{0.4\textwidth}{\small{\bf Automaton} $A_j$\\
\framebox[2.25in][l]{\parbox{1.5in}{{\bf Input} $X$\\
\T Precondition: None\\
\T Effects: {\em EXA}$_j$}}
}
%TexCad Options
%\grade{\off}
%\emlines{\off}
%\beziermacro{\off}
%\reduce{\on}
%\snapping{\on}
%\quality{1.00}
%\graddiff{0.01}
%\snapasp{1}
%\zoom{1.00}
\unitlength 1.00mm
\linethickness{0.4pt}
\begin{picture}(85.00,9.00)
\put(17.00,9.00){\line(0,-1){5.00}}
\put(17.00,4.00){\line(-1,0){2.00}}
\put(15.00,4.00){\line(5,-3){5.00}}
\put(23.00,9.00){\line(0,-1){5.00}}
\put(23.00,4.00){\line(1,0){2.00}}
\put(25.00,4.00){\line(-5,-3){5.00}}
\put(17.00,9.00){\line(1,0){6.00}}
\put(77.00,9.00){\line(0,-1){5.00}}
\put(77.00,4.00){\line(-1,0){2.00}}
\put(75.00,4.00){\line(5,-3){5.00}}
\put(83.00,9.00){\line(0,-1){5.00}}
\put(83.00,4.00){\line(1,0){2.00}}
\put(85.00,4.00){\line(-5,-3){5.00}}
\put(77.00,9.00){\line(1,0){6.00}}
\end{picture}\BB\\
\parbox{0.05\textwidth}{~}
\parbox[b]{0.45\textwidth}{\small{\bf Process} $P_i$\\
\framebox[2.25in][l]{\parbox{2in}{{\bf if} Enabled({\em PXA}$_i$)\\
\T Send($P_j$, ``Initiate $X$");\\
\T Transition({\em EXA}$_i$);\\
\T Receive($P_j$, ``Done $X$");\\
{\bf endif}
}}
}
\parbox[b]{0.4\textwidth}{\small{\bf Process} $P_j$\\
\framebox[2.25in][l]{\parbox{2in}{{\bf if} NBReceive($P_i$, ``Initiate $X$")\\
\T Transition({\em EXA}$_j$);\\
\T Send($P_i$, ``Done $X$");\\
{\bf endif}
}}
}
\end{center}\BBB
\caption{Converting an Input/Output Combination to Code}\BB
\label{figure:IOtoCode}
\end{figure}

\subsection{Converting Input/Output Combination Actions to Code}
\label{section:IOCombinations-to-code}

Implementation of an I/O combination is trickier because it must rely on asynchronous 
messages to implement the combination atomically.
We give a 
technique for implementing an I/O combination in the special case when only two 
automata participate in the combination.  This is sufficient for most existing
algorithms that have been specified as I/O automata.

The rule for converting an I/O combination to code is illustrated in 
Figure~\ref{figure:IOtoCode}. Here, automata $A_i$ and $A_j$ correspond to processes 
(or nodes) $P_i$ and $P_j$. The Send() and Receive() calls in the pseudocode for 
processes $P_i$ and $P_j$ stand for sending and receiving asynchronous messages
using MPI. \remove{They 
are implemented by MPI's \emph{MPI-Send} and \emph{MPI-Recv} primitives, respectively.} The 
NBReceive() in the process $P_j$ is a non-blocking receive of a message\remove{, implemented 
by calling {\em MPI-Iprobe} and then calling \emph{MPI-Recv} if there is a pending message 
of type \emph{``Initiate X''} from $P_i$}. If a message of the type \emph{``Initiate~X''} 
has not arrived at $P_j$, then the \textbf{if} block is skipped.

An I/O combination is always initiated at the process that represents the output end of the 
combination ($P_i$ in Figure~\ref{figure:IOtoCode}). When the call to 
\emph{Enabled(PXA$_i$)} returns \emph{true}, $P_i$ sends a message to $P_j$ 
initiating the combination. Any argument that $X$ has is passed to $P_j$ in the 
same message. Next $P_i$ performs the local state transition associated with $X$ by 
invoking the \emph{Transition(EXA$_i$)} procedure. Then, $P_i$ waits for an 
acknowledgment message \emph{``Done~X''} from $P_j$. This step ``synchronizes'' the 
execution of $X$ at the two participating processes.

At the input end of the I/O combination, $P_j$ watches for requests from $P_i$ to initiate 
$X$. While the NBReceive call returns \emph{false}, $P_j$ can continue executing other 
actions. When $P_j$ receives an \emph{``Initiate~X''} message, it executes its local state 
transition for $X$ and then sends the acknowledgment message to $P_i$.

The \emph{Transition} procedures representing the effects regular actions are atomic with
respect to other actions, since they may affect local variables only and may not
communicate with other processes. Informally, it is easy to see that the code implementing
an input/output combination, as presented in Figure~\ref{figure:IOtoCode}, is also atomic.
Atomicity would be violated only if it were possible for some process to observe that process
$P_i$ had executed \emph{Transition(EXA$_i$)} but process $P_j$ had not executed
\emph{Transition(EXA$_j$)}, or vice versa. The synchronizing message \emph{``Done~X''}
rules out both possibilities.

In our ESDS implementation it is sufficient to consider I/O
combinations that involve actions of two automata.
In the future we plan to extend the
the mechanism of negotiation presented here to the more
general case when multiple automata participate in the I/O combination at
the input end. The approach may be as follows.
The initiating process broadcasts \emph{``Initiate~X''}
messages to participating automata, gathers \emph{``Done~X''} messages from
all of them, and then broadcasts a \emph{``Proceed~X''} message to indicate
that all of the effects clauses in the combination have been executed.  Upon
receiveing the \emph{``Proceed~X''} message, the participants continue with
other actions.

\subsubsection{Deadlock Avoidance}

\label{section:Deadlock-avoidance}

\begin{figure}

\centerline{\psfig{figure=Reserv.eps,bbllx=52bp,bblly=13bp,bburx=395bp,bbury=350bp,height=160bp,clip=}}

\caption{Reservation Status Finite State Automaton for Process $P_i$}
\label{figure:Reservation-status}
\end{figure}

As presented, the translation from I/O automata to programs is safe, but suffers from deadlock.
If two automata running concurrently enter the output part of two different
input/output combinations and 
simultaneously attempt to initiate a combination with each other, it is possible for them to 
block at the Receive($A_i$, \emph{``Done~X''}) and Receive($A_j$, \emph{``Done~X''}) lines,
repectively, and wait for each other indefinitely.

We resolve this deadlock problem by setting up a \emph{reservation system} for 
performing input/output combinations. Each process $P_i$ maintains its 
\emph{reservation status} in a state variable. The states of the reservation status at
$P_i$ are as follows: 

\begin{description}

\Item[free] This is the initial state of reservation status. In this state $P_i$ is 
free to initiate or accept reservation requests.

\Item[reserved$_j$, $j~\neq~i$] In the \emph{reserved}$_j$ state $P_i$ is 
waiting for process $P_j$ to initiate an I/O combination.

\item[holding-reservation$_j$, $j~\neq~i$] In the \emph{holding-reservation}$_j$
state $P_i$ may initiate an I/O combination with $P_j$.

\item[negotiating$_j$, $j~\neq~i$] In the \emph{negotiating}$_j$ state $P_i$ is waiting
for $P_j$ to respond to a reservation request.

\end{description}

The complete finite state automaton for the reservation status of one process
is depicted in Figure~\ref{figure:Reservation-status}.

The reservation system imposes restrictions on processes' ability to engage in input/output combinations. Process $P_i$ can initiate an input/output combination with process $P_j$ only after $P_i$ has obtained a reservation with $P_j$.  In order to obtain the reservation, $P_i$ must set its reservation status to \emph{holding-reservation}$_j$, and $P_j$ must set its reservation status to \emph{reserved}$_j$.  Figure~\ref{figure:Reservation-status} specifies the rules for requesting and granting reservations.

When a process $P_i$ wants to initiate an input/output combination with process $P_j$, its first step is to send a message to the receiving process $P_j$ requesting a reservation. $P_i$ is allowed to do this only when its reservation status is \emph{free}. After the request for a reservation is sent to $P_j$, $P_i$ enters the \emph{negotiating}$_j$ state and waits for a response to the request. If the reservation were granted by $P_j$, $P_i$ enters the \emph{holding-reservation}$_j$ state, while $P_j$ enters the \emph{reserved}$_j$ state. In this case, $P_i$ is free to initiate an input/output combination as described in Section~\ref{section:IOCombinations-to-code}. If the reservation request were rejected, $P_i$ ``bounces back'' to the \emph{free} state.

Whenever the reservation status of a process is \emph{free} and there is an incoming reservation request, the process may grant the request. Although we do not require that the process grant the reservation every time, it is necessary to grant them for the system to make progress. Furthermore, reservations should be granted fairly to prevent starvation.

We informally argue that, under the reservation system, deadlock cannot occur. The key is Invariant~\ref{inv:Pi-IOcombination-j}, which specifies the relationship between the reservation status values of two processes.

\begin{invariant}

\label{inv:Pi-IOcombination-j}

Let $\mathcal{P}$ be the set of process identifiers in the target program. Then for all $i,j$ in $\mathcal{P}$ s.t. $i~\neq~j$, $P_i =$ {\it holding-reservation}$_j$ implies $P_j =$ {\it reserved}$_i$.

\end{invariant}

We provide a sketch of the proof of Invariant~\ref{inv:Pi-IOcombination-j}. Process $P_i$ must receive a reservation acceptance message from $P_j$ before it can enter the \emph{holding-reservation}$_j$ state. $P_j$ must enter the \emph{reserved}$_i$ state to send a reservation acceptance message to $P_i$. $P_j$ remains in \emph{reserved}$_i$ until it executes an I/O combination with $P_i$. When executing this combination, $P_i$ must leave the \emph{holding-reservation}$_j$ state. It follows that while $P_i$ is in the \emph{holding-reservation}$_j$ state, $P_j$ must be in \emph{reserved}$_i$ state.

By Invariant~\ref{inv:Pi-IOcombination-j}, process $P_j$ cannot initiate an I/O combination when process $P_i$ is in the \emph{holding-reservation}$_j$ state. So when $P_i$ initiates an I/O combination with $P_j$, $P_j$ is not blocked but is able to participate. Therefore, the I/O combination executes successfully.

\subsubsection{Optimizing Abstract I/O Channels Away} 
\label{section:Optimizing-channels-away}

\begin{figure}[t]
%\horline
\iocodesize

\textbf{State}
\begin{description}

\item $\channel_{i,j}$, a multiset of messages

\end{description}

\textbf{Actions}

\iocode{
\ef{\textbf{Input} $\send_{i,j}(m)$}
{%Effect: \\
  $\setinsert{\channel_{i,j}}{m}$
}

}{

\prcef{\textbf{Output} $\receive_{i,j}(m)$}
{%Precondition: \\
  $m \in \channel_{i,j}$
}{%Effect: \\
  $\setdelete{\channel_{i,j}}{m}$
}
}
%\horline
\caption{Automaton for channel from process $i$ to process $j$}
\label{fig:channel}
\end{figure}

The reservation system for avoiding deadlock can be costly because it reduces the potential 
for concurrency. For algorithms specified using I/O automata
that use channels with asynchronous 
message delivery for communication between its distributed components, an 
implementation that can provide more concurrency is possible. The I/O automata specification
must obey the following restrictions. For any pair of component automata $A_i$ and $A_j$
that communicate with each other, the specification must have
explicit asynchronous half-duplex channel
automata $C_{i,j}$ and $C_{j,i}$ between them. $C_{i,j}$ is used for sending messages from
$A_i$ to $A_j$, and $C_{j,i}$ for sending messages from $A_j$ to $A_i$.

A channel from automaton $A_i$ to automaton $A_j$ 
is modeled  with $\send_{i,j}$ and $\receive_{i,j}$ actions 
and a single state variable $\channel_{i,j}$ representing messages
in transit. The channel has a simple specification (cf.~\cite{Lynch}),
which is given in Figure~\ref{fig:channel}.

The $\send_{i,j}$ and $\receive_{i,j}$ actions form I/O combinations with the
automata that wish to communicate via
the channel.  For instance, $C_{i,j}$ can share a \emph{send}$_{i,j}$ I/O
combination with $A_i$ and a \emph{receive}$_{i,j}$ I/O combination
with $A_j$. In an execution, a \emph{send}$_{i,j}$
event and a subsequent \emph{receive}$_{i,j}$ event model
asynchronous messages from $A_i$ to $A_j$.

The message passing model used by MPI already contains the asynchronous channel discipline.
An implementation of an algorithm that is specified using channels can use asynchronous 
\emph{MPI-Send} at $A_i$ and \emph{MPI-Receive} at $A_j$ instead of implementing explicit channel
automata $C_{i,j}$. This optimization omits a significant portion of the code generated by the basic
conversion framework. Specifically, it removes two I/O combinations (one at each of the 
sending and the receiving ends of the channel $C_{i,j}$) and a separate process for the channel 
automaton. Thus, removal of the I/O combinations provides more concurrency.

\subsubsection{Abstract Algorithm Relaxation Through Introduction of I/O Channels}
\label{section:Algorithm-relaxation-with-channels}

When appropriate, the algorithm designer can take advantage of the optimization in
Section~\ref{section:Optimizing-channels-away} by relaxing the abstract algorithm.
Whenever algorithm correctness does not require the synchrony imposed on the system by
an I/O combination, the combination can be replaced with an explicit asynchronous channel.
The channel is then optimized away during conversion of the algorithm to a distributed program. This approach can lead to a significant performance improvement in the target 
program due to added concurrency.

\subsection{Object-Oriented Implementation of the I/O Automata Framework}

\label{section:IO-to-Code}

%In the previous sections of this chapter we presented a framework that is useful in 
%converting abstract algorithms to distributed implementations. 

We developed a set of C++ 
objects that encapsulate the functions of the framework common to all I/O automata.
These objects have been designed in accordance with the conversion techniques 
we defined, and are intended to be used as a foundation in converting specific algorithms to 
programs. The implementation takes advantage of polymorphism to provide generic
service to any converted I/O automata algorithm; it includes an integral scheduling
mechanism using either random or round-robin discipline, and it incorporates
an exponential backoff scheme for avoiding livelock in the reservation system.
The implementation details are given in~\cite{Cheiner}. We mention the two most important
components here.

\begin{description}

\item[IOAutomaton class.] This class encapsulates components needed in all implementations of I/O Automata. It handles scheduling of locally-controlled actions for execution (subject to them being enabled) and the reservation system for I/O combinations. \emph{IOAutomaton} is the superclass of all classes that represent I/O automata.

\item[IOAction class.] \emph{IOAction} is the superclass of all classes that represent locally-controlled actions. These classes must implement the \emph{Enabled} and \emph{Transition} methods, used by the \emph{IOAutomaton} class to schedule and execute actions.

\end{description}

\newpage
\section{The ESDS Algorithm}

\label{ESDS-algorithm}

\begin{figure}[t]
\begin{minipage}{3in}
\iocodesize
\textbf{State}
\begin{description}
\Item $\wait_f$, a subset of $\opdescs$, initially empty
\Item $\rept_f$, a subset of $\opdescs \times V$, initially empty 
\end{description}
\textbf{Actions}
\B
\ef{\textbf{Input} $\request_c(x)$}
{%Effect: \\
  $\setinsert{\wait_f}{x}$
}
\prc{\textbf{Output} $\send_{f,r}(\reqmsg{x})$}
{%Precondition: \\
  $x \in \wait_f$
}
\ef{\textbf{Input} $\receive_{r,f}(\respmsg{x,v})$}
{%Effect: \\
  if $x \in \wait_f$ then $\setinsert{\rept_f}{(x,v)}$
}
\prcef{\textbf{Output} $\response_c(\op, v)$}
{%Precondition: \\
  $(x, v) \in \rept_f$ \\
  $x \in \wait_f$
}{%Effect: \\
  $\setdelete{\wait_f}{x}$ \\
  $\rept_f \gets \rept_f - \{(x, v'):(x, v') \in \rept_f \}$
}
\end{minipage}
\caption{\emph{ESDSAlg}: Automaton for front end $f$}
\label{fig:frontend}
\end{figure}

\begin{figure}[h]
\iocodesize
\textbf{State}
\begin{description}
\Item $\pending_r$, a subset of $\opdescs$; the messages which require a response
\Item $\rcvd_r$, a subset of $\opdescs$; all operations that have been received
\Item $\done_r[i]$ for each replica $i$, a subset of $\opdescs$; 
	the operations $r$ knows that $i$ has ``done'' 
\Item $\solid_r[i]$ for each replica $i$, a subset of $\opdescs$; 
	the operations that $r$ knows are ``stable at $i$'' 
\Item $\minlabel_r: \opdescs \to \labels \cup \{\infty\}$; 
	the smallest label $r$ has seen for $x \in \opdescs$ 
\Item Derived from $\done_r[r]$ and $\minlabel_r$: $\val_r: \done_r[r] \to V$; 
	the value for $x \in \done_r[r]$ using the $\minlabel_r$ order 
\end{description}
\textbf{Actions}
\B
\iocode{
\ef{\textbf{Input} $\receive_{f,r}(\reqmsg{x})$}
{%Effect: \\
  $\setinsert{\pending_r}{x}$ \\
  $\setinsert{\rcvd_r}{x}$
}
\prcef{\textbf{Internal} $\doit_r(x, l)$}
{%Precondition: \\
  $x \in \rcvd_r$ \\
  $x \notin \done_r[r]$ \remove{(Equivalently, $\minlabel_r(x) = \infty$)} \\
  $x.\prev \subseteq \done_r[r].\id$ \\
  $l > \minlabel_r(y)$ for all $y \in \done_r[r]$ \\
  ($l \in \labels$, equivalently $l \neq \infty$)
}{%Effect: \\
  $\setinsert{\done_r[r]}{x}$ \\
  $\minlabel_r(x) \gets l$ \\
  $\solid_r[r] \gets \solid_r[r] \cup \bigcap_i \done_r[i]$
}
\prcef{\textbf{Output} $\send_{r,f}(\respmsg{x, v})$}
{%Precondition: \\
  $x \in \pending_r$ \\
  $x \in \done_r[r]$ \\
  $x.\strict \implies x \in \bigcap_i \solid_r[i]$ \\
  $v = \val_r(x)$ \\
  $f = \frontend(\client(x.\id))$
}{%Effect: \\
  $\setdelete{\pending_r}{x}$
}
}{
\prc{\textbf{Output} $\send_{r,r'}(\gossipmsg{R, D, L, S})$}
{%Precondition: \\
  $R = \rcvd_r$; $D = \done_r[r]$; \\ 
  $L = \minlabel_r$; $S = \solid_r[r]$
}
\ef{\textbf{Input} $\receive_{r',r}(\gossipmsg{R, D, L, S})$}
{%Effect: \\
  $\rcvd_r \gets \rcvd_r \cup R$ \\
  $\done_r[r'] \gets \done_r[r'] \cup D \cup S$ \\
  $\done_r[r] \gets \done_r[r] \cup D \cup S$ \\
  $\done_r[i] \gets \done_r[i] \cup S$ for all $i \neq r, r'$ \\
  $\minlabel_r \gets \min(\minlabel_r, L)$ \\
  $\solid_r[r'] \gets \solid_r[r'] \cup S$ \\
  $\solid_r[r] \gets \solid_r[r] \cup S \cup (\bigcap_i \done_r[i])$
}
}
\caption{\emph{ESDSAlg}: Automaton for replica $r$}
\label{fig:replica}
\end{figure}

For reference, we provide a description of \emph{ESDSAlg} component
automata. The description follows that of Fekete~\emph{et~al.}~\cite{esds}.
In the algorithm, each process that maintains a copy of the
data objects is modeled as a {\it replica\/}. The replicas
maintain an {\em order} on the operations they know about.
A known prefix of the order at each replica is such that it is
consistent with the eventual total order on operations.
Replicas periodically exchange their knowledge in {\em gossip}
messages. Operations can be {\em strict}, which means that the
response to such operations must be consistent with the eventual
order, or {\em non-strict}, which means that responses
may not reflect the eventual order. Clients may also specify
the {\em prev} set of operations that must be executed
before the new operation.

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%5

\subsection{The Frontends and Channels}

Clients are represented in the algorithm by \emph{frontends}.
The frontend automaton is shown in Figure~\ref{fig:frontend}.
When a client $c$ submits a request (via the $\request_c$
input action), 
its frontend simply relays the request to one or more 
replicas that maintain a copy of the data object, 
and, when it receives a response, relays that back to the client
via the $\response_c$ action.
Frontends use state variables $\wait_f$ and $\rept_f$
for this purpose. 

Channels (of the type specified in Figure~\ref{fig:channel})
are used for request/response messages between frontends and
replicas and for ``gossip'' between replicas.

\subsection{The Replicas}

The replica automaton is given in Figure~\ref{fig:replica}.
The automaton is specified for replica $r$, and it
 has a number of state components. 
The component $\rcvd_r$ is the set of operation descriptors 
of all requests that this replica has received, 
either directly from a frontend, 
or else through gossip from other replicas.
The component $\pending_r$ is the set of operation descriptors
of all requests to which a reply by the replica is pending.
The component $\done_r$ is an array of sets of operation descriptors,  
one for each replica.  
Each set represents the operations known to have been ``done'' 
at the corresponding replica, 
that is, the operations for which the replica can compute a value. 
The component $\solid_r$ is also an array of sets of operation descriptors, 
again one for each replica. 
The interpretation of $x \in \solid_r[i]$ is that replica $r$ 
knows that replica $i$ knows 
that every replica has $x$ in its $\done$ set. 



Replicas assign \emph{labels} uniquely\footnote{
Process identifiers can be used to break ties.} 
to operations from a well-ordered set $\labels$. 
Each replica keeps a function 
$\minlabel_r\colon \opdescs \to \labels \cup \{ \infty \}$ that
encodes the minimum label that the replica knows has 
been assigned to an operation (by any replica), 
where $l < \infty$ for all $l \in \labels$. 
As information is gossiped between replicas, 
the value of $\minlabel_r(x)$ may be reduced
when $r$ learns of a lower label for $x$
(however, \cite{esds} gives an invariant stating that once $x \in \solid_r[r]$, 
no further reduction is possible).


The function $\minlabel_r$ defines a partial order $\LC_r$
(on operation identifiers ${\cal O}.id$),
where $\LC_r = \{(y.\id, x.\id): \minlabel_r(y) < \minlabel_r(x) \}$. 
Because labels are assigned uniquely, 
$\LC_r$ defines a total order for $\done_r[r]$.
A replica uses this order to compute the value of an operation, 
$\val_r(x) =$ $\val(x,$ $\done_r[r], \LC_r)$ for $x \in \done_r[r]$. 

Replicas use gossip messages to keep each other informed about the
operations issued to other replicas, about
the operations received and processed, 
and the labels associated with each operation. 
Hence a gossip message essentially contains the state of a replica 
at a given point in time, 
which will be ``merged'' with the state of the receiving replica. 

For a complete treatment of the algorithm the reader is referred
to~\cite{esds}.

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%


\section{Implementation of Experimental ESDS Systems}

\label{chapter:ESDS-prototype-design}


Using the framework and the classes implementing the I/O automata foundation,
we have created distributed 
implementations of two versions of the unoptimized abstract ESDS algorithm, 
\emph{ESDSAlg} and \emph{SimpleESDSAlg}.  We have also created and 
implemented a more practical optimized version \emph{ESDSOptAlg} of the abstract algorithm.

\begin{figure}
%\centerline{\psfig{figure=ESDSDsgn.eps,bbllx=13bp,bblly=13bp,bburx=455bp,bbury=450bp,clip=,width=435bp}}
\begin{center}

%TexCad Options
%\grade{\off}
%\emlines{\off}
%\beziermacro{\off}
%\reduce{\on}
%\snapping{\on}
%\quality{1.00}
%\graddiff{0.01}
%\snapasp{1}
%\zoom{1.00}
\unitlength 1.00mm
\linethickness{0.4pt}
\begin{picture}(125.00,41.00)
\put(1.00,16.00){\dashbox{1.00}(42.00,10.00)[cc]
{\parbox{1.2in}{\footnotesize Application-specific ESDS data objects}}}
\put(1.00,31.00){\dashbox{1.00}(30.00,10.00)[cc]
{\parbox{1.10in}{\footnotesize Application-specific\\ESDS user client}}}
\put(36.00,31.00){\dashbox{1.00}(60.00,10.00)[cc]{{\footnotesize Application-specific ESDS Server}}}
\put(72.00,21.00){\oval(50.00,10.00)[]}
\put(73.00,21.00){\makebox(0,0)[cc]
{\parbox{1.5in}{\footnotesize ESDS-specific\\I/O Automata Objects}}}
\put(30.00,6.00){\oval(58.00,10.00)[]}
\put(30.00,6.00){\makebox(0,0)[cc]{{\footnotesize Generic ESDS Data Objects}}}
\put(63.00,1.00){\framebox(34.00,10.00)[cc]
{\parbox{1.2in}{\footnotesize Generic Distributed\\I/O Automata Objects}}}
\put(15.00,11.00){\vector(0,1){5.00}}
\put(15.00,26.00){\vector(0,1){5.00}}
\put(40.00,26.00){\vector(0,1){5.00}}
\put(52.00,11.00){\vector(0,1){5.00}}
\put(80.00,11.00){\vector(0,1){5.00}}
\put(70.00,26.00){\vector(0,1){5.00}}
\put(106.00,36.00){\dashbox{1.00}(15.00,5.00)[cc]{}}
\put(100.00,32.00){\makebox(0,0)[lc]
{\parbox{1.05in}{\centering\footnotesize Application-specific\\objects}}}
\put(113.00,23.50){\oval(15.00,5.00)[]}
\put(100.00,17.00){\makebox(0,0)[lc]
{\parbox{1.05in}{\centering\footnotesize ESDS-specific\\objects}}}
\put(106.00,7.00){\framebox(15.00,5.00)[cc]{}}
\put(100.00,03.00){\makebox(0,0)[lc]
{\parbox{1.05in}{\centering\footnotesize Generic I/O\\Automata objects}}}
\end{picture}

\end{center}
\BBB
\caption{\emph{ESDSImpl} Structure}\BB
\label{figure:ESDS-structure}
\end{figure}


Our implementation of the algorithm is independent of the
data object that implements the serial datatype.  A designer can use it 
as a building block for any type of data service. It is only necessary
to implement the data object and add to it the few features needed to 
make it work with ESDS; no modification to ESDS itself is required.
We have implemented several such applications. They are described
in Section~\ref{section:Applications}.

Figure~\ref{figure:ESDS-structure} depicts the hierarchy of the objects that comprise the 
system. Arrows in Figure~\ref{figure:ESDS-structure} represent the relationship ``is 
used by.'' The objects are divided into three groups. The generic I/O automata objects are 
the base \emph{IOAutomaton} and \emph{IOAction} classes. They encapsulate 
functions shared by all I/O automata. 
The ESDS-specific objects implement \emph{ESDSAlg} (see section 4.2 in
~\cite{Cheiner} for details). These objects are 
independent of the particular data service application and do not require modification 
when one wishes to implement a new data service. Finally, the application-specific 
objects implement a particular data service. Application-specific objects have to be 
written for each such service.


\subsection{Mapping Component Automata to MPI Nodes}

\label{Component-automata-to-processes}

\begin{figure}[t]
%\centerline{\psfig{figure=Struct.eps,bbllx=55bp,bblly=13bp,bburx=455bp,bbury=510bp,clip=}}
\begin{center}

%TexCad Options
%\grade{\off}
%\emlines{\off}
%\beziermacro{\off}
%\reduce{\on}
%\snapping{\on}
%\quality{1.00}
%\graddiff{0.01}
%\snapasp{1}
%\zoom{1.00}
\unitlength 1.00mm
\linethickness{0.4pt}
\begin{picture}(117.00,58.00)
\put(28.00,41.50){\oval(24.00,5.00)[]}
\put(28.00,42.00){\makebox(0,0)[cc]{{\small Frontend}}}
\put(59.00,41.50){\oval(24.00,5.00)[]}
\put(91.00,41.50){\oval(24.00,5.00)[]}
\put(59.00,42.00){\makebox(0,0)[cc]{{\small Frontend}}}
\put(91.00,42.00){\makebox(0,0)[cc]{{\small Frontend}}}
\put(2.00,33.00){\framebox(114.00,14.00)[lb]{{$^{\bf ~User~frontends}$}}}
\put(28.00,22.50){\oval(24.00,5.00)[]}
\put(28.00,22.00){\makebox(0,0)[cc]{{\small Replica}}}
\put(91.00,22.50){\oval(24.00,5.00)[]}
\put(91.00,22.00){\makebox(0,0)[cc]{{\small Replica}}}
\put(59.00,13.50){\oval(24.00,5.00)[]}
\put(59.00,13.00){\makebox(0,0)[cc]{{\small Replica}}}
\put(2.00,8.00){\framebox(114.00,20.00)[lb]{{$^{\bf ~Replica~servers}$}}}
\put(1.00,53.00){\framebox(20.00,5.00)[cc]{{\small User client}}}
\put(25.00,53.00){\framebox(20.00,5.00)[cc]{{\small User client}}}
\put(49.00,53.00){\framebox(20.00,5.00)[cc]{{\small User client}}}
\put(73.00,53.00){\framebox(20.00,5.00)[cc]{{\small User client}}}
\put(97.00,53.00){\framebox(20.00,5.00)[cc]{{\small User client}}}
\put(1.00,1.00){\dashbox{1.00}(116.00,47.00)[lb]{{$^{\bf ~MPI~Processes}$}}}
\put(11.00,53.00){\vector(0,-1){5.00}}
\put(11.00,48.00){\vector(0,1){5.00}}
\put(35.00,53.00){\vector(0,-1){5.00}}
\put(59.00,53.00){\vector(0,-1){5.00}}
\put(83.00,53.00){\vector(0,-1){5.00}}
\put(107.00,53.00){\vector(0,-1){5.00}}
\put(35.00,48.00){\vector(0,1){5.00}}
\put(59.00,48.00){\vector(0,1){5.00}}
\put(83.00,48.00){\vector(0,1){5.00}}
\put(107.00,48.00){\vector(0,1){5.00}}
\put(28.00,33.00){\vector(0,-1){5.00}}
\put(28.00,28.00){\vector(0,1){5.00}}
\put(59.00,33.00){\vector(0,-1){5.00}}
\put(90.00,33.00){\vector(0,-1){5.00}}
\put(59.00,28.00){\vector(0,1){5.00}}
\put(90.00,28.00){\vector(0,1){5.00}}
\put(40.00,23.00){\vector(1,0){39.00}}
\put(79.00,23.00){\vector(-1,0){39.00}}
\put(38.00,20.00){\vector(3,-2){9.00}}
\put(47.00,14.00){\vector(-3,2){9.00}}
\put(81.00,20.00){\vector(-3,-2){10.00}}
\put(71.00,13.33){\vector(3,2){10.00}}
\put(59.00,19.00){\makebox(0,0)[cc]{{\small\em gossip}}}
\end{picture}

\end{center}
\BB\BB\BB
\caption{\emph{ESDSImpl} Processes}\BB
\label{figure:ESDS-process-structure}
\end{figure}

Version 1.1 of the MPI standard does not allow dynamic management of nodes.  The number of available processes is determined statically at invocation and cannot change during execution. Thus, the application clients in our data service implementation could not be integrated in the MPI framework because they need to be created and destroyed dynamically. We used sockets instead of MPI mechanisms for communication between application clients and ESDS frontends.

The mapping of ESDS components to system processes is depicted in Figure~\ref{figure:ESDS-process-structure}. In the figure ESDS replicas and frontends run inside the MPI environment, and the application clients connect to the system from outside the MPI environment. At the invocation of the program the ESDS system administrator specifies the number of MPI nodes that will participate in the execution. Three MPI nodes are reserved for system use (they are not depicted in Figure~\ref{figure:ESDS-process-structure}). The rest are divided between ESDS replicas and frontends. The administrator specifies how many nodes to allocate for each purpose.

After the invocation the number of replicas and frontends remains static throughout the execution. Replicas use MPI messages to receive requests from frontends, send gossip message to each other, and send responses back to the frontends. Client processes are dynamically created and destroyed by users. Clients use sockets to connect to one of the frontends. Once the connection is established, the client submits an operation to the frontend and waits for the response.

\subsection{Communication Between Clients, FrontEnds, and Replicas}

\label{section:Comm-between-automata}

As we have already stated, \emph{ESDSAlg} uses asynchronous channel automata for communication between replicas and frontends and for gossip among replicas. We implemented two different systems of communication among frontends and replicas. The first version (\emph{SimpleESDSImpl}) was produced using our scheme for converting I/O combinations to distributed programs (see Section~\ref{section:IOCombinations-to-code}).

The second version (\emph{ESDSImpl}) takes advantage of the fact that \emph{ESDSAlg} relies on asynchronous channels for communication among frontends and replicas. It uses reliable FIFO channels implemented by MPI, as discussed in Section~\ref{section:Algorithm-relaxation-with-channels}. \emph{ESDSImpl} is a more efficient implementation of ESDS than \emph{SimpleESDSImpl} because it avoids the overhead of synchronizing communications among frontends and replicas.

Integration of this approach into \emph{ESDSImpl} implementation is straightforward.  Instead of negotiating with the receiving automaton for synchronized execution, each automaton is free to send an asynchronous request, gossip, or response message and continue executing normally. The pending messages accumulate in the MPI subsystem, which implements reliable FIFO channels and thereby relieves the programmer of that responsibility. In this implementation, the system takes advantage of the distributed nature of the application.

Different methods of interprocess communication constitute the only difference between \emph{ESDSImpl} and \emph{SimpleESDSImpl}. For convenience, both implementations are combined into a single program. The desired method of communication can be set with a switch in the program's configuration file.

\subsection{Optimizing ESDS: ESDSOptImpl}

\label{section:Optimized-prototype}

In addition to implementing \emph{ESDSAlg}, we implemented some of the 
optimizations suggested in~\cite{esds}. In this section we describe the 
optimizations. We also present an I/O automaton for the 
optimized ESDS replica.

We make several algorithmic contributions.
The optimized algorithm \emph{ESDSOptAlg} contains a new 
gossiping scheme. It also caches current stable state at each replica. These 
modifications reduce unnecessary exchange of information between replicas and the
amount of work needed to compute values for new requests. These optimizations make
the ESDS algorithm more practical.

The complete \emph{ESDSOptAlg} replica automaton is shown in 
Figure~\ref{fig:optimized-replica}. 

\begin{figure}

\iocodesize

%\FF

\textbf{Data types}
\begin{description}

\item $P = \{ 1, \ldots, n \}$, the set of replica IDs

\end{description}

\textbf{State}
\begin{description}

\B\item $\pending_r$, a subset of $\opdescs$; the messages which require a
response
\B\item $\rcvd_r$, a subset of $\opdescs$; all operations that have been
received
\B\item $\done_r[i]$ for each replica $i$, a subset of $\opdescs$; 
the operations $r$ knows that $i$ has ``done''   
\B\item $\solid_r[i]$ for each replica $i$, a subset of $\opdescs$; 
the operations that $r$ knows are ``stable at $i$''   
\B\item $\gossip_r[i]$ for each replica $i$, a subset of $\opdescs$; 
the operations that $r$ needs to gossip to $i$
\B\item $\minlabel_r\colon \opdescs \to \labels \cup \{\infty\}$; 
the smallest label $r$ has seen for $x \in \opdescs$
\B\item Derived from $\solid_r[r]$ and $\minlabel_r$: $\maxstab_r \in$ 
$\solid_r[r]$ s.t. $\forall y \in \solid_r[r]$,
$\minlabel_r(\maxstab_r) \geq \minlabel_r(y)$
\B\item $\stabstate_r \in \states$, initially $\sigma_0$; 
the state resulting from doing all the operations up to and
including $\maxstab_r$
\B\item $\stabval_r\colon \solid_r[r] \to V$, initially empty;
the values of the stable operations in the eventual total order
\B\item Derived from $\done_r[r]$ and $\minlabel_r$: $\val_r\colon \done_r[r]
\to V$; the value for $x \in \done_r[r]$ using the $\minlabel_r$ order 

\end{description}

%}
%\caption{Automaton for optimized replica $r$}
%\label{fig:optimized-replica}
%\end{figure}

\textbf{Actions}
\B
\iocode{

\ef{\textbf{Input} $\receive_{f,r}(\reqmsg{x})$}
{%Effect: \\
  $\setinsert{\pending_r}{x}$ \\
  $\setinsert{\rcvd_r}{x}$ \\
  for all $i$ : \\
   \> $\setinsert{\gossip_r[i]}{x}$ 
}

\prcef{\textbf{Internal} $\doit_r(x, l)$}
{%Precondition: \\
  $x \in \rcvd_r - \done_r[r]$\remove{
  $x \in \rcvd_r$ \\
  $x \notin \done_r[r]$ \remove{(Equivalently, $\minlabel_r(x) = \infty$)}} \\
  $x.\prev \subseteq \done_r[r].\id$ \\
  for all $y \in \done_r[r]$ : \\\remove{ \\
  ($l \in \labels$, equivalently $l \neq \infty$)}
    \> $l > \minlabel_r(y)$
}{%Effect: \\
  $\setinsert{\done_r[r]}{x}$ \\
  $\minlabel_r(x) \gets l$ \\
  for all $i$ : \\
     \>$\setinsert{\gossip_r[i]}{x}$ \remove{\\ 
  %% This doesn t do anything unless there is only one replica.
  %% In this case, there is no gossiping at all.
  $\solid_r[r] \gets \solid_r[r] \cup \bigcap_i \done_r[i]$}
}

\prcef{\textbf{Output} $\send_{r,f}(\respmsg{x, v})$}
{%Precondition: \\
  $x \in \pending_i \cap \done_r[r]$ \remove{
  $x \in \pending_r$ \\
  $x \in \done_r[r]$} \\
  $x.\strict \implies x \in \bigcap_i \solid_r[i]$ \\
  %$v = \val_r(x)$ \\
  $v = \!\!\left\{ \begin{array}{ll}
\!\!\!\!\stabval_r(x) & \!\!\!\!\text{if~} x \in \solid_r[r] \\
\!\!\!\!\val_r(x) & \!\!\!\!\text{otherwise}
      \end{array} \right. $\hspace*{-10mm} \\
  $f = \frontend(\client(x.\id))$
}{%Effect: \\
  $\setdelete{\pending_r}{x}$
}

\prcef{\textbf{Output} $\send_{r,r'}(\gossipmsg{R, D, L, S})$}
{%Precondition: \\
  $R = \rcvd_r \cap \gossip_r[r]$; \\
  $D = \done_r[r] \cap \gossip_r[r]$; \\ 
  $S = \solid_r[r] \cap \gossip_r[r]$; \\
  $L = \minlabel_r$; $r \neq r'$
}{%Effect: \\
  $\gossip_r[r ] \gets \{\}$
}

}{

\prcef{\textbf{Internal} $\solidify_r$}
{%Precondition: \\
  $|P| = 1$
}{%Effect: \\
  $\solid_r[r] \gets \solid_r[r] \cup (\bigcap_i \done_r[i])$\\
  for $y$ s.t. $\minlabel_r(y)$\\
                   \>\>\> $\leq \minlabel_r(\maxstab_r)$\\
\> and $\stabval_r(y)$ is undefined,\\
\> in $\minlabel_r$ order:\\
\>\> $(\stabstate_r, \stabval_r(y)) \gets$\\
\>\>\> $\datatrans(\stabstate_r, y.\op)$
}


\ef{\textbf{Input} $\receive_{r',r}(\gossipmsg{R, D, L, S})$}
{%Effect: \\
  for all $i$ : $\gossip_r[i] \gets \gossip_r[i]$\\
             \>\> $\cup (R - \rcvd_r) \cup$ \\
             \>\> $\cup (S - (\bigcap_j \done_r[j])) \cup$ \\
             \>\> $\cup (S - (\solid_r[r ] \cap \solid_r[r])) \cup$ \\
             \>\> $\cup (D - (\done_r[r ] \cap \done_r[r])) \cup$ \\
             \>\> $\cup \{ x: \minlabel_r(x) > L(x) \}$ \\
%            \>\> for all $i$ \\
  $\rcvd_r \gets \rcvd_r \cup R$ \\
  $\done_r[r'] \gets \done_r[r'] \cup D \cup S$ \\
  $\done_r[r] \gets \done_r[r] \cup D \cup S$ \\
  $\done_r[i] \gets \done_r[i] \cup S$ for all $i \neq r, r'$ \\
  $\minlabel_r \gets \min(\minlabel_r, L)$ \\
  $\solid_r[r'] \gets \solid_r[r'] \cup S$ \\
  for all $i$ : $\gossip_r[i] \gets \gossip_r[i] \cup$ \\
          \>\> $\cup ((\bigcap_j \done_r[j]) - \solid_r[r])$ \\
  $\solid_r[r] \gets \solid_r[r] \cup S \cup (\bigcap_i \done_r[i])$\\
  for $y$ s.t. $\minlabel_r(y)$\\
        \>\>\>$ \leq \minlabel_r(\maxstab_r)$\hspace*{-10mm}\\
\> and $\stabval_r(y)$ is undefined,\\
\> in $\minlabel_r$ order:\\
\>\> $(\stabstate_r, \stabval_r(y)) \gets$\\
\>\>\> $\datatrans(\stabstate_r, y.\op)$
}
}
\caption{Automaton for optimized replica $r$}
\label{fig:optimized-replica}
\end{figure}

\subsubsection{Incremental Gossip}

\label{section:Incremental-gossip}

\emph{ESDSAlg} is composed of identical replicas. Each replica $r$ periodically sends information about all operations it has seen to other replicas in gossip messages (Fig.~\ref{fig:replica}). Thus, a typical gossip message contains a lot of information that has been gossiped previously between the same two replicas. Furthermore, the amount of redundant gossip increases linearly with the number of new operations. \emph{ESDSImpl}, as a faithful implementation of \emph{ESDSAlg}, requires gossip messages of unbounded size, and thus cannot be used continuously for long time periods without exhausting system resources or leading to unacceptable deterioration of system performance.

If we assume that replicas do not fail and that replicas communicate via reliable FIFO channels (as is the case with \emph{ESDSImpl}), we can modify the replica automaton to send only the incremental gossip updates. Each replica keeps track of changes in its state and gossips only new information. This change improves system performance, but reduces the system's ability to tolerate lost gossip messages.

The optimized replica automaton has a new state component $\gossip_r$, which is an array
of sets of operation descriptors, one set for each replica. Each $\gossip_r[i]$ set contains
operations that replica $r$ needs to send to replica $i$ in the next gossip message.
An operation enters all $\gossip_r[i]$ sets whenever the replica learns new information
about the operation. Set $\gossip_r[i]$ is emptied every time replica $r$ sends a gossip
message to replica $i$.

\textbf{Remark:} Explicit sequencing of gossip messages combined 
with retransmission and removal of duplicates is sufficient to make 
the optimization work with \emph{unreliable} channels that allow 
message losses, duplicate messages, and out of order delivery.
We may consider this enhancement in the future.

\subsubsection{Removal of Self-Gossip}

\label{section:Removal-self-gossip}

\emph{ESDSAlg} assumes that each replica sends gossip messages to itself as well as to other replicas.  This behavior is inefficient in a practical implementation, but if we removed it from the \emph{ESDSAlg} replica automaton, its behavior would be incorrect when there is only one replica in the system.  The reason is that \emph{ESDSAlg} updates a replica's set of operations that it knows to be stable only during receipt of gossip messages.  In a one-replica system execution without self-gossip messages the operations would never stabilize, violating the requirement of eventual serializability. This optimization adds another action \emph{solidify}$_r$ to the replica automaton to preserve correctness.  The new action detects one-replica executions and updates the set of stable operations independently from gossip actions.

\subsubsection{Memoizing Stable State}

\label{section:Memoizing-stable-state}

\emph{ESDSAlg} ignores the cost of local computation at the replicas.  A replica $r$ gets the current value the value for operation $op_n$ from the initial state $\sigma_0$ by re-computing it as $\datatransextend(\sigma_0, \mkseq{\op_1, \op_2, \ldots, \op_n})$ for $\op_1, \op_2, \ldots, \op_n$ in $\minlabel_r$ order (the function $\datatransextend$ applies $\op_1, \op_2, \ldots, \op_n$, in that order, to $\sigma_0$~\cite{esds}). \emph{ESDSImpl} faithfully implements the same inefficient behavior. Testing \emph{ESDSImpl} under heavy operation load confirmed that the time consumed by recomputation can be significant. In addition, the algorithm requires all operations to stay in memory indefinitely to enable recomputation. These problems make the na\"{\i}ve implementation of the algorithm unsuitable for practical applications.

In~\cite{esds}, an optimization is suggested
that involves {\em memoizing\/} stable state at each replica. Here we specify a 
more aggressive variation of this optimization. We apply operations to the
stable state of a replica as soon as they have stabilized at that replica, whereas
in~\cite{esds}, it is suggested to wait until the operation stabilizes globally before
applying it.  Our version of the optimization results in faster stabilization of operations and a corresponding increase in performance.

We add a state component $\stabstate_r$ to each replica that keeps track of the 
\emph{stable state}, which is the result of applying all operations, 
whose total order has been fixed at all replicas, to the initial state. 
To compute the current state, replica $r$ needs only to apply all 
operations in $\done_r[r]$ that have not yet stabilized to the stable state.

The computation of the new stable state takes place every time replicas
receive gossip messages (see Fig.~\ref{fig:optimized-replica}). Among 
all operations that have not yet entered the stable state, replica $r$ 
finds one with the highest minlabel that has entered
the $\solid_r[r]$ set. Call this operation $\maxstab_r$. All operations 
with minlabels lower than that of $\maxstab_r$ are guaranteed to never 
change minlabels again, and no operation with a lower minlabel can 
be received later. This means that the order of operations up to 
and including $\maxstab_r$ can never be altered again at replica $r$. 
Thus, the replica applies all operations with minlabels lower than 
that of $\maxstab_r$ to the old stable state to compute the new 
stable state.

\textbf{Remark:} Combined with the multipart timestamp optimization (see next section),
this optimization makes it possible to discard almost all information about operations
as soon as they enter the stable state.  We have not implemented this.


\subsubsection{Multipart Timestamps}

\label{section:multipart-timestamps}

In \emph{ESDSAlg}, an operation may be \emph{causally dependent}
on other operations previously processed by the system. To represent
this dependence, the operation's state contains a \emph{prev} component, which is a set of
operation \emph{id}s that must be executed before it.
We substituted a more efficient method for tracking causal dependencies between
operations in place of \emph{prev} sets. Our approach is based on a technique
called \emph{multipart timestamps}.

A multipart timestamp $t$ is a n-tuple $(t_1,\ldots,t_n)$ of nonnegative integer counters. In the context of ESDS, $n$ is the number of replicas. A partial order is defined on multipart timestamps: $t \leq s$ iff $t_j \leq s_j$ for $j \in [1..n]$. Two multipart timestamps are \emph{merged} by taking their component-wise maximum.

In this optimization we remove \emph{prev} sets from operation state and redefine the protocol for keeping track of dependencies between operations with multipart timestamps. 

In the new protocol the state of operation $j$ includes two multipart timestamps, \emph{prev-ts} and \emph{op-ts} (\emph{op-ts} is initialized to all zeros). Replica state also gets two multipart timestamps,  \emph{val-ts} and \emph{rep-ts}, both initially all zeros. The meanings of these new state components are as follows:

\begin{itemize}

\item \emph{op-ts} is assigned to each new operation by the receiving replica in the manner described below. \emph{Op-ts} is guaranteed to be unique for each operation.

\item \emph{prev-ts} plays the same role for an operation $j$ that the \emph{prev} set played in the unoptimized version. It specifies that any other operation with an \emph{op-ts} smaller than $j$'s \emph{prev-ts} must be done before $j$. In other words, for each pair of operations $i$ and $j$, $j$.\emph{op-ts} $< i$.\emph{prev-ts} implies that $j$ is in $i$'s \emph{prev} set.

\item \emph{val-ts} is the merge of \emph{op-ts} timestamps of all operations done at the replica.

\item \emph{rep-ts} is the current replica timestamp, used to assign values to \emph{op-ts} of newly submitted operations in the protocol below.

\end{itemize}

The protocol works as follows:

\begin{enumerate}

\item When replica $r$ receives a new operation $i$ from a front end, it increments
$r$.\emph{rep-ts[r]}, assigns $r$.\emph{rep-ts} to $i$.\emph{op-ts}, and sends the value of
$i$.\emph{op-ts} to the front end. The front end then forwards $i$.\emph{op-ts}
to the client.

\item When a client wants to specify that operations $i_1, \ldots, i_k$ must precede
operation $j$, it merges $i_1$.\emph{op-ts}$, \ldots, i_k$.\emph{op-ts} and assigns
the result to $j$.\emph{prev-ts}.  This method of specifying causal constraints
restricts the kinds of constraints that the client can specify. In particular,
it requires that \emph{all} operations with \emph{op-ts} $\leq$ $j$.\emph{prev-ts}
must be done before $j$, not just $i_1, \ldots, i_k$. Thus, multipart timestamps
trade off flexibility for efficiency.

\item When replica $r$ does operation $i$ (i.e., moves it into $done_r[r]$), it
merges $i$.\emph{op-ts} into $r$.\emph{val-ts}.

\item Gossip messages from replica $r$ to replica $r'$ contain $r$.\emph{rep-ts}.  Upon
receipt of the gossip message, replica $r'$ merges $r$.\emph{rep-ts} into
$r'$.\emph{rep-ts}. For all operations $i_k \in done_r[r]$ included in the
gossip message, $r'$ merges $i_k$.\emph{op-ts} into $r'$.\emph{val-ts}.

\item When replica $r$ wants to do operation $i$ and needs to check that $i$'s
dependencies have been satisfied, it checks that $i$.\emph{prev-ts} $\leq r$.\emph{val-ts}.

\end{enumerate}

The multipart timestamp optimization does not introduce any changes to the
abstract description of the optimized replica automaton in
Figure~\ref{fig:optimized-replica}.  The optimization only changes
the way the precondition $x.\prev \subseteq \done_r[r].\id$ of the $\doit_r$
action is implemented in \emph{ESDSOptImpl}.

\textbf{Remark:} To complete the multipart timestamp implementation, it is necessary to take care of the case when a client submits an operation to more than one replica simultaneously and gets different \emph{op-ts} values for the operation. This has been relegated to future work, as it is not essential to our goal of implementing a working timestamp prototype\remove{ and measuring its advantages against the \emph{prev} set approach}.

\subsection{Fault Tolerance in ESDS}

We have already remarked on how to make our implementation
able to cope with unreliable channels by introducing
gossip message sequencing, retransmission and removal
of duplicates.

Our implementation also includes some {\em ad-hoc\/} fault tolerance mechanisms for handling
fail-stop faults and restarts of replicas. These mechanisms make strong timing assumptions
about the environment.  In particular, they rely on time bounds on communication latency.
We are in the process of relaxing these assumtions and specifying these fault tolerance
mechanisms formally.

\subsection{Applications Using ESDS as a Generic Building Block}
\label{section:Applications}

In order to test our implementations and to provide a proof-of-concept
of the suitability of our implementation as a generic building block,
we have implemented four data service applications that use ESDS.

\subsubsection{String Concatenation Service}

The {\it String Concatenation Service} is a simple data service application.  The
data object is a single string that supports two operations, \emph{Read} and
\emph{Concatenate}. The \emph{Read} operation gives the current value of the string.  
The \emph{Concatenate} operation appends its argument to the string and gives back the 
new value.
%
Because of its simplicity, the String Concatenation Service was 
used for testing \emph{ESDSImpl} and \emph{ESDSOptImpl} 
during development and for running empirical measurements of \emph{ESDSOptImpl} 
performance. 


\subsubsection{Counter Service}

The {\em Counter Service} is another simple data service application, similar to String 
Concatenation in its level of sophistication. The data object is a integer counter variable 
that supports two operations, \emph{Read} and \emph{Add}. The \emph{Read} 
operation gives the current value of the variable.  The \emph{Add} operation adds an 
integer argument to the current counter value and gives back the new counter value.

The Counter Service differs from the String Concatenation Service in one potentially
important respect. Its update operation \emph{Add} commutes with other \emph{Add}s, whereas 
the \emph{Concatenate} operation of the String Concatenation Service does not 
commute with other \emph{Concatenate} operations (unless one of them has the empty 
string as an argument).
The Counter Service was created with the purpose of testing 
whether commutative update operations like \emph{Add} lead to a smaller percentage of 
inconsistent responses than non-commutative update operations like \emph{Concatenate} 
(as we will see in Section~\ref{chapter:Empirical-testing}, it does not).



\subsubsection{Distributed Spreadsheets}

The purpose of creating a third, more sophisticated client has been to demonstrate
the viability of ESDS as a platform for creating diverse and capable data
service applications.
The {\em Distributed Spreadsheets} client creates an environment 
where several people can simultaneously enter spreadsheet data into the same Microsoft 
Excel workbook. Their additions are sent to ESDS replicas, which maintain the current 
state of the workbook and can refresh each user's copy on demand. A possible use of this 
combination of Excel and ESDS is to allow multiple users to enter disjoint data into a 
single Excel file concurrently, see the updates of others automatically, and not worry 
about overwriting other user's additions with their own.


\subsubsection{A More Extensive ESDS Application: Bank Accounts}

The original work on lazy replication and ESDS~\cite{esds} suggests that directory and information services (and similar applications) are the most suitable candidates for ESDS-based implementation because immediate consistency is not important to users of such systems.  To show that ESDS could potentially be used for a wider variety of applications, we have implemented an ESDS application that keeps track of bank accounts. The bank application demonstrates how ESDS-based services can utilize strict and non-strict operations and multipart timestamp-based dependencies.

The service maintains a database of customer accounts in a bank, implemented as an application layer on top of ESDS. Branches of the bank are assumed to be located at physically different sites.  At least one replica node resides at each branch. Operations submitted at a particular branch are forwarded to the local front end.  During normal operation, the front end submits these operations to the local replica. However, if that replica happens to be down, the branch can continue to function by having the front end submit the operations to replicas at remote branches.

The database maintains a set of data tuples in the form of (\emph{name}, \emph{amount}). In addition to opening a new account and closing an established one, there are three basic operation which can be carried out on an account. The operations with their corresponding ESDS specifications are as follows: 
(1)~\emph{Withdrawal}: strict = \emph{true}, \emph{prev-ts} = \emph{full}$_r$,
(2)~\emph{Deposit}: strict = \emph{false}, \emph{prev-ts} = \emph{empty},
(3)~\emph{Balance}:

\begin{itemize}
\item Local,  Hurried:	strict = \emph{false},	\emph{prev-ts} = \emph{empty}
\item Local,  Quick:	strict = \emph{false},	\emph{prev-ts} = \emph{local-full}$_r$
\item Global, Prompt:	strict = \emph{false},	\emph{prev-ts} = \emph{full}$_r$
\end{itemize}


By \emph{empty}, \emph{local-full}$_r$, and \emph{full}$_r$ values of \emph{prev-ts} we mean the following. Assume that the last operation submitted to replica $r$ had been assigned timestamp $(t_1,\ldots,t_n)$. Then
\emph{empty} = $(0,\ldots,0)$, \emph{local-full}$_r$ = $(0,\ldots,0,t_r,0,...,0)$, and \emph{full}$_r$ = $(t_1,\ldots,t_n)$.
%
%
The interpretations of these values of \emph{prev-ts} as \emph{prev} sets are as follows. An \emph{empty} value means that the \emph{prev} set contains no operations. A \emph{local-full}$_r$ value means that the \emph{prev} set includes all operations previously submitted at replica $r$ (but no others), and a \emph{full}$_r$ value means that the \emph{prev} set includes all previously submitted operations at all replicas that replica $r$ knows about.

A \emph{Deposit} operation always succeeds, and it is independent of its ordering relative to other operations on the same account. Thus, \emph{Deposit} is specified without any dependency constraints. On the other hand, a \emph{Withdrawal} of amount $m$ can result in different answers to the client, depending on whether the account has sufficient funds. If the case when it does not, the \emph{Withdrawal} operation does not change the amount in the account and returns an error message.  Otherwise, it decreases the amount in the account by $m$. Permitting two \emph{Withdrawal} operations on the same account to occur concurrently at different replicas would allow the client to withdraw money she does not have. Therefore, we have implemented \emph{Withdrawal} as a strict operation.

It is up to the customer to determine what level of inconsistency she can tolerate in a \emph{Balance} operation in exchange for lower latency. Using the Hurried option, there is no guarantee that previously submitted operations for the account will be visible by the \emph{Balance} lookup. With Quick \emph{Balance} lookup, all previously submitted operations at the local branch will be visible, but there is no guarantee with respect to operations submitted at other branches. Using Prompt \emph{Balance} lookup, all operations on the account known at the local replica will be visible, but there is no guarantee that deposit operations carried out at other branches and not yet gossiped to the local branch will be visible.

\section{Empirical Testing and Analysis}

\label{chapter:Empirical-testing}

We have evaluated our implementation on a network of workstations.
All tests were done on a 10 Mbps Ethernet LAN of 12 Sun workstations 
running SunOS 4.1.4 using the \emph{ESDSOptImpl} 
implementation.
The workstations were not dedicated to this project, and their loads fluctuated.
To account for this variance, we performed each 
test 10 times and averaged the results.
We ran some tests with over 20 replicas; 
however, for performance testing we used 10 replicas only, so that each
replica ran on a separate processor.
Each test run consisted of three hundred operations.
We measured two performance characteristics of the prototype, average
response time, and average throughput, which we now define.

%\B
%\begin{Definition}\label{Def1}\rm
The \emph{response time} for an operation is
the elapsed time between submission of the operation to a replica and response
from the replica in a given execution of the implementation.
%\end{Definition}
%\BBB\B

%\begin{Definition}\label{Def2}\rm
The average \emph{throughput} is the number of
operations the system processes per unit time in a given execution of the implementation.
%\end{Definition}
%\B
  
We would also like to know how the percentage of strict operations among all operations 
submitted to the system affects performance and the degree of ``inconsistency'' in responses.
For a given execution of the implementation, we say that a response to an operation $x$ is 
\emph{inconsistent} if the value returned to the user for $x$ differs from
the value of $x$ in the 
eventual total order of operations. More formally,
%\B
%\begin{Definition}\label{Def3}\rm
let $response_r(x, v_x)$ be a response sent by replica $r$ to a 
front end.  Let $val_{to}(x)$ be the value of $x$ in the eventual total order of all 
operations. Then $response_r(x, v_x)$ is \emph{inconsistent} if $v_x \neq val_{to}(x)$. 
%\end{Definition}\BBB\B
%
%\begin{Definition}\label{Def4}\rm
In a finite execution of an implementation, the \emph{degree of 
inconsistency} is the percentage of inconsistent responses among all responses to user 
operations returned by the system during the execution.
%\end{Definition}
%\B

We have conducted three series of tests using \emph{ESDSOptImpl}.
The first series was designed to test the flexibility of the software,
regardless of the underlying hardware.
The second series was used to determine how system performance,
characterized by response time and throughput, 
depends on the number of replicas. For the first two series, we used the
String Concatenation Service with empty strings; the datatype and content of the
application data were irrelevant for these experiments. All operations in these
tests were non-strict.

The third series measured the changes in system performance and
degree of inconsistency in response to varying the percentage
of the submitted strict operations. We used the Counter Service
and the \emph{Add} operation for our test setup to find out 
whether commutative update operations like \emph{Add} affect the percentage of 
inconsistent responses when compared to non-commutative update operations.


\begin{figure}[t]
%\centerline{\psfig{figure=30thlong.eps,width=400bp}}
\centerline{\psfig{figure=30thlong.eps,width=5.5in,height=2.5in}}
%\vspace*{-0.45in}
\caption{System Throughput $AP$ (submission rate is $330$ operations/sec)}
\label{chart:30thlong}
\end{figure}

Three quantities were measured for each run:
(1) the average time $T_{fe}$ from the submission of an operation by front end $f$ to one 
of the replicas until the receipt of replica response by $f$,
(2) the average time $T_r$ from the receipt of operation $x$ by replica $r$ till the
response from $r$ with a value for $x$, and (3) the total time $\tau$ for the system
to process and respond to all three hundred operations.

>From these data, we obtained two different measures of the response time
$AT_{fe}$ and $AT_r$, and one 
measure of system throughput $AP$, as follows. For each number of replicas from $N=1$ to 
$N=10$, we averaged $T_{fe}$ over 10 runs to get the average time $AT_{fe}$ it 
took a front end to receive a response from a replica after it sent the request message.
The average of $T_r$ over 10 runs produced the average time $AT_r$ it took a replica to
process a request and send back a response. Finally, the average value of $300/\tau$
over 10 runs gives the average system throughput $AP$.

We determined experimentally that a single replica can keep up with 
requests if each comes approximately once in 30 milliseconds. If this 
frequency increases, a single replica cannot keep up and
messages accumulate in the MPI message queues.
Therefore, in our testing, we ranged the frequency of requests
from one every 30 milliseconds to approximately one every 30 times $N$ milliseconds,
where $N$ is the number of replicas.

The complete results are presented in~\cite{Cheiner}.
Some key observations follow.

\remove{
\begin{figure}[b]
%\centerline{\psfig{figure=300th.eps,bbllx=13bp,bblly=13bp,bburx=420bp,bbury=250bpclip=}}
\parbox{3in}{%
%\hspace*{0.51in}\centerline{\psfig{figure=300th.eps,width=4.5in}}
\hspace*{0.05in}\centerline{\psfig{figure=300th.eps,bbllx=13bp,%
bblly=13bp,bburx=420bp,bbury=250bp,clip=,width=2.3in}}
}
\parbox{3in}{%
%\hspace*{1.01in}\centerline{\psfig{figure=p300th.eps,width=4.5in}}
\hspace*{05mm}\centerline{\psfig{figure=p300th.eps,bbllx=13bp,%
bblly=13bp,bburx=420bp,bbury=250bp,clip=,width=2.3in}}
}\BBB\BB\\
\parbox{3in}{%
\caption{System Throughput $AP$ (submission rate is 33 operations/sec)}
\label{chart:300th}
}\hspace*{1em}
\parbox{3in}{%
\caption{System Throughput $AP$ (submission rate is 33*$N$ operations/sec, $N$ = Number of replicas)}
\label{chart:p300th}
}
\end{figure}
}


\begin{figure}[b]
%\centerline{\psfig{figure=300th.eps,bbllx=13bp,bblly=13bp,bburx=420bp,bbury=250bp,clip=}}
\parbox{2.45in}{%
\hspace*{1mm}\vspace*{2.5mm}\centerline{\psfig{figure=p300th.eps,bbllx=13bp,%
bblly=3bp,bburx=420bp,bbury=250bp,clip=,height=1.5in,width=2.4in}}
}
\parbox{2.45in}{%
\hspace*{2mm}\centerline{\psfig{figure=300rl.eps,height=1.8in,width=2.6in,clip=}}
\\~
}\BBB\BB\\
\parbox{2.45in}{%
\caption{System Throughput $AP$ (submission rate is 33*$N$ operations/sec, $N$ = Number of replicas)}
\label{chart:p300th}
}\hspace*{1em}
\parbox{2.45in}{%
\caption{Response Time at the Replicas $AT_r$ (submission rate is 33 operations/sec)}
\label{chart:300rl}
}
\end{figure}


We ran tests in which the number of replicas exceeded the number
of available processors. This demonstrated the flexibility of
our implementation to run in a variety of settings.
The test increased the number of replicas from $N=1$ to $N=20$,
while  operations were submitted at the constant rate of 330 
operations/sec. The collected data are plotted in Figure~\ref{chart:30thlong}.
The system throughput increases with the number of replicas participating in the system and 
performing submitted operations. However, the throughput drops off 
when the system runs out of physical processes (at $N=10$).
The overhead of context switches and the forced serialization of communications between 
replicas sharing a processor adversely impact system performance. 
In all other examples, we have limited the number of replicas
to the number of available processors.

%We now describe how system throughput $AP$ and response time at 
%replicas $AT_r$ depend on the number of replicas and the rate of operation submission.
%Figure~\ref{chart:300th} shows that additional replicas have no significant
%effect on throughput when the frequency of requests is such that
%a single replica is able to handle them one at a time (i.e. about 33 operations
%per second).
%The system throughput is close to its theoretical limit of thirty three 
%operations per second, and it actually declines slightly as
%the number of replicas is increased; the change may be attributed to
%differences between individual servers' performance and load.

%\begin{figure}
%\centerline{\psfig{figure=p300th.eps,bbllx=13bp,bblly=13bp,bburx=420bp,bbury=250bp,clip=}}
%\caption{System Throughput (submission rate is 33*$N$ operations/sec, $N$ is the 
%number of replicas)}
%\label{chart:p300th}
%\end{figure}

We now describe how system throughput $AP$ and response time at 
replicas $AT_r$ depend on the number of replicas and the rate of operation submission.
Figure~\ref{chart:p300th} shows that throughput rises
(nearly linearly) by adding new replicas
when the frequency of requests starts at thirty three operations/sec 
for one replica and is increased proportionally to the number of replicas.
However, throughput does not come close to the estimated limit of three hundred and thirty 
operations per second. This might be justified by the increasing gossip overhead.

%\begin{figure}
%\centerline{\psfig{figure=30th.eps,bbllx=13bp,bblly=13bp,bburx=410bp,bbury=250bp,clip=}}
%\caption{System Throughput (submission rate is $330$ operations/sec)}
%\label{chart:30th}
%\end{figure}

\remove{
\begin{figure}[t]
\parbox{3in}{%
%\hspace*{0.51in}\centerline{\psfig{figure=30th.eps,width=4.5in}}
\hspace*{0.05in}\centerline{\psfig{figure=30th.eps,bbllx=13bp,%
bblly=13bp,bburx=410bp,bbury=250bp,clip=,width=215bp,rheight=2in}}
}\BBB
\parbox{3in}{%
\hspace*{05mm}
%\vspace*{3.5in}\centerline{\psfig{figure=p300rl.eps,width=3.2in}}
%\hspace*{0.05in}\centerline{\psfig{figure=300rl.eps,bbllx=13bp,%
%bblly=13bp,bburx=420bp,bbury=250bp,clip=,width=225bp,rheight=2in}}
\centerline{\psfig{figure=300rl.eps,width=3.4in,rheight=2.2in}}
\\~
}\\
\parbox{3in}{%
\caption{System Throughput $AP$ (submission rate is 330 operations/sec)}
\label{chart:30th}
}\hspace*{1em}
\parbox{3in}{%
\caption{Response Time at the Replicas $AT_r$ (submission rate is 33 operations/sec)}
\label{chart:300rl}
}
\end{figure}
%\centerline{\psfig{figure=300rl.eps,width=400bp}}
%\label{chart:300rl}
}


%Figure~\ref{chart:30th} demonstrates a pattern similar to that in
%Figure~\ref{chart:p300th}, except that the frequency of requests
%is held constant at a high level. The increase in throughput is expected,
%but it is not clear what is the reason for the drop-off for the tenth replica.

%\begin{figure}
%\centerline{\psfig{figure=300rl.eps,width=400bp}}
%\vspace*{-0.45in}
%\caption{Response Time at the Replicas (submission rate is $33$ operations/sec)}
%\label{chart:300rl}
%\end{figure}


Figure~\ref{chart:300rl} shows response time for a constant, low frequency
of requests and with increasing number of replicas. This frequency is just short
of saturating one replica. Not surprisingly, the best response time shows up in
a one-replica execution: the replica can keep up with the submissions, there
is no gossip, and all operations stabilize immediately.
As the number of replicas grows, operations take longer to
stabilize (recall that all replicas
must perform the operation before it stabilizes); moreover, the time spent
on gossip processing is also increased. Each replica is busy
re-applying non-stable operations and gossiping some of the time.
As a consequence, we observe an increase in $AT_r$ for $N=1$ through $N=4$.
However, the system reaches a steady state for $N \geq 4$. After that point, the
load on individual replicas is sufficiently low, and they can keep up with
both gossip and new operations simultaneously.
%(We have performed similar tests with gossiping disabled and observed
%a dramatic decrease in response time as the number of replicas is increased.)


%\begin{figure}
%\centerline{\psfig{figure=p300rl.eps,width=400bp}}
%\vspace*{-0.45in}
%\caption{Response Time at the Replicas (submission rate is $33*N$ operations/sec, 
%$N$ is the number of replicas)}
%\label{chart:p300rl}
%\end{figure}

%\begin{figure}
%\centerline{\psfig{figure=png300rl.eps,width=400bp}}
%\vspace*{-0.45in}
%\caption{Response Time at the Replicas (submission rate is $33*N$ operations/sec, 
%$N$ is the number of replicas, gossip is disabled)}
%\label{chart:png300rl}
%\end{figure}


% Hmm, our explanation for $AT_{fe}$ at moderate request frequency
% doesn't make sense - why should $AT_{fe}$ not simply correspond
% $AT_r$ with a constant for overhead tucked on? Need to repeat these
% tests and/or come up with a better explanation. - O.C.

%We have also performed a similar series of tests to measure
%response time at the front end ($AT_{fe}$). 
%If the requests are very frequent, then
%the response time at fronends degrades for small number
%of replicas that become quickly saturated and take a long
%time to respond.



\begin{figure}[t]
%\centerline{\psfig{figure=pvsc4.eps,bbllx=13bp,bblly=13bp,bburx=380bp,bbury=260bp,clip=}}
\parbox{2.45in}{%
\hspace*{0.02in}\centerline{\psfig{figure=pvsc4.eps,bbllx=13bp,%
bblly=13bp,bburx=380bp,bbury=260bp,clip=,height=1.8in,width=2.4in}}
}
\parbox{2.45in}{%
\hspace*{2mm}\vspace*{1mm}\centerline{\psfig{figure=pvsc6.eps,bbllx=13bp,%
bblly=13bp,bburx=380bp,bbury=260bp,clip=,height=1.8in,width=2.4in}}
}\\
\parbox{2.45in}{%
\caption{Tradeoff Between Response Time and Consistency (4 Replicas)}
\label{chart:pvsc4}
}\hspace*{1em}
\parbox{2.45in}{%
\caption{Tradeoff Between Response Time and Consistency (6 Replicas)}
\label{chart:pvsc6}
}
\end{figure}

Finally, we performed a test to observe the trade-off between
the response time and consistency when the number of required
strictly consistent responses increases.
This test was conducted using the Counter Service application, using \emph{Add} 
operations. The results for four and six replicas are summarized in Figures 
\ref{chart:pvsc4} and \ref{chart:pvsc6}, respectively.

The percentage of inconsistent responses goes down linearly as
the percentage of strict operations climbs.  On the other hand,
since strict operations require the system to stabilize the operation's
value at all replicas before responding, the latency of responses to
strict operations is significantly higher than to non-strict operations.  
This difference between strict and non-strict operation latencies is
reflected in the linear increases of average latency with percentage of strict
operations in Figures~\ref{chart:pvsc4} and~\ref{chart:pvsc6}.  The 
coefficient of the linear increase is higher for the larger number of replicas (6),
possibly because the time 
required to synchronize all replicas with respect to a particular operation increases with 
the number of replicas.

The trade-off between consistency and performance is clearly demonstrated by these results.
Users willing to tolerate transient inconsistencies in system responses can
submit primarily non-strict operations. For these users the system will function
in the region on the left side of Figures~\ref{chart:pvsc4} and~\ref{chart:pvsc6},
where transient inconsistency is high but response latency is low.  Conversely,
users who require strict consistency in responses to some or all of their operations
will pay the cost of higher response latency.

We have run the same experiment using the String Concatenation service in place
of the Counter service and found that the results did not differ from those
shown in Figures~\ref{chart:pvsc4} and~\ref{chart:pvsc6}. Thus, we found no evidence that
commuting operations like Counter service's \emph{Add} substantially change the
percentage of inconsistent responses compared to non-commuting operations like
String Concatenation service's \emph{Concatenate}.

%\begin{figure}
%\centerline{\psfig{figure=pvsc4.eps,bbllx=13bp,bblly=13bp,bburx=380bp,bbury=260bp,clip=}}
%\caption{Tradeoff Between Response Time and Consistency (4 Replicas)}
%\label{chart:pvsc4}
%\end{figure}
%
%\begin{figure}
%\centerline{\psfig{figure=pvsc6.eps,bbllx=13bp,bblly=13bp,bburx=380bp,bbury=260bp,clip=}}
%\caption{Tradeoff Between Response Time and Consistency (6 Replicas)}
%\label{chart:pvsc6}
%\end{figure}

\section{Conclusions}

\label{chapter:Conclusions}

We have defined a set of techniques for converting source algorithms specified as I/O 
automata compositions into target distributed programs written in an imperative 
language. We demonstrated that the techniques support object-oriented design for target 
programs by implementing a set of C++ objects that encapsulate common properties of 
I/O automata and can be used in designing the target program. Our techniques are 
applicable to commonly occurring algorithms that use asynchronous channels or 
input/output combinations that involve two automata for communications between 
distributed components.
%An interesting topic for future work is to generalize these 
%techniques to cover all types of I/O automata compositions.


Using our conversion techniques, we implemented the abstract ESDS algorithm 
\emph{ESDSAlg} as a distributed program \emph{ESDSImpl}. 
The modular design of \emph{ESDSImpl} allowed us to create 
modules specific for ESDS only once and then build several distinct data services without
any need to further modify these modules. In this way, we show how \emph{ESDSAlg} can
be used as a building
block for distributed data service implementations. A data service built on
top of ESDS inherits its characteristics. Strict consistency in such a service may
be relaxed by specifying operations as non-strict and requiring only the
explicitly stated causal
relations between individual operations (given by \emph{prev} sets) to be preserved.
The benefit of such relaxation is improved system performance. At the same time, the
system is assured of reaching a globally consistent state, as specified and proven
in~\cite{esds}. This is a property of \emph{ESDSImpl}
that differentiates it from the implementation in~\cite{Ladin92},
in which additional measures must be taken by an application designer
to ensure that replica states do not diverge irrevocably.

We have added several optimizations to \emph{ESDSAlg}, producing an optimized
abstract algorithm \emph{ESDSOptAlg} and a corresponding \emph{ESDSOptImpl}
implementation. \emph{ESDSOptImpl} fixes some inefficiencies of \emph{ESDSImpl}
and makes ESDS a more practical system.
Some current and future work on other optimizations to ESDS is discussed
in the next section.
%Considered improvements include introducing the
%multi-part timestamps (similar to \cite{Ladin92})
%and dealing with \emph{unreliable} channels that
%allow message losses, duplicate messages, and out of order delivery.

The empirical tests on \emph{ESDSOptImpl} show how its 
performance, characterized by response time and throughput, is affected by changing the 
number of replicas participating in the execution and by changing the system load.
The data also
confirm that ESDS performance reflects an inherent tradeoff 
between performance and consistency.

\section{Ongoing and Future Work}
\label{section:Future-work}

Several theoretical and practical aspects of our work are open for further exploration.
On the theoretical side, it remains to be shown that \emph{ESDSImpl} (as well as
\emph{ESDSOptImpl} and \emph{ESDSOptAlg}) implement \emph{ESDSAlg} 
(in the mathematical sense). This requires formalization of our framework
for converting I/O automata to distributed programs.
More ambitiously, it would be interesting to develop a framework
for showing that a practical implementation of an algorithm, 
derived with our techniques and treated as a mathematical object,
correctly implements the I/O automata specification of the
original algorithm.

As we have remarked, the mechanism for converting I/O combinations to
code (Section~\ref{section:IOCombinations-to-code}) needs to be extended to
the more general case when multiple automata participate in the combination. 
Section~\ref{section:multipart-timestamps} contains suggestions for
further work on the multipart timestamp optimization.

A possible application of ESDS is a wide-area network data service.
To be useful in practice as a WAN service, 
ESDS must accomodate dynamic changes in the number
of replicas and tolerate server and communication failures. We are currently
experimenting with a version of ESDS that tolerates simulated fail-stop crashes
of replicas and allows crashed replicas to rejoin the system after recovery.

Finally, our implementation can be viewed as a proof-of-concept
of ESDS as a generic component of distributed operating systems.
Serious applications that can benefit from such service include
directory services and distributed type repositories useful
for distributed object systems. It would be interesting to
formalize ESDS as a distributed operating system service and use it
in implementing a real application.

\newpage
\newcommand{\Bibitem}{\BB\bibitem}

\begin{thebibliography}{9}

\begin{small}

\Bibitem{AlsbergD76}  P. Alsberg and J. Day. A principle for resilient sharing of 
distributed resources. In \textsl{Proceedings of the 2nd International Conference on 
Software Engineering}, pp. 627-644, Oct. 1976.


\Bibitem{BalKT92} H. Bal, M. Kaashoek, and A. Tanenbaum. Orca: Alanguage for 
parallel programming of distributed systems. \textsl{IEEE Transactions on Software 
Engineering}, 18(3):190-205, March 1992

\Bibitem{BernsteinHG87} P. Bernstein, V. Hadzilacos, and N. Goodman. 
\textsl{Concurrency Control and Recovery in Database Systems}. Addison-Wesley, 
1987.


\Bibitem{BirrellLNS82} A. Birrell, R. Levin, R. Needham, and M. Schroeder. Grapevine: 
An Exercise in Distributed Computing. \textsl{Communications of the ACM}, 
25(4):260-274, 1982.

\Bibitem{Cheiner} O. Cheiner,  {\it Implementation and Evaluation of
an Eventually-Serializable Data Service},
M. Eng. Thesis, Electrical Engineering and Computer Science 
Massachusetts Institute of Technology, 1997.

\Bibitem{MPI-intro} J. Dongarra, S. Otto, M. Snir, and D. Walker. A message passing 
standard for MPP and workstations. \textsl{Communications of the ACM}, Vol.39, No. 7 
(July 1996), pp. 84-90


\Bibitem{esds} A. Fekete, D. Gupta, V. Luchangco, N. Lynch, and A. Shvartsman. 
Eventually-Serializable Data Services. \textsl{Principles of Distributed Systems} 1996, pp. 300-310.

\Bibitem{FischerM82} M. Fischer and A. Michael. Sacrificing Serializability to Attain 
High Availability of Data in an Unreliable Network. In \textsl{Proceedings of the ACM 
Symposium on Database Systems}, pp. 70-75, March 1982.

\Bibitem{Garcia-MolinaLBKS85} H. Garcia-Molina, N. Lynch, B. Blaustein, C. 
Kaufman, and O. Schmueli. Notes on  a Reliable Broadcast Protocol. \textsl{Technical 
Memorandum}, Computer Corporation America, October 1985.

\Bibitem{MPICH} W. Gropp and E. Lusk. User's guide for MPICH, a portable 
implementation of MPI. \textsl{Technical Report ANL-96/6}, Argonne National 
Laboratory, 1994.

\Bibitem{Gifford79} D. Gifford. Weighted Voting for Replicated Data. In 
\textsl{Proceedings of the 7th ACM Symposium on Principles of Operating Systems 
Principles}, pp. 150-162, December 1979.


\Bibitem{Herlihy86} M. Herlihy. A Quorum-Consensus Replication Method for Abstract 
Data Types. \textsl{ACM Transactions on Computer Systems}. 4(1):32-53, February 
1986.


\Bibitem{Ladin92} R. Ladin, B. Liskov, L. Shrira, and S. Ghemawat. Lazy replication: 
Exploiting the semantics of distributed services. \textsl{ACM Transactions on Computer 
Systems}, 10(4):360-391, Nov. 1992.


\Bibitem{Lamport79} L. Lamport. How to make a multiprocessor computer that correctly 
executes multiprocess programs. \textsl{IEEE Transactions on Computers}, 28(9):690-
691, September 1979.


\Bibitem{Lynch} N. Lynch. \textsl{Distributed Algorithms}.
Morgan Kaufmann Publishers, San Mateo, CA, 1996. 

\Bibitem{I/O-Automata} N. Lynch and M. Tuttle. An introduction to
Input/Output automata. \textsl{CWI-Quarterly}, 2(3):219--246, September 1989.
Centrum voor Wiskunde en Informatica, Amsterdam, The Netherlands.


\Bibitem{OkiL88} B. Oki and B. Liskov. Viewstamp Replication: A New Primary Copy 
Method to Support Highly-Available Distributed Systems. In \textsl{Proceedings of the 
7th ACM Symposium on Principles of Distributed Computing}, August 1988.


\Bibitem{Stonebraker79} M. Stonebraker. Concurrency Control and Consistency of 
Multiple Copies of Data in Distributed INGRES. \textsl{IEEE Transaction on Software 
Engineering}, 5(3):188-194, May 1979.

\Bibitem{Thomas79} R. Thomas. A Majority Consensus Approach to Concurrency 
Control for Multiple Copy Databases. \textsl{ACM Transactions on Database Systems}, 
4(2):180-209, June 1979.


\Bibitem{ISO500}{\em International Standard 9594-1,
        Information Processing Systems---Open Systems Interconnection---The Directory},
        {ISO} and {IEC}, 1988.
 
\Bibitem{DNS}IETF, \emph{RFC 1034} and \emph{RFC 1035},
 {D}omain {N}ame {S}ystem, 1990.

\Bibitem{MPI-1.1} The Message Passing Interface  Forum. \textsl{The MPI message-
passing interface standard}. http://www.mcs.anl.gov/mpi/standard.html, May 1995.


\Bibitem{OSF/DCE-UUID} Open Software Foundation. \textsl{OSF DCE Application 
Development Guide}. Prentice Hall, Englewood Cliffs, NJ, 1993

\end{small}
\end{thebibliography}

\setlength{\parindent}{0.0in}

\vspace*{10bp}
{\Large\bfseries Acknowledgements}\\[7bp]

The authors are grateful to Nancy Lynch, Victor Luchangco, and Istvan Derenyi
for valuable technical suggestions and to Istvan Derenyi for his contributions
to the implementation of the Bank Accounts service.

\vspace*{20bp}
{\Large\bfseries Author Contact Information}\\[7bp]

Oleg M. Cheiner\\
Department of Computer Science,\\
Carnegie Mellon University, 5000 Forbes Ave.\\
Pittsburgh, PA 15213-3890, USA,\\
Email: {\tt oleg@cmu.edu}.\\
The work of the first author was substantially done\\
at the Massachusetts Institute of Technology.\\[7bp]

Alex A. Shvartsman\\
Laboratory for Computer Science,\\ 
Massachusetts Institute of Technology,\\
545 Technology Square NE43-368,\\
Cambridge, MA 02139, USA,\\
Email:{\tt alex@theory.lcs.mit.edu}\\
\emph{and}\\
Department of Computer Science and Engineering,\\
University of Connecticut,\\
191 Auditorium Road, U-155, Storrs, CT 06269, USA,\\
Email: {\tt aas@cse.uconn.edu}.\\


\end{document}


