%% This is an example first chapter.  You should put chapter/appendix that you
%% write into a separate file, and add a line \include{yourfilename} to
%% main.tex, where `yourfilename.tex' is the name of the chapter/appendix file.
%% You can process specific files by typing their names in at the 
%% \files=
%% prompt when you run the file main.tex through LaTeX.
\chapter{Introduction}

By harnessing the power of tens, hundreds, or even thousands of
individual computers, it is possible to solve many problems that might
not be tractable on a single system.  Such distributed systems are a
compelling application of large networks.  These systems usually work
on a heterogeneous network by design; large networks rarely consist of
computers that all share the same configuration of hardware and
software.  To effectively scale a distributed system, it is usually
necessary for the system to perform the same computation on many
different host configurations.  By partitioning data across the nodes
in the network, it is possible to improve the performance of many
algorithms.  Some algorithms, such as those that feature a high level
of interdependency within the data they process, are not amenable to
such division of labor.  For these algorithms, another approach to
their distribution is required.

Distributed systems often maintain only a basic level of history for
the data they process.  This may include simply logging the particular
node to which a piece of data was delegated for processing.  Complex
data histories may be constructed in distributed systems, but this
often requires careful consideration of the role and need for this
``data provenance'' at design-time.  It would be more efficient from a
software design perspective if a system was able to handle such
provenance without significant modification of existing code.

\section{Heterogeneous Distributed Systems}

As the archetypal large network, the Internet has enabled the
construction of many distributed computing efforts.  Architectures
like SETI@home \cite{anderson2002seti}, the related BOINC platform
\cite{anderson2004boinc}, or even Google's internal usage of MapReduce
\cite{dean2004mapreduce} demonstrate the value that the Internet has
added to distributed systems.  By allowing the many machines connected
to the Internet to work together, many otherwise intractable problems
may be solved.

Whether on the Internet, as in the case of the former two systems, or
in private data centers, as in Google's case, these distributed
systems all operate on heterogeneous networks.  There is typically
little standardization of the hardware and software configurations of
nodes on these networks.  Even systems that are superficially similar
may differ greatly in their actual specifications.  The ability to
adapt to these differences is a major requirement of a viable
distributed architecture.

Distributed systems usually rely on partitioning input data so it may
be processed identically on a number of nodes. This data is then
distributed across the network for processing.  Once this is complete,
results are sent to a central server that merges them to produce a
meaningful result.  Usually, the same algorithm is applied on all
nodes in the network.  Due to the heterogeneous nature of distributed
systems, this may result in compiling as many implementations of the
same code as are necessary for it to run on most hardware and software
in the network.

Partitioning data is not a practical solution for all problems we
might wish to distribute.  Highly interdependent data with a high
degree of dimensionality is relatively difficult to partition, due to
Bellman's classic Curse of Dimensionality
\cite{bellman1961processes,parsons2004clustering}.  General-purpose
reasoning is one such example of a problem that can suffer from the
Curse due to sparsity of knowledge and potentially complex
interactions of rules.  Although small-scale reasoning has been
distributed with some success \cite{urbani2009reasoning}, larger, more
complex rule-sets may not be distributed so easily.  In applying a
particular rule, a general-purpose reasoner may depend on a number of
logical statements to conclude a result.  Any particular partitioning
of input data must respect these interdependencies; partitioning data
to account for this may be a costly operation.  Furthermore, these
interdependent facts must be able to fit into a single data partition.

One of the best-known algorithms for forward-chaining reasoning, the
Rete algorithm \cite{doorenbos1995thesis,forgy1979rete}, suggests a
possible solution to the data partitioning problem.  In the Rete
algorithm, each sub-pattern of a rule is matched against the knowledge
base separate from every other sub-pattern.  The results of a
particular pattern-match are then merged with other results in a
manner that unifies bound variables, and a rule may be applied.

In practice, the process of pattern-matching is treated as an entirely
separate operation from that of merging the results of these
pattern-matches.  Unlike traditional distributed computing approaches,
the Rete algorithm effectively divides the \textit{algorithm} into
smaller, more tractable parts, rather than the data.  By applying this
concept to distributed systems we may resolve the data partitioning
problem by bypassing the problem altogether!  Instead of partitioning
data across a network of hosts running an identical algorithm, we
might try to partition the algorithm itself across the nodes.

We must be careful not to blindly apply such an approach on an
existing distributed architecture as this might prove to be
impossible.  Existing distributed systems have usually assumed a
strong client-server relationship.  Although they may have been built
with the weak-connectivity of the Internet in mind, systems like BOINC
and MapReduce still require an algorithmic client to communicate with
a central server upon starting or completing a particular subtask.
This is usually a desirable feature, as these systems were constructed
with data partitioning in mind.  By separating the data into tractable
chunks, only the initial data and final results need to be
communicated between a client and a server, making such a system
quite feasible.

This client-server relationship is not so desirable when we try to
partition an algorithm instead.  Algorithm partitioning requires
communication between nodes performing the different parts of a given
algorithm.  In order to implement a distributed system that
distributes the algorithm rather than the data, we must be careful to
use a decentralized model for intra-node communication rather than a
centralized one.  Different parts of the algorithm may need to receive
or send results to each other.  Relying on a central server to relay
such communications may be infeasible for large networks.

\section{Retrofitting Provenance}

As data partitions are assigned to different nodes in a traditional
distributed system, it is important to track how that data was
generated and manipulated.  Distributed systems must usually be
constructed with redundancy in mind; nodes performing a distributed
computation may perform an incorrect calculation or fail completely.
Thus, most systems, including SETI@home \cite{anderson2002seti}, BOINC
\cite{anderson2004boinc} and MapReduce \cite{dean2004mapreduce}, all
maintain at least a minimal level of the history of how data was
generated and manipulated, also known as provenance.

This minimal level of data provenance is generally collected to build
redundancy and resilience into a distributed system.  To defend
against nodes that return results for data that they did not actually
process, it is necessary to keep track of what data has been assigned
to each node.  Similarly, it is necessary to know when the data was
assigned to a node, so that the server may be able to double check the
results from a node or account for a node that may have failed to
return its results.

Unfortunately, this provenance may be insufficient for many more
complex tasks, as we may only be tracking which node was given what
data.  Although such basic provenance could be used to derive a more
complete history of data in the system, the quality and detail may be
unsuitable for more complex provenance use cases.

Complex, non-linear provenance has been integrated into
workflow-management software for grid computing
\cite{simmhan2005provenance}, but these systems have often been
designed by including provenance handling as a major software design
requirement.  Handling of complex provenance is generally more
difficult to add to systems after they have been initially designed.
Integrating provenance into the system design may also lead to a
less-flexible product, and extensive retrofitting may be needed to add
support for data provenance to applications running on distributed
systems that were not built to explicitly handle provenance.

A distributed system should ideally require minimal code redesign to
retrofit provenance into any application that uses the system.
Minimizing the cost of adding provenance would allow end-users and
developers to more easily integrate provenance in their systems,
especially if provenance handling may be deferred to the system
itself.  By transparently managing provenance on the programmer's
behalf, we may be able to reduce the complexity of applications
designed to use the distributed system as well as the number of
software bugs encountered when using provenance.

% TODO: Cite constant bug probability.

\section{Introducing Distributed Propagation}

One programming model for concurrent systems, data propagation
\cite{radul2009art}, offers a solution to both of these problems.
Data propagation is a modular, implicitly concurrent computational
architecture that connects state-bearing ``cells'' with monotonic
computational ``propagators'' that do not discard accumulated state.
By distributing the same computational propagator across the various
nodes of a network, it is possible to perform traditional distributed
computation, where the same operation is performed on all nodes.
Unlike traditional distributed systems, propagator networks also
enable more complex division of processing, as there are no
restrictions on the structure of propagator networks.

Propagation does not depend on centralized coordination of
computational resources; it is easy to decentralize a propagator
network and perform computation without any further involvement of the
host that started computation.  Application of several constraints on
propagator networks facilitate decentralized synchronization of data
across a network.  Although the initial presentation of propagator
networks assumes a tightly-linked computational network
\cite{radul2009thesis,radul2009art}, the set of constraints I propose
frees us from this assumption.  Instead, we need only assume reliable
broadcast and knowledge-convergence on a weakly-connected network.
This has already been proven by Awerbuch \cite{awerbuch1984reliable}
and Demers \cite{demers1987epidemic}.

Propagators may also facilitate the addition of provenance into
existing software designs.  Propagator networks are easily extended
and may be modified algorithmically due to their strong modularity.
We need only add several cells and propagators to handle data
provenance within an existing propagator network.  There is no need to
integrate any additional primitives within propagator networks to
handle provenance, nor is it necessary to fundamentally redesign
existing software to add provenance to a distributed propagator
network.  This results in obtaining provenance support ``for free''
with existing software codebases.

I have constructed an architecture that maintains and executes
distributed propagator networks, named DProp.\footnote{DProp is
  currently available in a Mercurial repository at
  http://dig.csail.mit.edu/hg/dprop/.}  Along with this architecture,
I have also built an accompanying library, PyDProp, that may be used
to easily add provenance to existing software making use of DProp.
Together, DProp and PyDProp may be used to construct any number of
interesting distributed propagator networks.  I am able to demonstrate
its viability not only for distributed processing, but also in adding
provenance to existing distributed applications that use DProp.

My protocol for distributed propagation has been designed to build on
the network architectural model known as Representational State
Transfer, or REST.  By basing the protocol design on REST, my protocol
for distributed propagation achieves a flexibility that might not be
possible in other implementations of such a protocol.  Several
constraints of a REST-based model not only map nicely onto the model
of data propagation, but also lend themselves to the efficient
implementation of distributed propagation that I describe.

% TODO: It would be good to give a negative comment about what my
% system can't do in this introduction.

\section{Thesis Overview}

In this thesis, I describe my proposed distributed propagation and
provenance system as follows:

Chapter~\ref{ch:background} introduces the concepts of data
propagation and the Representational State Transfer architectural
design.

Chapter~\ref{ch:design} details the design choices that influenced my
implementation of distributed propagation.

Chapter~\ref{ch:provenance} discusses how we may to extend existing
propagator networks to handle provenance.

Chapter~\ref{ch:issues} offers a discussion of some potential issues
with distributed propagation and provenance in the system I have
described.  I also discuss how I resolve these issues.

Chapter~\ref{ch:related} relates this work to other work relating to
data synchronization and distributed provenance.

Chapter~\ref{ch:summary} provides a summary of the
contributions of this work, as well as potential future work that may
be used to expand the distributed propagation system proposed here.

Finally, Appendix~\ref{ch:dprop} offers a more detailed description of
my implementation of distributed propagation and provenance, DProp.
