iccsa-16-factory

sections.tex (17320B)

---

 \section{Methods}
 
 \subsection{Model of computation}
 
 To infer fault tolerance model which is suitable for big data applications we
 use bulk-synchronous parallel model~\cite{valiant1990bridging} as the basis.
 This model assumes that a parallel programme is composed of several sequential
 steps that are internally parallel, and global synchronisation of all parallel
 processes occurs after each step. In our model all sequential steps are
 pipelined where it is possible. The evolution of the computational model is
 described as follows.
 
 Given a programme that is sequential and large enough to be decomposed into
 several sequential steps, the simplest way to make it run faster is to exploit
 data parallelism. Usually it means finding multi-dimensional arrays and loops
 that access their elements and trying to make them parallel. After transforming
 several loops the programme will still have the same number of sequential steps,
 but every step will (ideally) be internally parallel.
 
 After that the only possibility to speedup the programme is to overlap execution
 of code blocks that work with different hardware devices. The most common
 pattern is to overlap computation with network I/O or disk I/O. This approach
 makes sense because all devices operate with little synchronisation, and issuing
 commands in parallel makes the whole programme perform better. This behaviour
 can be achieved by allocating a separate task queue for each device and
 submitting tasks to these queues asynchronously with execution of the main
 thread. So, after this optimisation, the programme will be composed of several
 steps chained into the pipeline, each step is implemented as a task queue for a
 particular device.
 
 Pipelining of otherwise sequential steps is beneficial not only for code
 accessing different devices, but for code different branches of which are
 suitable for execution by multiple hardware threads of the same core, i.e.
 branches accessing different regions of memory or performing mixed arithmetic
 (floating point and integer). In other words, code branches which use different
 modules of processor are good candidates to run in parallel on a processor core
 with multiple hardware threads.
 
 Even though pipelining may not add parallelism for a programme that uses only
 one input file (or a set of input parameters), it adds parallelism when the
 programme can process multiple input files: each input generates tasks which
 travel through the whole pipeline in parallel with tasks generated by other
 inputs. With a pipeline an array of files is processed in parallel by the same
 set of resources allocated for a batch job, and possibly with greater efficiency
 for busy HPC clusters compared to executing a separate job for each input file,
 because the time that each subsequent job after the first spends in a queue is
 eliminated.
 
 Computational model with a pipeline can be seen as \emph{bulk-asynchronous
   model}, because of the parallel nature of otherwise sequential execution
 steps. This model is the basis of the fault-tolerance model developed here.
 
 \subsection{Fail over model}
 
 Although, fault-tolerance and high-availability are different terms, in essence
 they describe the same property---an ability of a system to switch processing
 from a failed component to its live spare or backup component. In case of
 fault-tolerance it is the ability to switch from a failed slave node to a spare
 one, i.e. to repeat computation step on a healthy slave node. In case of
 high-availability it is the ability to switch from a failed master node to a
 backup node with full restoration of execution state. These are the core
 abilities that constitute distributed system's ability to \emph{fail over}.
 
 The key feature that is missing in the current parallel programming and big data
 processing technologies is a possibility to specify hierarchical dependencies
 between parallel tasks. When one has such dependency, it is trivial to determine
 which task should be responsible for re-executing a failed task on a healthy
 node. To re-execute the root of the hierarchy, a backup root task is created and
 executed on a different node. There exists a number of engines that are capable
 of executing directed acyclic graphs of tasks in
 parallel~\cite{acun2014charmpp,islam2012oozie}, but graphs are not good to infer
 master-slave relationship between tasks, because a node in the graph may have
 multiple parent nodes.
 
 \subsection{Programming model}
 
 This work is based on the results of previous research:
 In~\cite{gankevich2015subordination,gankevich2015iccsa} we developed an
 algorithm that allows to build a tree hierarchy from strictly ordered set of
 cluster nodes. The sole purpose of this hierarchy is to make a cluster more
 fault-tolerant by introducing multiple master nodes. If a master node fails,
 then its subordinates try to connect to another node from the same or higher
 level of the hierarchy. If there is no such node, one of the subordinates
 becomes the master. In~\cite{gankevich2015spec} we developed a framework for big
 data processing without fault tolerance, and here this framework is combined
 with fault-tolerance techniques described in this paper.
 
 Each programme that runs on top of the tree hierarchy is composed of
 computational kernels---objects that contain data and code to process it. To
 exploit parallelism a kernel may create arbitrary number of subordinate kernels
 which are automatically spread first across available processor cores, second
 across subordinate nodes in the tree hierarchy. The programme is itself a kernel
 (without a parent as it is executed by a user), which either solves the problem
 sequentially on its own or creates subordinate kernels to solve it in parallel.
 
 In contrast to HPC applications, in big data applications it is inefficient to
 run computational kernels on arbitrary chosen nodes. More practical approach is
 to bind every kernel to a file location in a parallel file system and transfer
 the kernel to that location before processing the file. That way expensive data
 transfer is eliminated, and the file is always read from a local drive. This
 approach is more deterministic compared to existing ones, e.g. MapReduce
 framework runs jobs on nodes that are ``close'' to the file location, but not
 necessarily the exact node where the file is located~\cite{dean2008mapreduce}.
 However, this approach does not come without disadvantages: scalability of a big
 data application is limited by the strategy that was employed to distribute its
 input files across cluster nodes. The more nodes used to store input files, the
 more read performance is achieved. The advantage of our approach is that the I/O
 performance is more predictable, than one of hybrid approach with streaming
 files over the network.
 
 % From mathematical point of view kernel $K$ can be described as a vector-valued functional which recursively maps a kernel to $n$-component vector of kernels:
 % \begin{equation*}
 %     K(f): \mathbb{K} \rightarrow \mathbb{K}^n
 %     \qquad
 %     \mathbb{K}^n = \left\{ f: \mathbb{K} \rightarrow \mathbb{K}^n \right\}.
 % \end{equation*}
 % Dummy kernel $\mathbb{O}: \mathbb{K} \rightarrow \mathbb{K}^0$, which stops recursion, is used to call the first kernel and finish execution of the programme. An argument to each kernel is interpreted using the following rules.
 % \begin{enumerate}
 %     \item If a kernel is a new kernel, then its argument is its parent kernel.
 %     \item If a kernel is a parent of the kernel that produced it or some other existing kernel, then the argument is the kernel that produced it.
 % \end{enumerate}
 
 % Engine that executes kernels is implemented as a simple loop. It starts with calling the first kernel with dummy kernel as an argument, then calls each kernel that was produced by this call and so forth. The loop finishes when a dummy kernel is returned as a result of the call.
 
 % Since kernel call may return multiple kernels they are executed in parallel. Parallel execution quickly produces a pool of kernels which permit execution in an unspecified order. Several threads concurrently retrieve kernels from the pool and may ``spill'' remaining kernels to neighbouring cluster nodes.
 
 % Kernels are implemented as closures---function objects containing all their arguments, a reference to parent kernel and user-supplied data. The data is either processed upon kernel call, or subordinate kernels are created to process it in parallel. When the processing is complete a parent kernel closure with its subordinate kernel as an argument is called to collect data from it.
 
 \subsection{Handling master node failures}
 
 A possible way of handling a failure of a node where the first kernel is located
 (a master node) is to replicate this kernel to a backup node, and make all
 updates to its state propagate to the backup node by means of a distributed
 transaction. This approach requires synchronisation between all nodes that
 execute subordinates of the first kernel and the node with the first kernel
 itself. When a node with the first kernel goes offline, the nodes with
 subordinate kernels must know what node is the backup one. However, if the
 backup node also goes offline in the middle of execution of some subordinate
 kernel, then it is impossible for this kernel to discover the next backup node
 to return to, because this kernel has not discovered the unavailability of the
 master node yet. One can think of a consensus-based algorithm to ensure that
 subordinate kernels always know where the backup node is, but distributed
 consensus algorithms do not scale well to the large number of nodes and they are
 not reliable~\cite{fischer1985impossibility}. So, consensus-based approach does
 not play well with asynchronous nature of computational kernels as it may
 inhibit scalability of a parallel programme.
 
 Fortunately, the first kernel usually does not perform operations in parallel,
 it is rather sequentially launches execution steps one by one, so it has only
 one subordinate at a time. Such behaviour is described by bulk-synchronous
 parallel programming model, in the framework of which a programme consists of
 sequential supersteps which are internally parallel~\cite{valiant1990bridging}.
 Keeping this in mind, we can simplify synchronisation of its state: we can send
 the first kernel along with its subordinate to the subordinate node. When the
 node with the first kernel fails, its copy receives its subordinate, and no
 execution time is lost. When the node with its copy fails, its subordinate is
 rescheduled on some other node, and in the worst case a whole step of
 computation is lost.
 
 Described approach works only for kernels that do not have a parent and have
 only one subordinate at a time, and act similar to manually triggered
 checkpoints. The advantage is that they
 \begin{itemize}
     \item save results after each sequential step when memory footprint of a
       programme is low,
     \item they save only relevant data,
     \item and they use memory of a subordinate node instead of stable storage.
 \end{itemize}
 
 \section{Results}
 
 Master node fail over technique is evaluated on the example of wave energy
 spectra processing application. This programme uses NDBC
 dataset~\cite{ndbc-dataset} to reconstruct frequency-directional spectra from
 wave rider buoy measurements and compute variance. Each spectrum is
 reconstructed from five variables using the following
 formula~\cite{earle1996nondirectional}.
 \begin{equation*}
     S(\omega, \theta) = \frac{1}{\pi}
     \left[
         \frac{1}{2} + 
         r_1 \cos \left( \theta - \alpha_1 \right) +
         r_2 \sin \left( 2 \left( \theta - \alpha_2 \right) \right)
     \right]
     S_0(\omega).
 \end{equation*}
 Here $\omega$ denotes frequency, $\theta$ is wave direction, $r_{1,2}$ and
 $\alpha_{1,2}$ are parameters of spectrum decomposition and $S_0$ is
 non-directional spectrum; $r_{1,2}$, $\alpha_{1,2}$ and $S_0$ are acquired
 through measurements. Properties of the dataset which is used in evaluation are
 listed in Table~\ref{tab:ndbc-dataset}.
 
 \begin{table}
     \centering
     \caption{NDBC dataset properties.}
     \begin{tabular}{ll}
         \toprule
         Dataset size & 144MB \\
         Dataset size (uncompressed) & 770MB \\
         No. of wave stations & 24 \\
         Time span & 3 years (2010--2012) \\
         Total no. of spectra & 445422 \\
         \bottomrule
     \end{tabular}
     \label{tab:ndbc-dataset}
 \end{table}
 
 The algorithm of processing spectra is as follows. First, current directory is
 recursively scanned for input files. Data for all buoys is distributed across
 cluster nodes and each buoy's data processing is distributed across processor
 cores of a node. Processing begins with joining corresponding measurements for
 each spectrum variables into a tuple, then for each tuple frequency-directional
 spectrum is reconstructed and its variance is computed. Results are gradually
 copied back to the machine where application was executed and when the
 processing is complete the programme terminates.
 
 \begin{table}
     \centering
     \caption{Test platform configuration.}
     \begin{tabular}{ll}
          \toprule
          CPU & Intel Xeon E5440, 2.83GHz \\
          RAM & 4Gb \\
          HDD & ST3250310NS, 7200rpm \\
          No. of nodes & 12 \\
          No. of CPU cores per node & 8 \\
          \bottomrule
     \end{tabular}
     \label{tab:cluster}
 \end{table}
 
 In a series of test runs we benchmarked performance of the application in the
 presence of different types of failures:
 \begin{itemize}
     \item failure of a master node (a node where the first kernel is run),
     \item failure of a slave node (a node where spectra from a particular
       station are reconstructed) and
     \item failure of a backup node (a node where the first kernel is copied).
 \end{itemize}
 A tree hierarchy with sufficiently large fan-out value was chosen to make all
 cluster nodes connect directly to the first one so that only one master node
 exists in the cluster. In each run the first kernel was launched on a different
 node to make mapping of kernel hierarchy to the tree hierarchy optimal. A victim
 node was made offline after a fixed amount of time early after the programme
 start. To make up for the node failure all data files have replicas stored on
 different cluster nodes. All relevant parameters are summarised in
 Table~\ref{tab:benchmark} (here ``root'' and ``leaf'' refer to a node in the
 tree hierarchy). The results of these runs were compared to the run without node
 failures (Figure~\ref{fig:benchmark-bigdata}).
 
 \begin{table}
     \centering
     \caption{Benchmark parameters.}
     \begin{tabular}{llll}
          \toprule
          Experiment no. & Master node & Victim node & Time to offline, s \\
          \midrule
          1 & root &      & \\
          2 & root & leaf & 30 \\
          3 & leaf & leaf & 30 \\
          4 & leaf & root & 30 \\
          \bottomrule
     \end{tabular}
     \label{tab:benchmark}
 \end{table}
 
 The benchmark showed that only a backup node failure results in significant
 performance penalty, in all other cases the performance is roughly equals to the
 one without failures but with the number of nodes minus one. It happens because
 a backup node not only stores the copy of the state of the current computation
 step but executes this step in parallel with other subordinate nodes. So, when a
 backup node fails, the master node executes the whole step once again on
 arbitrarily chosen healthy subordinate node.
 
 % \begin{figure}
 %     \centering
 %     \includegraphics{factory-3000}
 %     \caption{Performance of hydrodynamics HPC application in the presence of node failures.}
 %     \label{fig:benchmark-hpc}
 % \end{figure}
 
 \begin{figure}
     \centering
     \includegraphics{spec}
     \caption{Performance of spectrum processing application in the presence of different types of node failures.}
     \label{fig:benchmark-bigdata}
 \end{figure}
 
 \section{Discussion}
 
 Described algorithm guarantees to handle one failure per computational step,
 more failures can be tolerated if they do not affect the master node. The system
 handles simultaneous failure of all subordinate nodes, however, if both master
 and backup nodes fail, there is no chance for an application to survive. In this
 case the state of the current computation step is lost, and the only way to
 restore it is to restart the application.
 
 Computational kernels are means of abstraction that decouple distributed
 application from physical hardware: it does not matter how many nodes are online
 for an application to run successfully. Computational kernels eliminate the need
 to allocate a physical backup node to make master node highly-available, with
 computational kernels approach any node can act as a backup one. Finally,
 computational kernels can handle subordinate node failures in a way that is
 transparent to a programmer.
 
 The disadvantage of this approach is evident: there is no way of making existing
 middleware highly-available without rewriting their source code. Although, our
 programming framework is lightweight, it is not easy to map architecture of
 existing middleware systems to it: most systems are developed keeping in mind
 static assignment of server/client roles, which is not easy to make dynamic.
 Hopefully, our approach will simplify design of future middleware systems.