% $Id: arch.tex,v 1.11 2006/04/11 03:57:24 gokhale Exp $


\section{Architecture of the Fault Tolerant DRE Middleware}
\label{sec:arch}

%%PGR
Providing fault-tolerance for our MLRM middleware required a careful
design and a number of improvements to move us from a
non-fault-tolerant solution to one that could survive multiple
cascading failures. A number of areas that needed improvement became
apparent as we developed both the individual pieces making up the MLRM
functionality as well as the interactions between the MLRM pieces.
This section describes the individual improvements we made and how
they worked together to provide a comprehensive solution.


%%PGR  If we can fit the figure in we may need this but I don't see it at
%%the moemnt. 
%% Paul,  I am going to attempt modifying it - andy
Our FT design for the MLRM is incorporated into the conventional
layers of middleware depicted in Figure~\ref{fig:mw-stacks}.  The rest
of the section describes how individual pieces of our complete FT-MLRM
map to these different middleware layers and resolves the FT
challenges manifested in systems such as the dynamic resource
manager.  

\begin{figure}[!htb]
  \centering
  \includegraphics[width=0.75\columnwidth,keepaspectratio]{mw-stacks}%
%  \vspace{-0.6in}
  \caption{MLRM Functionality Mapping on Middleware Layers}
  \label{fig:mw-stacks}
\end{figure}

As explained in Section~\ref{sec:case-study}, the MLRM and
our FT solution uses a number of individual base technologies, such as
CIAO, OpenCCM, JacORB, TAO, MEAD and MySQL.  A number of middleware
improvements were necessary to make a foundation strong enough to
stand up a fault-tolerant MLRM.  At the fault-tolerant middleware
layer we needed to improve the MEAD infrastructure to better deal with
application state transfer (necessary for replica reconstitution),
duplicate message detection (necessary when multiple replicas may all
reply to an incoming message), peer-to-peer calling conventions, and
support for multi-ORB and multi-language solutions.  Another
improvement made to MEAD, not used by replicas, but used by processes
interacting with replicas was support for limiting the use of group
communications via a ``gateway.''  Above MEAD, at the component
middleware layer, improvements were necessary to support state
exchange for components.  Moving further towards the application layer
we made small changes to MySQL clustering to better support the
fault-model in which the MLRM is deployed.  Our enhancements result in
a reusable framework with fast and bounded failure recovery times.


\subsection{Co-existence of Protocols: Singleton Gateway Approach} 
\label{sec:coexist}

In order to keep replicas consistent, our fault-tolerance
infrastructure ensures that messages are reliably delivered to each
replica in the same order, using a Group Communication System (GCS)
system.  Any interactions with a replica, after the initial middleware
layer (i.e., CCM) bootstrapping, pass through GCS.  In the simple case
of a client interacting with a replicated server, all interactions are
necessarily over GCS. The existing FT Middleware solution forced all
communications in a fault-tolerant system to use GCS.

Our situation is more complex in the MLRM case, a three-layered
hierarchical architecture, shown in Figure~\ref{fig:mlrm-arch}. We are
(currently) replicating only the top layer.  Due to the need for
consistency, elements in the middle layer interact with the top layer
using GCS.  However, since the middle and bottom layers are not
replicated, they do not require the same consistency guarantees and do
not need to use GCS to communicate with one another.  Furthermore, we
do not want to force the extra overhead of the GCS infrastructure onto
the lowest layer, which can include hundreds of components, each with
real-time performance requirements and frequent communication needs.
To provide acceptable performance for components that require GCS and
to more efficiently use resources, we implemented functionality that
limits the use of GCS to where it is strictly necessary.  We call
these applications that use both GCS and non-GCS communication as
``gateways''. 

These are different from ``gateways'' used elsewhere in fault tolerance
solutions, since we don't introduce any extra architectural elements.
Instead, as illustrated in Figure~\ref{fig:gateway}, we simply introduce
an extra role that existing elements take on.  Replicated elements all
communicate group communication.  Most non-replicated elements communicate
using non-group communication, such as TCP.  Those elements that interact
directly with replicated elements speak both group communication (with
the replicated elements) and non-group communication with everything
else. To implement this, we modified the interception
approach used by our MEAD software (explained next), which intercepts
all interprocess communication and directs it over group communication
(Spread), to  support both the default communication and group
communication and to selectively route messages based on whether it is
going to a replicated element.

