\chapter{A Design for Distributed Propagation}

\label{ch:design}

There are several different ways we may construct a distributed
propagator network, based on where one draws the boundary between each
node on the physical network.  For example, we may draw boundaries
between cells and the propagators that depend on them, which would
imply that cells would notify propagators on other hosts.  We could
also draw boundaries between propagators and the cells they update,
which would imply that propagators would send update messages to other
hosts.

Regardless of how the propagator network is divided, propagators and
cells \textit{must} reside on the nodes of the network, and not in the
network itself.  The physical infrastructure of a network is incapable
of performing computation or storing content; it may only serve as a
web of links across which data may be transferred.  As a result, only
the edges between propagators and cells may be placed on the physical
network.

Although we cannot place propagators in the network, we may still
construct propagators that contain a network link within them.  This
network ``bridging'' may be derived from our understanding of compound
propagators.  Just as a single compound propagator ``contains'' a new
network of propagators and cells, we could easily think of the
inverse, in which a collection of propagators and cells is turned into
a compound propagator.  Since the edges that connect propagators and
cells may be placed across a physical network, we may construct a
compound propagator that includes such an edge, effectively
``bridging'' the network.  As a result of this, we may also assume
that \textit{any} propagator may cross the network, as
compound-propagators are normally opaque.

My approach to constructing distributed propagation framework builds
on this basic concept of propagators that bridge the network.  I treat
HTTP communication as part of a network-spanning propagator designed
to maintain identical copies of a cell on multiple hosts.  I do this
by intercepting update messages sent to a local cell as part of its
merge operation and broadcasting the intercepted messages to all other
hosts with a copy of that cell.  I then may ensure convergence of this
mechanism by occasionally attempting to re-synchronize all remote
copies of a cell to ensure that no updates were lost on the network.

\section{Constraining the Merge Operation}

\label{sec:mergeprops}

Before we may consider our specific implementation of this approach to
distributed propagation, we must first determine exactly what costs
are incurred by simply copying the update messages and sending them
across a network.  Implementing propagation over a network introduces
uncertainty in the order of updates and lacks guaranteed message
reception.  Unfortunately, propagator networks may have issues with
messages that are in the wrong order, and the behavior of these
networks assumes that messaging is reliable.  As a result,
propagation-based computation may not be deterministic over a computer
network without ensuring that certain properties are present in the
data propagation model.  In order to remedy these issues and ensure
deterministic computation, we must enforce four properties on the
merge operations used by cells that receive messages from other hosts:

\begin{enumerate}
\item \textbf{Idempotence:} If we already know a fact and are told it
  again, we do not know anything new.  Thus, a cell's contents should
  appear the same no matter how many times a particular update is
  received, provided that it has been received at least once.
\item \textbf{Monotonicity:} A cell never forgets.  This means that a
  cell may never ``unmerge'' an update.  A cell may be able to mark
  the results of an update as ``contradictory'' or ``disproven'', as
  both changes would refine the content of the cell, but it should not
  be able to undo the original update.
\item \textbf{Commutativity:} A cell's contents should be independent
  of the order of the updates that were merged to create them.  If we
  are told ``it's a warm day'' and ``it's a sunny day'', we will know
  both of these facts regardless of the order in which we are told.
\item \textbf{Associativity:} A cell that merges two messages should
  have the same contents as a cell that merges a single message that
  is the ``sensible merge'' of the two messages.  For example, a cell
  that is told ``$T>40^{\circ}$F'' and ``$T>50^\circ$F'' should appear
  the same as if it had only been told the latter, as the potential
  values of $T$ implied by $T>50^\circ$F are a subset of the
  values of $T$ implied by $T>40^\circ$.
\end{enumerate}

These constraints do not impose any restriction on the number of
\textit{effective} merge operations that may be implemented; merge
operations that do not adhere to one of these constraints may be
modified so that they do so.  For example, a non-idempotent operation
may be ``retrofitted'' to become idempotent by adding a unique
identifier as part of each update.  If this unique identifier is
stored with the data following the merge, a propagator reading the
cell may modify its ``perception'' of a cell's contents based on the
number of distinct identifiers of the otherwise duplicate updates.

