Author: Peter Braam
Owner: Peter Braam
Version: 0.1
Date: 2014-11-15
Audience: developers, SDP project leads

The SKA imaging software transforms pre-processed radio telescope antenna data (aka u,v data) into images (aka the science products) suitable for scientific research. The u,v data is segmented in numerous measurement sets, each of considerable size, continuously generated by the telescope (and pre-processing stages). Tens to hundreds of thousands of measurement sets will be processed simultaneously, and each measurement set produces an image file. The output data of this process is much smaller than its input.

The image reconstruction primarily takes place based on the data in a single measurement set.

The project targets exa-scale clusters of nodes with centralized and/or local storage. The nodes may have SMP systems or accelerators or both. Hierarchical structures may exist - for example exposing full bandwidth networks intra-rack and limited bandwidth networks cluster wide.

Experience with radio telescopes has shown that the precise imaging algorithm will only be determined when the telescope comes into operation. The expected age of a radio telescope will be many 10s of years, and computer hardware is likely to go undergo fundamental changes.

• modifiability: there will be a clear separation of layers in the software, distinguishing:
  • the kernels efficiently implementing key operations (e.g. gridding, fourier transformations, etc) for particular hardware
  • the algorithmic layer, expressing the algorithm its data movement and its use of kernels concisely
  • a dataflow manager able to schedule the algorithm on a particular cluster architecture.
• ideally the software shall be self-optimizing, both at the scheduling and kernel level

The design shall be complete when

• Milestone 2 deliverables are addressed
• An inventory of future changes is included
• experienced programmers can modify the packages
• inexperienced programmers can write simple DNA programs and run them on a cluster
• profiling, debugging and running in painless for all

• develop basic GPU gridding code interfacing with a file channel
• create a sorted uv data set, if possible compare with processing un-sorted sets
• use a pre-computed analytical GCF, use a data flow channel to deliver the right portion of the GCF for different w planes
• target SMP and GPU
• make a GPU dot product program
• adapt the dot product program for vectors bigger than memory
• change the dot product program to index the vectors non-consecutively, compare sort by - channel with in place non-ordered approach
• create and visualize a model for the performance as a function of input data, cluster architecture, buffer sizes, ordering or not ordering input data (if applicable), kernel performance
• demonstrate extreme cases (i.e. very small working set compared to RAM and very big working sets compared to RAM) of performance with the model for dot product and gridding
• run the program on many measurement sets from many GPU nodes on Wilkes, visualize the efficiency of the job scheduling vs an ideal model
• arrange for high availability to omit a failed node from the computations, if possible let other nodes process the data sets not processed by the failed node
• update DNA DSL design, including the performance modeling features
• build downloadable packages and a project web site. Make an announcement. Deliver a talk (at CHPC).

Not part of milestone2 are:

• build better syntax or compiler yet than used for MS1
• refine distributed computation of the dot product
• build intra kernel multi-node parallelism for gridding
• build an infiniband communication platform
• model measure the performance of the collective operation
• use different storage platforms

We will be delivering:

• A summary of data flow programming
• An explanation of the Halide software
• A Milestone 2 report shall be delivered
• Software packaged

More specific deliverables are documented in the [Sprint2BL tab of the planning spreadsheet]https://docs.google.com/a/braamresearch.com/spreadsheets/d/19pojIxnIGg8Ew7ib9KWhS6isRAh_De_7ifjbLuIBboc/edit#gid=1418027851

We will first describe 3 use cases:

• program startup
• log and profiling information collection
• syntax of a simplified DSL to write DNA programs

After this we give a functional specification of many key functions

An analysis is made of various state that is created and possible situations that can lead to errors and must be detected.

The design section is finished with highlighting a few implementation issues

Milestone 2 implements programs as a hybrid of

• Cloud Haskell for job scheduling, data flow and high availability
• C, C++, Cuda and Halide kernels for computation

In future milestones the program will likely by coded in a more widely used language like Python, but not change essentially in terms of the binaries they generate.

We want to start the programs meeting the following requirements

• Unix process startup on a single node for testing
• SLURM based multi-node runs
• Great simplicity
• A nearly identical workflow for debugging and profiling between the local and SLURM case

The resources are a set of processes inside which a master process starts actors and channels. The allocated resources together with metadata like the process identifiers, ranks etc give a description cluster architecture descriptor.

An sbatch file is created to request resources for a job. This file may be adapted for different clusters.