Implementing this solution turned out to be more difficult than
initially anticipated. Up to that point we had replicated only
single-threaded applications. Without the constraints imposed by being
a replica many gateway applications ended up being multi-threaded. We
improved our middleware to be able to deal with multi-threaded
applications. This allowed our gateways to work but also allowed our
passive replication scheme to be used in multi-threaded applications.

\begin{figure}[!htb]
  \centering
  \vspace{-.4in}
  \hspace{-1.6in}
  \includegraphics[width=0.7\columnwidth,keepaspectratio]{gateway}%
  \vspace{-.6in}
  \caption{Gateway Approach to Mix-mode Communication}
  \label{fig:gateway}
\end{figure}


\subsection{MEAD Enhancements}
\label{sec:mead}

The MEAD system~\cite{Mead:CCPE} that we have developed at Carnegie
Mellon University is a fault-tolerance infrastructure that supports
the transparent replication and recovery of components in stateful
CORBA applications.  In order to ensure the system-wide consistency of
replicas of every CORBA component, MEAD exploits the guarantees of the
underlying Spread group communication system~\cite{Amir:Spread} to
convey the replicated application's messages.  In addition, to ensure
exactly-once semantics for processing requests and replies in the
presence of replication, MEAD performs the detection and suppression of
duplicate requests and replies before delivering messages to the
application.  Below we describe our enhancements to MEAD to resolve
the challenges described in Section~\ref{sec:challenges}.

%%PGR I don't really see this one
% - all the features that were added for queuing,
%    for working on mix mode FT i.e., active and passive together,
%    interworking with higher layer component middleware solns

%%PGR
\vspace{-.13in}
\paragraph{ORB and CCM interactions.} Originally, MEAD had support for
CORBA 2, non-CCM style interactions.  In order to interoperate with the
component world a number of changes were necessary.  In a CCM context
the ORB may be up before the hosted objects are ready to interact with
the rest of the system.  The ORB is ready for bookkeeping interactions
rather than application messages.  Initially MEAD assumed that when
the ORB was up it was ready to receive messages.  In order to ensure
this was the case we implemented a notification from the ORB
indicating all its components were ready.  This allowed MEAD to make
invocations to get and set the state when they could be processed by
the ORB.

%%PGR
\vspace{-.13in}
\paragraph{Multiplatform support.} Initially our FT middleware
solution worked only with the TAO C++ ORB.  This was not an acceptable
solution as pieces of the system, which needed to be replicated, also
use the JacORB Java ORB.  We upgraded our middleware to support both
TAO/\-CIAO/\-C++ and also JacORB/\-Java.  This allowed interactions
between replicated C++ components and replicated Java objects. 

\vspace{-.13in}
\paragraph{Multiprotocol support.}
MEAD exploits library interpositioning to intercept a
process' network system calls by using the dynamic linker's run-time
support. Using the \texttt{LD\_PRELOAD} environment variable, we load
the MEAD system as a shared object library into the address space of
each GCS-hosted component and of the singleton, ahead of all of the
other dynamically linked system and application libraries. The MEAD
library overrides some socket and network functions (such as
\texttt{socket}, \texttt{connect}, \texttt{bind},
\texttt{read}, \texttt{write}, etc.) to perform
the transparent re-routing of the CORBA application's IIOP messages
over the Spread GCS \cite{Amir:Spread}.

The MEAD way of performing the TCP-GCS bridging introduces additional
considerations for the implementation of its infrastructure.  The
singleton that forms the bridge between unreplicated TCP clients and
replicated servers must also support GCS clients, albeit in a mutually
exclusive manner. It is perfectly possible for the end-to-end
operations passing through the singleton to arise from either TCP or
GCS clients.

Because MEAD operates at the network/socket level of the CORBA
application (in user space, underneath the ORB, but not within the
kernel), its operation hinges upon the sequence of network-level
interactions that CORBA clients and server undergo during connection
establishment, during normal communication of requests and replies,
and during connection teardown. For synchronous CORBA client-server
communication, the MEAD library underneath either the GCS client or
server is typically driven off interrupt-driven I/O (through a
\texttt{select} system-call with a timeout value set to
\texttt{NULL}, indicating that call blocks until
interrupted by a signal indicating incoming I/O) whenever it receives
notifications of received GCS messages that are waiting for
processing.