Operations lacking the other constraints may be modified similarly.
Extensive research in the fields of computer networking and
distributed databases has attempted to meet the need for serialization
of communications (implying non-commutativity and, indirectly,
non-monotonicity) despite the lack of guaranteed message reception
offered by Internet
protocols.\cite{cerf1974tcp,birrell1982grapevine,terry1995bayou,singhal1990updatetransport}
Packets on the Internet may be received out of order or even not at
all.  Any application desiring serialization of messages, such as
distributed databases, must be able to deal with the commutativity of
messages on the Internet.  We may apply the results of this research
in a more general fashion to enforce commutativity constraints on
non-commutative functions.

\section{Choosing a Distribution Application Model}

\label{sec:netarch}

Like many other network applications, distributed propagators could be
implemented using one of two primary distributed application models:

\begin{enumerate}
\item Cells could be managed using a \textit{client-server model}.  In
  such a model, a particular cell would be maintained by a single
  host, the server.  All other hosts interested in the cell would be
  clients of the server.

  As the maintainer of the cell, the server would be the only host to
  decide whether or not an update is accepted.  It would also be
  responsible for merging all data into the canonical version of the
  cell that it stores and would update all interested clients when the
  contents of the cell changed.  As clients would not have local
  copies of the cell, any client that needed to read the cell's
  contents would need to request the server's canonical version.
  Although clients are able to cache the contents of the cell, these
  caches may only contain static content; they may not be modified, as
  only the server controls what is merged into the cell.
  
  Implementing a client-server model is easy when compared to the
  complexity of the peer-to-peer model described below, but it is
  highly reliant on a single server to perform merges and to service
  other requests.  Failure of the server could potentially cause a
  cascading halt to all computation in the propagator network, as
  propagators would no longer be able to modify the contents of the
  cell, and, without a cache, no propagator would be able to read the
  contents of the cell either.
\item Cells could alternately be managed using a \textit{peer-to-peer
  model}.  Every cell would be duplicated across all hosts interested
  in the cell, called the peers of the cell.  Unlike the client-server
  model, no single host would have special status.  Rather than
  allowing one server to handle updates and merge the cell, each peer
  does so for its \textit{own} copy of the cell.  As long as all peers
  implement the same merge operation on the cell and may reliably
  forward any update messages generated by local propagators to remote
  peers, an update should eventually reach all hosts and the cell's
  contents should converge across the network.
  
  A peer-to-peer model requires a more complex implementation than the
  client-server model; there are no explicit constraints on the
  network topology of a peer-to-peer system.  In contrast, a
  client-server model enforces a star-like topology on the network.
  In exchange for this additional complexity, peer-to-peer systems
  have a robustness that is not present in a client-server model.  In
  a peer-to-peer model, a node failure need not halt all computation.
  As each peer is in charge of maintaining its own copy of the cell,
  the peer may continue updating and referring to that copy even if
  that peer is completely isolated from the remainder of the network
  by a node or network failure.
  
  Choosing a peer-to-peer architecture does not mean that we must
  abandon a RESTful design for our application.  Although REST assumes
  a client-server model, a peer-to-peer distributed propagator network
  may be viewed as a virtual network that operates \textit{above}
  traditional RESTful communications.  Rather than abandoning the
  client-server restriction of REST, each peer effectively runs
  separate client and server applications that happen to share the
  same stored resource, but do not interact at all in any other
  respect.  This is not all that different from how a representation
  of a resource in existing RESTful systems may be generated in part
  by fetching an external resource.
\end{enumerate}

It is important to note that neither model requires timely
communication in a distributed propagator implementation.  The
commutativity constraint on the cell merge operation, as described in
Section~\ref{sec:mergeprops}, ensures that the contents of the cell
will be the same, regardless of the order in which messages are
received.

This lack of a timeliness constraint may seem strange because it
allows global inconsistencies between the cells being synchronized.
Propagator networks admit such inconsistencies by design.  Progress
can still be made on a partitioned problem with only local
consistency, and each individual global inconsistency will eventually
be resolved.  Ensuring ultimate convergence is up to the programmer.

Rather than ensuring timeliness, we need only guarantee that any
update will \textit{eventually} arrive at every other cell in the
network.  That said, some timeliness guarantees may be appropriate in
practice to ensure that results are generated in a reasonable
time-frame.

