\ifdraft
\svnInfo $Id$
\fi

The central problem for a content-sharing system is searching: it must
be able to translate a search query from a user into a list of files
that fit the description and a method for obtaining them. In Arpeggio,
we concern ourselves with file \emph{metadata} only.  As discussed
previously, each file shared on the network has an associated set of
metadata. We only allow searches to be performed on the small amount
of metadata associated with the file, rather than the contents, which
are potentially much larger.

Limiting searches to file metadata is an important restriction that
enables us to use a number of algorithms that would not be suitable
for full-text search of documents. Indeed,
Li~et~al.~\cite{li03:_feasib_of_peer_to_peer} investigated the
communications costs that would be required to perform full-text
search of the web using a distributed index, and found that the size
of the data set made the problem infeasible using current
techniques. Our work does not contradict these findings; rather, it
addresses a different problem. Peer-to-peer indexing for metadata
remains feasible, because the datasets we consider have a small number
of metadata keywords per file (on the order of 5--10), which is much
smaller than the full text of an average Web page.

\section{Design Evolution}
\label{sec:indexing:design-evolution}

Our indexing techniques are designed to ensure that the system scales
with the addition of new nodes, and to ensure that the indexing cost
is shared equally among the nodes rather than causing one node to bear
a disproportionate amount of the load. To justify our design
decisions, we present the evolution of our indexing system as a series
of possible designs, exploring the problems with each and how to
address them.


\subsection{Inverted Indexes}
\label{sec:indexing:design-evolution:inverted-indexes}

The standard building block for performing searches is, of course, the
\emph{inverted index}: a map from each search term to a list of the
documents that contain it. Such indexes can be searched by scanning
over all entries in an index to identify documents that match the full
query, or finding the intersection of the indexes corresponding to
each search term.

There are two main approaches to distributing an inverted index. In
\emph{partition-by-document} indexing, each node in the system stores
all of the inverted index entries corresponding to some subset of the
indexed documents. To perform a query, the system must contact a set
of nodes that collectively hold index entries for every document; if
index entries are not replicated, this means sending the query to
every node in the system. In \emph{partition-by-keyword} indexing,
each node stores the inverted index entries corresponding to some
range of \emph{keywords}.

Both approaches have been employed in a number of
systems. Partition-by-document indexing is used for web searches by
Google~\cite{barroso03:_web_searc_for_planet}, and for indexing
scholarly publications in OverCite~\cite{stribling06:_overc}. Li et
al.~\cite{li03:_feasib_of_peer_to_peer} observed that naive
implementations of partition-by-document indexing require three orders
of magnitude less bandwidth to perform a search than naive
implementations of partition-by-keyword indexing. Though this appears
to be a compelling argument for a partition-by-document index, we
instead base our system on partition-by-keyword indexing. We note that
there are a number of optimizations that can be employed to reduce the
bandwidth cost, which we explore in
Sections~\ref{sec:indexing:design-evolution:partition-by-keyword}~and~\ref{sec:indexing:design-evolution:index-side-filtering}. Moreover,
partition-by-document indexing requires contacting nodes from each
index partition. In the two partition-by-document systems described
above, this is reasonable: Google's index servers are likely to be in
the same cluster, and thus enjoy low latency; OverCite has a
relatively small number of nodes, each of which holds as much as
\nicefrac{1}{2} of the whole index. In a large-scale peer-to-peer
system, it is not practical to contact so many nodes or store so much
index data on each node. Partition-by-keyword indexing, on the other
hand, maps nicely to the interface provided by consistent hashing and
distributed hash tables, allowing us to perform queries while
contacting a minimal number of nodes.

\subsection{Partition-by-Keyword Distributed Indexing}
\label{sec:indexing:design-evolution:partition-by-keyword}

A partition-by-keyword inverted index can be implemented in a
relatively straightforward way on top of a distributed hash table:
each keyword maps to the server responsible for storing its inverted
index. To perform a query, the client can fetch each index list from
the DHT, and calculate their intersection, as shown in
Figure~\ref{fig:indexing:design-evolution:partition-by-keyword:pseudocode}. No
modifications to the standard DHT \proc{Get}/\proc{Put} interface are
required.

