Skip to content

Commit 5fbca60

Browse files
author
mfarrera
committed
Change in dask_init to run with dask-mpi plus scripts to run scalability tests
1 parent 63213d2 commit 5fbca60

7 files changed

Lines changed: 233 additions & 21 deletions

File tree

‎scripts/csd3-slurm/create-slurm-jobs.sh‎

Lines changed: 12 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -4,12 +4,14 @@
44
# Script should be run from $ARLROOT/scripts/csd3-slurm
55

66
TEMPLATE=slurm_submit.peta4-skylake.template
7-
OUTPUT=slurm_job.peta4
7+
TEMPLATE=slurm_submit.peta4-skylake-dask.template
8+
OUTPUT=slurm_job.peta4.dask
89
NUMNODES_LIST=' 8 16'
9-
NUMNODES_LIST='2 4'
10+
#NUMNODES_LIST='2 4'
1011
NODETYPE_LIST='skylake-himem'
11-
NFREQ_LIST='203 407'
12-
#NFREQ_LIST='41 71 101'
12+
NFREQ_LIST='All'
13+
NFREQ_LIST_IN='41 71 101'
14+
NFREQ_LIST_IN='101 203 403'
1315

1416
JOB_FOLDER=../../scripts/csd3-slurm/
1517
WORK_DIRECTORY=../../workflows/mpi/
@@ -28,18 +30,22 @@ for NUMNODES in $NUMNODES_LIST
2830
elif [ $NODETYPE == 'skylake-himem' ]
2931
then
3032
NUMCORES_LIST='16 32'
33+
NUMCORES_LIST='16'
3134
else
3235
echo 'Wrong nodetype'
3336
fi
3437
for NUMCORES in $NUMCORES_LIST
3538
do
3639
NUMTASKS=$((NUMNODES*NUMCORES))
37-
sed -e "s/@@NUMNODES@@/$NUMNODES/" -e "s/@@NUMTASKS@@/$NUMTASKS/" -e "s/@@NFREQ@@/$NFREQ/" -e "s/@@NODETYPE@@/$NODETYPE/" $TEMPLATE > ./tmp/$OUTPUT.$NODETYPE.$NUMNODES.$NUMTASKS.$NFREQ
40+
# Next line for MPI jobs
41+
#sed -e "s/@@NUMNODES@@/$NUMNODES/" -e "s/@@NUMTASKS@@/$NUMTASKS/" -e "s/@@NFREQ@@/$NFREQ/" -e "s/@@NODETYPE@@/$NODETYPE/" $TEMPLATE > ./tmp/$OUTPUT.$NODETYPE.$NUMNODES.$NUMTASKS.$NFREQ
42+
# Next line for Dask jobs
43+
sed -e "s/@@NUMNODES@@/$NUMNODES/" -e "s/@@NUMTASKS@@/$NUMCORES/" -e "s/@@NFREQ_LIST@@/$NFREQ_LIST_IN/" -e "s/@@NODETYPE@@/$NODETYPE/" $TEMPLATE > ./tmp/$OUTPUT.$NODETYPE.$NUMNODES.$NUMTASKS.$NFREQ
3844
echo 'Created' ./tmp/$OUTPUT.$NODETYPE.$NUMNODES.$NUMTASKS.$NFREQ
3945
cd $WORK_DIRECTORY
4046
echo 'Current directory' $PWD
4147
echo 'Submiting' $JOB_FOLDER/tmp/$OUTPUT.$NODETYPE.$NUMNODES.$NUMTASKS.$NFREQ
42-
sbatch $JOB_FOLDER/tmp/$OUTPUT.$NODETYPE.$NUMNODES.$NUMTASKS.$NFREQ
48+
#sbatch $JOB_FOLDER/tmp/$OUTPUT.$NODETYPE.$NUMNODES.$NUMTASKS.$NFREQ
4349
# rm $OUTPUT.$NODETYPE.$NUMNODES.$NUMTASKS.$NFREQ
4450
cd $JOB_FOLDER
4551
done

‎scripts/csd3-slurm/process-slurm-job-results.sh‎

Lines changed: 11 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -7,14 +7,15 @@
77
#OUTPUT=slurm_job.peta4
88
NUMNODES_LIST='2 4 8 16'
99
NODETYPE_LIST='skylake-himem'
10-
NFREQ_LIST='41 71 101 203 407'
10+
NFREQ_LIST='21 41 71 101 203 407'
1111