The choice between a client-server model and a peer-to-peer model is
ultimately a tradeoff between system complexity and fault tolerance.
As we seek to build an architecture that is comparable to existing
distributed systems, we must consider that, as the number of nodes in
a network increases, so do the chances that at least one node will
fail.  This means that fault tolerance is likely to be of greater
benefit than simplicity of the system's design.  As a result, I have
chosen to implement distributed propagation using a peer-to-peer
model, rather than a client-server model.

\section{Resource Representations}

\label{sec:resources}

Now that I have established the basic distributed application model to
be used for distributed propagation, we must now examine propagation
with respect to the architectural model of REST.  As resources and
their representations are the central focus of a RESTful design, it is
important to identify the types of resources and representations used
in the system.  As mentioned previously, there are two primary
components of the propagator paradigm that may be candidate resources
in a RESTful implementation: cells and propagators.  One additional
construct of propagation, that of ``update messages'', may be a third
possible resource type.  Finally, as a peer-to-peer system must be
able to keep track of the neighboring peers of a node, we may also
consider this collection of peers to be a fourth potential resource
type.  I examine each of these potential classes of resources in turn.

Propagators are the class of object least suited to being turned into
a resource.  Although we may be able to represent a propagator using a
basic description of a propagator, this seems to be of limited use
within distributed propagation.  Outside of enabling introspection of
the propagator network, there is not much need to actually manipulate
propagators themselves.  As propagators are opaque computational
blocks rather than entities that may be directly manipulated, it seems
more appropriate to not include them as resources within the system.

Cells seem to be a far more appropriate class of resources in a
RESTful distributed propagator design.  Not only do they bear state,
as resources do, but they are the \textit{primary} source of state in
a propagator network.  Cells are also more closely linked with
communication of data than propagators are, as they are refined by way
of updates that are sent to them.  Together, these facts suggest that
we should represent cells as resources.  Cell updates, on the other
hand, are tied to a specific cell and have no modifiable state.
Although we could model updates as a resource associated with the cell
itself, it seems more appropriate to consider them to be a ``partial''
representation of a cell, due to their use in actually modifying the
cell's state.

Peers also appear to be an appropriate class of resources in our
system.  In order to maintain a peer-to-peer network, it is necessary
to have as much redundancy of communication as possible to route
around temporary network failures.  This means there must be a
mechanism for \textit{node discovery}.  I do not address the issue of
discovery of the initial node and treat it as out of the scope of this
thesis.  Instead, when I refer to node discovery, I mean the discovery
of additional nodes already present on the peer-to-peer network.  To
enable such discovery mechanisms, we must have a way of representing
the peers in a network, and we thus consider peers to be another set
of resources in our system.

\section{Naming Cells}

\begin{table}
\begin{center}
\begin{tabular}[c]{llp{6.5cm}}
Relative URL & Description \\
\hline
\texttt{/Cells} & The collection of all distributed cells on the server. \\
\texttt{/Cells/\{UUID\}} & The distributed cell identified by \textit{UUID}. \\
\texttt{/Cells/\{UUID\}/Peers} & The collection of peers of the cell identified by \textit{UUID}.\\
\hline
\end{tabular}
\caption{The URL structure used by DProp.}
\label{tab:urls}
\end{center}
\end{table}

\begin{figure}
\begin{center}
\includegraphics{distributing-propagator-network.pdf}
\rule[5pt]{12cm}{1pt}
\includegraphics{distributed-propagator-network.pdf}
\caption{Distributing propagator networks by dividing shared cells.}
\label{fig:distprop}
\end{center}
\end{figure}

The issue of cell naming in our peer-to-peer RESTful model is not a
simple one.  A RESTful architecture identifies resources using some
globally-unique identifier.  In implementations that use HTTP, this
identifier comes in the form of a URL, or Uniform Resource
Locator.\cite{rfc1738} URLs provide for such identifiers by
concatenating a globally unique server name with a unique name of the
resource local to the server itself.  URLs allow us to construct a
hierarchical ordering of resources, as they are built on the concept
of ``paths'' on a server.  We use such a hierarchical structure to
associate the collection of a cell's peers with that cell, as
described in Table~\ref{tab:urls}.

% TODO: the other image.