\#!/bin/bash
\#
\#SBATCH --job-name=ddp-ska  
\#SBATCH --output=ddp.%j.txt  
\#SBATCH --partition=tesla  
\#SBATCH --time=5:00  
\#SBATCH --mem-per-cpu=MaxMemPerCPU  
\#SBATCH --nodes=3   # XXX start with 3  
\#SBATCH --tasks=3     # XXX same  
\#SBATCH --qos=gpu0  

srun  -N <number of nodes> -n <number of tasks>  <dna program>  [--base_rank=<base rank>] \[--base_port=<port>\] <program args>

For testing single node, Unix processes are supported:

dna-program --nprocs=<number of processes> [--base_rank=<base rank>] [--base_port=<port>] <program args>&

The program will determine if the job is running under Slurm through testing for some environment variables.

This section architecture describes how logging information will be stored and how shell Cloud Haskell processes will be started. Here we describe in more detail what the software does to implement this.

The following key requirements drive the design:

• for reasons of scalability a distributed collection of well organized files must be created, not a central log
• a remote distributed query system is built to analyze the collection of logs, profiling data etc.
• fatal errors are forwarded to the coordinating node, and will generally lead to aborting the program
• profiling information can be unified across languages that are used
• data flow messages are gathered
• debugging and profiling can be done while a programming is running and can be turned on and off
• there are levels indicating the detail of information and subsystem settings indicating what subsystem is being profiled or debugged

DNA programs will use a directory on the node's file system under which all data is stored. By default this directory is $HOME/.dna but can be set to any directory $DNADIR. It must be writeable by the processes that are started. The data under the directory can be visible through a distributed file system (or after a future milestone queried remotely).

The the following sub directory structure will be used:

.../.dna/configs     \#\# containing named configuration files
.../.dna/logs/<process id>/<rank>  \#\# logs and profiling files generated by 
.../.dna/names  \#\# containing symbolic links with time stamp and full invocation information pointing to the rank

During startup each of the processes will store its data in specific directories inside a directory that is by default

In the Slurm case the environment variable SLURM_JOBID gives a process id.

In the Shell case the Unix process id of the spawning process (parent pid) gives a process id.

we will call this dnaPid

The semantics are that under SLURM a rank number is determined from the SLURM environment variable: SLURM_PROCID.

In the Unix case the slave dna programs started first (without the master flag) will try to create directories dnaPid/n for n=1,2,3, etc. When this succeeds for a value n, that n is the rank of the slave. The -master flag will become rank 0 process for the parents process id.

We will call this the dnaRank

If the dnaRank equals 0, the process becomes a cloud Haskell master process after a few 100ms waiting to allow clients to start. All other processes become cloud Haskell slave processes.

The port numbers either equal a base port (default 40,000) or an explicitly passed baseport argument plus the dnaLocalRank. In the Unix case this is simply the rank, in the Slurm case this is obtained from SLURM through the environment variable SLURM_LOCALID.

The following lead to error conditions:

• the environment variable SLURM_JOBID and the --master argument are both defined
• starting the processes repeatedly from an interactive shell will succeed once. After running once the PID will reported to have been used before - the error will be detected when the master starts. (NOTE: this can be fixed by making multiple instances of the shell's PID as PID, PID.1, PID.2 etc. This will not be explored now.)

The architecture specifies that the log and profiling information will be stored under $HOME/.dna/logs. In this directory there are the following subdirectories

• u - holding directories with integer names being unix process id's
• s - holding directories named SLURM job id's
• job names - the name is structured as

<YYYY>:<MM>:<DD>:<HH>:<SS>-{S,U}-<ARG0>+..<ARGN>

and is a symbolic link to a directory s/<SLURM_JOBID>, or u/<PROCESS ID>

• inside these directories there are subdirectories for every rank named: <rank>-<relative node>:<CH nodeid>
• there is a human readable format of the cluster architecture description.
• each of the task directories has event files for profiling and stdout and stderr files
• stderr is also forwarded to the master processes (see below)

This section describes a DNA SLURM backend. It is extremely simple.

The Slurm system will allocate a set of processes that offer the resources for the program. At large scale the program will start hierarchically, and an instance of program is responsible for forking tasks on its children. It may pass a subset of the available processes to each child to continue spawning in a hierarchical manner.

Key requirements:

• A newly started parent must establish connections with its allocated children
• A newly started child process must learn of its parent
• A newly started child must have a process id known to the parent to set up further communication

The Unix program is started first:

• if an argument --nrpocs=N is detected it is treated as a Unix processes. The process started becomes the coordinator process. and the coordinator forks N-1 slave processes (using IP address 127.0.0.1 and ports base_port+1...base_port+N). The coordinator has dna_rank 0 (orbase_rank`) and communicates the rank to each child process (in an increasing sequence like the port numbers) and collects the slave's CH node address to spawn processes on slaves. Slaves do not get the program arguments, these are solely handled by the coordinator process. Before starting to execute the main program the coordinator reports any slaves that failed to start (e.g. because the port was in use).
• the Slurm case is similar.
• An error case arises if both a SLURM_JOBID and an --nprocs argument are present.
• The debugging, profiling and logging directories are set up

Formation of the CAD:

• The CAD will initially be the set of CH nodes.
• In the Unix case this is a simple list formed as 127.0.0.1:base_port...127.0.0.1:base_port+N
• In the Slurm case the nodelist is derived from the Slurm NODELIST variable, properly formatted by a utility function.

Scheduling and job start:

• A set of CH processes is created for actors and channels to run, so that their process id's are known to build the data flow graph. A select function is called multiple times to return groups of processes on which the same actors or channels will run. By default, select functions select nodes round robin for scheduling processes, but processes can be placed more strategically.
• The scheduling select functions can handle failures for processes to start and re-assign work to other nodes.
• The scheduling of actors and channels are now started and acknowledge to the master that they are waiting for a start message upon receipt of which the computation can start. It is not possible to start a computation safely without first instantiating the data flow graph.
• The master sends a start message.

We note here that the scheduling may involve a substantial computation to get optimal results in the case of a complex data flow.

To be copied from the milestone design.

startUnixProcess

** NOT YET REVIEWED OR FINISHED **

-- jobdir: the directory $HOME/.dna/logs/{U,S}/pname
-- processdir: the directory $HOME/.dna/logs/{U,S}/pname/rank
-- name  
makerank
        until error == 0 
            mkdir "$HOME/.dna/u/PARENT_PID/rank"
        return rank
baseman = map ('+') ARGS
case
    defined SLURM_JOBID:
            rank = SLURM_LOCALID
            pidname = SLURM_JOBID
            insert1 = S
            insert2 = show SLURM_JOBID
    _ :             
            pidname = PARENT_PID
            insert1 = U
            insert2 = show PARENT_PID
            
name = time +"-" + insert1 + "-" basename
if err = mkdir "$HOME/.dna/logs/insert1/insert2"
    exit -EEXISTS 
ln -s "$HOME/.dna/logs/insert1/insert2" "$HOME/.dna/names/name"

if (SLURM_JOBID)
    rank = findrank pidname
mkdir baseLogDir + "/" + rank
chdir processdir

create backend

##Getting and building Halide on cluster

Halide team periodically builds binary releases of Halide, which may be used locally on desktops provided sufficiently recent gcc (4.9.x) compiler is installed.

Wilkes cluster has no gcc-4.9.x installed yet, and installing it locally (taking into account the sheer number of dependencies) looks problematic. Hence, we choose to rebuild Halide against gcc-4.8.1, which can be used on cluster provided we load the corresponding environment using cluster's module command.

Also clang is required to build Halide. We didn't try to use binary releases from LLVM team because we need clang to be built against the same gcc version we use on cluster.

Thus, building clang on cluster looks as follows:

1. Execute module load gcc/4.8.1 command. This command is cluster-specific and is necessary to set necessary environment variables to bring gcc-4.8.1 in scope.

2. Download and build clang/llvm the way stated in official documentation. Take into account the following:

• Use release branches instead of trunks or download release sources archives and unpack them manually. We tested building Halide against 3.4.2 release.
• No any special configuration options are required. It is good practice to specify --prefix /place/to/put/clang option though to point to well-defined location, so that /place/to/put/clang/bin could be added to PATH environment variable.
• If using archived sources, please maintain directory layout described in the documentation (clang goes under llvm/tools subdirectory and compiler-rt goes to llvm/projects llvm subdirectory.
• Please, limit the number of job slots for make command. make -j8 goes pretty smooth.
• To use thus built clang add /place/to/put/clang/bin to your PATH.

Currently, Halide gridding code assumes Halide distribution is placed on the same directory level as RC repository is and resides in halide directory. For example, to be able to build Halide gridding code with no problems please execute

git clone https://github.com/SKA-ScienceDataProcessor/RC.git

and

git clone https://github.com/halide/Halide.git halide

from the same directory. Note we renamed Halide to halide here. This way it is made compatible with our current build infrastructure and with possible (desktop/future cluster) usage of Halide binary releases which unpacks to halide directory.

Now go to halide directory and make it with the following:

make -j8 CLANG="clang --gcc-toolchain=/usr/local/Cluster-Apps/gcc/4.8.1"

This is required when Halide is built on cluster because clang has bug/feature (for ages) preventing it from picking correct gcc environment.