Exact reproducibility of stochastic simulations for parallel and serial algorithms simultaneously Random Stream Chunking

Preamble 

Figure: Visual description of
random stream chunking, M.Suzen (2017)

The advent of using computational sciences approaches, i.e., data science and machine learning, in the industry becomes more common practice in almost all organisations due to the democratisation of data science tools and availability of inexpensive cloud infrastructure. This brings the requirement or even compulsory practice of a code being so called parallelised. Note that parallelisation is used as an umbrella term here for using multiple compute resources in accelerating otherwise a serial computation and could mean a distributed or multi-core computing, i.e., multiple CPUs/GPUs. Here we provide a simple yet a very powerful approach to preserve reproducibility of parallel and serial implementation of the same algorithm that uses random numbers, i.e., stochastic simulations. We assume Single Instruction Multiple Data (SIMD) setting. 

Terminology on repeatability, reproducibility and  replicability 

Even though we only use reproducibility as a term as an umbrella term, there are clear distinctions, recommended by ACM, here. We summarise this from computational science perspective :

repeatability : Owner re-run the code to produce the same results on own environment. 

reproducibility: Others can  re-run the code to produce the same results on other's environment. 

replicability: Others can re-code the same thing differently and produce the same results on other's environment. 

In the context of this post, since parallel and serial settings would constitute different environments, the practice of getting the same results, this falls under reproducibility.   

Single Instruction Multiple Data (SIMD) setting. 

This is the most used, and probably the only one you would ever need, approach in parallel processing. It implies using the same instruction, i.e., algorithm or function/module, for the distinct portions of the data. This originates from applied mathematics techniques so called domain decomposition method. 

Simultaneous Random Stream Chunking 

The approach in ensuring exact reproducibility of a stochastic algorithm in both serial and parallel implementation lies in default chunking in producing the random numbers both in serial and parallel code. This approach used in the Bristol Python package Here is the mathematical definition. 

Defintion Random Stream Chunking Given a random number generator $\mathscr{R}(s_{k})$ with as seed $s_{k}$ is used over $k$-partitions. These partitions are always corresponds to $k$ datasets, MD portion of SIMD. In the case of parallel algorithms, each $k$ corresponds to a different compute resource     $\mathscr{C}_{k}$. 

By this definition we ensure that both parallel and serial processing receiving exactly the same random number, this is summarised in the Figure.

Conclusion

We outline a simple yet effective way of ensuring exact reproducibility of serial and parallel simulations simultaneously. However, reproducibility of stochastic simulations are highly hardware dependent as well, such as GPUs and NPUs, and their internal streams might not be that easy to control partitions, but generic idea presented here should be applicable.

Please cite this article as: 

 @misc{suezen24rep, 

     title = {Exact reproducibility of stochastic simulations for parallel and serial algorithms simultaneously}, 

     howpublished = {\url{https://memosisland.blogspot.com/2024/02/exact-reproducibility-of-stochastic.html}, 

     author = {Mehmet Süzen},

     year = {2024}

}

  

Distributed processing with BASH via LSF' blaunch

LSF is currently a de-facto standard in managing computing queues. Many inexpensive CPU or GPU clusters uses LSF. Majority of current academic HPC codes are using MPI for their core computations. While the parallelisation/distribution of the computational tasks are achived by the core MPI codes, LSF batch job script, usually can invoke via shell, is still a serial code. See here. This may bring a bottleneck in case that the batch script needs to run large set of commands in sequence, that can be done in distributed way.  

Idle CPUs

An example task could be that your parallel code generates large set of output files and you require to further process them, such as compressing all data files produced by processes. Once you launch $n$ processes, after completion of the MPI code, running compression over the $n$ data files will be processed in a single process. This will not only waste $n-1$ cpu time waiting idle, it will waste our wall clock time $n-1$ fold while compression is performed sequentially. To overcome this limitation LSF has introduced a tool called blaunch, for distributing the tasks. The documentation provided by IBM is really fantastic. However it lacks a simple example for novice users. In this post I provide a simple example that compresses $n$ data files using $n$ distributed processes. 

Job distribution via blaunch

The first step in doing this task, we need to figure our what hostnames we are currently running with the LSF job script. This can be obtained using $LSF_HOSTS environment variable. The second step is to write this information into hosts file that can be read by blaunch. A compression script should be provided in a seperate file. This is an important point. Because whole concept of blaunch is that it distributes the given command to all available processes. Note that, LSF process is running your script not the distributed host processes, at least what it is transparently. Knowing this fact, we can only differentiate the process ID outside of our job submit script using $LSF_PM_TASKID.

Here is a possible portion of the job script, using BASH,

1 2 3 4 5

cd /mydata rm hosts.txt export myhosts=($LSB_HOSTS) for hostName in ${myhosts[*]};do echo $hostName >> hosts.txt ; done blaunch -u hosts.txt compress.sh

As you can see we don't use  $LSF_PM_TASKID in the job script, rather in the compression script that is going to be launch in one of the distributed process:

1 2 3 4 5

export trajF=(`find . -name '*.dat'`) # ${trajF[ii]}; ii=`expr $ii + 1`;done # This line should be removed (see comments) export ii=`expr $LSF_PM_TASKID - 1` echo ${trajF[$ii]} xz -4e --format=lzma ${trajF[$ii]}

We assume that commands are available in the BASH environment. Note that we use bash array to identify file names and your output convention should match with the identification.

 Conclusion

We have provided a simple working example of launching parallel shell script tasks using blaunch. A similar approach is presented here. An alternating approach to this could be using GNU Parallel , which is discussed here.

How to benchmark communication cost in the beowulf (parallel system)

Quite long time ago, I've proposed one simple way of estimating
communication cost of an MPI program (or multiple-instruction)
appeared in the beowulf mailing list. Here is the outline of
the procedure.
[original post]

Definition Communication cost (Comm) is the ratio between the real
time used by all processors (processes actually if you use MPI)
and global real cpu time (GRU).

 Comm=GRU/GR = (GR-GU-GS)/GR 

Undefined symbols are as follows:

Comm= Communication cost
GR=global real time
LR=local worked real time 
LU=local user
LS=local system
GS=global system
GU=global user
n=# of procs
GU=n*LU
GS=n*LS
GR=n*LR 

In the actual MPI program this measure's implementation could be outlined
as follows

 BEGIN PROGRAM Myprogram
      begin distribution
       initial_timings=timings;
                ...do whatever is your distributed computing....
       end_timings=timings;
      end distribution
       if(myrank == 0) {compute communication cost}        
END Myprogram

BEGIN PROC timings
    for(n=0;n < numproc;numproc) {
       if(n==myrank) { measure timing of myrank and send to rank 0 }
     }
END PROC