A complication arises when we need a way to identify multiple copies
of the same resource, such as when we wish to identify all copies of a
cell that has been copied to multiple peers.  When I discuss
distributing a propagator design, like that depicted at the top of
Figure~\ref{fig:distprop}, we end up creating copies of cells for each
host, and synchronizing them, as at the bottom.\footnote{In
  Figure~\ref{fig:distprop}, synchronized cell copies are depicted by
  clusters of cells connected by dotted lines.}  In practice, the URLs
for each copy of a cell identify the \textit{specific} copy of a cell
(e.g. $\mathrm{a_A}$ or $\mathrm{b_C}$); they do not identify which
original cell (e.g. a or b) a cell copy is associated with.

A na\"{\i}ve solution that resolves this problem relies on having,
\textit{a priori}, a globally unique identifier for the collection of
cells as a whole.  Although we could use a URL for this identifier, we
may not have a server name prior to constructing the cell; we may be
creating the cell on a host that is temporarily disconnected from the
network.  Instead of using URLs then, I choose to use universally
unique identifiers, or UUIDs \cite{rfc4122}, to identify each
collection of cells.  These UUIDs are meant to be globally unique,
like URLs, but they do not require a namespace to be associated with
them.

We integrate these UUIDs into the URLs we construct to identify each
copy of a cell.  Although this is not strictly necessary, as we could
construct a mapping from URLs to UUIDs, this eases the task of
uniquely identifying cells within the server namespace.  It also
allows for the easy derivation of the UUID of a cell from the URL of a
copy of a cell.  We use this approach when initializing a cell.

\section{Maintaining the Network}

As mentioned in Section~\ref{sec:resources}, we consider peers to be
resources in a RESTful architecture, so that we may maintain the
peer-to-peer network using HTTP.  By tracking the peers that a
particular cell has been copied to, a particular peer will know which
other peers hold copies of a cell that it must forward updates to.
Making the collection of known copies of a cell available via HTTP
also allows new peers to easily connect to an existing network by
simply adding their copy to the set of known cell copies on an
existing peer with a simple POST or PUT request.  They may also
bootstrap their knowledge of copies of the cell by requesting the set
of copies from the existing peer.

The set of peers that hold copies of a cell may be independent of
those holding any other cell in the propagator network it belongs to.
This means that the set of peers is only dependent on the cell itself.
We may model the collection of peers as a single resource named
``Peers'' associated with each cell.  In fact, as long as we do not
need to modify each peer itself, and only need to know their URLs, we
may model the collection of cell copies as a single resource, merely
updating the list of cell copies that make up that resource as needed.

Each peer is responsible for maintaining its own list of other copies
of a cell; it may choose to keep track of any number of copies of the
cell.  Thus, the topology of the peer-to-peer network of cell copies
may take any shape.  Some topologies may be more preferable than
others.  Topologies that are eventually connected are required to
ensure the convergence of cell contents and that all propagators will
fire; cell contents may not globally converge if update messages
cannot propagate to all peers sharing a cell.  Increasing the number
of peers visible from a single node is also preferable: network
diameter will decrease, and update messages will be more timely.  It
will also increase fault tolerance, as there is added redundancy of
the update messages sent out.  For this reason, I choose to maintain a
clique in my model of distributed propagation.

By assuming that all peers are connected in a clique, we gain several
benefits.  As we may assume that every node is a distance of 1 from
every other node, we may avoid implementing message routing mechanisms
in our network.  In a clique, sending a message to all neighboring
nodes means that we have forwarded an update message to \textit{all}
nodes.  However, even if temporary network fluctuations result in a
network topology that is not a clique, I also implement a
synchronization mechanism that will still ensure global convergence.
If a clique is not used with the algorithms I use for distributed
propagation, it is not possible to guarantee timeliness of message
arrival for nodes that are farther than distance 1 from the original
node.

% E-mail geoff

% Do we mention that the merge operation should be identical?

% Tie the two together afterwards.

% RESTAR

\section{Cell Initialization}

\label{sec:cellinit}

\begin{figure}
\begin{center}
\includegraphics{create-cell.pdf}
\caption{Creating a distributed propagator cell.}
\label{fig:create-cell}
\end{center}
\end{figure}

Now that I have established the basic architectural guidelines for
distributed propagation, how can we actually construct a cell and
initialize it?  Initialization of the cell is actually quite simple,
and is outlined in Figure~\ref{fig:create-cell}.