The singleton infrastructure is assigned to a separate group of its
own, which allows it to communicate over the GCS protocols within the
replicated domain.  The singleton's addressing information (a group
identifier for GCS access and an IP address and port number for TCP
access) is made available to its GCS and TCP clients.

Because MEAD conveys the application's messages through GCS, any
component supported by MEAD will have its connections funneled/mapped
by MEAD onto a single socket connection to the local Spread GCS
daemon. Effectively, when a CORBA client/server/singleton is supported
over MEAD, there is only way into and out of that component through
MEAD, and this ``GCS channel'' (accessed via a
\texttt{SP\_receive} blocking call on the Spread GCS API) supplies
only totally ordered messages. As long as it delivers (to its hosted
clients and server) only the ordered sequence of messages that it
receives off this GCS channel, MEAD can ensure replica consistency
because every MEAD-hosted server replica will ``see'' the same
requests and replies.

%%PGR
\vspace{-.13in}
\paragraph{Duplicate suppression.}
Another update to MEAD involved duplicate suppression. When multiple
replicas reply, MEAD must suppress duplicate messages from reaching
the intended recipient. Initially an application could be either a
client or a server. In a peer-to-peer environment this assumption was
too constraining and was removed. Furthermore, we also enabled this
duplicate suppression information to be moved from an existing replica
to a new replica as part of the state.

\vspace{-.13in}
\paragraph{Dealing with non determinism.}
Any connection from the TCP domain to the singleton will take the form
of a separate socket connection at the MEAD infrastructure. These
sockets are channels of communication that are distinct from MEAD's
preferred GCS channel, into and out of the singleton. In fact, from
MEAD's viewpoint, every TCP connection represents out-of-band
communication with the singleton that can potentially compromise its
consistency because the TCP messages (especially when there exist
multiple TCP connections) are not totally ordered across the
system. While we could opt to handle the TCP communication also in an
interrupt-driven fashion, as we did with the GCS communication, the
two sets of interrupts can become arbitrarily interleaved and make for
tricky concurrent programming.

We chose the simpler and more controllable solution of using
polling-based I/O for the TCP connections. Although this keeps the
MEAD infrastructure in a ``busy-wait'' state, polling periodically for
TCP-based I/O, this allows for the careful scheduling of when TCP
communication can interleave with the GCS communication, through the
appropriate adjustment of the polling frequency. Effectively, the
file-descriptors associated with the TCP connections are inserted into
a \texttt{select} system-call with a fixed timeout value; we note that
this does change application semantics because a standard CORBA
application (without replication or GCS involved) would use
interrupt-driven I/O.

Because the GCS channel and the TCP sockets are managed separately by
MEAD, the infrastructure underlying the singleton can differentiate
between the two kinds of messages and handle the bridging seamlessly.
Because the singleton must handle multiple TCP or GCS clients
simultaneously, MEAD needs to perform concurrent duplicate
detection-and-suppression across all of the various virtual end-to-end
connections that pass through the singleton. For each incoming
connection, MEAD maintains some book-keeping information, such as the
last-seen request identifiers on that connection, in order to ensure
state consistency. 


\subsection{Enhancements to Component Middleware and Deployment Engine.} 

Our fault tolerance enhancements span the ORB and component
middleware.  CIAO is an open-source\footnote{CIAO is available from
\url{www.dre.vanderbilt.edu/CIAO}.} implementation of the OMG
Lightweight CCM and Real-time CORBA~\cite{RTCorba:02} specifications
built atop The ACE ORB (TAO)~\cite{Schmidt:97q}.  CIAO's architecture
(shown in Figure~\ref{fig:ciao-dance}) is designed based on (1)
patterns for composing component-based middleware and
(2) reflective middleware techniques to enable mechanisms within the
component-based middleware to support different QoS
aspects. 

We also make use of the deployment facilities in the middleware.  In
Lightweight CCM, component assemblies are deployed and configured via
the OMG D\&C~\cite{DandC:03} specification, which manages the mapping
of application components onto nodes in a target environment.  The
information about the component assemblies and the target environment
in which the application components will be deployed are captured in
the form of standard XML descriptors.  To support automatic deployment
and configuration of components based on their descriptors, we
developed the {\em Deployment And Configuration Engine}
(DAnCE)~\cite{Schmidt:05y}, whose architecture is shown in
Figure~\ref{fig:ciao-dance}.  

