Spack (Ruche) compatible

Author: ksiero@man.poznan.pl

CMakeLists.txt

@@ -24,6 +24,7 @@
 cmake_minimum_required(VERSION 3.9)
 project(pdi_examples LANGUAGES C)
 
+find_package(spdlog)
 find_package(MPI REQUIRED COMPONENTS C)
 find_package(paraconf REQUIRED COMPONENTS C)
 find_package(PDI 0.7 REQUIRED COMPONENTS C)

README.md

@@ -35,7 +35,7 @@ cd tutorial
 
 Before compilation, configure the tutorial by detecting all dependencies:
 ```bash
-pdirun cmake .
+cmake .
 ```
 
 \attention
@@ -44,7 +44,7 @@ If you installed PDI in a standard path, the `pdirun` prefix is never required.
 Once you have correctly modified each exercise according to instructions, you
 can compile it by running:
 ```bash
-pdirun make ex?
+make ex?
 ```
 Where `?` is the number of the exercise.
 
@@ -53,15 +53,15 @@ Where `?` is the number of the exercise.
 
 You can run each exercise with the following command:
 ```bash
-pdirun mpirun -n 4 ./ex?
+/usr/bin/srun -n 4 ./ex?
 ```
 Where `?` is the number of the exercise and 4 represents the number of MPI
 processes to use.
 
 To store the logs for later comparison, you can use the following command (for
 example for ex2.):
 ```bash
-pdirun mpirun -n 1 ./ex2 > ex2.result.log
+/usr/bin/srun -n 1 ./ex2 > ex2.result.log
 ```
 
 Now you're ready to work, **good luck**!

deisa_example/CMakeLists.txt

@@ -0,0 +1,12 @@
+cmake_minimum_required(VERSION 3.9)
+project(Coupling LANGUAGES C CXX)
+
+find_package(MPI REQUIRED COMPONENTS CXX C)
+find_package(paraconf REQUIRED COMPONENTS C )
+find_package(PDI 1.4 REQUIRED COMPONENTS C)
+
+set(CMAKE_C_STANDARD 99)
+
+add_executable(simulation simulation.c)
+target_link_libraries(simulation m MPI::MPI_C  paraconf::paraconf PDI::pdi)
+

deisa_example/Launcher.sh