Cell creation begins much as it would in a local propagator network;
storage is allocated for the cell and the cell is associated with the
appropriate merge operation.  Once this is complete, we must associate
the cell with its appropriate distributed network, creating it if
necessary.

If the cell is a completely new cell for which no network exists, we
must generate a new UUID for the particular collection of cells, if we
are not given one.  We then associate this UUID with the cell.
Finally, we make the cell available as an HTTP resource by assigning
it an appropriate URL, derived from the cell's server-based namespace,
appending its UUID as in Table~\ref{tab:urls}.  Local updates may be
merged as soon as the cell is initialized, before its UUID is
assigned.  Remote connections are only possible once the URL and the
UUID have been associated with the cell.

If we are instead joining an existing collection of cells, we must
provide the URL of at least one of the existing cell copies in that
collection.  Creation of a cell from a remote copy resembles the
process of creating a local cell, except that the UUID is specified as
part of the URL provided as an argument.  Once the cell has been made
available to remote hosts, we then submit an HTTP POST request to the
``Peers'' collection of the cell initially passed to the constructor.
This POST request contains the newly minted URL of the new peer.  It
effectively adds the peer to the network so that it may receive
updates from the remote cell copy.

We then submit an HTTP GET request for the cell resource itself.  In
requesting the cell, we desire a representation of the cell's contents
as if it was an update message.  This allows us to merge the content
of the remote cell copy with any content present in the newly created
cell.  Once this merge is complete, local propagators may fire and
begin computation, as with any other merge operation.

We finish the initialization process by initializing the list of cell
copies.  We submit an HTTP GET request for the ``Peers'' collection of
the initial remote cell and then set the contents of the local
``Peers'' collection to the contents of the remote cell.  This makes
the local cell aware of all other cells in the network (assuming that
the network is currently a clique).  Finally, to ensure that the
network is fully repaired to a clique again, the cell submits HTTP
POST requests to each peer in the newly initialized set of ``Peers'',
adding its own URL to the collection of peers of that copy of the
cell.  At the end of initialization, the new peer is aware of all
peers in the network, and all peers in the network are now aware of
the new peer.

Due to the associativity and idempotency constraints that we have
placed on the merge operation of the cell, we need not ``lock'' the
cell before we have merged the contents of the remote cell or learned
its set of peers.  We may begin processing without any data whatsoever
and update the cell before synchronization with the initial remote
cell has taken place, as the contents of the cell are guaranteed to be
monotonically increasing as well!  As the merge from the
``initialization'' will still cause local propagators to fire, we need
not preclude computation before the cell becomes non-empty.  There is
\textit{one} benefit to allowing the cell to synchronize as soon as
possible, because any updates merged before the cell has fully
connected to all peers will not propagate over the network in a timely
fashion.

% More details?

% We want a diagram here.

\section{Cell Updates}

\label{sec:update}

\begin{figure}
\begin{center}
\includegraphics{update-cell.pdf}
\caption{Updating a distributed propagator cell.}
\label{fig:update-cell}
\end{center}
\end{figure}

The process of updating a distributed cell is superficially similar to
the process of updating a cell on a non-distributed propagator
network.  The cell has a merge operation that takes the particular
update and determines the new contents of the cell.  In a distributed
cell, an unmodified merge operation that would have been applied to a
cell on a non-distributed propagator network is used to complete a
generic stub method which will serve as the actual merge operation of
the cell.  This generic stub copies the received update and forwards
it to all other copies of the cell by sending a POST request to all
other copies present in the local ``Peers'' collection.  In effect,
the merge operation intercepts the update before performing the local
merge.  By forwarding the update to all other peers, we may delegate
the merge to each peer, rather than assuming that the \textit{local}
copy of the cell controls which updates are \textit{globally} merged.
This may be seen in Figure~\ref{fig:update-cell}.

A remote peer that receives an update in a POST request also performs
a merge, but uses a different stub method.  Instead of forwarding the
update to all other cells, this stub method prevents rogue updates by
confirming that the sender is actually a peer before merging the
update.  Aside from this action, the stub method behaves much as the
non-distributed merge would: it wakes up local propagators to perform
computation, just as if the update had been received from another
local propagator.

% Relate to REST and the representations.