\begin{figure}[tbp]
  \centering
  \begin{boxedminipage}{3.5in}
  \begin{codebox}
    \Procname{$\proc{Query}(criteria)$}
      \li $indexes \gets \emptyset$ 
      \li \For $keyword,\; value$ \textbf{in} $criteria$
      \li \Do $indexes$.\proc{Append}(\proc{Get}($keyword$))
      \End
      \li \Return $\proc{Intersection}(indexes)$
  \end{codebox}
  
  \end{boxedminipage}
  \caption{Pseudocode for distributed inverted index}
  \label{fig:indexing:design-evolution:partition-by-keyword:pseudocode}
\end{figure}

This naive approach scales poorly to large numbers of documents. Each
keyword index list must be transferred in its entirety; these keyword
index lists can become prohibitively long, particularly for very
popular keywords, so retrieving the entire list may generate
tremendous network traffic.

There is clearly room for optimization, as the desired intersection of
the indexes is typically much smaller than any one of the index lists,
yet in the scheme above the full index lists are transferred. Bloom
filters~\cite{bloom70:_space_time_trade_offs_in} use hashes to encode
a summary of the contents of a set in a small data structure; set
inclusion queries are probabilistic, with no false negatives but some
chance of a false positive. Reynolds and
Vahdat~\cite{reynolds03:_effic_peer_to_peer_keywor_searc} proposed a
protocol in which one index server transmits a Bloom filter of its
results to another, allowing the second to determine a conservative
superset of the intersection to send to the client. Li et
al.~\cite{li03:_feasib_of_peer_to_peer} proposed several further
optimizations, including compression of the index lists and adaptive
intersection
algorithms~\cite{demaine00:_adapt_set_inter_union_and_differ}. These
optimizations can substantially reduce the amount of bandwidth
required to find the index intersection; however, we do not employ
them in Arpeggio because our restrictions on the type of data being
indexed enables the following more effective optimization.

\subsection{Index-Side Filtering}
\label{sec:indexing:design-evolution:index-side-filtering}

Performance of a keyword-based distributed inverted index can be
improved by performing \emph{index-side filtering} instead of
performing an index intersection at the querying node or a distributed
index join. Because our application postulates that metadata is small,
the entire contents of each item's metadata can be kept in the index
as a \emph{metadata block}, along with information on how to obtain
the file contents. To perform a query involving a keyword, we send the
full query to the corresponding index node, and it performs the
filtering and returns only relevant results. This dramatically reduces
network traffic at query time, since only one index needs to be
contacted and only results relevant to the full query are
transmitted. This appears to be similar to the search algorithm used
by the Overnet network~\cite{overnet},
which uses the Kademlia DHT, though details of its operation are
difficult to find. The same technique is also used in
eSearch~\cite{tang04:_hybrid_global_local_index_for}.

\begin{figure}[tbp]
  \centering
  \begin{boxedminipage}{3.5in}
  \begin{codebox}
    \Procname{$\proc{Query}(criteria)$}
  \li $index \gets \proc{Random-Keyword}(criteria)$
    \li \Return $\proc{Filtered-Get}(index, criteria)$
  \end{codebox}
  
  \end{boxedminipage}
  \caption{Pseudocode for distributed inverted index}
  \label{fig:indexing:design-evolution:index-side-filtering:pseudocode}
\end{figure}


