Dask is used to parallelize analysis in PMDA. It provides a flexible approach to task-based parallelism and can scale from multi-core laptops to large compute clusters.
By default, all the available cores on the local machine (laptop or
workstation) are used with the
n_jobs=-1 keyword but any number
can be set, e.g.,
n_jobs=4 to split the trajectory into 4 blocks.
Internally, this uses the processes (multiprocessing) scheduler
of dask. If you want to make use of more advanced scheduler features
or scale your analysis to multiple nodes, e.g., in an HPC (high
performance computing) environment, then use the
scheduler, as described next. If
n_jobs==1 a synchronous
(single threaded) scheduler is used 1.
With the distributed scheduler on can run analysis in a distributed fashion on HPC or ad-hoc clusters (see setting up a dask.distributed network) or on a single machine. (In addition, distributed also provides diagnostics in form of a dashboard in the browser and a progress bar.)
Local cluster (single machine)¶
import distributed lc = distributed.LocalCluster(n_workers=8, processes=True) client = distributed.Client(lc)
rmsd_ana = rms.RMSD(u.atoms, ref.atoms).run()
Because the local cluster contains 8 workers, the RMSD trajectory analysis will be parallelized over 8 trajectory segments.
In order to run on a larger cluster with multiple nodes (see setting up a dask.distributed network) one needs to know how to connect to the running scheduler (e.g., address and port number or shared state file). Assuming that the scheduler is running on 192.168.0.1:8786, one would initialize the distributed.Client and this is enough to use distributed for all analysis (it configures the scheduler to be distributed):
import distributed client = distributed.Client('192.168.0.1:8786') rmsd_ana = rms.RMSD(u.atoms, ref.atoms).run()
In this way one can spread an analysis task over many different nodes.