<commit_before>\documentclass{article}
\usepackage{url,color,xspace,verbatim,subfig,ctable,multirow,listings}
\usepackage[utf8]{inputenc}
\usepackage[T1]{fontenc}
\usepackage{txfonts}
\usepackage{rotating}
\usepackage{paralist}
\usepackage{subfig}
\usepackage{graphics}
\usepackage{enumitem}
\usepackage{times}
\usepackage{amssymb}
\usepackage[colorlinks=true]{hyperref}
\usepackage[ruled,vlined]{algorithm2e}

% ==================================================

\graphicspath{{figs/}}
\urlstyle{sf}

% tikz stuff
\usepackage{tikz}
\usepackage{pgfplots}
% configuration
\usetikzlibrary{shapes,positioning,calc,snakes,arrows,shapes}
\pgfplotsset{width=.9\linewidth}

\lstset{
  language=C,
  basicstyle=\ttfamily \small,
  flexiblecolumns=false,
  basewidth={0.5em,0.45em},
  boxpos=t,
}

\newcommand{\etal}{{\it et al.}\xspace}
\newcommand{\naive}{na\"{\i}ve\xspace}
\newcommand{\Naive}{Na\"{\i}ve\xspace}
\newcommand{\textc}[1]{{\color{gray} {\footnotesize #1}}}

\definecolor{skRed}{RGB}{155,25,25}
\newcommand{\stefan}[1]{
  {\color{skRed}[{\color{red}{SK}} #1]}}

\setcounter{section}{0} % Start sections with 1, not 0
\begin{document}

\title{Adaptive broadcast tree for multicores}

% email address
\newcommand{\eaddr}{<EMAIL>}
\newcommand{\email}{\href{mailto:\eaddr}{\eaddr}}

\author{<NAME>\\
  \email \\
  Systems Group, ETH Zurich}

\maketitle

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
\section{Introduction}

Atomic broadcast is a building-block for many higher-level distributed
algorithm such as state-machine replication. Such distributed
algorithms are increasingly used to overcome scalability challenges on
multicore machines. Examples are databases~\cite{Salomie2011,
  Wiesmann2000} and operating systems~\cite{fos:osr09, tornado:osdi99,
  barrelfish:sosp09}.

% What do you want to enable?
We want to investigate how to achieve better performance of atomic
broadcast across a wide range of multicore machines without manually
tuning the implementation to concrete machines.

% What problem are you solving, and why is it hard?
The problem is complexity and diversity of modern multicore
machines. Multicores come in many flavors depending on vendor and
generation of the machine. All of these machines have fundamentally
different characteristics such as number of nodes, NUMA
characteristics, interconnect topologies, propagation time on
interconnect and use of shared caches. These characteristics are hard
to understand, but important to consider for application performance. 

% New ideas
Instead of manually tuning algorithms to characteristics of individual
machines, we automatically configure our atomic broadcast
implementation based on a machine model.
% How will you go about it?
We will base our atomic broadcast on overlay networks, which we
automatically configure based on the machine model. One questions is
which information should actually be encoded in the machine model. %
Our idea is to then explore different approaches (such as MST
algorithms, clustering) to \emph{find good overlay networks} based on
the network model to base the atomic broadcast on. Then, for every
node, we need to decide which message to send first. This is a
\emph{scheduling} problem.

% How do we show it works
We auto-configure a wide range of different multicore machines and
show the performance of our implementation comes close to the
performance of hand-tuned implementations for all of these machines.

% Hypothesis
My hypothesis is that automatic tuning of atomic broadcast (and other
distributed algorithms) to machine characteristics is worth
doing. Obviously, we cannot achieve performance of hand-tuned
implementations, but we can get close enough to argue that loosing a
bit of performance to avoid tedious manual reconfiguration is worth if
for many applications (while others, where performance \emph{really}
matters, might still want to manually configure their stuff).

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
\section{Related work}

A multicore aware broadcast has been implemented in~\cite{Tu2008}. It
takes the cache architecture of the machine (and only one particular?)
into consideration for building the broadcast. It is however for
larger messages (starting at 4K) and it does not seem to be performing
very well (numbers are given in microseconds -6). 4K messages are 52
microseconds, which is 150000 cycles. For synchronization primitives,
a cache line is sufficient.

Another tree implementation with MPI in mind was done
in~\cite{Graham2008}. They have comparable numbers to what we found,
but they cannot automatically tune to different machine
characteristics. Our hypothesis is that we can do achieve higher
average performance across a wide range of different multicore
machines without manually adapting the implementation.

Alistarh\etal~\cite{Alistarh2012} show composition of shared-memory
algorithms that perform efficiently under different conditions at
negligible overhead. A similar approach can potentially be applied to
deal with diversity of multicore machines. However, we would have
to extend it to support both shared-memory and message-passing
implementations. 

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
\section{The multicore network model}

% selecting hard
Characteristics of the network influence the performance of
distributed algorithms. Due to the heterogeneous combinations of
complex networks on multicores, selection of algorithms is
particularly hard.

% model
We express the network characteristics in a multicore network
model. This model at this stage is by no means complete. We are going to
extend it over time to address more complexities of current and future
multicores.

% graph
We represent the multicore network as a graph. The nodes in this graph
correspond to cores, and edges the communication channels between
nodes. The machines we consider for now only have bi-directional
communication links. Other machines (like the SCC) have
uni-directional links, which is why we model the graph as directed
graph.

% shared memory -> fully meshed
This graph is fully meshed for shared memory machines. Every core can
communicate with every other node via the cache coherence protocol.

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
\subsection{Node characteristics} 
\label{sec:model_nodes}

Every node (i.e.\ core) in the system has characteristics associated
with it:

\begin{description}
\item[synchrony] Parts of a multicore machine are synchronous. Hence,
  one characteristic of a node is an identifier for the synchrony
  cluster of a node.
\item[send time $t_{send}$] Time required to send a message, i.e.\ marshalling
  and queuing the message
\item[receive $t_{recv}$] Time required to receive a message, i.e.\
  dequeuing, unmarshalling and demultiplexing a channel.
\item[failure] Can nodes fail? If so, how? (Byzantine, crash)
\end{description}

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
\subsection{Link characteristics} 
\label{sec:model_links}

The network model is defined by communication channels, which in turn
are defined by a set of attributes (here sorted by priority):

\begin{description}
\item[latency] (quantitative) between each pair of nodes (per-edge)
\item[breakdown] (quantitative) of message send cost:
  dominating propagation time vs.\ dominating send and receive time on
  multicores (per-node)
\item[bounded delivery]
\item[reliability] (loss, link failure, in-order delivery, phantoms,
  corruption, duplicated messages): 
  multicore interconnect networks often are
\end{description}

% Assumptions
We assume that the cost of sending a message to the same node is 0,
which is not true, since even using a highly optimized implementation
for local communication (like LMP) has a non-negligible cost.

% Limitations
What we do not consider yet is link congestion. 

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
\subsection{Hierarchical model}
\label{sec:model_hierarchy}

Multicore machines are hierarchical. Processor cores are grouped in
NUMA nodes with shared caches. These NUMA nodes are then connected to
each other by increasingly complex interconnect networks. 

We encode NUMA node affinity in our model. This allows to break down
programming of a multicore machine into two distinct instances. We
solve one instance of the problem between NUMA nodes, and then, in a
second step, within each NUMA node. These two steps can be executed
independently and different algorithms can be used in both case.

If cores on the same NUMA node share a cache, it makes sense to
leverage this hardware feature to implement shared-memory algorithms
on top of it. %
Communication across NUMA nodes is realized sending messages across
the packet-based interconnect network. Using message-passing based
implementations is the natural choice. %
Shared memory algorithms and message-passing based algorithms can be
composed together to program hierarchical machines similarly to what
was previously shown for composing of shared-memory
algorithms~\cite{Alistarh2012}.

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
\subsection{Acquiring the model}

% Model given
We assume that the model is given. We just define one ourselves for
now or build one manually for a particular machine. In a real-world
scenario, this does not work. The model needs to be created
automatically for a particular machine (due to the diversity of
machines, and also the pace at which hardware changes)

We now make up a machine model for evaluation. This model is based on
a 8x4x1 AMD Shanghai machine. The topology of such a machine is shown
in Figure~\ref{fig:gruyere}.

Each of the eight nodes consists of four cores. We approximate the
cost of sending messages within nodes as $1$ and across nodes as
$10*num(hops)$.

Send and receive time are $10$ each, i.e.\ in the order of propagation
over a cross-NUMA link.

\begin{figure}[htb]
  \centering
  \begin{tikzpicture}[
    xscale=.7,
    yscale=.5,
    every node/.style={draw,fill=red!10,minimum width=2cm,minimum height=1cm}
    ]
    \node (c1) at ( 0, 4) {node 0};
    \node (c2) at ( 4, 4) {node 1};
    \node (c3) at ( 8, 4) {node 2};
    \node (c4) at (12, 4) {node 3};

    \node (c5) at ( 0, 0) {node 4};
    \node (c6) at ( 4, 0) {node 5};
    \node (c7) at ( 8, 0) {node 6};
    \node (c8) at (12, 0) {node 7};

    % top horizontal
    \draw[thick] (c1) -- (c2);
    \draw[thick] (c2) -- (c3);
    \draw[thick] (c3) -- (c4);
    % bottom horizontal
    \draw[thick] (c5) -- (c6);
    \draw[thick] (c6) -- (c7);
    \draw[thick] (c7) -- (c8);
    % vertical
    \draw[thick] (c1) -- (c5);
    \draw[thick] (c2) -- (c6);
    \draw[thick] (c3) -- (c7);
    \draw[thick] (c4) -- (c8);
    % cross
    \draw[thick] (c3) -- (c8);
    \draw[thick] (c4) -- (c7);
  \end{tikzpicture}
  \caption{Interconnect topology of 8x4x1 AMD Shanghai machine}
  \label{fig:gruyere}
\end{figure}

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
\section{Atomic Broadcast for Multicores}

For a specific model, we now need to find ``the right'' implementation
of algorithms to program it. We build our atomic broadcast based on a
subset of links offered by the model such that all nodes of a given
multicore machine can be reached. Atomic broadcast guarantees that 1)
if one node receives a message, all other nodes receive it as well,
and 2) that all nodes see the message in the same order.

First, we discuss what kind of topologies we use to communicate
between nodes. We evaluate tree based implementations. Other
potentially interesting topologies are rings.

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
\subsection{Tree Topology}

We now look at tree-based atomic broadcast implementations. The first
challenge is to find efficient spanning trees for arbitrary
machines. We now discuss different strategies to select trees based on
the model representing the machine.

%-------------------------------------------------
\subsubsection{Binary Tree}

A very \naive strategy to build a tree from machine models is the use
of a binary tree with root in core 0. Every node $n$ has two children
$2(n+1)-1$ and $2(n+1)$.

The benefit of such a tree is that the time complexity is guaranteed
to be $O(\log{n})$. The downside is that it is not considering the
machine topology. Unnecessary communication across NUMA nodes might be
introduced. 

We show an example for such a tree for a 8x4x1 AMD Barcelona machine
(gruyere) in Figure~\ref{fig:qrm_tree_gruyere}.

\begin{figure}
  \input{qrm_tree_gruyere}
  \caption{\Naive binary tree of cores for broadcast on gruyere}
  \label{fig:qrm_tree_gruyere}
\end{figure}

%-------------------------------------------------
\subsubsection{Minimum-Spanning Tree}
\label{sec:mst_tree}

We can apply a minimum spanning tree algorithm to the machine model
graph to find a number of links such that these links connect all
nodes at a minimum cost.

We have implemented such an algorithm and found out, that it is not a
good solution in the general case, since it does not consider
parallelism. Naturally, we would like to parallelize send operations
to different nodes, which allows them to send messages in parallel. 

An example is obvious in Figure~\ref{fig:mst_gruyere_operations}. The
level of parallelism is only two. Cores 00 to 15 are dealt with in
parallel to cores 16 to 31. Other than that, there is no further
parallelism although it is possible to ``split up'' communication with
every new ``round'' (i.e.\ send operations of nodes).

An important variable in constructing such a tree is the degree of
nodes. If the send cost $t_s$ is smaller than the propagation time
$t_p$, $d = \lfloor \frac{t_p}{t_s} \rfloor$ messages can be send to
hide propagation of messages. Hence, a tree with degree $d$ should be
ideal to minimize latency. For multicore interconnect networks,
however, we expect $d<1$. Thus, we do not look into this in detail
right now.

Reducing the degree of nodes also helps for the select operation. It
reduces the number of memory locations to poll, and therefore the
latency of detecting messages.

We show the multicast tree automatically generated from a fully meshed
machine model by applying a minimum spanning tree algorithm in
Figure~\ref{fig:mst_gruyere}.

\begin{figure}
\begin{tikzpicture}[>=latex,line join=bevel,scale=.5]
  \pgfsetlinewidth{1bp}
\input{graphs/mst_gruyere}
\end{tikzpicture}
\caption{Multicast tree automatically found for a 8x4x1 multicore}
\label{fig:mst_gruyere}
\end{figure}

%-------------------------------------------------
\subsubsection{Clustering}

We discussed earlier in Section~\ref{sec:model_hierarchy} that
multicore machines are hierarchical.%

We divide the space into clusters. In every round, we further split up
all cluster into smaller clusters and connect them with the minimal
link. 

We find this link by searching for links that have start and
end node in different regions of the most recent split. Out of these
links, we pick the cheapest ones. We ignore links whose start or end
node have a too high degree in the spanning tree already.

Assuming that the choice of node IDs represents the topology to some
extend, splitting them up into distinct clusters avoids contention, as
two separate sub-trees are processed in different areas of the
interconnect network.

Another idea is to employ clustering on NUMA domains such that every
NUMA domain corresponds to one cluster.

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
\subsection{Scheduling}

% Tree -> scheduling
Assuming the tree (or some other kind of overlay network), there is a
\emph{scheduling problems} for the order in which to send messages to
children. The intuition is to send on long-delay links first to hide
the more expensive latencies, as this link is dominating total send
cost.

In Figure~\ref{fig:mst_gruyere_operations}, we visualize sending of a
broadcast message in \naive order. We use the MST given in
Figure~\ref{fig:mst_gruyere}.  When core 08 sends its six messages, it
does so in random order (here: in increasing node ID order). This is
inefficient, as the latency of the broadcast is dominated by cores 12,
24 and 28 rather than cores 9, 10 and 11. A better schedule for
sending messages to the children of core 8 would be to start with the
child from which the longest path emerges.

\begin{figure}[htb]
  \centering
  \begin{tikzpicture}[scale=.35,transform shape]
    % Insert visualization
    \input{visu_only_mst}
  \end{tikzpicture}
  \caption{Visualization of operations for sending a message along the
    tree. Red boxes represent send operations, blue boxes represent
    receive operations and arrows messages sent between cores. The
    tree has been acquired using an MST algorithm.}
  \label{fig:mst_gruyere_operations}
\end{figure}

In Figure~\ref{fig:mst_gruyere_operations_sorted} we schedule sending
of messages based on link cost. We sort the outgoing edges by their
weights and send on the most expensive links first. See
Algorithm~\ref{algo:scheduling_link_weight}

\begin{algorithm}[htb]
\SetCommentSty{textc}
\SetKwFunction{weight}{edge\_weight}%
\SetKwFunction{sort}{sort\_by\_edge\_weight}%
\SetKwFunction{icsend}{send}%
  %
  \KwData{Node $n$ sending messages, \\ %
    broadcast tree as graph $(V, E)$, \\ %
    weight function \weight{$E$}
    %
  }
  % 
  \BlankLine
  $nb \leftarrow \{ c: \exists (self, c) \in E \}$\tcp*{Get children}
  $nb \leftarrow $ \sort{$nb$, \weight{}}\tcp*{Sort by edge weight}
  \For(\tcp*{For all children}){$c \leftarrow nb$}{
    \icsend{$c$}
  }
  % 
  \caption{Scheduling in order of outgoing link weight}
  \label{algo:scheduling_link_weight}
\end{algorithm}

\begin{figure}[htb]
  \centering
  \begin{tikzpicture}[scale=.35,transform shape]
    % Insert visualization
    \input{visu_mst_sort}
  \end{tikzpicture}
  \caption{Visualization of operations for sending a message along the
    tree. Red boxes represent send operations, blue boxes represent
    receive operations and arrows messages sent between cores. After
    running the MST algorithm, we optimize scheduling in every node to
    send messages to in decreasing order of their latency.}
  \label{fig:mst_gruyere_operations_sorted}
\end{figure}

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
\section{Implementation}

% Our implementation
Assuming the tree, we now implement an atomic broadcast for multicore
machines. Our implementation is based on the Barrelfish UMP
interconnect driver. It provides reliable channels with in-order
delivery. Flow control is build in for reliability. Communication
starts at the root node, which acts as a
sequentializer. See~\ref{algo:ab} for details.

\begin{algorithm}[htb]
\SetCommentSty{textc}
\SetKwInOut{Assumptions}{assumptions}
\Assumptions{Underlying communication channel is reliable and in-order}
\SetKwProg{Fn}{Function}{}{end}%
\SetKwFunction{receive}{on\_receive}%
\SetKwFunction{waitchild}{wait\_for\_children}%
\SetKwFunction{send}{send}%
\SetKwFunction{icsend}{send\_bc\_request}%
\SetKwFunction{icsendack}{send\_bc\_ack}%
\SetKwFunction{handlemessages}{handle\_other\_messages}%
  %
  \KwData{List of processes $p$, broadcast tree as graph $(V, E)$}
  \KwResult{Tree based atomic broadcast using a sequentializer}
  % 
  \BlankLine
  \Fn(\tcp*{Receive a message}){\receive{$client$, $m$}}{
    \For(\tcp*{For all children}){$c \leftarrow \{ c: \exists (self, c) \in E \} $}{
      \icsend{c}
    }
    \waitchild{}\;
    \icsendack{$client$}\;
  }
  % 
  \BlankLine
  \Fn(\tcp*{Send a message}){\send{void}}{
    \tcc{Need to wait for acknowledgment before returning to
      caller. Otherwise, sender might see his own request before some
      other request, that the sequentializer decided to handle first}
    \icsend{$V_{root}$}\tcp*{Relay msg (sequentializer)}
    \While{no answer received}{
      \handlemessages{}\tcp*{Otherwise, deadlocks}
    }
  }
  \caption{Atomic broadcast on reliable communication channels}
  \label{algo:ab}

\end{algorithm}

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
\subsection{Issues with the Barrelfish implementation}

When implementing the protocol in Barrelfish, we found several
practical problems, which we will discuss briefly in this section.
We kick-start our protocol connecting every process with all other
processes to get a fully-meshed network of channels.
We use a round-based algorithm to open channels. Then, every node
knows exactly the source of an incoming connection. We formalize this
algorithm in Algorithm~\ref{algo:ab_bind}.
This is required for Barrelfish UMP communication channels since on
bind, no source identifier is send along\footnote{Check if this is
  actually true, and even if it is true, if is a Barrelfish problem,
  and not a general one}.

\begin{algorithm}[htb]
  %
  \SetKwInOut{Assumptions}{assumptions}
  \Assumptions{Processes have unique contiguous
    identifiers starting at 0}
  \BlankLine
  %
  \SetKwArray{c}{channels}
  \SetKwFunction{connectNode}{connectNode}
  \SetKwFunction{listen}{listen}
  \SetKwFunction{barrier}{barrier}
  % 
  \KwData{process id $p$, round $r$, %
    each process an array of channels \c}
  \KwResult{Fully-meshed network of processes}
  % 
  \BlankLine
  %
  $r \leftarrow 0$\;
  \For{$i \leftarrow 0$ \KwTo $num(p)$}{
    \eIf{$p=i$}{
      \For{$o \leftarrow i+1$ \KwTo $num(p)$}{
        \c{$o$} $\leftarrow$ \connectNode{$o$}\;
      }
    }{
        \c{$i$} $\leftarrow$ \listen{}\;
    }
    \barrier{} \tcp*{Otherwise, reordering possible}
  }
  \caption{Establish fully-meshed network of channels}
  \label{algo:ab_bind}
\end{algorithm}

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
\section{Evaluation}

\stefan{Compare to pure shared-memory implementations, otherwise
  reviewers might argue that there is no need for distributed
  algorithms.}

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
\subsection{Atomic broadcast}

%-------------------------------------------------
\subsubsection{Simulation}

This section shows results for a simulation based on the model
introduced in Section~\ref{sec:mst_tree}. The time given is a
fictional time unit at this point.

\begin{table}[htb]
  \centering
  \begin{tabular}{llll}
    \toprule
    topology & scheduling & time & factor \\
    \midrule
    MST & naive         & 241 & 1.00 \\
    MST & longest first & 151 & 0.63 \\
    \bottomrule
  \end{tabular}
  \caption{Simulation results for simplified machine model}
  \label{tab:sim_results}
\end{table}

%-------------------------------------------------
\subsubsection{Real hardware}

Current numbers are listed in Table~\ref{tab:bc_measurements}.

\begin{table}[htb]
  \centering
  \begin{tabular}{lrrrr}
    \toprule
    broadcast algorithm & \multicolumn{2}{c}{nos5} & \multicolumn{2}{c}{gruyere} \\
      & cycles & error & cycles & error \\
    \midrule
    sequential &  4096.1 &  105.2 & 136751.9 &   3061.5 \\
    batch      &  2408.0 &  181.9 &  57360.8 &   5271.5 \\
    \bottomrule
  \end{tabular}
  \caption{Broadcast measurements}
  \label{tab:bc_measurements}
\end{table}

Plot~\ref{pgfplot:201303141819} shows the cost of flooding a sub-tree
(with the 50\% worst measurements dropped). The group communication is
based on a binary tree. Core 0 is the root, cores 1 and 2 are its
children etc.

The average cost is really high (probably a scheduling issue). But the
minimal numbers show what is possible. The numbers achieved are easily
explainable. The cost is increasing logarithmic with the number of
nodes reached by the broadcast (as expected). Every level in the tree
adds an additional 2500 to 3000 cycles to the tree. Node 0 takes
significantly longer. I don't know yet why that is.

\begin{figure}
  \caption{Execution time for a broadcast with ACK on gruyere. The
    cost is for execution for the sub-graph starting at the given
    node. Refer to Figure~\ref{fig:qrm_tree_gruyere} for a
    visualization of the broadcast tree used. }
  \label{pgfplot:201303141819}
  \begin{tikzpicture}
    \begin{axis}[
      xlabel=core id,
      scaled y ticks = false, % prevent 10^x stuff
      y tick label style={/pgf/number format/fixed},
      ylabel={cost for subtree [cycles]}]
    \addplot[
      color=red,
      very thin,
      mark=*,
      mark options={%
        scale=.4
      },
      error bars/y dir=both,
      error bars/y explicit] coordinates {
      (0,9932.3) +- (564.4,564.4)
      (1,8240.1) +- (298.5,298.5)
      (2,8259.6) +- (549.4,549.4)
      (3,6206.3) +- (225.1,225.1)
      (4,5914.9) +- (399.2,399.2)
      (5,6033.5) +- (462.2,462.2)
      (6,6070.4) +- (434.3,434.3)
      (7,4530.4) +- (188.6,188.6)
      (8,3163.4) +- (246.1,246.1)
      (9,3403.0) +- (285.3,285.3)
      (10,3650.7) +- (336.5,336.5)
      (11,3559.0) +- (318.8,318.8)
      (12,3558.1) +- (284.8,284.8)
      (13,3567.5) +- (304.5,304.5)
      (14,3476.3) +- (290.2,290.2)
      (15,2711.7) +- (151.9,151.9)
      (16,614.0) +- (17.5,17.5)
      (17,606.2) +- (17.3,17.3)
      (18,636.2) +- (72.5,72.5)
      (19,641.3) +- (46.3,46.3)
      (20,583.8) +- (46.7,46.7)
      (21,577.8) +- (38.3,38.3)
      (22,561.8) +- (18.1,18.1)
      (23,603.5) +- (56.6,56.6)
      (24,589.0) +- (33.4,33.4)
      (25,582.9) +- (33.2,33.2)
      (26,571.7) +- (16.3,16.3)
      (27,581.6) +- (36.6,36.6)
      (28,564.0) +- (16.4,16.4)
      (29,564.0) +- (20.8,20.8)
      (30,551.6) +- (13.0,13.0)
      (31,596.2) +- (51.4,51.4)
    };
    \end{axis}
  \end{tikzpicture}

\end{figure}

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
\section{Future work}

\begin{itemize}
\item Arguing that group communication/atomic broadcast is important
  and showing that we can do it well might be enough (appeal to atomic
  broadcasts as the foundation for other distributed algorithms, and
  the MPI primitives, that can/need to be mapped to
  group-communication (have a list of them)
\item Need higher-level applications, such as:
  \begin{itemize}
  \item capability system? \stefan{Ask Simon how expensive revocation
      currently is}
  \item a database with replication and consistency maintenance
    \begin{itemize}
    \item SharedDB? (buffers between operators, but no synchronization
      except for access to buffers, which are multiple writers, one
      reader)
    \item Crescando (some kind of state machine replication)
    \end{itemize}
  \end{itemize}
\item MPI collectives (as in~\cite{Tu2008})
\item Try other topologies (Frans Kaashoek), e.g.\ rings
\item \emph{Load balancing in a tree} We could build several rings,
  that do not always include all coordinators and work with read- and
  write-sets. We can select these rings such that they nicely
  integrate with the physical topology. They can also have different
  sizes (e.g.\ a smaller one for read operations and larger ones for
  writes).
\item \emph{Aggregation for convergecast} %
  Convergecast is a way of collecting information from nodes (i.e.\ it
  is kind of the reverse of a broadcast). On the return path,
  information can be aggregated. One example is batching, where several
  pieces of information are grouped in the same message to reduce packet
  processing overhead. Another example is aggregation, i.e.\
  pre-processing of data. One example is pre-calculating the average
  value of child messages if the sink node is not interested in
  individual values. %
  Similarly to what has been done in wireless sensor networks (where it
  also matters to reduce the number of messages, but for other reasons:
  power consumption), we can do aggregation in nodes. In difference to
  traditional distributed systems, this works, because it is easy to
  deploy custom software on every node in the network. Classical
  distributed systems do not typically allow this. Furthermore, reducing
  the number of messages at the price of higher complexity does not make
  sense in classical systems. %
  Examples for aggregation: number of nodes agreeing to something, find
  capabilities (concatenate core ids). 
\end{itemize}

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
\subsection{Octopus}

Octopus is a coordination service implementation for the Barrelfish
Multikernel operating system based on a key-value store
data-backend. It is, however, centralized. The idea is to make Octopus
a decentralized application.

For that, we extend Octopus such that it can be replicated within a
multicore machine. This will retain scalability for future manycore
machines and potentially outperform a centralized version of Octopus
especially in contented systems.

%--------------------------------------------------
\subsubsection{Key-value store}

One foundation block of Octopus is a key-value store. Since there is
only one copy, the data in there is always consistent. If replicated,
some sort of consistency guarantee would have to be implemented. One
example is \emph{linearizability}, i.e.\ ``accesses occur one at a time
in, some sequential order that is consistent with the order of
invocations and responses''~\cite{lynch}.

One concrete idea is to have one copy of the Octopus data-store
replicated on every NUMA node. One core on every node acts as
\emph{coordinator}, we call the other nodes \emph{slaves}.
Coordinators communicate on a ring. Communication on a local NUMA node
can be realized using read-only shared memory for read access. Write
access needs to be globally ordered and hence communication with other
replicas is required. 

Slaves direct their updates to the local coordinator. Coordinators
send the update request $r$ along the ring and wait for them to be
send back. While passing on requests, coordinators can set a flag to
indicate conflicts. A conflict is another colliding request $r_o$ that
is already in flight. I think we will have to send activity vectors
along with this information! Otherwise, the coordinators requesting
$r$ and $r_o$ will both prevent the other thing from happening and
none of them will be applied! Maybe we need a two-phase commit.

The idea of having a ring is somewhat related to SharedDB. It
optimizes for throughput, not latency!

%-------------------------------------------------
\subsubsection{Synchronization}

It is doubtful that a key-value store as a backend for synchronization
primitives makes sense. As we want to show efficiency, we probably
rather want to implement new synchronization algorithms from scratch.

The coordinator knows how many of its slaves need to enter the
barrier. It can then just aggregate local barrier\_enters calls and
only communicate this information to the others once all children are
ready. Local communication can be shared-memory based, while global
communication should probably rather be message-passing based.
<commit_msg>Moved the barrier ideas to a different section and added some more content<commit_after>\documentclass{article}
\usepackage{url,color,xspace,verbatim,subfig,ctable,multirow,listings}
\usepackage[utf8]{inputenc}
\usepackage[T1]{fontenc}
\usepackage{txfonts}
\usepackage{rotating}
\usepackage{paralist}
\usepackage{subfig}
\usepackage{graphics}
\usepackage{enumitem}
\usepackage{times}
\usepackage{amssymb}
\usepackage[colorlinks=true]{hyperref}
\usepackage[ruled,vlined]{algorithm2e}

% ==================================================

\graphicspath{{figs/}}
\urlstyle{sf}

% tikz stuff
\usepackage{tikz}
\usepackage{pgfplots}
% configuration
\usetikzlibrary{shapes,positioning,calc,snakes,arrows,shapes}
\pgfplotsset{width=.9\linewidth}

\lstset{
  language=C,
  basicstyle=\ttfamily \small,
  flexiblecolumns=false,
  basewidth={0.5em,0.45em},
  boxpos=t,
}

\newcommand{\etal}{{\it et al.}\xspace}
\newcommand{\naive}{na\"{\i}ve\xspace}
\newcommand{\Naive}{Na\"{\i}ve\xspace}
\newcommand{\textc}[1]{{\color{gray} {\footnotesize #1}}}

\definecolor{skRed}{RGB}{155,25,25}
\newcommand{\stefan}[1]{
  {\color{skRed}[{\color{red}{SK}} #1]}}

\setcounter{section}{0} % Start sections with 1, not 0
\begin{document}

\title{Adaptive broadcast tree for multicores}

% email address
\newcommand{\eaddr}{<EMAIL>}
\newcommand{\email}{\href{mailto:\eaddr}{\eaddr}}

\author{<NAME>\\
  \email \\
  Systems Group, ETH Zurich}

\maketitle

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
\section{Introduction}

Atomic broadcast is a building-block for many higher-level distributed
algorithm such as state-machine replication. Such distributed
algorithms are increasingly used to overcome scalability challenges on
multicore machines. Examples are databases~\cite{Salomie2011,
  Wiesmann2000} and operating systems~\cite{fos:osr09, tornado:osdi99,
  barrelfish:sosp09}.

% What do you want to enable?
We want to investigate how to achieve better performance of atomic
broadcast across a wide range of multicore machines without manually
tuning the implementation to concrete machines.

% What problem are you solving, and why is it hard?
The problem is complexity and diversity of modern multicore
machines. Multicores come in many flavors depending on vendor and
generation of the machine. All of these machines have fundamentally
different characteristics such as number of nodes, NUMA
characteristics, interconnect topologies, propagation time on
interconnect and use of shared caches. These characteristics are hard
to understand, but important to consider for application performance. 

% New ideas
Instead of manually tuning algorithms to characteristics of individual
machines, we automatically configure our atomic broadcast
implementation based on a machine model.
% How will you go about it?
We will base our atomic broadcast on overlay networks, which we
automatically configure based on the machine model. One questions is
which information should actually be encoded in the machine model. %
Our idea is to then explore different approaches (such as MST
algorithms, clustering) to \emph{find good overlay networks} based on
the network model to base the atomic broadcast on. Then, for every
node, we need to decide which message to send first. This is a
\emph{scheduling} problem.

% How do we show it works
We auto-configure a wide range of different multicore machines and
show the performance of our implementation comes close to the
performance of hand-tuned implementations for all of these machines.

% Hypothesis
My hypothesis is that automatic tuning of atomic broadcast (and other
distributed algorithms) to machine characteristics is worth
doing. Obviously, we cannot achieve performance of hand-tuned
implementations, but we can get close enough to argue that loosing a
bit of performance to avoid tedious manual reconfiguration is worth if
for many applications (while others, where performance \emph{really}
matters, might still want to manually configure their stuff).

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
\section{Related work}

A multicore aware broadcast has been implemented in~\cite{Tu2008}. It
takes the cache architecture of the machine (and only one particular?)
into consideration for building the broadcast. It is however for
larger messages (starting at 4K) and it does not seem to be performing
very well (numbers are given in microseconds -6). 4K messages are 52
microseconds, which is 150000 cycles. For synchronization primitives,
a cache line is sufficient.

Another tree implementation with MPI in mind was done
in~\cite{Graham2008}. They have comparable numbers to what we found,
but they cannot automatically tune to different machine
characteristics. Our hypothesis is that we can do achieve higher
average performance across a wide range of different multicore
machines without manually adapting the implementation.

Alistarh\etal~\cite{Alistarh2012} show composition of shared-memory
algorithms that perform efficiently under different conditions at
negligible overhead. A similar approach can potentially be applied to
deal with diversity of multicore machines. However, we would have
to extend it to support both shared-memory and message-passing
implementations. 

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
\section{The multicore network model}

% selecting hard
Characteristics of the network influence the performance of
distributed algorithms. Due to the heterogeneous combinations of
complex networks on multicores, selection of algorithms is
particularly hard.

% model
We express the network characteristics in a multicore network
model. This model at this stage is by no means complete. We are going to
extend it over time to address more complexities of current and future
multicores.

% graph
We represent the multicore network as a graph. The nodes in this graph
correspond to cores, and edges the communication channels between
nodes. The machines we consider for now only have bi-directional
communication links. Other machines (like the SCC) have
uni-directional links, which is why we model the graph as directed
graph.

% shared memory -> fully meshed
This graph is fully meshed for shared memory machines. Every core can
communicate with every other node via the cache coherence protocol.

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
\subsection{Node characteristics} 
\label{sec:model_nodes}

Every node (i.e.\ core) in the system has characteristics associated
with it:

\begin{description}
\item[synchrony] Parts of a multicore machine are synchronous. Hence,
  one characteristic of a node is an identifier for the synchrony
  cluster of a node.
\item[send time $t_{send}$] Time required to send a message, i.e.\ marshalling
  and queuing the message
\item[receive $t_{recv}$] Time required to receive a message, i.e.\
  dequeuing, unmarshalling and demultiplexing a channel.
\item[failure] Can nodes fail? If so, how? (Byzantine, crash)
\end{description}

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
\subsection{Link characteristics} 
\label{sec:model_links}

The network model is defined by communication channels, which in turn
are defined by a set of attributes (here sorted by priority):

\begin{description}
\item[latency] (quantitative) between each pair of nodes (per-edge)
\item[breakdown] (quantitative) of message send cost:
  dominating propagation time vs.\ dominating send and receive time on
  multicores (per-node)
\item[bounded delivery]
\item[reliability] (loss, link failure, in-order delivery, phantoms,
  corruption, duplicated messages): 
  multicore interconnect networks often are
\end{description}

% Assumptions
We assume that the cost of sending a message to the same node is 0,
which is not true, since even using a highly optimized implementation
for local communication (like LMP) has a non-negligible cost.

% Limitations
What we do not consider yet is link congestion. 

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
\subsection{Hierarchical model}
\label{sec:model_hierarchy}

Multicore machines are hierarchical. Processor cores are grouped in
NUMA nodes with shared caches. These NUMA nodes are then connected to
each other by increasingly complex interconnect networks. 

We encode NUMA node affinity in our model. This allows to break down
programming of a multicore machine into two distinct instances. We
solve one instance of the problem between NUMA nodes, and then, in a
second step, within each NUMA node. These two steps can be executed
independently and different algorithms can be used in both case.

If cores on the same NUMA node share a cache, it makes sense to
leverage this hardware feature to implement shared-memory algorithms
on top of it. %
Communication across NUMA nodes is realized sending messages across
the packet-based interconnect network. Using message-passing based
implementations is the natural choice. %
Shared memory algorithms and message-passing based algorithms can be
composed together to program hierarchical machines similarly to what
was previously shown for composing of shared-memory
algorithms~\cite{Alistarh2012}.

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
\subsection{Acquiring the model}

% Model given
We assume that the model is given. We just define one ourselves for
now or build one manually for a particular machine. In a real-world
scenario, this does not work. The model needs to be created
automatically for a particular machine (due to the diversity of
machines, and also the pace at which hardware changes)

We now make up a machine model for evaluation. This model is based on
a 8x4x1 AMD Shanghai machine. The topology of such a machine is shown
in Figure~\ref{fig:gruyere}.

Each of the eight nodes consists of four cores. We approximate the
cost of sending messages within nodes as $1$ and across nodes as
$10*num(hops)$.

Send and receive time are $10$ each, i.e.\ in the order of propagation
over a cross-NUMA link.

\begin{figure}[htb]
  \centering
  \begin{tikzpicture}[
    xscale=.7,
    yscale=.5,
    every node/.style={draw,fill=red!10,minimum width=2cm,minimum height=1cm}
    ]
    \node (c1) at ( 0, 4) {node 0};
    \node (c2) at ( 4, 4) {node 1};
    \node (c3) at ( 8, 4) {node 2};
    \node (c4) at (12, 4) {node 3};

    \node (c5) at ( 0, 0) {node 4};
    \node (c6) at ( 4, 0) {node 5};
    \node (c7) at ( 8, 0) {node 6};
    \node (c8) at (12, 0) {node 7};

    % top horizontal
    \draw[thick] (c1) -- (c2);
    \draw[thick] (c2) -- (c3);
    \draw[thick] (c3) -- (c4);
    % bottom horizontal
    \draw[thick] (c5) -- (c6);
    \draw[thick] (c6) -- (c7);
    \draw[thick] (c7) -- (c8);
    % vertical
    \draw[thick] (c1) -- (c5);
    \draw[thick] (c2) -- (c6);
    \draw[thick] (c3) -- (c7);
    \draw[thick] (c4) -- (c8);
    % cross
    \draw[thick] (c3) -- (c8);
    \draw[thick] (c4) -- (c7);
  \end{tikzpicture}
  \caption{Interconnect topology of 8x4x1 AMD Shanghai machine}
  \label{fig:gruyere}
\end{figure}

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
\section{Atomic Broadcast for Multicores}

For a specific model, we now need to find ``the right'' implementation
of algorithms to program it. We build our atomic broadcast based on a
subset of links offered by the model such that all nodes of a given
multicore machine can be reached. Atomic broadcast guarantees that 1)
if one node receives a message, all other nodes receive it as well,
and 2) that all nodes see the message in the same order.

First, we discuss what kind of topologies we use to communicate
between nodes. We evaluate tree based implementations. Other
potentially interesting topologies are rings.

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
\subsection{Tree Topology}

We now look at tree-based atomic broadcast implementations. The first
challenge is to find efficient spanning trees for arbitrary
machines. We now discuss different strategies to select trees based on
the model representing the machine.

%-------------------------------------------------
\subsubsection{Binary Tree}

A very \naive strategy to build a tree from machine models is the use
of a binary tree with root in core 0. Every node $n$ has two children
$2(n+1)-1$ and $2(n+1)$.

The benefit of such a tree is that the time complexity is guaranteed
to be $O(\log{n})$. The downside is that it is not considering the
machine topology. Unnecessary communication across NUMA nodes might be
introduced. 

We show an example for such a tree for a 8x4x1 AMD Barcelona machine
(gruyere) in Figure~\ref{fig:qrm_tree_gruyere}.

\begin{figure}
  \input{qrm_tree_gruyere}
  \caption{\Naive binary tree of cores for broadcast on gruyere}
  \label{fig:qrm_tree_gruyere}
\end{figure}

%-------------------------------------------------
\subsubsection{Minimum-Spanning Tree}
\label{sec:mst_tree}

We can apply a minimum spanning tree algorithm to the machine model
graph to find a number of links such that these links connect all
nodes at a minimum cost.

We have implemented such an algorithm and found out, that it is not a
good solution in the general case, since it does not consider
parallelism. Naturally, we would like to parallelize send operations
to different nodes, which allows them to send messages in parallel. 

An example is obvious in Figure~\ref{fig:mst_gruyere_operations}. The
level of parallelism is only two. Cores 00 to 15 are dealt with in
parallel to cores 16 to 31. Other than that, there is no further
parallelism although it is possible to ``split up'' communication with
every new ``round'' (i.e.\ send operations of nodes).

An important variable in constructing such a tree is the degree of
nodes. If the send cost $t_s$ is smaller than the propagation time
$t_p$, $d = \lfloor \frac{t_p}{t_s} \rfloor$ messages can be send to
hide propagation of messages. Hence, a tree with degree $d$ should be
ideal to minimize latency. For multicore interconnect networks,
however, we expect $d<1$. Thus, we do not look into this in detail
right now.

Reducing the degree of nodes also helps for the select operation. It
reduces the number of memory locations to poll, and therefore the
latency of detecting messages.

We show the multicast tree automatically generated from a fully meshed
machine model by applying a minimum spanning tree algorithm in
Figure~\ref{fig:mst_gruyere}.

\begin{figure}
\begin{tikzpicture}[>=latex,line join=bevel,scale=.5]
  \pgfsetlinewidth{1bp}
\input{graphs/mst_gruyere}
\end{tikzpicture}
\caption{Multicast tree automatically found for a 8x4x1 multicore}
\label{fig:mst_gruyere}
\end{figure}

%-------------------------------------------------
\subsubsection{Clustering}

We discussed earlier in Section~\ref{sec:model_hierarchy} that
multicore machines are hierarchical.%

We divide the space into clusters. In every round, we further split up
all cluster into smaller clusters and connect them with the minimal
link. 

We find this link by searching for links that have start and
end node in different regions of the most recent split. Out of these
links, we pick the cheapest ones. We ignore links whose start or end
node have a too high degree in the spanning tree already.

Assuming that the choice of node IDs represents the topology to some
extend, splitting them up into distinct clusters avoids contention, as
two separate sub-trees are processed in different areas of the
interconnect network.

Another idea is to employ clustering on NUMA domains such that every
NUMA domain corresponds to one cluster.

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
\subsection{Scheduling}

% Tree -> scheduling
Assuming the tree (or some other kind of overlay network), there is a
\emph{scheduling problems} for the order in which to send messages to
children. The intuition is to send on long-delay links first to hide
the more expensive latencies, as this link is dominating total send
cost.

In Figure~\ref{fig:mst_gruyere_operations}, we visualize sending of a
broadcast message in \naive order. We use the MST given in
Figure~\ref{fig:mst_gruyere}.  When core 08 sends its six messages, it
does so in random order (here: in increasing node ID order). This is
inefficient, as the latency of the broadcast is dominated by cores 12,
24 and 28 rather than cores 9, 10 and 11. A better schedule for
sending messages to the children of core 8 would be to start with the
child from which the longest path emerges.

\begin{figure}[htb]
  \centering
  \begin{tikzpicture}[scale=.35,transform shape]
    % Insert visualization
    \input{visu_only_mst}
  \end{tikzpicture}
  \caption{Visualization of operations for sending a message along the
    tree. Red boxes represent send operations, blue boxes represent
    receive operations and arrows messages sent between cores. The
    tree has been acquired using an MST algorithm.}
  \label{fig:mst_gruyere_operations}
\end{figure}

In Figure~\ref{fig:mst_gruyere_operations_sorted} we schedule sending
of messages based on link cost. We sort the outgoing edges by their
weights and send on the most expensive links first. See
Algorithm~\ref{algo:scheduling_link_weight}

\begin{algorithm}[htb]
\SetCommentSty{textc}
\SetKwFunction{weight}{edge\_weight}%
\SetKwFunction{sort}{sort\_by\_edge\_weight}%
\SetKwFunction{icsend}{send}%
  %
  \KwData{Node $n$ sending messages, \\ %
    broadcast tree as graph $(V, E)$, \\ %
    weight function \weight{$E$}
    %
  }
  % 
  \BlankLine
  $nb \leftarrow \{ c: \exists (self, c) \in E \}$\tcp*{Get children}
  $nb \leftarrow $ \sort{$nb$, \weight{}}\tcp*{Sort by edge weight}
  \For(\tcp*{For all children}){$c \leftarrow nb$}{
    \icsend{$c$}
  }
  % 
  \caption{Scheduling in order of outgoing link weight}
  \label{algo:scheduling_link_weight}
\end{algorithm}

\begin{figure}[htb]
  \centering
  \begin{tikzpicture}[scale=.35,transform shape]
    % Insert visualization
    \input{visu_mst_sort}
  \end{tikzpicture}
  \caption{Visualization of operations for sending a message along the
    tree. Red boxes represent send operations, blue boxes represent
    receive operations and arrows messages sent between cores. After
    running the MST algorithm, we optimize scheduling in every node to
    send messages to in decreasing order of their latency.}
  \label{fig:mst_gruyere_operations_sorted}
\end{figure}

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
\section{Implementation}

% Our implementation
Assuming the tree, we now implement an atomic broadcast for multicore
machines. Our implementation is based on the Barrelfish UMP
interconnect driver. It provides reliable channels with in-order
delivery. Flow control is build in for reliability. Communication
starts at the root node, which acts as a
sequentializer. See~\ref{algo:ab} for details.

\begin{algorithm}[htb]
\SetCommentSty{textc}
\SetKwInOut{Assumptions}{assumptions}
\Assumptions{Underlying communication channel is reliable and in-order}
\SetKwProg{Fn}{Function}{}{end}%
\SetKwFunction{receive}{on\_receive}%
\SetKwFunction{waitchild}{wait\_for\_children}%
\SetKwFunction{send}{send}%
\SetKwFunction{icsend}{send\_bc\_request}%
\SetKwFunction{icsendack}{send\_bc\_ack}%
\SetKwFunction{handlemessages}{handle\_other\_messages}%
  %
  \KwData{List of processes $p$, broadcast tree as graph $(V, E)$}
  \KwResult{Tree based atomic broadcast using a sequentializer}
  % 
  \BlankLine
  \Fn(\tcp*{Receive a message}){\receive{$client$, $m$}}{
    \For(\tcp*{For all children}){$c \leftarrow \{ c: \exists (self, c) \in E \} $}{
      \icsend{c}
    }
    \waitchild{}\;
    \icsendack{$client$}\;
  }
  % 
  \BlankLine
  \Fn(\tcp*{Send a message}){\send{void}}{
    \tcc{Need to wait for acknowledgment before returning to
      caller. Otherwise, sender might see his own request before some
      other request, that the sequentializer decided to handle first}
    \icsend{$V_{root}$}\tcp*{Relay msg (sequentializer)}
    \While{no answer received}{
      \handlemessages{}\tcp*{Otherwise, deadlocks}
    }
  }
  \caption{Atomic broadcast on reliable communication channels}
  \label{algo:ab}

\end{algorithm}

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
\subsection{Issues with the Barrelfish implementation}

When implementing the protocol in Barrelfish, we found several
practical problems, which we will discuss briefly in this section.
We kick-start our protocol connecting every process with all other
processes to get a fully-meshed network of channels.
We use a round-based algorithm to open channels. Then, every node
knows exactly the source of an incoming connection. We formalize this
algorithm in Algorithm~\ref{algo:ab_bind}.
This is required for Barrelfish UMP communication channels since on
bind, no source identifier is send along\footnote{Check if this is
  actually true, and even if it is true, if is a Barrelfish problem,
  and not a general one}.

\begin{algorithm}[htb]
  %
  \SetKwInOut{Assumptions}{assumptions}
  \Assumptions{Processes have unique contiguous
    identifiers starting at 0}
  \BlankLine
  %
  \SetKwArray{c}{channels}
  \SetKwFunction{connectNode}{connectNode}
  \SetKwFunction{listen}{listen}
  \SetKwFunction{barrier}{barrier}
  % 
  \KwData{process id $p$, round $r$, %
    each process an array of channels \c}
  \KwResult{Fully-meshed network of processes}
  % 
  \BlankLine
  %
  $r \leftarrow 0$\;
  \For{$i \leftarrow 0$ \KwTo $num(p)$}{
    \eIf{$p=i$}{
      \For{$o \leftarrow i+1$ \KwTo $num(p)$}{
        \c{$o$} $\leftarrow$ \connectNode{$o$}\;
      }
    }{
        \c{$i$} $\leftarrow$ \listen{}\;
    }
    \barrier{} \tcp*{Otherwise, reordering possible}
  }
  \caption{Establish fully-meshed network of channels}
  \label{algo:ab_bind}
\end{algorithm}

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
\section{Evaluation}

\stefan{Compare to pure shared-memory implementations, otherwise
  reviewers might argue that there is no need for distributed
  algorithms.}

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
\subsection{Atomic broadcast}

%-------------------------------------------------
\subsubsection{Simulation}

This section shows results for a simulation based on the model
introduced in Section~\ref{sec:mst_tree}. The time given is a
fictional time unit at this point.

\begin{table}[htb]
  \centering
  \begin{tabular}{llll}
    \toprule
    topology & scheduling & time & factor \\
    \midrule
    MST & naive         & 241 & 1.00 \\
    MST & longest first & 151 & 0.63 \\
    \bottomrule
  \end{tabular}
  \caption{Simulation results for simplified machine model}
  \label{tab:sim_results}
\end{table}

%-------------------------------------------------
\subsubsection{Real hardware}

Current numbers are listed in Table~\ref{tab:bc_measurements}.

\begin{table}[htb]
  \centering
  \begin{tabular}{lrrrr}
    \toprule
    broadcast algorithm & \multicolumn{2}{c}{nos5} & \multicolumn{2}{c}{gruyere} \\
      & cycles & error & cycles & error \\
    \midrule
    sequential &  4096.1 &  105.2 & 136751.9 &   3061.5 \\
    batch      &  2408.0 &  181.9 &  57360.8 &   5271.5 \\
    \bottomrule
  \end{tabular}
  \caption{Broadcast measurements}
  \label{tab:bc_measurements}
\end{table}

Plot~\ref{pgfplot:201303141819} shows the cost of flooding a sub-tree
(with the 50\% worst measurements dropped). The group communication is
based on a binary tree. Core 0 is the root, cores 1 and 2 are its
children etc.

The average cost is really high (probably a scheduling issue). But the
minimal numbers show what is possible. The numbers achieved are easily
explainable. The cost is increasing logarithmic with the number of
nodes reached by the broadcast (as expected). Every level in the tree
adds an additional 2500 to 3000 cycles to the tree. Node 0 takes
significantly longer. I don't know yet why that is.

\begin{figure}
  \caption{Execution time for a broadcast with ACK on gruyere. The
    cost is for execution for the sub-graph starting at the given
    node. Refer to Figure~\ref{fig:qrm_tree_gruyere} for a
    visualization of the broadcast tree used. }
  \label{pgfplot:201303141819}
  \begin{tikzpicture}
    \begin{axis}[
      xlabel=core id,
      scaled y ticks = false, % prevent 10^x stuff
      y tick label style={/pgf/number format/fixed},
      ylabel={cost for subtree [cycles]}]
    \addplot[
      color=red,
      very thin,
      mark=*,
      mark options={%
        scale=.4
      },
      error bars/y dir=both,
      error bars/y explicit] coordinates {
      (0,9932.3) +- (564.4,564.4)
      (1,8240.1) +- (298.5,298.5)
      (2,8259.6) +- (549.4,549.4)
      (3,6206.3) +- (225.1,225.1)
      (4,5914.9) +- (399.2,399.2)
      (5,6033.5) +- (462.2,462.2)
      (6,6070.4) +- (434.3,434.3)
      (7,4530.4) +- (188.6,188.6)
      (8,3163.4) +- (246.1,246.1)
      (9,3403.0) +- (285.3,285.3)
      (10,3650.7) +- (336.5,336.5)
      (11,3559.0) +- (318.8,318.8)
      (12,3558.1) +- (284.8,284.8)
      (13,3567.5) +- (304.5,304.5)
      (14,3476.3) +- (290.2,290.2)
      (15,2711.7) +- (151.9,151.9)
      (16,614.0) +- (17.5,17.5)
      (17,606.2) +- (17.3,17.3)
      (18,636.2) +- (72.5,72.5)
      (19,641.3) +- (46.3,46.3)
      (20,583.8) +- (46.7,46.7)
      (21,577.8) +- (38.3,38.3)
      (22,561.8) +- (18.1,18.1)
      (23,603.5) +- (56.6,56.6)
      (24,589.0) +- (33.4,33.4)
      (25,582.9) +- (33.2,33.2)
      (26,571.7) +- (16.3,16.3)
      (27,581.6) +- (36.6,36.6)
      (28,564.0) +- (16.4,16.4)
      (29,564.0) +- (20.8,20.8)
      (30,551.6) +- (13.0,13.0)
      (31,596.2) +- (51.4,51.4)
    };
    \end{axis}
  \end{tikzpicture}

\end{figure}

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
\section{Future work}

\begin{itemize}
\item Arguing that group communication/atomic broadcast is important
  and showing that we can do it well might be enough (appeal to atomic
  broadcasts as the foundation for other distributed algorithms, and
  the MPI primitives, that can/need to be mapped to
  group-communication (have a list of them)
\item Need higher-level applications, such as:
  \begin{itemize}
  \item capability system? \stefan{Ask Simon how expensive revocation
      currently is}
  \item a database with replication and consistency maintenance
    \begin{itemize}
    \item SharedDB? (buffers between operators, but no synchronization
      except for access to buffers, which are multiple writers, one
      reader)
    \item Crescando (some kind of state machine replication)
    \end{itemize}
  \end{itemize}
\item MPI collectives (as in~\cite{Tu2008})
\item Try other topologies (Frans Kaashoek), e.g.\ rings
\item \emph{Load balancing in a tree} We could build several rings,
  that do not always include all coordinators and work with read- and
  write-sets. We can select these rings such that they nicely
  integrate with the physical topology. They can also have different
  sizes (e.g.\ a smaller one for read operations and larger ones for
  writes).
\item \emph{Aggregation for convergecast} %
  Convergecast is a way of collecting information from nodes (i.e.\ it
  is kind of the reverse of a broadcast). On the return path,
  information can be aggregated. One example is batching, where several
  pieces of information are grouped in the same message to reduce packet
  processing overhead. Another example is aggregation, i.e.\
  pre-processing of data. One example is pre-calculating the average
  value of child messages if the sink node is not interested in
  individual values. %
  Similarly to what has been done in wireless sensor networks (where it
  also matters to reduce the number of messages, but for other reasons:
  power consumption), we can do aggregation in nodes. In difference to
  traditional distributed systems, this works, because it is easy to
  deploy custom software on every node in the network. Classical
  distributed systems do not typically allow this. Furthermore, reducing
  the number of messages at the price of higher complexity does not make
  sense in classical systems. %
  Examples for aggregation: number of nodes agreeing to something, find
  capabilities (concatenate core ids). 
\end{itemize}

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
\subsection{Octopus}

Octopus is a coordination service implementation for the Barrelfish
Multikernel operating system based on a key-value store
data-backend. It is, however, centralized. The idea is to make Octopus
a decentralized application.

For that, we extend Octopus such that it can be replicated within a
multicore machine. This will retain scalability for future manycore
machines and potentially outperform a centralized version of Octopus
especially in contented systems.

%--------------------------------------------------
\subsubsection{Key-value store}

One foundation block of Octopus is a key-value store. Since there is
only one copy, the data in there is always consistent. If replicated,
some sort of consistency guarantee would have to be implemented. One
example is \emph{linearizability}, i.e.\ ``accesses occur one at a time
in, some sequential order that is consistent with the order of
invocations and responses''~\cite{lynch}.

One concrete idea is to have one copy of the Octopus data-store
replicated on every NUMA node. One core on every node acts as
\emph{coordinator}, we call the other nodes \emph{slaves}.
Coordinators communicate on a ring. Communication on a local NUMA node
can be realized using read-only shared memory for read access. Write
access needs to be globally ordered and hence communication with other
replicas is required. 

Slaves direct their updates to the local coordinator. Coordinators
send the update request $r$ along the ring and wait for them to be
send back. While passing on requests, coordinators can set a flag to
indicate conflicts. A conflict is another colliding request $r_o$ that
is already in flight. I think we will have to send activity vectors
along with this information! Otherwise, the coordinators requesting
$r$ and $r_o$ will both prevent the other thing from happening and
none of them will be applied! Maybe we need a two-phase commit.

The idea of having a ring is somewhat related to SharedDB. It
optimizes for throughput, not latency!

%-------------------------------------------------
\subsubsection{Synchronization}

It is doubtful that a key-value store as a backend for synchronization
primitives makes sense. As we want to show efficiency, we probably
rather want to implement new synchronization algorithms from scratch.

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
\subsection{Barriers}

The idea is to explore barrier implementations. The idea is to
leverage information about the NUMA topology for better performance of
barrier intensive workloads. Every NUMA node has a coordinator node,
that knows how many threads on that node want to enter the
barrier. There is only local communication (possibly using a
share-memory implementation to exploit the shared cache) until the
point where all NUMA local threads did enter the barrier.  Then there
is ``global'' communication with other coordinators on other nodes
using a message-passing based algorithm. We believe that this scales
better than pure shared-memory implementations due to less
interconnect contention.

{'fraction_non_alphanumeric': 0.12852498919083383, 'fraction_numerical': 0.022739282911909298, 'mean_word_length': 4.051609771881573, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 13, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1000924', 'n_tokens_mistral': 18789, 'n_tokens_neox': 17325, 'n_words': 7468}