\begin{figure}[!htb]
  \centering
  \includegraphics[width=\columnwidth,keepaspectratio]{ciao-dance}%
  \caption{CCM Implementation and DAnCE Architecture}
  \label{fig:ciao-dance}
\end{figure}

Our middleware enhancements comprises the following:
\vspace{-.1in}
\paragraph{Reconciling object reference semantics with GCS.}
In a GCS all members look alike to the outside world, which implies
that a single object reference (IOR) should be available to clients.
The concept of interoperable group reference (IOGR) does not work in
GCS since the semantics of a group reference are not understood by
GCS.  This required our middleware to create exactly the same IOR for
every replica even though they are different objects (or components).
Moreover, when a new replica joins a group to maintain the quorum, we
require it to have the same IOR exposed to the GCS.  We modified the
portable object adapter within TAO to use the {\tt USER\_ID} and {\tt
  PERSISTENT\_ID} POA policies.  Moreover, the user id is chosen based
on the unique instance name given to every replica and which
corresponds to the group id used by the GCS for that replica group.
We achieve this effect in a seamless manner without manual
programmatic effort by delegating the job of configuring the policies
on the objects (or components) using the DAnCE engine and supplying it
with the right set of XML descriptors.

\vspace{-.1in}
\paragraph{Reconciling procedural and OO models in state synchronization.}
The MEAD FT framework is a library that uses procedural programming
while TAO/CIAO are OO frameworks dealing with objects/components.
MEAD works only at the process level.  This disparity required us to
devise solutions that seamlessly interworked between MEAD and
CIAO/TAO.  We used DAnCE engine as the enabling technology.  Within
DAnCE, the NodeApplication is a process, which performs the job of an
application or component server.  We interfaced MEAD with the
NodeApplication process and provided two global functions called {\tt
  get\_state} and {\tt set\_state}.  MEAD cannot differentiate the
state of individual components (or objects).  As a result, MEAD
invokes the global functions in the NodeApplication process passing it
the group id of the replica group.

Within the CIAO/TAO framework, we implemented the state management
functions that can be invoked on the components (objects).   Rather
than requiring the component developers to change their IDL
descriptions of the components, which in turn will require recreating
the stubs and glue code, we only require the component executor (or
object servant) to implement the component-specific state management
functionality.

Within the NodeApplication process the implementation of the state
management global functions take care of coalescing the marshaled
state from the replicas and passing it to MEAD during a get operation,
and the other way round when setting the state.  To do this, MEAD
simply leverages the DAnCE's domain application manager, which in turn
instructs all the NodeApplication processes to get/set the state
during recovery.


\subsection{Non-CORBA Fault tolerance: MySQL}
\label{sec:mysql}

Replicated data bases were provided by a nearly stock MySQL clustering
service. In a stock MySQL clustering solution the emphasis is on data
consistency. To this end if one of three cluster data bases finds
itself alone it assumes that the other two data bases are still up and
it has just lost connectivity to them. In order to maintain
consistency in what it perceives to be a network partition it will
shut itself down. This is obviously the wrong behavior if two of three
pools have been destroyed the last DB should not shut itself
down. Given that our fault-model assumes there will be no partitions
we want the DBs to continue running no matter how many have been
lost. We modified the cluster servers to not kill themselves when they
detect a loss of a majority of the servers. 

\subsection{Putting it Together}
\label{sec:solve-puzzle}

Figure~\ref{fig:replmlrm} shows the MLRM replicated across three pools of 
computational resources.  The top-level elements are replicated across 
all three pools.  This way even a large scale failure of a group of machines
will not result in the failure of the MLRM functionality.  Pool level 
MLRM elements do not need to be replicated in the same way, since there is
by definition already instances in each pool.  They are gateway
elements though, since they have to interact with the replicated top-level
elements and the node level elements.

\begin{figure}[!htb]
  \centering
  \includegraphics[width=0.7\columnwidth,keepaspectratio]{ReplMLRM}%
  \caption{Replicated MLRM}
  \label{fig:replmlrm}
\end{figure}


%%% Local Variables: 
%%% mode: latex
%%% TeX-master: "middleware-2006"
%%% End: 