Though this seems like a trivial change, it represents an important
change in the role of the DHT from passive storage element to active
processing component. The procedure of transmitting the query to the
index server for execution is reminiscent of the use of searchlets in
Diamond~\cite{huston04:_diamon}, though far simpler. It requires
extending the DHT interface beyond the standard
\proc{Get-Block}/\proc{Put-Block} hash table abstraction; our
implementation accesses the underlying Chord \proc{Lookup} function to
build more powerful functions involving network-side processing. As a
result, our algorithm cannot be deployed atop an existing DHT service,
such as OpenDHT~\cite{rhea05:_opend} without modification. This
presents a deployment problem as several BitTorrent
clients~\cite{forum:_bittor_track_dht_protoc_specif_v1,azureus}
already operate their own (mutually incompatible) DHTs; we cannot use
them to store index data without modifying all existing clients. We
discuss this in Section~\ref{sec:implementation:deployability}.

\subsection{Keyword-Set Indexing}
\label{sec:keyword-set-indexing}

While index-side filtering reduces network usage, query load may still
be unfairly distributed, overloading the nodes that are responsible
for indexing popular keywords. To overcome this problem, we propose to
build inverted indexes not only on individual keywords but also on
keyword \emph{sets}: pairs, triples, etc.\ up to some maximum size. As
before, each unique file has a corresponding metadata block that holds
all of its metadata. Now, however, an identical copy of this metadata
block is stored in an index corresponding to each subset of at most
$K$ metadata terms. The maximum set size $K$ is a parameter of the
network. This is the Keyword-Set Search system (KSS) introduced by
Gnawali~\cite{gnawali02:_keywor_set_searc_system_for}.

Essentially, this scheme allows us to precompute the index
intersection for all queries of up to $K$ keywords.  It is not a
problem if a particular query has more than $K$ keywords: the query
can be sent to the index server for a randomly chosen $K$-keyword
subset of the query, and the index server can filter the results to
only send back responses that match the full query.  This approach has
the effect of querying smaller and more distributed indexes whenever
possible, thus alleviating unfair query load caused by queries of more
than one keyword.

The majority of searches contain multiple keywords, as shown in
Reynolds and Vahdat's analysis of web
queries~\cite{reynolds03:_effic_peer_to_peer_keywor_searc} and our
analysis of peer-to-peer query traffic in
Section~\ref{sec:evaluation:index-properties}. As a result, most
queries can be satisfied by the smaller, more specific
multiple-keyword indexes, so the larger single-keyword indexes are no
longer critical to result quality.  To reduce storage requirements,
maximum index size can be limited, preferentially retaining entries
that exist in fewest other indexes, i.e.\ those with fewest total
keywords.

% Network-side filtering. Explain keyword/filterable metadata.
% unnecessary network traffic is eliminated by filtering inside of the
% network instead of on the originator of the query
In Arpeggio, we combine KSS indexing with index-side filtering, as
described above: indexes are built for keyword sets and results are
filtered on the index nodes. We make a distinction between
\emph{keyword metadata}, which is easily enumerable and excludes
stopwords, and therefore can be used to partition indexes with KSS,
and \emph{filterable metadata}, which is not used to partition indexes
but can further constrain a search. Index-side filtering allows for
more complex searches than KSS alone. A user wish to restrict his or
her keyword query to files of size greater than 1 MB, files in
\texttt{tar.gz} format, or MP3 files with a bitrate greater than 128
Kbps, for example. It is not practical to encode this information in
keyword indexes, but the index obtained via a KSS query can easily be
filtered by these criteria. The combination of KSS indexing and
index-side filtering increases both query efficiency and precision.

Nevertheless, the index structure imposes limitations on the type of
queries supported. Because keywords or keyword sets must be used to
identify the index to be searched, the system cannot support general
substring queries (though it is possible to use index-side filtering
to restrict the output of a keyword search to those also matching a
substring or regular expression query). We argue that this is not a
major limitation, as it is shared by many widely-used systems that
exist today. In addition to being a limitation inherent to systems
based on partition-by-keyword distributed indexes, the use of Bloom
filters in Gnutella's query routing
protocol~\cite{rohrs02:_query_routin_for_gnutel_networ} also limits
the user to keyword queries.


\section{Indexing Cost}
\label{sec:indexing:indexing-cost}

Techniques such as KSS improve the distribution of indexing load,
reducing the number of very large indexes --- but they do so by
creating more index entries. In order to show that this solution is
feasible, we argue that the increase in total indexing cost is
reasonable.