1212
RESULTS_OUTPUT=mpitest
1313

1414
WORK_DIRECTORY=../../workflows/mpi
1515

1616
cd $WORK_DIRECTORY
17-
cd results/mpi
17+
#cd results/mpi
18+
cd results/dask/some-results
1819
echo 'Current directory' $PWD
1920
for NUMNODES in $NUMNODES_LIST
2021
do
@@ -29,19 +30,26 @@ for NUMNODES in $NUMNODES_LIST
2930
elif [ $NODETYPE == 'skylake-himem' ]
3031
then
3132
NUMCORES_LIST='16 32'
32-
NUMCORES_LIST='32'
33+
NUMCORES_LIST='16'
3334
else
3435
echo 'Wrong nodetype'
3536
fi
3637
for NUMCORES in $NUMCORES_LIST
3738
do
39+
# For MPI tests
3840
NUMTASKS=$((NUMNODES*NUMCORES))
3941
#sed -e "s/@@NUMNODES@@/$NUMNODES/" -e "s/@@NUMTASKS@@/$NUMTASKS/" -e "s/@@NFREQ@@/$NFREQ/" -e "s/@@NODETYPE@@/$NODETYPE/" $TEMPLATE > $OUTPUT.$NODETYPE.$NUMNODES.$NUMTASKS.$NFREQ
4042
#cat $RESULTS_OUTPUT-$NODETYPE-$NUMNODES-$NUMTASKS-$NFREQ.out |grep "0: predict finished in" | awk {'print $NUMNODES $NUMTASKS $NFREQ $5'}
4143
cat $RESULTS_OUTPUT-$NODETYPE-$NUMNODES-$NUMTASKS-$NFREQ.out |grep ^"0: predict finished in" | awk {"print ${NUMNODES} \"\t\" ${NUMTASKS} \"\t\" $NFREQ \"\t\" \$5"} >> predict-results.txt
4244
cat $RESULTS_OUTPUT-$NODETYPE-$NUMNODES-$NUMTASKS-$NFREQ.out |grep ^"0: invert finished in" | awk {"print ${NUMNODES} \"\t\" ${NUMTASKS} \"\t\" $NFREQ \"\t\" \$5"} >> invert-results.txt
4345
cat $RESULTS_OUTPUT-$NODETYPE-$NUMNODES-$NUMTASKS-$NFREQ.out |grep ^"0: continuum imaging finished in" | awk {"print ${NUMNODES} \"\t\" ${NUMTASKS} \"\t\" $NFREQ \"\t\" \$6"} >> contimg-results.txt
4446
cat $RESULTS_OUTPUT-$NODETYPE-$NUMNODES-$NUMTASKS-$NFREQ.out |grep ^"0: ical finished in" | awk {"print ${NUMNODES} \"\t\" ${NUMTASKS} \"\t\" $NFREQ \"\t\" \$5"} >> ical-results.txt
47+
# For Dask tests
48+
NUMTASKS=$NUMCORES
49+
cat $RESULTS_OUTPUT-$NODETYPE-$NUMNODES-$NUMTASKS-$NFREQ.out |grep ^"predict finished in" | awk {"print ${NUMNODES} \"\t\" ${NUMTASKS} \"\t\" $NFREQ \"\t\" \$4"} >> predict-results.txt
50+
cat $RESULTS_OUTPUT-$NODETYPE-$NUMNODES-$NUMTASKS-$NFREQ.out |grep ^"invert finished in" | awk {"print ${NUMNODES} \"\t\" ${NUMTASKS} \"\t\" $NFREQ \"\t\" \$4"} >> invert-results.txt
51+
cat $RESULTS_OUTPUT-$NODETYPE-$NUMNODES-$NUMTASKS-$NFREQ.out |grep ^"continuum imaging finished in" | awk {"print ${NUMNODES} \"\t\" ${NUMTASKS} \"\t\" $NFREQ \"\t\" \$5"} >> contimg-results.txt
52+
cat $RESULTS_OUTPUT-$NODETYPE-$NUMNODES-$NUMTASKS-$NFREQ.out |grep ^"ical finished in" | awk {"print ${NUMNODES} \"\t\" ${NUMTASKS} \"\t\" $NFREQ \"\t\" \$4"} >> ical-results.txt
4553