@@ -0,0 +1,36 @@
+#!/bin/bash
+
+DIR=$PWD
+
+### prescript.py  is used to create the configuration file that is shared betwwen the simulation and the Dask cluster
+# sys.argv[1] : global_size.height
+# sys.argv[2] : global_size.width
+# sys.argv[3] : parallelism.height
+# sys.argv[4] : parallelism.width
+# sys.argv[5] : generation 
+# sys.argv[6] : nworkers
+
+source $WORKDIR/spack/share/spack/setup-env.sh
+spack load cmake@3.22.1
+spack load pdiplugin-deisa
+spack load /hbohtbo #pdiplugin-mpi
+
+NWORKER=4
+
+PARALLELISM1=2
+PARALLELISM2=2
+
+DATASIZE1=1024
+DATASIZE2=1024
+
+GENERATION=5
+
+mkdir -p $WORKDIR/Deisa
+WORKSPACE=$(mktemp -d -p $WORKDIR/Deisa/ Dask-run-XXX)
+cd $WORKSPACE
+cp $DIR/simulation.yml $DIR/*.py  $DIR/Script.sh $DIR/Launcher.sh  $DIR/*.c $DIR/CMakeLists.txt  .
+pdirun cmake .
+make -B simulation
+echo Running $WORKSPACE 
+`which python` prescript.py $DATASIZE1 $DATASIZE2 $PARALLELISM1 $PARALLELISM2 $GENERATION $NWORKER 
+sbatch Script.sh 

deisa_example/README.md

@@ -0,0 +1,24 @@
+# Dask-Enabled In Situ Analytics
+This repository contains an example on how deisa can be used. 
+
+## Requirement :
+- [Deisa PDI plugin](https://github.com/GueroudjiAmal/deisa) 
+- [Dask Distributed Deisa version repo](https://github.com/GueroudjiAmal/distributed)
+
+## How it works ?
+
+A simulation can be instrumented in PDI to make its internal data available for **_DEISA_**. The plugin retrieves it, creates corresponding keys, and sends it to Dask workers. 
+
+Internally, a **_DEISA Bridge_** is created per MPI process. Once a piece of data is shared with PDI, the Bridge sends it to a worker that has been chosen in a round-robin fashion. 
+
+**_DEISA_** python library implements a **_DEISA Adaptor_**. This component is used from the Dask client-side to create Dask arrays describing the data generated by the simulation. The **_DEISA Adaptor_** waits for an array descriptor to be sent from the **_DEISA Bridge_** in MPI rank 0. This descriptor is a dictionary with data names as keys and a dictionary containing the sizes, dimensions, and chunk sizes, as values. 
+The **_DEISA Adaptor_** uses this information to create Dask arrays, that can be retrieved by calling `get_data()` method.    
+
+## Files :
+- simulation.c : is a toy example of a C simulation code, here we have used heat 2D from [PDI examples](https://pdi.dev/master/PDI_example.html).
+- simulation.yml :  is the PDI configuration code. 
+- dask-interface.py : contains **_DEISA_** python libaray (Bridge and Adaptor classes).
+- Client.py containts an example of a python script for analytics. It it run is the Dask Client.
+- prescript.py creates a file Config.yml that contains simulation configuration such as the size of data, the number of timesteps and the domain decomposition.
+- Launcher.sh and Script.sh can be used to launch the simulation and Dask cluster in [Ruche](https://mesocentre.pages.centralesupelec.fr/user_doc/)
+ 

deisa_example/Script.sh

@@ -0,0 +1,58 @@
+#!/bin/bash
+
+#SBATCH -J dask-cluster
+#SBATCH -A dask_coupling
+#SBATCH --time=01:00:00
+#SBATCH --nodes=1
+#SBATCH --partition=cpu_med
+#SBATCH --exclusive
+
+
+NPROC=4                          # Total number of processes
+NPROCPNODE=4                     # Number of processes per node
+NWORKERPNODE=4                  # Number of Dask workers per node
+
+SCHEFILE=scheduler.json
+
+source $WORKDIR/spack/share/spack/setup-env.sh
+spack load pdiplugin-deisa
+spack load /hbohtbo #pdiplugin-mpi
+spack load py-bokeh 
+
+# Launch Dask Scheduler in a 1 Node and save the connection information in $SCHEFILE
+echo launching Scheduler 
+srun --cpu-bind=verbose --ntasks=1 --nodes=1 -l \
+    --output=scheduler.log \
+    dask-scheduler \
+    --interface ib0 \
+    --scheduler-file=$SCHEFILE   &
+
+# Wait for the SCHEFILE to be created 
+while ! [ -f $SCHEFILE ]; do
+    sleep 3
+    echo -n .
+done
+
+# Connect the client to the Dask scheduler
+echo Connect Master Client  
+`which python` client.py &
+client_pid=$!
+
+# Launch Dask workers in the rest of the allocated nodes 
+echo Scheduler booted, Client connected, launching workers 
+srun  --cpu-bind=verbose  -l \
+     --output=worker-%t.log \
+     dask-worker \
+     --interface ib0 \
+     --local-directory /tmp \
+     --nprocs $NWORKERPNODE \
+     --scheduler-file=${SCHEFILE} &
+     
+# Launch the simulation code
+echo Running Simulation 
+pdirun srun  --ntasks=$NPROC --ntasks-per-node=$NPROCPNODE  -l ./simulation  &
+
+# Wait for the client process to be finished 
+wait $client_pid
+
+

deisa_example/client.py

@@ -0,0 +1,35 @@
+import yaml
+from dask_interface import Initialization
+import dask
+import dask.array as da
+from dask.distributed import performance_report
+
+# Get configuration
+with open(r'config.yml') as file:
+    data = yaml.load(file, Loader=yaml.FullLoader)
+    Sworkers = data["workers"]
+
+# Scheduler file name 
+scheduler_info = 'scheduler.json'
+
+# Initialize the Deisa Adaptor 
+Adaptor = Initialization(Sworkers, scheduler_info)
+
+# Check if client version is compatible with scheduler version
+Adaptor.client.get_versions(check=True)
+
+# Get data descriptor as a dict of Dask arrays
+arrays = Adaptor.get_data()
+
+# py-bokeh is needed if you wanna see the perf report 
+with performance_report(filename="dask-report.html"):
+    # Get the Dask array global_t
+    gt = arrays["global_t"]    
+    #gt = gt.rechunk({1: 'auto', 2: 'auto'})
+    print(gt.chunks)
+    # Construct a lazy task graph 
+    cpt = (gt.sum() - gt.mean())*5.99 /  gt.mean() 
+    # Submit the task graph to the scheduler
+    s = Adaptor.client.compute(cpt, release=True)
+    # Print the result, note that "s" is a future object, to get the result of the computation, we call `s.result()` to retreive it.  
+    print(s.result())

deisa_example/dask_interface.py

@@ -0,0 +1,138 @@
+import sys
+import dask
+import numpy as np
+import dask.array as da
+from dask.distributed import Client, Event, get_client, comm, Queue, Future, Variable
+from dask.delayed import Delayed
+import time
+import asyncio
+import json
+import itertools
+
+
+class metadata:
+    index = list()
+    data = ""
+    shap = None
+    typ = ""
+    def __init__(self, name):
+        self.name = name
+
+def connect(sched_file):
+    sched = ''.join(chr(i) for i in sched_file)
+    with open(sched[:-1]) as f:
+        s = json.load(f)
+    adr = s["address"]
+    client  = get_client(adr)
+    return client
+
+
+def init(sched_file, rank, size, arrays, deisa_arrays_dtype):
+    client = connect(sched_file)
+    return Bridge(client, size, rank, arrays, deisa_arrays_dtype)
+
+class Bridge:
+    workers = []
+    def __init__(self, Client, Ssize, rank, arrays, deisa_arrays_dtype):
+        self.client  = Client
+        self.rank = rank
+        listw = Variable("workers").get()
+        if Ssize > len(listw): # more processes than workers
+            self.workers = [listw[rank%len(listw)]]
+        else:
+            k = len(listw)//Ssize # more workers than processes
+            self.workers = listw[rank*k:rank*k+ k]
+        self.arrays = arrays
+        for ele in self.arrays:
+            self.arrays[ele]["dtype"] = str(deisa_arrays_dtype[ele])
+            self.arrays[ele]["timedim"] = self.arrays[ele]["timedim"][0]
+            self.position = [self.arrays[ele]["starts"][i]//self.arrays[ele]["subsizes"][i] for i in range(len(np.array(self.arrays[ele]["sizes"])))]
+        if rank==0:
+            Queue("Arrays").put(self.arrays) # If and only if I have a perfect domain decomposition 
+        
+            
+    def create_key(self, timestep, name):
+        self.position[self.arrays[name]["timedim"]]= timestep
+        position = tuple(self.position)
+        return ("deisa-"+name, position)
+
+    def publish_data(self, data, data_name, timestep):
+        event = Event("Done")
+        if (timestep==0):
+            event.wait()
+        key = self.create_key(timestep, data_name)
+        shap = list(data.shape)
+        new_shape = tuple(shap[:self.arrays[data_name]["timedim"]]+[1]+shap[self.arrays[data_name]["timedim"]:])
+        data.shape = new_shape #will not copy, if not possible raise an error so handle it :p 
+        #data = data.reshape(new_shape)
+        f = self.client.scatter(data, direct = True, workers=self.workers, keys=[key], deisa=True)
+        while (f.status != 'finished'):
+            f = self.client.scatter(data, direct = True, workers=self.workers, keys=[key], deisa=True)
+        data=None
+class Adaptor :
+    adr = ""
+    client = None
+    workers = []
+    queues = []
+    def __init__(self, Sworker, scheduler_info):
+        with open(scheduler_info) as f:
+            s = json.load(f)
+        self.adr = s["address"]
+        self.client  = Client(self.adr, serializers=['dask', 'pickle']) # msgpack pour grand message ne serialize pas
+        dask.config.set({"distributed.deploy.lost-worker-timeout": 60, "distributed.workers.memory.spill":0.97, "distributed.workers.memory.target":0.95, "distributed.workers.memory.terminate":0.99 })
+        self.workers = [comm.get_address_host_port(i,strict=False) for i in self.client.scheduler_info()["workers"].keys()]
+        while (len(self.workers)!= Sworker):
+            self.workers = [comm.get_address_host_port(i,strict=False) for i in self.client.scheduler_info()["workers"].keys()]
+        Variable("workers").set(self.workers)
+        
+
+    def create_array(self, name, shape, chunksize, dtype, timedim):
+        chunks_in_each_dim = [shape[i]//chunksize[i] for i in range(len(shape))]
+        l = list(itertools.product(*[range(i) for i in chunks_in_each_dim]))
+        items = []
+        for m in l:
+            f=Future(key=("deisa-"+name,m), inform=True, deisa=True)
+            d = da.from_delayed(dask.delayed(f), shape=chunksize, dtype=dtype)
+            items.append([list(m),d])
+        ll = self.array_sort(items)
+        arrays = da.block(ll)
+        return arrays
+    
+    def create_array_list(self, name, shape, chunksize, dtype, timedim): #list arrays, one for each time step.
+        chunks_in_each_dim = [shape[i]//chunksize[i] for i in range(len(shape))]
+        l = list(itertools.product(*[range(i) for i in chunks_in_each_dim]))
+        items = []
+        for m in l:
+            f=Future(key=("deisa-"+name,m), inform=True, deisa=True)
+            d = da.from_delayed(dask.delayed(f), shape=chunksize, dtype=dtype)
+            items.append([list(m),d])
+        ll = self.array_sort(items)
+        for i in ll:
+            arrays.append(da.block(i))
+        return arrays
+    
+    def array_sort(self, ListDs):
+        if len(ListDs[0][0]) == 0:
+            return ListDs[0][1]
+        else:
+            dico = dict()
+            for e in ListDs:
+                dico.setdefault(e[0][0],[]).append([e[0][1:], e[1]])
+            return [self.array_sort(dico[k]) for k in sorted(dico.keys())]
+
+    def get_data(self, as_list=False):
+        arrays = dict()
+        self.arrays_desc = Queue("Arrays").get()
+        for name in self.arrays_desc:
+            if not as_list:
+                arrays[name] = self.create_array(name,self.arrays_desc[name]["sizes"], self.arrays_desc[name]["subsizes"], self.arrays_desc[name]["dtype"], self.arrays_desc[name]["timedim"])
+            else: #TODO test this 
+                arrays[name] = self.create_array_list(name,self.arrays_desc[name]["sizes"], self.arrays_desc[name]["subsizes"], self.arrays_desc[name]["dtype"], self.arrays_desc[name]["timedim"])
+        #Barrier after the creation of all the dask arrays 
+        e = Event("Done")
+        e.set()
+        return arrays
+
+def Initialization(Sworker, scheduler_info):
+    return Adaptor(Sworker, scheduler_info)
+

deisa_example/prescript.py

@@ -0,0 +1,18 @@
+import yaml
+import sys
+# sys.argv[1] : global_size.height
+# sys.argv[2] : global_size.width
+# sys.argv[3] : parallelism.height
+# sys.argv[4] : parallelism.width
+# sys.argv[5] : generation 
+# sys.argv[6] : nworkers
+
+with open('config.yml', 'w') as file:
+    data = {"global_size":   {"height": int(sys.argv[1]), "width": int(sys.argv[2])},
+             "parallelism":  { "height": int(sys.argv[3]), "width": int(sys.argv[4])},
+             "MaxtimeSteps": int(sys.argv[5]),
+             "workers":   int(sys.argv[6])}
+if data:
+    with open('config.yml','w') as file:
+        yaml.safe_dump(data, file) 
+