Using keyword set indexes rather than keyword indexes increases the
number of index entries for a file with $m$ metadata keywords from
$m$ to $I(m)$, where
\begin{displaymath}
  I(m) = \sum_{i = 1}^K \binom{m}{i} =
  \begin{cases}
    2^m - 1 & \text{if $m \le K$} \\
    O(m^K) & \text{if $m > K$}
  \end{cases}
\end{displaymath}
For files with many metadata keywords, $I(m)$ is polynomial in $m$.
If $m$ is small compared to $K$ (as for files with few
keywords), then $I(m)$ is exponential in $m$ --- but this is no worse,
since $m$ is so small anyway. The graph in
Figure~\ref{fig:I-growth} shows that $I(m)$ grows polynomially with
respect to $m$, and its degree is determined by $K$.
%This trade-off is desirable because the number of \proc{Insert} requests
%on a network is smaller than the number of \proc{Query} requests.
% Cite Gummadi or some other measurement paper for this statistic.
As discussed in~Section~\ref{sec:evaluation:choosing-k}, for many
applications the desired value of $K$ will be small (around 3 or 4),
and so $I(m)$ will be a polynomial of low degree in $m$.
% A major advantage of keyword-set indexing is that the query traffic to
% nodes responsible for popular keywords will be abated by queries going
% to more specific indexes.

\begin{figure}[tb]
  \centering
  \input{figures/egraph2d.tex}
  \caption{Growth of $I(m)$ for various values of $K$}
  \label{fig:I-growth}
\end{figure}

\section{Index Maintenance}
\label{sec:indexing:index-maintenance}

Peers are constantly joining and leaving the network. In addition to
causing index servers to become unavailable, this churn also causes
new files to become available and old files to disappear as the nodes
containing them join and leave the network. Thus, the search index
must respond dynamically to the shifting availability of the data it
is indexing and the nodes on which the index resides.  Furthermore,
certain changes in the network, such as nodes leaving without
notification, may go unnoticed, and polling for these changing
conditions is too costly, so the index must be maintained by passive
means.

\subsection{Metadata Expiration}
\label{sec:indexing:index-maintenance:metadata-expiration}

Instead of actively polling for the departure of source nodes, or
expecting source nodes to send a notification before leaving, the
index servers expire file metadata entries on a regular basis so that
long-absent files will not be returned by a search.  Nevertheless,
during the expiration delay indexes may contain out-of-date references
to files that are no longer accessible. Thus, a requesting peer must
be able to gracefully handle failure to contact source peers. To
counteract expiration, source peers periodically \emph{refresh}
metadata that is still valid by re-inserting the metadata blocks for
their files, thereby resetting its expiration counter.

The choice of expiration time balances index freshness
against maintenance cost: a short expiration time means that the index
will not contain many invalid entries, but that peers must constantly
send refresh messages for all their files. Though choosing a short
expiration time for index freshness seems desirable at first glance,
we argue in Section~\ref{sec:content-distribution:postfetching} that
there can be value in long expiration times for metadata, as it not
only allows for low refresh rates, but for tracking of attempts to
access missing files in order to artificially replicate them to
improve availability.


\subsection{Index Gateways}
\label{sec:indexing:index-maintenance:index-gateways}

If each node directly maintains its own files' metadata in the
distributed index, and multiple nodes are sharing the same file, the
metadata block for each file will be inserted repeatedly. Consider a
file $F$ that has $m$ metadata keywords and is shared by $s$
nodes. Then each of the $s$ nodes will attempt to insert the file's
metadata block into the $I(m)$ indexes in which it belongs, as in
Figure~\ref{fig:gateway-diagram:without}. The total cost for inserting
the file is therefore $\bigTheta{s \cdot I(m)}$ messages. Since
metadata blocks simply contain the keywords of a file, not information
about which peers are sharing the file, each node will be inserting
the \emph{same} metadata block repeatedly. This is both expensive and
redundant. Moreover, the cost is further increased by each node
repeatedly renewing its insertions to prevent their expiration.