4654
done
4755
done

‎scripts/csd3-slurm/slurm_submit.peta4-skylake-dask.template‎

Lines changed: 178 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,178 @@
1+
#!/bin/bash
2+
#!
3+
#! Example SLURM job script for Peta4-Skylake (Skylake CPUs, OPA)
4+
#! Last updated: Mon 13 Nov 12:25:17 GMT 2017
5+
#!
6+
7+
#!#############################################################
8+
#!#### Modify the options in this section as appropriate ######
9+
#!#############################################################
10+
11+
#! sbatch directives begin here ###############################
12+
#! Name of the job:
13+
#SBATCH -J cpujob
14+
#! Redirect stderr and stdout
15+
#SBATCH --output results/dask/mpitest-@@NODETYPE@@-@@NUMNODES@@-@@NUMTASKS@@.out
16+
#SBATCH --error results/dask/mpitest-@@NODETYPE@@-@@NUMNODES@@-@@NUMTASKS@@.err
17+
#! Which project should be charged (NB Wilkes projects end in '-GPU'):
18+
#SBATCH -A SKA-SDP-SL2-CPU
19+
#! How many whole nodes should be allocated?
20+
#SBATCH --nodes=@@NUMNODES@@
21+
#! How many (MPI) tasks will there be in total? (<=nodes*12)
22+
#! The skylake/skylake-himem nodes have 32 CPUs (cores) each. ntasks=64
23+
#! We will allocate one process per node
24+
#SBATCH --ntasks=@@NUMNODES@@
25+
#! How much wallclock time will be required?
26+
#SBATCH --time=24:22:00
27+
#! What types of email messages do you wish to receive?
28+
#SBATCH --mail-type=FAIL
29+
#! Uncomment this to prevent the job from being requeued (e.g. if
30+
#! interrupted by node failure or system downtime):
31+
##SBATCH --no-requeue
32+
33+
# Request the nodes in exclusive mode
34+
#SBATCH --exclusive
35+
# Request the nodes to be on the same switch
36+
#SBATCH --switches=1
37+
38+
#! For 6GB per CPU, set "-p skylake"; for 12GB per CPU, set "-p skylake-himem":
39+
#SBATCH -p @@NODETYPE@@
40+
41+
#! sbatch directives end here (put any additional directives above this line)
42+
43+
#! Notes:
44+
#! Charging is determined by core number*walltime.
45+
#! The --ntasks value refers to the number of tasks to be launched by SLURM only. This
46+
#! usually equates to the number of MPI tasks launched. Reduce this from nodes*32 if
47+
#! demanded by memory requirements, or if OMP_NUM_THREADS>1.
48+
#! Each task is allocated 1 core by default, and each core is allocated 5990MB (skylake)
49+
#! and 12040MB (skylake-himem). If this is insufficient, also specify
50+
#! --cpus-per-task and/or --mem (the latter specifies MB per node).
51+
52+
#! Number of nodes and tasks per node allocated by SLURM (do not change):
53+
numnodes=$SLURM_JOB_NUM_NODES
54+
numtasks=$SLURM_NTASKS
55+
mpi_tasks_per_node=$(echo "$SLURM_TASKS_PER_NODE" | sed -e 's/^\([0-9][0-9]*\).*$/\1/')
56+
#! ############################################################
57+
#! Modify the settings below to specify the application's environment, location
58+
#! and launch method:
59+
60+
#! Optionally modify the environment seen by the application
61+
#! (note that SLURM reproduces the environment at submission irrespective of ~/.bashrc):
62+
. /etc/profile.d/modules.sh # Leave this line (enables the module command)
63+
module purge # Removes all modules still loaded
64+
module load rhel7/default-peta4 # REQUIRED - loads the basic environment
65+
66+
export ARLROOT=/home/mf582/scratch/ARL/algorithm-reference-library
67+
export ARL=/home/mf582/scratch/ARL/algorithm-reference-library
68+
69+
#! Insert additional module load commands after this line if needed:
70+
echo "source ${ARLROOT}/scripts/csd3-slurm/activate-env-csd3.sh"
71+
source $ARLROOT/scripts/csd3-slurm/activate-env-csd3.sh
72+
73+
#module load intel/bundles/psxe/2017.4
74+
75+
#! Full path to application executable:
76+
application="$ARLROOT/workflows/mpi/imaging-pipelines_arlexecute.py"
77+
78+
#! Run options for the application:
79+
#! the number of frequencies and the json file?
80+
#!options="--nfreqwin=@@NFREQ@@"
81+
82+
#! Work directory (i.e. where the job will run):
83+
workdir="$SLURM_SUBMIT_DIR" # The value of SLURM_SUBMIT_DIR sets workdir to the directory
84+
# in which sbatch is run.
85+
workdir="$ARLROOT/workflows/mpi/results/dask"
86+
87+
#! Are you using OpenMP (NB this is unrelated to OpenMPI)? If so increase this
88+
#! safe value to no more than 32:
89+
export OMP_NUM_THREADS=1
90+
91+
#! Number of MPI tasks to be started by the application per node and in total (do not change):
92+
np=$[${numnodes}*${mpi_tasks_per_node}]
93+
94+
#! The following variables define a sensible pinning strategy for Intel MPI tasks -
95+
#! this should be suitable for both pure MPI and hybrid MPI/OpenMP jobs:
96+
export I_MPI_PIN_DOMAIN=omp:compact # Domains are $OMP_NUM_THREADS cores in size
97+
export I_MPI_PIN_ORDER=scatter # Adjacent domains have minimal sharing of caches/sockets
98+
#! Notes:
99+
#! 1. These variables influence Intel MPI only.
100+
#! 2. Domains are non-overlapping sets of cores which map 1-1 to MPI tasks.
101+
#! 3. I_MPI_PIN_PROCESSOR_LIST is ignored if I_MPI_PIN_DOMAIN is set.
102+
#! 4. If MPI tasks perform better when sharing caches/sockets, try I_MPI_PIN_ORDER=compact.
103+
104+
105+
#! Uncomment one choice for CMD below (add mpirun/mpiexec options if necessary):
106+
107+
#! Choose this for a MPI code (possibly using OpenMP) using Intel MPI.
108+
#CMD="mpirun -ppn $mpi_tasks_per_node -np $np $application $options"
109+
#CMD="mpiexec -trace -np $np python $application $options"
110+
#CMD="mpiexec -np $np python $application $options"
111+
#! NUMTASKS is not correct here, it should be numthreads numtasks referes to the total threads
112+
#CMD="mpiexec -np $np dask-mpi --no-nanny --nthreads @@NUMTASKS@@ --memory-limit 0.75 --scheduler-file TEXT --interface ib0 --local-directory TEXT"
113+
#CMD="mpiexec -np $np dask-mpi --no-nanny --nthreads @@NUMTASKS@@ --memory-limit 0.75 --interface ib0"
114+
115+
#CMD2="python $application $options"
116+
#CMD="mpirun -ppn $mpi_tasks_per_node -np $np $application $options"
117+
#mpi_tasks_per_node=$[${mpi_tasks_per_node}/2]
118+
#CMD2="mpirun -ppn $mpi_tasks_per_node -np $np $application $options"
119+
120+
#! Choose this for a pure shared-memory OpenMP parallel program on a single node:
121+
#! (OMP_NUM_THREADS threads will be created):
122+
#CMD="$application $options"
123+
124+
#! Choose this for a MPI code (possibly using OpenMP) using OpenMPI:
125+
#CMD="mpirun -npernode $mpi_tasks_per_node -np $np $application $options"
126+
127+
128+
###############################################################
129+
### You should not have to change anything below this line ####
130+
###############################################################
131+
132+
cd $workdir
133+
echo -e "Changed directory to `pwd`.\n"
134+
135+
# Make sure there is no scheduler.json file and tmp directory for threads in workflows/mpi
136+
# We can make it cleaner by passing it as a parameter to dask-mpi ...
137+
#rm -fr ../../dask-worker-space
138+
#rm ../../scheduler.json
139+
140+
JOBID=$SLURM_JOB_ID
141+
mkdir $JOBID
142+
cd $JOBID
143+
echo -e "JobID: $JOBID\n======"
144+
echo "Time: `date`"
145+
echo "Running on master node: `hostname`"
146+
echo "Current directory: `pwd`"
147+
148+
if [ "$SLURM_JOB_NODELIST" ]; then
149+
#! Create a machine file:
150+
export NODEFILE=`generate_pbs_nodefile`
151+
cat $NODEFILE | uniq > machine.file.$JOBID
152+
echo -e "\nNodes allocated:\n================"
153+
echo `cat machine.file.$JOBID | sed -e 's/\..*$//g'`
154+
fi
155+
156+
echo -e "\nnumtasks=$numtasks, numnodes=$numnodes, mpi_tasks_per_node=$mpi_tasks_per_node (OMP_NUM_THREADS=$OMP_NUM_THREADS)"
157+
schedulerfile="$workdir/$JOBID/scheduler.json"
158+
tmpdir="$workdir/$JOBID/dask-worker-space"
159+
machines="$workdir/$JOBID/machine.file.$JOBID"
160+
CMD="mpiexec -np $np -machinefile $machines dask-mpi --no-nanny --nthreads @@NUMTASKS@@ --memory-limit 0.75 --scheduler-file $schedulerfile --interface ib0 --local-directory $tmpdir"
161+
echo -e "\nExecuting command:\n==================\n$CMD\n"
162+
163+
sleep 10
164+
eval $CMD &> mpiexec-${JOBID}.log &
165+
sleep 10
166+
echo -e "\n After command:\n==================\n$CMD\n"
167+
168+
#NFREQ_LIST='7 21 41'
169+
NFREQ_LIST='@@NFREQ_LIST@@'
170+
for NFREQ in $NFREQ_LIST
171+
do
172+
options="--nfreqwin=$NFREQ"
173+
options="--nfreqwin=$NFREQ --schedulerfile=${schedulerfile}"
174+
CMD2="python $application $options > ../mpitest-@@NODETYPE@@-@@NUMNODES@@-@@NUMTASKS@@-$NFREQ.out"
175+
echo -e "\nExecuting command:\n==================\n$CMD2\n"
176+
eval $CMD2
177+
done
178+