Some of the previously mentioned constraints on the cell merge
operation prove useful here.  If we ensure that the basic cell merge
operation is the same for all copies of a cell, we need only forward
the update, rather than the new contents of the cell.  This may reduce
the amount of bandwidth used by a cell update operation.  Furthermore,
the commutativity constraint relaxes the need for timely updates or a
strict global ordering of operations.  Instead, we may receive two
remote updates in any order and not be concerned that we will lose
synchronization with any other peer.

\section{Failure Recovery Through Re-synchronization}

\label{sec:resync}

Although the proposed system appears to be relatively robust, we must
define one more operation to ensure that the architecture is truly
fault tolerant.  As described thus far, distributed propagation may
suffer from the loss of a link between two copies of a cell as this
may lead to a loss of synchronization when messages sent by one copy
are not received by the other.  If the system uses HTTP over TCP as
its underlying communication protocol, we may gain the benefits of the
fault tolerance mechanisms of TCP and reduce the chances of such
de-synchronization.  Unfortunately, simply introducing TCP is not
enough to ensure that network failures do not impact the integrity of
the content of a cell, and it may not guarantee global consistency.

Instead, we achieve these goals by implementing a recurring operation
to re-synchronize the contents of the local cell with remote copies of
that cell.  This operation sends a GET request to each remote cell,
requesting a representation of the cell as if it was an update.  It
then applies the response to the local cell as if it was an update
from the remote cell.  Because of the idempotency and associativity
constraints on the merge operation, we do not need to concern
ourselves with the fact that the representation of the cell as an
update may include the effects of updates that have already been
merged into the local copy of the cell.  Similarly, we need not worry
that the update representation might not explicitly include updates
that had no substantive effect on the cell's contents.

We may also re-synchronize the list of peers in the network in the
same way; this ensures the maintenance of the clique.  As the list of
peers is not a cell, care must be taken to correctly merge the list of
peers.  We merge the contents of the remote ``Peers'' collection into
the local collection by taking the union of the two collections.  This
ensures that the list of peers is monotonically increasing.
Unfortunately, this means that this system is unable to
\textit{remove} peers from the collection.  This is not currently a
significant issue; propagator networks do not currently permit the
destruction of existing parts of the propagator network.

Performing these occasional re-synchronization operations may consume
more bandwidth than is necessary, especially if there are no global
inconsistencies.  As we transfer the entire contents of the cell as an
update, if a cell is rather large, we may transfer a significant
amount of data, even if the update results in no change to the local
copy of the cell.  We may resolve this problem by making use of the
\textit{cacheability} constraint of RESTful architectures.  HTTP
offers several methods to control client-side caching; these methods
may be controlled through manipulation of the headers of HTTP
messages, collections of meta-data that are sent with each HTTP
request and response.

Each HTTP response to a GET request may include an appropriate
entity-tag, or ETag.  These ETags uniquely identify a particular state
of a resource, much as a hash of its contents does.  This ETag may
then be supplied back to the server in subsequent GET requests by
using the If-None-Match header.  If the ETag of the current state of
the resource on the server matches the one provided in the
If-None-Match header, the server need only respond with a simple ``304
Not Modified'' response that has no other content.  This implies that
any cached copy of the resource bearing the ETag provided in the
If-None-Match header is still valid.  If the ETag does not match, or
the If-None-Match header is not included, then the server will return
the full content of the resource and a ``200 OK'' response.

We use the ETag and If-None-Match headers to reduce bandwidth usage
during re-synchronization.  Each peer maintains a hash of the current
cell state, being the entity tag for that cell.  When a cell performs
re-synchronization, it assumes that both copies are already equal to
each other and provides its local ETag in the If-None-Match header.
The receiving cell may then return a 304 response if the ETag matches,
or the full contents of the cell if it does not.

This re-synchronization mechanism allows for processing to continue
when cells have become temporarily disconnected.  If a link to a
particular peer fails and no updates are received by a cell, both the
remainder of the network and the disconnected peer may continue to
compute, as each cell is managed and updated locally.  Once the cell
is able to reconnect to the network, the re-synchronization mechanism
will eventually reconcile any global inconsistencies that arise due to
the failure of a link.  If a node fails outright, we may simply treat
this as a special case of link failure, in which the node performs no
further computation or modification of its local cell state, and may
choose to simply reinitialize with the network.