\begin{figure}[tb]
  \centering
  \subfloat[Without gateway]{
    \includegraphics{figures/gateway-without}
    \label{fig:gateway-diagram:without}}
  \subfloat[With gateway]{
    \includegraphics{figures/gateway-with}
    \label{fig:gateway-diagram:with}}
  \caption{Two source nodes $S_{1,2}$, inserting file metadata block
    $M_F$ with keywords $\{a,b\}$ to three index nodes $I_{1,2,3}$,
    with and without a gateway node $G$}
  \label{fig:gateway-diagram}
\end{figure}

To minimize this redundancy, we introduce an \emph{index gateway} node
that aggregates index insertion requests. Index gateways are not
required for correct index operation, but they increase the efficiency
of index insertion using the standard technique of introducing an
intermediary. With gateways, rather than directly inserting a file's
metadata blocks into the index, each peer sends a single copy of the
block to the gateway responsible for the block (found via a Chord
\proc{Lookup} of the block's hash), as in
Figure~\ref{fig:gateway-diagram:with}. The gateway then inserts the
metadata block into all of the appropriate indexes, but only if
necessary.  If the block already exists in the index and is not
scheduled to expire soon, then there is no need to re-insert it into
the index. A gateway only needs to refresh metadata blocks when the
blocks in the network are due to expire soon, but the copy of the
block held by the gateway has been more recently refreshed.

Gateways dramatically decrease the total cost for multiple nodes to
insert the same file into the index. Using gateways, each source node
sends only one metadata block to the gateway, which is no more costly
than inserting into a centralized index. The index gateway only
contacts the $I(m)$ index nodes once, thereby reducing the total cost
from $\bigTheta{s \cdot I(m)}$ to $\bigTheta{s + I(m)}$. For files
that are shared by only a single source node, using the gateway only
negligibly increases the insertion cost, from $I(m)$ to $I(m) +
1$. Another benefit is that it spreads the cost of insertion among the
nodes in the network. A common scenario is that a node joining the
network will want to register all of the files it has available with
the index. Without index gateways, it would have to contact every
index server; with index gateways, it only needs to contact the
gateways, and the cost of contacting the index servers is spread
amongst the gateways --- which, thanks to consistent hashing, are
likely to be distributed equally around the network.


\subsection{Index Replication}
\label{sec:indexing:index-maintenance:index-replication}

In order to maintain the index despite node failure, index replication
is also necessary. As described in Section~\ref{sec:chord}, DHTs
typically accomplish this by storing multiple copies of data on
replicas determined via the lookup algorithm, or storing erasure-coded
fragments of the data in the same way. The choice between erasure
coding and replication affects not only the probability of data loss
during node failures, but also performance: replication can provide
lower read latency because only a single replica needs to be
contacted~\cite{dabek04:_desig_dht_for_low_laten}. Because metadata
blocks are small and our application requires reading from indexes to
have low latency, we opt for replication over erasure
coding. Moreover, it is not straightforward to implement server-side
processing elements such as index-side filtering in an erasure-coded
system.

Furthermore, because replicated indexes are independent, any node in
the index group can handle any request pertaining to the index (such
as a query or insertion) without interacting with any other
nodes. Arpeggio requires only \emph{weak consistency} of indexes, so
index insertions can be propagated periodically and in large batches
as part of index replication.  Expiration can be performed
independently by each index server.




% References to dig up

% Klemm Lindeman Vernon Waldhorst Characterizing the query behavior in
% p2p file sharing systems, IMC 04

% Rasti Stutzbach Rejaie On the long-term evolution of the two-tier
% gnutella overlay IEEE Global Internet Symposium 06

%%% Local Variables: 
%%% mode: latex
%%% TeX-master: "thesis"
%%% TeX-command-default: "rubber"
%%% End: 

% LocalWords:  metadata DHTs DHT lookup