‎scripts/csd3-slurm/submit-slurm-jobs.sh‎

Lines changed: 4 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1,14 +1,15 @@
11
#! /bin/bash
22
# skylake nodes have 6GB per CPU and 12 cores per node
33
# skylake-himem have 12GB per CPU and 32 CPUs per node
4-
TEMPLATE=slurm_submit.peta4-skylake.template
54
OUTPUT=slurm_job.peta4
5+
OUTPUT=slurm_job.peta4.dask
66
NUMNODES_LIST='2 4 8 16'
7-
NUMNODES_LIST='2 4'
7+
#NUMNODES_LIST='2 4'
88
NODETYPE_LIST='skylake skylake-himem'
99
NODETYPE_LIST='skylake-himem'
1010
NFREQ_LIST='21 41 71 101 203'
1111
NFREQ_LIST='203 407'
12+
NFREQ_LIST='All'
1213

1314
JOB_FOLDER=../../scripts/csd3-slurm/tmp/
1415
WORK_DIRECTORY=../../workflows/mpi
@@ -27,14 +28,14 @@ for NUMNODES in $NUMNODES_LIST
2728
elif [ $NODETYPE == 'skylake-himem' ]
2829
then
2930
NUMCORES_LIST='16 32'
31+
NUMCORES_LIST='16'
3032
else
3133
echo 'Wrong nodetype'
3234
fi
3335

3436
for NUMCORES in $NUMCORES_LIST
3537
do
3638
NUMTASKS=$((NUMNODES*NUMCORES))
37-
#sed -e "s/@@NUMNODES@@/$NUMNODES/" -e "s/@@NUMTASKS@@/$NUMTASKS/" -e "s/@@NFREQ@@/$NFREQ/" -e "s/@@NODETYPE@@/$NODETYPE/" $TEMPLATE > $OUTPUT.$NODETYPE.$NUMNODES.$NUMTASKS.$NFREQ
3839
echo $PWD
3940
echo 'Submiting' $JOB_FOLDER$OUTPUT.$NODETYPE.$NUMNODES.$NUMTASKS.$NFREQ
4041
sbatch $JOB_FOLDER$OUTPUT.$NODETYPE.$NUMNODES.$NUMTASKS.$NFREQ

‎workflows/mpi/imaging-pipelines_arlexecute.py‎

Lines changed: 19 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -70,12 +70,13 @@ def init_logging():
7070
format='%(asctime)s,%(msecs)d %(name)s %(levelname)s %(message)s',
7171
datefmt='%H:%M:%S',
7272
level=logging.INFO)
73-
log = logging.getLogger()
74-
logging.info("Starting imaging-pipeline")
7573

7674
parser = argparse.ArgumentParser(description='Imaging pipelines in Dask.')
7775
parser.add_argument('--nfreqwin', type=int, nargs='?', default=7,
7876
help='The number of frequency windows')
77+
parser.add_argument('--schedulerfile', type=str, nargs='?',
78+
default='./scheduler.json',
79+
help='The scheduler file if starting with dask-mpi')
7980

8081
args = parser.parse_args()
8182

@@ -86,16 +87,24 @@ def init_logging():
8687
threads_per_worker=16
8788
memory=384
8889
nworkers=1
89-
90+
filename=args.schedulerfile
9091
client = get_dask_Client(threads_per_worker=threads_per_worker,
9192
processes = threads_per_worker == 1,
9293
memory_limit=memory * 1024 * 1024 * 1024,
93-
n_workers=nworkers)
94+
n_workers=nworkers, with_file=True,
95+
scheduler_file=filename)
9496
arlexecute.set_client(client)
9597
nodes = findNodes(arlexecute.client)
9698
print("Defined %d workers on %d nodes" % (nworkers, len(nodes)))
9799
print("Workers are: %s" % str(nodes))
98100
#arlexecute.set_client(use_dask=True)
101+
102+
init_logging()
103+
log = logging.getLogger()
104+
logging.info("Starting imaging-pipeline")
105+
106+
# Initialise logging on the workers. This appears to only work using
107+
# the process scheduler.
99108
arlexecute.run(init_logging)
100109

101110

@@ -111,12 +120,15 @@ def init_logging():
111120

112121
#nfreqwin=7
113122
nfreqwin=args.nfreqwin
114-
ntimes=5
123+
#ntimes=5
115124
rmax=300.0
116-
frequency=numpy.linspace(1.0e8,1.2e8,nfreqwin)
125+
#frequency=numpy.linspace(1.0e8,1.2e8,nfreqwin)
126+
ntimes=11
127+
frequency=numpy.linspace(0.9e8,1.1e8,nfreqwin)
117128
channel_bandwidth=numpy.array(nfreqwin*[frequency[1]-frequency[0]])
118129
times = numpy.linspace(-numpy.pi/3.0, numpy.pi/3.0, ntimes)
119-
phasecentre=SkyCoord(ra=+0.0 * u.deg, dec=-40.0 * u.deg, frame='icrs', equinox='J2000')
130+
#phasecentre=SkyCoord(ra=+0.0 * u.deg, dec=-40.0 * u.deg, frame='icrs', equinox='J2000')
131+
phasecentre=SkyCoord(ra=+30.0 * u.deg, dec=-60.0 * u.deg, frame='icrs', equinox='J2000')
120132

121133
bvis_list=simulate_list_arlexecute_workflow('LOWBD2',
122134
frequency=frequency,

‎workflows/mpi/tests/test-dask.py‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,3 @@
1+
2+
from dask.distributed import Client
3+
client = Client(scheduler_file='./scheduler.json')

‎wrappers/arlexecute/execution_support/dask_init.py‎

Lines changed: 6 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -12,7 +12,8 @@
1212

1313

1414
def get_dask_Client(timeout=30, n_workers=None, threads_per_worker=1, processes=True, create_cluster=True,
15-
memory_limit=None, local_dir='.'):
15+
memory_limit=None, local_dir='.', with_file=False,
16+
scheduler_file='./scheduler.json'):
1617
""" Get a Dask.distributed Client for the scheduler defined externally, otherwise create
1718
1819
The environment variable ARL_DASK_SCHEDULER is interpreted as pointing to the scheduler.
@@ -30,7 +31,10 @@ def get_dask_Client(timeout=30, n_workers=None, threads_per_worker=1, processes=
3031
if scheduler is not None:
3132
print("Creating Dask Client using externally defined scheduler")
3233
c = Client(scheduler, timeout=timeout)
33-
34+
elif with_file:
35+
print("Creating Dask Client using externally defined scheduler in file %s" % scheduler_file)
36+
c = Client(scheduler_file=scheduler_file, timeout=timeout)
37+
3438
elif create_cluster:
3539
if n_workers is not None:
3640
if memory_limit is not None:

0 commit comments

Comments
 (0)